433 lines
18 KiB
Python
433 lines
18 KiB
Python
# api.py
|
|
import requests
|
|
import json
|
|
import numpy as np # 导入numpy库用于处理numpy数组
|
|
import time
|
|
|
|
|
|
# 换应用记得修改应用编码
|
|
|
|
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/v1.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)
|
|
res.raise_for_status() # 如果响应状态码不是2xx，则抛出HTTPError
|
|
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"):
|
|
"""
|
|
函数功能：读取普通表单的所有数据
|
|
|
|
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
|
|
}
|
|
res = requests.post(api, headers=headers, json=formData)
|
|
print(formData)
|
|
return res.json()
|
|
|
|
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):
|
|
"""
|
|
函数功能：读取流程表单的所有数据，并加入重试机制。
|
|
|
|
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,
|
|
}
|
|
|
|
while True:
|
|
if attempt >= max_retries:
|
|
print(f"请求失败，已达最大重试次数 {max_retries}，无法获取流程实例数据,跳过本次请求。")
|
|
break
|
|
|
|
try:
|
|
res = requests.post(api, headers=headers, json=formData)
|
|
res.raise_for_status() # 如果返回状态码不是2xx，抛出异常
|
|
return res.json()
|
|
|
|
except requests.exceptions.RequestException as e:
|
|
print(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:
|
|
print(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:
|
|
print(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
|
|
|
|
}
|
|
#
|
|
# prams = {
|
|
# "systemToken": systemToken,
|
|
# "appType": appType,
|
|
# "userId": userId,
|
|
# "language": "zh_CN",
|
|
# }
|
|
|
|
while True:
|
|
if attempt >= max_retries:
|
|
print(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:
|
|
print(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):
|
|
"""
|
|
函数功能：获取表单结构 # 宜搭废弃
|
|
|
|
Args:
|
|
token (str): 登录验证token，用于API调用的身份验证。
|
|
formUuid (str): 表单的UUID，标识要获取结构的表单。
|
|
appType
|
|
|
|
Returns:
|
|
响应值: 如果请求成功，则返回服务器的JSON响应；如果请求失败或无响应，则返回一个表示获取成功信息的字符串。
|
|
"""
|
|
|
|
api = f'https://api.dingtalk.com/v1.0/yida/forms/formFields'
|
|
headers = {
|
|
"Content-Type": "application/json",
|
|
"x-acs-dingtalk-access-token": token
|
|
}
|
|
retries = 0
|
|
data_get = {}
|
|
payload = {
|
|
"formUuid": formUuid,
|
|
"appType": appType,
|
|
"systemToken": systemToken,
|
|
"userId": "2268275546837446",
|
|
}
|
|
while retries < max_retries:
|
|
res = requests.post(api, headers=headers, json=payload)
|
|
if res.status_code == 200:
|
|
data_get = res.json()
|
|
break
|
|
else:
|
|
retries += 1
|
|
time.sleep(0.1)
|
|
return data_get
|
|
|
|
|
|
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)
|