# api.py import requests import json import numpy as np # 导入numpy库用于处理numpy数组 import time # 换应用记得修改应用编码 class API: # 常用变量 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): """ 函数功能：读取流程表单的所有数据，并加入重试机制。 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, } 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] # print(res_new) 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, appType="APP_UYZ0KG6L0CCNV80GZ66O", systemToken="XA966F81JAJOFCVVVKO64E9MIIZV1EWE5SFMKJ2"): """ 函数功能：删除表单实例 添加了重试机制：最多5次，每次等待2秒 Args: token (str): 登录验证token id (str): 表单实例ID appType (str): 应用类型 systemToken (str): 系统token Returns: 响应值: 成功返回JSON响应，失败返回错误信息 """ max_retries = 5 retries = 0 while True: try: api = f'https://api.dingtalk.com///v1.0/yida/forms/instances?appType={appType}&systemToken={systemToken}&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: return f'{id}已删除' else: return res.json() except Exception as e: retries += 1 if retries >= max_retries: return f"删除失败，已达到最大重试次数: {str(e)}" time.sleep(2) # 等待2秒后重试 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)