脚本

张阳脚本/宜搭刷数据/api.py

@@ -0,0 +1,409 @@
+# 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)