367 lines
15 KiB
Python
367 lines
15 KiB
Python
import datetime
|
||
starttime = datetime.datetime.now()
|
||
|
||
# 基础信息
|
||
import pandas as pd
|
||
import numpy as np
|
||
import requests
|
||
import json
|
||
import time
|
||
import re
|
||
from datetime import datetime
|
||
from dateutil.relativedelta import relativedelta
|
||
from pathlib import Path
|
||
from urllib.parse import quote
|
||
from datetime import date, timedelta
|
||
import datetime
|
||
from io import BytesIO
|
||
|
||
ROOT = Path('.').absolute() # 当前工作目录
|
||
textField_lrzoowld = "正常" # 运行状态
|
||
textField_lrzoowlb = "" # 信息说明
|
||
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 read_processes_instances(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 read_processes_instances_new(token, formUuid, page, n,searchField):
|
||
""" 函数功能：读取流程表单的所有数据 """
|
||
|
||
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 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 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 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
|
||
|
||
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
|
||
|
||
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()
|
||
FORMID = "FORM-PE866MD1MJMU0WGLYRFLYEN5YN9L1I55Z7ZUK22" # 读取续约流程表单
|
||
|
||
|
||
|
||
LIST_DATA = []
|
||
|
||
tinydict_two = {'30天联系情况': 'radioField_kurxyhvq', '60天联系情况': 'radioField_kurxyhvp', '90天联系情况': 'radioField_kuntp6fn', '120天联系情况': 'radioField_kuntp6fm'}
|
||
""" 通过员工名称获取员工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 data in form_data_two.get('data'):
|
||
ALL_DATA_staff.append(data)
|
||
""" 获取全量数据 """
|
||
textField_kycfic6o = ['普通客户（VIP）','区域KA（MVP）','全国KA（FMVP）','重要客户（SVIP）']
|
||
for sle in range(0,len(textField_kycfic6o)):
|
||
# 读取续约流程表单
|
||
form_data = read_processes_instances_new(token=TOKEN, formUuid=FORMID, page=1, n=100, searchField={'textField_kycfic6o': textField_kycfic6o[sle]})
|
||
PAGES = form_data.get('totalCount')//100 + 1
|
||
if PAGES >50 :
|
||
PAGES = 50
|
||
for i in range(1, PAGES+1):
|
||
form_data = read_processes_instances_new(token=TOKEN, formUuid=FORMID, page=i, n=100, searchField={'textField_kycfic6o': textField_kycfic6o[sle]})
|
||
for v in range(0,len(form_data.get('data'))):
|
||
if form_data['data'][v]['data']['dateField_ksirro5l'] > 1708225911641 :
|
||
tinydict = {'0天处理情况': 1, '30天联系情况': 2, '60天联系情况': 3, '120天联系情况': 4}
|
||
# 过期日期的时间戳
|
||
expire_timestamp = form_data['data'][v]['data']['dateField_ksirro5l']/1000
|
||
|
||
# 获取距离过期日期前120天，前90天，前60天，前30天的日期
|
||
expire_date = datetime.datetime.fromtimestamp(expire_timestamp)
|
||
before_120_days = expire_date - datetime.timedelta(days=120)
|
||
before_90_days = expire_date - datetime.timedelta(days=90)
|
||
before_60_days = expire_date - datetime.timedelta(days=60)
|
||
before_30_days = expire_date - datetime.timedelta(days=30)
|
||
print(expire_date,before_120_days,before_90_days,before_60_days,before_30_days)
|
||
# 获取当前日期
|
||
current_date = datetime.datetime.now()
|
||
code_name = 0
|
||
# 比较当前日期和（前120天、前90天、前60天、前30天）的日期
|
||
if current_date < before_60_days:
|
||
code_name = 4
|
||
print("90-120天：120天联系情况")
|
||
elif current_date < before_30_days:
|
||
code_name = 3
|
||
print("30-60天：60天联系情况")
|
||
elif current_date < expire_date:
|
||
code_name = 2
|
||
print("0-30天：30天联系情况")
|
||
else:
|
||
code_name = 1
|
||
print("已过期")
|
||
try:
|
||
# 获取当前所处节点
|
||
res_new = get_approval_records(token=TOKEN, processInstanceId=form_data['data'][v]['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')
|
||
code_list = tinydict[records_new[0]['showName']] - code_name
|
||
print(records_new[0]['showName'],code_list)
|
||
if code_list>0 and code_name!=0:
|
||
# data_new = {tinydict_two[records_new[0]['showName']]:'系统'}
|
||
data_new = {}
|
||
res=aggree_approval(token=TOKEN, taskId=taskId_new, processInstanceId=form_data['data'][v]['processInstanceId'], formData=data_new,res_new =get_staffID(TOKEN,ALL_DATA_staff,records_new[0]['operatorName'])[0] )
|
||
print(res.json())
|
||
except Exception as e:
|
||
textField_lrzoowld = "正常" # 运行状态
|
||
textField_lrzoowlb = str(e) # 信息说明
|
||
else:
|
||
tinydict = {'0天处理情况': 1, '30天联系情况': 2, '60天联系情况': 3, '90天联系情况': 4, '120天联系情况': 5}
|
||
# 过期日期的时间戳
|
||
expire_timestamp = form_data['data'][v]['data']['dateField_ksirro5l']/1000
|
||
|
||
# 获取距离过期日期前120天，前90天，前60天，前30天的日期
|
||
expire_date = datetime.datetime.fromtimestamp(expire_timestamp)
|
||
before_120_days = expire_date - datetime.timedelta(days=120)
|
||
before_90_days = expire_date - datetime.timedelta(days=90)
|
||
before_60_days = expire_date - datetime.timedelta(days=60)
|
||
before_30_days = expire_date - datetime.timedelta(days=30)
|
||
print(expire_date,before_120_days,before_90_days,before_60_days,before_30_days)
|
||
# 获取当前日期
|
||
current_date = datetime.datetime.now()
|
||
code_name = 0
|
||
# 比较当前日期和（前120天、前90天、前60天、前30天）的日期
|
||
if current_date < before_90_days:
|
||
code_name = 5
|
||
print("90-120天：120天联系情况")
|
||
elif current_date < before_60_days:
|
||
code_name = 4
|
||
print("60-90天：90天联系情况")
|
||
elif current_date < before_30_days:
|
||
code_name = 3
|
||
print("30-60天：60天联系情况")
|
||
elif current_date < expire_date:
|
||
code_name = 2
|
||
print("0-30天：30天联系情况")
|
||
else:
|
||
code_name = 1
|
||
print("已过期")
|
||
try:
|
||
# 获取当前所处节点
|
||
res_new = get_approval_records(token=TOKEN, processInstanceId=form_data['data'][v]['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')
|
||
code_list = tinydict[records_new[0]['showName']] - code_name
|
||
print(records_new[0]['showName'],code_list)
|
||
if code_list>0 and code_name!=0:
|
||
# data_new = {tinydict_two[records_new[0]['showName']]:'系统'}
|
||
data_new = {}
|
||
res=aggree_approval(token=TOKEN, taskId=taskId_new, processInstanceId=form_data['data'][v]['processInstanceId'], formData=data_new,res_new =get_staffID(TOKEN,ALL_DATA_staff,records_new[0]['operatorName'])[0] )
|
||
print(res.json())
|
||
except Exception as e:
|
||
textField_lrzoowld = "正常" # 运行状态
|
||
textField_lrzoowlb = str(e) # 信息说明
|
||
|
||
|
||
try:
|
||
import requests
|
||
import json
|
||
import numpy as np
|
||
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 start_instance_process(token: str, send_data):
|
||
|
||
"""发送宜搭表单 -- 发起流程表单
|
||
|
||
Args:
|
||
token
|
||
data:需要发送的数据字典
|
||
"""
|
||
|
||
yida_api = "https://api.dingtalk.com/v1.0/yida/processes/instances/start"
|
||
|
||
headers = {
|
||
"Content-Type": "application/json",
|
||
"x-acs-dingtalk-access-token": token
|
||
}
|
||
|
||
payload = {
|
||
"appType": "APP_TNVBVZ3K8G56HG03Z45Q",
|
||
"systemToken": "CH7669818R0WN18TYTYJ42PE6GY22WZN0BYWKD1",
|
||
"userId": "yida_pub_account",# 超级管理员账号
|
||
"language": "zh_CN",
|
||
"formUuid": "FORM-96D58EF2219240C7B1F55F9CA463CD2D4MGC",
|
||
"formDataJson": json.dumps(send_data),
|
||
"processCode":"TPROC--5Q966D918T1I1AZM68NASC6TS13P3QOL3PZRLC"
|
||
}
|
||
|
||
res = requests.post(yida_api, headers=headers, json=payload)
|
||
return res
|
||
|
||
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()
|
||
import datetime
|
||
endtime = datetime.datetime.now()
|
||
implement = (endtime - starttime).seconds
|
||
send_data = {
|
||
"textField_ls01al4o": implement, #运行耗时
|
||
"textField_lrzoowl8": "yidaxuyuezidong_tongyi", # 程序名称
|
||
"textField_lrzoowl9": "每天早上8点 将续约服务表判断过期日进行自动同意", # 功能简述
|
||
"textField_lrzoowld": textField_lrzoowld, # 运行状态
|
||
"textField_lrzoowlb": textField_lrzoowlb # 信息说明
|
||
}
|
||
res_yujing = start_instance_process(TOKEN,send_data)
|
||
except:
|
||
pass |