# 基础函数配置 import pandas as pd import pandas as pd import requests from pathlib import Path from urllib.parse import quote import json import numpy as np import time from datetime import date, timedelta ROOT = Path('.').absolute() # 当前工作目录 def generateToken() -> str: """ 生成 token """ token_api = 'https://api.dingtalk.com/v1.0/oauth2/accessToken' # 该信息在钉钉开放应用中 data = { "appKey": "ding5kqocon5s9oph5uq", "appSecret": 'HL1jgsIIfLAC0eTH0A1m4mwxUDqbgsiPeCCGGE3ocM6qJBTIW7Ivt9drxF_Z4Kb_' } res = requests.post(token_api, json=data) token = res.json()['accessToken'] return token def transcation(token,FORMID,data_new): """ 函数功能：更新表单内容 """ 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" : "2268275546837446", # 曹伟 id "language" : "zh_CN", "useLatestVersion" : "false", "formInstanceId" : FORMID, "updateFormDataJson" : json.dumps(data_new, cls=NpEncoder), } res = requests.put(api, headers=headers, json=payload) return res.json() def aggree_approval(token: str, taskId: str, processInstanceId: str, formData: dict,res_new): """ 函数功能：同意审批节点 --F6客户服务 应用 """ api = f'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": "yida_pub_account", "userId": res_new, "language": "zh_CN", "taskId": int(taskId) } res = requests.post(api, headers=headers, json=payload) # print('同意审批节点') return res def read_instances_new(token, formUuid, page, n): """ 函数功能：读取流程表单的所有数据 """ 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": "APP_UYZ0KG6L0CCNV80GZ66O", "systemToken": "XA966F81JAJOFCVVVKO64E9MIIZV1EWE5SFMKJ2", "userId": "yida_pub_account", # 超级管理员账号 "language": "zh_CN", "formUuid": formUuid, # "searchFieldJson": json.dumps(searchField), # 如果增加上这一项会要求升级宜搭存储 "instanceStatus": "RUNNING" } res = requests.post(api, headers=headers, json=formData) return res.json() def get_approval_records(token: str, processInstanceId: str): """ 函数功能：获取流程表单的审批记录 --F6客户服务 应用 """ appType = "APP_UYZ0KG6L0CCNV80GZ66O" systemToken = "XA966F81JAJOFCVVVKO64E9MIIZV1EWE5SFMKJ2" 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 } res = requests.get(api, headers=headers) # print('获取流程表单的审批记录') return res.json() def read_instances_pt(token, formUuid, page, n): """ 函数功能：读取普通表单的所有数据 """ api = f'https://api.dingtalk.com//v1.0/yida/forms/instances/search' headers = { "Content-Type": "application/json", "x-acs-dingtalk-access-token": token } formData = { "appType" : "APP_UYZ0KG6L0CCNV80GZ66O", "systemToken" : "XA966F81JAJOFCVVVKO64E9MIIZV1EWE5SFMKJ2", "userId" : "yida_pub_account", "language" : "zh_CN", "formUuid" : formUuid, "currentPage" : page, "pageSize" : n } res = requests.post(api, headers=headers, json=formData) return res.json() def read_processes(token, formUuid, page, n): """ 函数功能：读取普通表单的所有数据 """ api = f'https://api.dingtalk.com//v1.0/yida/forms/instances/search' headers = { "Content-Type": "application/json", "x-acs-dingtalk-access-token": token } formData = { "appType": "APP_UYZ0KG6L0CCNV80GZ66O", "systemToken": "XA966F81JAJOFCVVVKO64E9MIIZV1EWE5SFMKJ2", "userId": "yida_pub_account", # 超级管理员账号 "language": "zh_CN", "formUuid": formUuid, 'currentPage':page, 'pageSize':n } res = requests.post(api, headers=headers, json=formData) return res.json() def instances_id(TOKEN,id): """ 函数功能：查询表单实例 """ api = f'https://api.dingtalk.com//v1.0/yida/forms/instances/{id}?appType=APP_UYZ0KG6L0CCNV80GZ66O&systemToken=XA966F81JAJOFCVVVKO64E9MIIZV1EWE5SFMKJ2&userId=yida_pub_account&language=zh_CN' headers = { "Content-Type": "application/json", "x-acs-dingtalk-access-token": TOKEN } res = requests.get(api, headers=headers) return res.json() def component(FORMID,TOKEN): """ 获取组件信息 """ api = f'https://api.dingtalk.com//v1.0/yida/forms/formFields' headers = { "Content-Type": "application/json", "x-acs-dingtalk-access-token": TOKEN } payload = { "formUuid" : FORMID, "appType" : "APP_UYZ0KG6L0CCNV80GZ66O", # "formDataJson" : json.dumps(formData, cls=NpEncoder), "systemToken" : "XA966F81JAJOFCVVVKO64E9MIIZV1EWE5SFMKJ2", # "language" : "zh_CN", "userId" : "yida_pub_account" } res = requests.get(api, headers=headers, json=payload) return res.json() def component_id(id,TOKEN): api = f'https://api.dingtalk.com/v1.0/yida/processes/instancesInfos/{id}?appType=APP_UYZ0KG6L0CCNV80GZ66O&systemToken=XA966F81JAJOFCVVVKO64E9MIIZV1EWE5SFMKJ2&userId=yida_pub_account&language=zh_CN' headers = { "Content-Type": "application/json", "x-acs-dingtalk-access-token": TOKEN } res = requests.get(api, headers=headers) return res.json() def find_indexes(daichuli_mendian_name,i_two): indexes = [i for i, x in enumerate(daichuli_mendian_name) if x == i_two] return indexes def get_staffID(TOKEN: str,ALL_DATA_staff,staff_name): res_new = [v['formData']['textField_lfrw3u59'] for v in ALL_DATA_staff if v['formData']['textField_lfrw3u58']== staff_name] # print('通过员工名称获取员工id') return res_new 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) TOKEN = generateToken() ALL_DATA_id = {} tinydict = {'0天处理情况': 1, '30天联系情况': 2, '60天联系情况': 3, '90天联系情况': 4, '120天联系情况': 5} CODE_NAME = ['radioField_kuntp6fm','radioField_ksydghrf','dateField_ksydghri','textField_ksydghrg','dateField_ksydghrj','radioField_ksydghrk','dateField_ksydghrl','textField_ksydghrm','textField_livc8bjj','dateField_lifr1fdv','radioField_kuntp6fn','radioField_kuhnydlp','dateField_kuhnydlr','textField_kuhnydls','dateField_kuhnydlt','radioField_kuhnydlu','dateField_kuhnydlv','textField_kuhnydlw','textField_livc8bjk','dateField_lifr1fdw','radioField_kurxyhvp','radioField_kuhnydmd','dateField_kuhnydmf','textField_kuhnydmg','dateField_kuhnydmh','radioField_kuhnydmi','dateField_kuhnydmj','textField_kuhnydmk','textField_livc8bjl','dateField_lifr1fdx','radioField_kurxyhvq','radioField_kuhnydn0','dateField_kuhnydn2','textField_kuhnydn3','dateField_kuhnydn4','radioField_kuhnydn5','dateField_kuhnydn6','textField_kuhnydn7','textField_livc8bjm','dateField_lifr1fdy','radioField_l85ppdia','dateField_l85ppdib','textField_lbodyv8i','textField_l85ppdic','dateField_l85ppdid','dateField_l85ppdif','textField_l85ppdig','radioField_r3yeqvd','textAreaField_972lhkt','employeeField_lhbkudx0','textField_lhbnuvtn','selectField_lb0gxmq8','selectField_lb5xhwk4','selectField_lc70femm','radioField_l85ppdie','textField_livcjyvr','dateField_livcjyvp','selectField_l31clxfy','selectField_l31clxfz','selectField_l31clxg0','selectField_l31clxg1','selectField_l31clxg2','selectField_l31clxg3','textareaField_l31clxg4','radioField_ljqi5we3','textareaField_liviovx0','radioField_la7x32mq','textField_la7x32mp','imageField_la7x32mt'] # 读取 [流程]续约服务流程 FORMID = "FORM-PE866MD1MJMU0WGLYRFLYEN5YN9L1I55Z7ZUK22" form_data = read_instances_new(token=TOKEN, formUuid=FORMID, page=1, n=100) PAGES = form_data.get('totalCount')//100 + 1 """ 获取全量数据 """ for v in range(1, PAGES+1): form_data = read_instances_new(token=TOKEN, formUuid=FORMID, page=v, n=100) for data in form_data.get('data'): ALL_DATA_id[data['data']['textField_ksirro5j']]=data["processInstanceId"] # ALL_DATA_data.append(data["instanceValue"]) # ALL_DATA_mengdian_name.append(data['formData']['textField_ksirro5j']) print(f'读取到流程表中 {len(ALL_DATA_id)} 条数据！') """ 获取待处理数据 """ for i in range(1, PAGES+1): form_data = read_instances_new(token=TOKEN, formUuid=FORMID, page=i, n=100) for data in form_data.get('data'): try: data['data']['multiSelectField_ll5x02m5'] ALL_DATA = "正确" except: ALL_DATA = "错误" if ALL_DATA == "正确": # 获取到数据开始处理 """ 通过员工名称获取员工id""" # 读取员工对应关系：宜搭员工-ID对应表 FORMID_two = "FORM-EA866E715PF9YA7ECCAGSABX91Q72PVA3WRFL6" # 宜搭员工-ID对应表 FORM-EA866E715PF9YA7ECCAGSABX91Q72PVA3WRFL6 # 读取流程表单数据 form_data_two = read_processes(token=TOKEN, formUuid=FORMID_two, page=1, n=100) PAGES_two = form_data_two.get('totalCount')//100 + 1 ALL_DATA_staff = [] """ 获取员工全量数据 """ for i in range(1, PAGES_two+1): # form_data = read_processes_instances(token=TOKEN, formUuid=FORMID, createFromTimeGMT=CREATE_FROM, createToTimeGMT=CREATE_TO, page=i, n=100, searchField={'textField_l7if5ff9': '否'}) form_data_two = read_processes(token=TOKEN, formUuid=FORMID_two, page=i, n=100) for dataBB in form_data_two.get('data'): ALL_DATA_staff.append(dataBB) # -*- coding: utf-8 -*- import psycopg2 import pandas as pd # 获得连接 conn = psycopg2.connect(database="f6_bi", user="BASIC$ro_caowei", password="!ro_caowei123", host="hgprecn-cn-nif1vnv0y002-cn-shanghai.hologres.aliyuncs.com", port="80") # 获得游标对象，一个游标对象可以对数据库进行执行操作 cursor = conn.cursor() import datetime now_time = datetime.datetime.now() yes_time = now_time + datetime.timedelta(days=-1) yes_time_nyr = int(yes_time.strftime('%Y%m%d'))# 获取前一天日期 # sql语句 建表 sql =f"""SELECT * FROM "public"."holo_ads_report_saas_profile_ngv_detail_d" WHERE "date_id" = '{yes_time_nyr}';""" # 执行语句 cursor.execute(sql) # 获取结果集的每一行 rows = cursor.fetchall() # 获取所有字段名 all_fields = cursor.description #执行结果转化为dataframe col = [] for i in all_fields: col.append(i[0]) data_NGV = pd.DataFrame(list(rows),columns=col) # data_NGV.to_excel(r'C:\Users\admin\Desktop\NGV明细.xlsx') # data_NGV = pd.read_excel(r'C:\Users\admin\Desktop\NGV明细.xlsx') # 关闭数据库连接 cursor.close() conn.close() '''进行处理操作''' # 获取可更新内容 updateFormDataJson data_json = instances_id(TOKEN,data["processInstanceId"]) updateFormDataJson = {} for list_name in CODE_NAME: try: updateFormDataJson[list_name] = data_json['formData'][list_name] except: pass print(updateFormDataJson) # 获取当前流程节点 res_new = get_approval_records(token=TOKEN, processInstanceId=data["processInstanceId"]) records_new = res_new.get('result') # 获取审批节点的 taskId records_new = [item for item in records_new if item.get('type') == "TODO"] taskId_new = records_new[0].get('taskId') print(taskId_new,records_new[0]['showName']) # 获取待同步表单实例id gongsi_id = component_id(data["processInstanceId"],TOKEN)["data"]['textField_kuntp6fj'] # 获取公司id gongsi_name = component_id(data["processInstanceId"],TOKEN)["data"]['textField_ksirm9xx'] # 获取公司id # NGV中获取当前公司id对应的内容 row_indexes = [] row_indexes = data_NGV[data_NGV['id_own_group'] == gongsi_id].index.tolist() org_name_list = [] org_id_list = [] for i_one in row_indexes: org_name_list.append(data_NGV.loc[i_one,"org_name"]) org_id_list.append(data_NGV.loc[i_one,"id_own_org"]) print(org_name_list,org_id_list) # 根据门店名称获取单号 + 校验公司名称是否相同，不同则继续查找 for i_two in org_name_list: # i_two_list = find_indexes(data['data']['textField_ksirro5j'],i_two)[0] if data['data']['textField_ksirm9xx'] == gongsi_name : print(data['data']['textField_ksirm9xx'] , i_two) try: # 开始同步数据 res_three = transcation(TOKEN,ALL_DATA_id[i_two],updateFormDataJson) print(res_three,"Gengx ") # print(res_three,daichuli_gongsi_name[i_two_list],ALL_DATA_id[i_two_list],ALL_DATA_mengdian_name[i_two_list]) # 开始同步节点 # print(ALL_DATA_id[i_two_list]) res_new_one = get_approval_records(token=TOKEN, processInstanceId=ALL_DATA_id[i_two]) records_new_one = res_new_one.get('result') # 获取审批节点的 taskId records_new_one = [item for item in records_new_one if item.get('type') == "TODO"] taskId_new_one = records_new_one[0].get('taskId') print(taskId_new_one,records_new_one[0]['showName']) if tinydict[records_new_one[0]['showName']] > tinydict[records_new[0]['showName']] : data_new = {} res=aggree_approval(token=TOKEN, taskId=taskId_new_one, processInstanceId=ALL_DATA_id[i_two], formData=data_new,res_new =get_staffID(TOKEN,ALL_DATA_staff,records_new_one[0]['operatorName'])[0] ) print(res.json(),"liouc") except: print(i_two,"报错") break updateFormDataJson_new = {'multiSelectField_ll5x02m5':""} # transcation(TOKEN,data["processInstanceId"],updateFormDataJson_new) print(f'读取到流程表中 {len(ALL_DATA_id)} 条数据！')