510 lines
21 KiB
Python
510 lines
21 KiB
Python
# api.py
|
||
import requests
|
||
import json
|
||
import numpy as np # 导入numpy库用于处理numpy数组
|
||
import time
|
||
from log_config import configure_task_logger, configure_error_task_logger
|
||
|
||
logger = configure_task_logger()
|
||
error_task_logger = configure_error_task_logger()
|
||
|
||
|
||
# 换应用记得修改应用编码
|
||
|
||
class YDAPI:
|
||
# 常用变量
|
||
appKey = "ding5kqocon5s9oph5uq"
|
||
appSecret = "HL1jgsIIfLAC0eTH0A1m4mwxUDqbgsiPeCCGGE3ocM6qJBTIW7Ivt9drxF_Z4Kb_"
|
||
|
||
def generateToken(self) -> str:
|
||
"""
|
||
函数功能：生成访问令牌（token）
|
||
|
||
Returns:
|
||
str: 返回生成的访问令牌字符串。此token用于后续API调用的身份验证。
|
||
"""
|
||
token_api = 'https://api.dingtalk.com/v1.0/oauth2/accessToken'
|
||
data = {
|
||
"appKey": f"{self.appKey}",
|
||
"appSecret": f'{self.appSecret}'
|
||
}
|
||
res = requests.post(token_api, json=data)
|
||
token = res.json().get('accessToken')
|
||
return token
|
||
|
||
def update_from(self, token, formInstanceId, data_new):
|
||
"""
|
||
函数功能：更新表单内容
|
||
|
||
Args:
|
||
token (str): 登录验证token，用于API调用的身份验证。
|
||
formInstanceId (str): 表单实例ID，读文件获取。
|
||
data_new (dict): 新的数据内容，用于替换现有表单实例中的数据。读文件获取。
|
||
|
||
Returns:
|
||
Response: 返回API请求的响应对象。
|
||
"""
|
||
|
||
api = f'https://api.dingtalk.com//v1.0/yida/forms/instances'
|
||
|
||
headers = {
|
||
"Content-Type": "application/json",
|
||
"x-acs-dingtalk-access-token": token
|
||
}
|
||
|
||
payload = {
|
||
"appType": "APP_UYZ0KG6L0CCNV80GZ66O",
|
||
"systemToken": "XA966F81JAJOFCVVVKO64E9MIIZV1EWE5SFMKJ2",
|
||
"userId": "yida_pub_account", # 曹伟 id
|
||
"language": "zh_CN",
|
||
"useLatestVersion": "false",
|
||
"formInstanceId": formInstanceId,
|
||
"updateFormDataJson": json.dumps(data_new, cls=NpEncoder),
|
||
|
||
}
|
||
|
||
res = requests.put(api, headers=headers, json=payload)
|
||
return res
|
||
|
||
def processes_instancesInfos(self, token, id, appType="APP_UYZ0KG6L0CCNV80GZ66O",
|
||
systemToken="XA966F81JAJOFCVVVKO64E9MIIZV1EWE5SFMKJ2", delay=2, max_retries=10):
|
||
"""
|
||
根据流程实例ID获取流程实例
|
||
:param token:登录验证token，用于API调用的身份验证。
|
||
:param id: 表单唯一标识符，用于指定需要读取哪个表单的数据。
|
||
:param appType:应用类型标识符，默认为 "APP_UYZ0KG6L0CCNV80GZ66O"
|
||
:param systemToken:系统token，默认为固定值
|
||
:param delay:等待时间2s
|
||
:param max_retries:最大重试次数，默认为10次
|
||
:return:
|
||
"""
|
||
attempt = 0
|
||
api = f'https://api.dingtalk.com/v2.0/yida/processes/instancesInfos/{id}'
|
||
headers = {
|
||
"Content-Type": "application/json",
|
||
"x-acs-dingtalk-access-token": token
|
||
}
|
||
formData = {
|
||
"appType": appType,
|
||
"systemToken": systemToken,
|
||
"userId": "yida_pub_account", # 超级管理员账号
|
||
"language": "zh_CN",
|
||
}
|
||
|
||
while True:
|
||
if attempt >= max_retries:
|
||
print(f"请求失败，已达最大重试次数 {max_retries},跳过本次请求")
|
||
break
|
||
try:
|
||
res = requests.get(api, headers=headers, params=formData)
|
||
|
||
return res.json()
|
||
except (requests.exceptions.RequestException, Exception) as e:
|
||
print(f"请求出现异常: {e}, 正在重试({attempt + 1}/{max_retries})...")
|
||
time.sleep(delay) # 等待指定的延迟时间后再次尝试
|
||
attempt += 1
|
||
|
||
def read_processes(self, token, formUuid, page, n, appType="APP_UYZ0KG6L0CCNV80GZ66O",
|
||
systemToken="XA966F81JAJOFCVVVKO64E9MIIZV1EWE5SFMKJ2", max_retries=10, delay=2):
|
||
"""
|
||
函数功能：读取普通表单的所有数据
|
||
|
||
Args:
|
||
token (str): 登录验证token，用于API调用的身份验证。
|
||
formUuid (str): 表单唯一标识符，用于指定需要读取哪个表单的数据。
|
||
page (int): 当前页码，用于分页查询，指定从哪一页开始获取数据。
|
||
n (int): 每页显示的数据条数，用于控制每页返回的数据量。
|
||
|
||
Returns:
|
||
dict: 返回从API获取的表单实例数据的JSON解析结果。通常包括表单实例的基本信息、字段值等。
|
||
"""
|
||
|
||
api = 'https://api.dingtalk.com/v1.0/yida/forms/instances/search'
|
||
headers = {
|
||
"Content-Type": "application/json",
|
||
"x-acs-dingtalk-access-token": token
|
||
}
|
||
formData = {
|
||
"appType": appType,
|
||
"systemToken": systemToken,
|
||
"userId": "yida_pub_account", # 超级管理员账号
|
||
"language": "zh_CN",
|
||
"formUuid": formUuid,
|
||
'currentPage': page,
|
||
'pageSize': n
|
||
}
|
||
attempt = 0
|
||
while True:
|
||
if attempt >= max_retries:
|
||
error_task_logger.error(f"请求失败，已达最大重试次数 {max_retries}，无法获取普通表单数据,跳过本次请求。")
|
||
break
|
||
|
||
try:
|
||
res = requests.post(api, headers=headers, json=formData)
|
||
# logger.info(f"HTTP Status Code: {res.status_code}")
|
||
# logger.info(f"Response Headers: {dict(res.headers)}")
|
||
# logger.info(f"Raw Response Text: {res.text}") # 注意：不要在线上环境打印敏感数据！
|
||
return res.json()
|
||
|
||
except requests.exceptions.RequestException as e:
|
||
logger.warning(f"请求异常: {e}，正在尝试第 {attempt + 1} 次重试...")
|
||
time.sleep(delay)
|
||
attempt += 1
|
||
|
||
def read_processes_instances(self, token, formUuid, page, n, appType="APP_UYZ0KG6L0CCNV80GZ66O",
|
||
systemToken="XA966F81JAJOFCVVVKO64E9MIIZV1EWE5SFMKJ2", instanceStatus="RUNNING",
|
||
max_retries=10, delay=2, createFromTimeGMT=None, createToTimeGMT=None,
|
||
modifiedFromTimeGMT=None,
|
||
modifiedToTimeGMT=None, searchFieldJson={},useAlias=False):
|
||
"""
|
||
函数功能：读取流程表单的所有数据，并加入重试机制。
|
||
|
||
Args:
|
||
token (str): 登录验证token，用于API调用的身份验证。
|
||
formUuid (str): 表单唯一标识符，用于指定需要读取哪个表单的实例数据。
|
||
page (int): 分页参数，指定请求的数据页码。
|
||
n (int): 每页显示的数据条数。
|
||
appType (str): 应用类型标识符，默认为 "APP_UYZ0KG6L0CCNV80GZ66O"
|
||
systemToken (str): 系统token，默认为固定值
|
||
instanceStatus (str): 流程实例状态，默认为"RUNNING"
|
||
max_retries (int): 最大重试次数，默认为10次
|
||
delay (int): 每次重试之间的延迟秒数，默认为2秒
|
||
|
||
Returns:
|
||
dict: 返回从API获取的流程表单实例数据的JSON解析结果。
|
||
|
||
Raises:
|
||
Exception: 如果达到最大重试次数仍未成功，则抛出异常。
|
||
"""
|
||
|
||
attempt = 0
|
||
api = f'https://api.dingtalk.com/v1.0/yida/processes/instances?pageNumber={page}&pageSize={n}'
|
||
headers = {
|
||
"Content-Type": "application/json",
|
||
"x-acs-dingtalk-access-token": token
|
||
}
|
||
formData = {
|
||
"appType": appType,
|
||
"systemToken": systemToken,
|
||
"userId": "yida_pub_account", # 超级管理员账号
|
||
"language": "zh_CN",
|
||
"formUuid": formUuid,
|
||
"instanceStatus": instanceStatus, # 运行中
|
||
"createFromTimeGMT": createFromTimeGMT,
|
||
"createToTimeGMT": createToTimeGMT,
|
||
"modifiedFromTimeGMT": modifiedFromTimeGMT,
|
||
"modifiedToTimeGMT": modifiedToTimeGMT,
|
||
"searchFieldJson": json.dumps(
|
||
searchFieldJson
|
||
),
|
||
"useAlias": useAlias,
|
||
}
|
||
# print(formData)
|
||
|
||
while True:
|
||
if attempt >= max_retries:
|
||
error_task_logger.error(f"请求失败，已达最大重试次数 {max_retries}，无法获取流程实例数据,跳过本次请求。")
|
||
break
|
||
|
||
try:
|
||
res = requests.post(api, headers=headers, json=formData)
|
||
# print(res.json())
|
||
res.raise_for_status() # 如果返回状态码不是2xx，抛出异常
|
||
return res.json()
|
||
|
||
except requests.exceptions.RequestException as e:
|
||
logger.warning(f"请求异常: {e}，正在尝试第 {attempt + 1} 次重试...")
|
||
time.sleep(delay)
|
||
attempt += 1
|
||
|
||
@staticmethod
|
||
def get_ids_query(token, formUuid, appType, systemToken, formInstanceIdList=None, max_retries=10, delay=2):
|
||
"""
|
||
函数功能：读取表单的所有数据，并加入重试机制。
|
||
|
||
Args:
|
||
token (str): 登录验证token，用于API调用的身份验证。
|
||
formUuid (str): 表单唯一标识符，用于指定需要读取哪个表单的实例数据。
|
||
page (int): 分页参数，指定请求的数据页码。
|
||
n (int): 每页显示的数据条数。
|
||
appType (str): 应用类型标识符，默认为 "APP_UYZ0KG6L0CCNV80GZ66O"
|
||
systemToken (str): 系统token，默认为固定值
|
||
instanceStatus (str): 流程实例状态，默认为"RUNNING"
|
||
max_retries (int): 最大重试次数，默认为10次
|
||
delay (int): 每次重试之间的延迟秒数，默认为2秒
|
||
|
||
Returns:
|
||
dict: 返回从API获取的流程表单实例数据的JSON解析结果。
|
||
|
||
Raises:
|
||
Exception: 如果达到最大重试次数仍未成功，则抛出异常。
|
||
"""
|
||
|
||
attempt = 0
|
||
api = f'https://api.dingtalk.com/v1.0/yida/forms/instances/ids/query'
|
||
headers = {
|
||
"Content-Type": "application/json",
|
||
"x-acs-dingtalk-access-token": token
|
||
}
|
||
formData = {
|
||
"appType": appType,
|
||
"systemToken": systemToken,
|
||
"userId": "yida_pub_account", # 超级管理员账号
|
||
"language": "zh_CN",
|
||
"formUuid": formUuid,
|
||
"formInstanceIdList": formInstanceIdList,
|
||
}
|
||
# print(formData)
|
||
|
||
while True:
|
||
if attempt >= max_retries:
|
||
error_task_logger.error(f"请求失败，已达最大重试次数 {max_retries}，无法获取流程实例数据,跳过本次请求。")
|
||
break
|
||
|
||
try:
|
||
res = requests.post(api, headers=headers, json=formData)
|
||
# print(res.json())
|
||
res.raise_for_status() # 如果返回状态码不是2xx，抛出异常
|
||
return res.json()
|
||
|
||
except requests.exceptions.RequestException as e:
|
||
logger.warning(f"请求异常: {e}，正在尝试第 {attempt + 1} 次重试...")
|
||
time.sleep(delay)
|
||
attempt += 1
|
||
|
||
def get_approval_records(self, token: str, processInstanceId: str, appType="APP_UYZ0KG6L0CCNV80GZ66O",
|
||
systemToken="XA966F81JAJOFCVVVKO64E9MIIZV1EWE5SFMKJ2", max_retries=10, delay=2):
|
||
"""
|
||
函数功能：获取流程表单的审批记录，适用于"F6客户服务"应用，并且包含重试机制。
|
||
|
||
Args:
|
||
token (str): 登录验证token，用于API调用的身份验证。
|
||
processInstanceId (str): 流程实例ID，用于标识需要获取审批记录的具体流程实例。
|
||
appType (str): 应用类型标识符，默认为 "APP_UYZ0KG6L0CCNV80GZ66O"
|
||
systemToken (str): 系统token，默认为固定值
|
||
max_retries (int): 最大重试次数，默认为10次
|
||
delay (int): 每次重试之间的延迟秒数，默认为2秒
|
||
|
||
Returns:
|
||
dict: 返回从API获取的审批记录的JSON解析结果。通常包括审批步骤、审批人、审批时间等信息。
|
||
"""
|
||
attempt = 0
|
||
userId = "yida_pub_account"
|
||
api = f'https://api.dingtalk.com/v1.0/yida/processes/operationRecords?appType={appType}&systemToken={systemToken}&userId={userId}&language=zh_CN&processInstanceId={processInstanceId}'
|
||
headers = {
|
||
"Content-Type": "application/json",
|
||
"x-acs-dingtalk-access-token": token
|
||
}
|
||
|
||
while True:
|
||
if attempt >= max_retries:
|
||
error_task_logger.error(f"请求失败，已达最大重试次数 {max_retries}，无法获取审批数据,跳过本次请求。")
|
||
break
|
||
|
||
try:
|
||
res = requests.get(api, headers=headers)
|
||
res.raise_for_status() # 如果响应状态码不是2xx，则抛出HTTPError
|
||
return res.json()
|
||
except (requests.exceptions.RequestException, Exception) as e:
|
||
logger.warning(f"请求出现异常: {e}, 正在重试({attempt + 1}/{max_retries})...")
|
||
time.sleep(delay) # 等待指定的延迟时间后再次尝试
|
||
attempt += 1
|
||
|
||
def get_process_design_structure(self, token: str, processID: str, appType="APP_UYZ0KG6L0CCNV80GZ66O",
|
||
systemToken="XA966F81JAJOFCVVVKO64E9MIIZV1EWE5SFMKJ2", max_retries=10, delay=2):
|
||
"""
|
||
函数功能：获取流程表单的审批记录，适用于"F6客户服务"应用，并且包含重试机制。
|
||
|
||
Args:
|
||
token (str): 登录验证token，用于API调用的身份验证。
|
||
processInstanceId (str): 流程实例ID，用于标识需要获取审批记录的具体流程实例。
|
||
appType (str): 应用类型标识符，默认为 "APP_UYZ0KG6L0CCNV80GZ66O"
|
||
systemToken (str): 系统token，默认为固定值
|
||
max_retries (int): 最大重试次数，默认为10次
|
||
delay (int): 每次重试之间的延迟秒数，默认为2秒
|
||
|
||
Returns:
|
||
dict: 返回从API获取的审批记录的JSON解析结果。通常包括审批步骤、审批人、审批时间等信息。
|
||
"""
|
||
attempt = 0
|
||
userId = "yida_pub_account"
|
||
api = f'https://api.dingtalk.com/v1.0/yida/processes/{processID}definitions/designs?systemToken={systemToken}&userId={userId}&appType={appType}"x-acs-dingtalk-access-token": token'
|
||
headers = {
|
||
# "Content-Type": "application/json",
|
||
"x-acs-dingtalk-access-token": token
|
||
}
|
||
|
||
while True:
|
||
if attempt >= max_retries:
|
||
error_task_logger.error(f"请求失败，已达最大重试次数 {max_retries}，无法获取流程实例数据,跳过本次请求。")
|
||
break
|
||
|
||
try:
|
||
res = requests.get(api, headers=headers)
|
||
res.raise_for_status() # 如果响应状态码不是2xx，则抛出HTTPError
|
||
return res.json()
|
||
except (requests.exceptions.RequestException, Exception) as e:
|
||
logger.warning(f"请求出现异常: {e}, 正在重试({attempt + 1}/{max_retries})...")
|
||
time.sleep(delay) # 等待指定的延迟时间后再次尝试
|
||
attempt += 1
|
||
|
||
def get_staffID(self, TOKEN: str, ALL_DATA_staff, staff_name):
|
||
"""
|
||
函数功能：根据员工姓名获取员工工号
|
||
|
||
Args:
|
||
TOKEN (str): 登录验证token，虽然在此函数中未直接使用，但可能是为了保持接口的一致性而保留。
|
||
ALL_DATA_staff (list): 包含所有员工数据的列表，每个元素是一个字典，代表一个员工的数据记录。
|
||
staff_name (str): 需要查询的员工姓名。
|
||
|
||
Returns:
|
||
list: 返回与指定员工姓名匹配的所有员工的工号列表。如果找不到匹配的员工，则返回空列表。
|
||
"""
|
||
res_new = [v['formData']['textField_lfrw3u59'] for v in ALL_DATA_staff if
|
||
v['formData']['textField_lfrw3u58'] == staff_name]
|
||
return res_new
|
||
|
||
def aggree_approval(self, token: str, taskId: str, processInstanceId: str, formData: dict, res_new):
|
||
"""_summary_
|
||
|
||
函数功能：同意审批节点 --F6客户服务 应用
|
||
|
||
Args:
|
||
token (str): 登录验证token
|
||
taskId (str): 获取到的审批节点ID
|
||
processInstanceId (str): 读取文件获得的实例ID
|
||
formData (dict): 数据样式
|
||
res_new (响应值): 从员工ID表里获取到员工名对应的员工ID
|
||
|
||
Returns:
|
||
响应值: 返回请求结果
|
||
"""
|
||
api = 'https://api.dingtalk.com/v1.0/yida/tasks/execute'
|
||
headers = {
|
||
"Content-Type": "application/json",
|
||
"x-acs-dingtalk-access-token": token
|
||
}
|
||
payload = {
|
||
"outResult": "AGREE",
|
||
"appType": "APP_UYZ0KG6L0CCNV80GZ66O",
|
||
"systemToken": "XA966F81JAJOFCVVVKO64E9MIIZV1EWE5SFMKJ2",
|
||
"remark": "同意(接口自动)",
|
||
"formDataJson": json.dumps(formData, cls=NpEncoder),
|
||
"processInstanceId": processInstanceId,
|
||
"userId": res_new,
|
||
"language": "zh_CN",
|
||
"taskId": int(taskId)
|
||
}
|
||
|
||
res = requests.post(api, headers=headers, json=payload)
|
||
return res
|
||
|
||
def transfer_approvel_node(self, token, processInstanceId, nowActionExecutorId, taskId):
|
||
"""_summary_
|
||
|
||
函数功能：转交审批节点
|
||
|
||
Args:
|
||
token (str): 登录验证token
|
||
processInstanceId (str): 读取文件获取到的实例ID
|
||
nowActionExecutorId (srt): 读取文件获取到的新的任务处理人工号
|
||
taskId (响应值): 任务ID
|
||
|
||
Returns:
|
||
响应值: 返回请求结果
|
||
"""
|
||
api = f'https://api.dingtalk.com//v1.0/yida/tasks/redirect'
|
||
|
||
headers = {
|
||
"Content-Type": "application/json",
|
||
"x-acs-dingtalk-access-token": token
|
||
}
|
||
payload = {
|
||
"processInstanceId": processInstanceId,
|
||
"byManager": "y",
|
||
"appType": "APP_UYZ0KG6L0CCNV80GZ66O",
|
||
"systemToken": "XA966F81JAJOFCVVVKO64E9MIIZV1EWE5SFMKJ2",
|
||
"language": "zh_CN",
|
||
"remark": "转交(接口自动)",
|
||
"nowActionExecutorId": nowActionExecutorId, # 新的任务处理人工号
|
||
"userId": "2268275546837446",
|
||
"taskId": int(taskId)
|
||
}
|
||
res = requests.post(api, headers=headers, json=payload)
|
||
return res
|
||
|
||
def delete_form(self, token, id):
|
||
"""
|
||
函数功能：删除表单实例
|
||
|
||
Args:
|
||
token (str): 登录验证token，用于API调用的身份验证。
|
||
id (str): 表单实例ID，标识要删除的具体表单实例。
|
||
|
||
Returns:
|
||
响应值: 如果请求成功，则返回服务器的JSON响应；如果请求失败或无响应，则返回一个表示删除成功的信息。
|
||
"""
|
||
api = f'https://api.dingtalk.com///v1.0/yida/forms/instances?appType=APP_UYZ0KG6L0CCNV80GZ66O&systemToken=XA966F81JAJOFCVVVKO64E9MIIZV1EWE5SFMKJ2&userId=2268275546837446&language=zh_CN&formInstanceId=' + id
|
||
|
||
headers = {
|
||
"Content-Type": "application/json",
|
||
"x-acs-dingtalk-access-token": token
|
||
}
|
||
|
||
res = requests.delete(api, headers=headers)
|
||
if res is None:
|
||
res = f'{id}已删除'
|
||
return res
|
||
else:
|
||
return res.json()
|
||
|
||
def get_form_structures(self, token, formUuid, appType="APP_UYZ0KG6L0CCNV80GZ66O",
|
||
systemToken="XA966F81JAJOFCVVVKO64E9MIIZV1EWE5SFMKJ2", max_retries=20):
|
||
"""
|
||
获取表单结构（宜搭）
|
||
注意：该接口必须使用 GET 方法，参数通过 query string 传递
|
||
"""
|
||
|
||
api_base = 'https://api.dingtalk.com/v1.0/yida/forms/formFields'
|
||
|
||
# 构造查询参数
|
||
params = {
|
||
"appType": appType,
|
||
"systemToken": systemToken,
|
||
"formUuid": formUuid,
|
||
"userId": "2268275546837446",
|
||
}
|
||
|
||
headers = {
|
||
"x-acs-dingtalk-access-token": token
|
||
}
|
||
|
||
retries = 0
|
||
while retries < max_retries:
|
||
try:
|
||
# 使用 GET，并传入 params（自动编码为 query string）
|
||
res = requests.get(api_base, headers=headers, params=params, timeout=10)
|
||
if res.status_code == 200:
|
||
return res.json()
|
||
else:
|
||
print(f"请求失败，状态码: {res.status_code}, 响应: {res.text}")
|
||
except Exception as e:
|
||
print(f"请求异常: {e}")
|
||
|
||
retries += 1
|
||
time.sleep(0.1)
|
||
|
||
return {} # 或抛出异常
|
||
|
||
|
||
class NpEncoder(json.JSONEncoder):
|
||
def default(self, obj):
|
||
if isinstance(obj, np.integer):
|
||
return int(obj)
|
||
elif isinstance(obj, np.floating):
|
||
return float(obj)
|
||
elif isinstance(obj, np.ndarray):
|
||
return obj.tolist()
|
||
else:
|
||
return super(NpEncoder, self).default(obj)
|