274 lines
11 KiB
Plaintext
274 lines
11 KiB
Plaintext
{
|
|
"cells": [
|
|
{
|
|
"cell_type": "code",
|
|
"execution_count": null,
|
|
"metadata": {},
|
|
"outputs": [],
|
|
"source": [
|
|
"import pandas as pd\n",
|
|
"import time\n",
|
|
"import json\n",
|
|
"from datetime import datetime\n",
|
|
"from dateutil.relativedelta import relativedelta\n",
|
|
"import calendar\n",
|
|
"from datetime import timedelta\n",
|
|
"import requests\n",
|
|
"import datetime\n",
|
|
"\n",
|
|
"\n",
|
|
"def generateToken() -> str:\n",
|
|
" \"\"\" \n",
|
|
" 生成 token，参数不需要修改\n",
|
|
" \"\"\"\n",
|
|
"\n",
|
|
" token_api = 'https://api.dingtalk.com/v1.0/oauth2/accessToken'\n",
|
|
"\n",
|
|
" # 该信息在钉钉开放应用中\n",
|
|
" data = {\n",
|
|
" \"appKey\": \"ding5kqocon5s9oph5uq\",\n",
|
|
" \"appSecret\": 'HL1jgsIIfLAC0eTH0A1m4mwxUDqbgsiPeCCGGE3ocM6qJBTIW7Ivt9drxF_Z4Kb_'\n",
|
|
" }\n",
|
|
"\n",
|
|
" res = requests.post(token_api, json=data)\n",
|
|
" token = res.json()['accessToken']\n",
|
|
"\n",
|
|
" return token\n",
|
|
"\n",
|
|
"\n",
|
|
"def get_time_range(n=2):\n",
|
|
" \"\"\" 获取近n个月的时间戳(单位是毫秒) 默认取最近2个月\"\"\"\n",
|
|
"\n",
|
|
" def delay_time(time_str, years=0, months=0, days=0, hours=0, minutes=0, seconds=0):\n",
|
|
" if type(time_str) == str:\n",
|
|
" time_str = datetime.strptime(time_str, '%Y-%m-%d %H:%M:%S')\n",
|
|
" ret = time_str + relativedelta(years=years, months=months, days=days, hours=hours, minutes=minutes, seconds=seconds)\n",
|
|
" return ret\n",
|
|
"\n",
|
|
" # 获得当前时间\n",
|
|
" now_time = datetime.now()\n",
|
|
" endTime = int(time.mktime(time.strptime(now_time.date().strftime('%Y/%m/%d')+' 00:00:00', '%Y/%m/%d %H:%M:%S'))) * 1000\n",
|
|
" # 2个月前 的时间\n",
|
|
" ret2 = delay_time(now_time, months=-n)\n",
|
|
" startTime = int(time.mktime(time.strptime(ret2.date().strftime('%Y/%m/%d')+' 00:00:00', '%Y/%m/%d %H:%M:%S'))) * 1000\n",
|
|
"\n",
|
|
" # print(f'时间区间:[{startTime}-{endTime}]')\n",
|
|
" return startTime, endTime\n",
|
|
"\n",
|
|
"\n",
|
|
"def get_dateRange_this_month():\n",
|
|
" \"\"\" 函数功能：获取当前月份的起始日期范围时间戳(单位毫秒)\"\"\"\n",
|
|
" now = datetime.now()\n",
|
|
" this_month_start = datetime(now.year, now.month, 1)\n",
|
|
" this_month_end = datetime(now.year, now.month, calendar.monthrange(now.year, now.month)[1])\n",
|
|
" this_month_start, this_month_end = int(time.mktime(this_month_start.timetuple())), int(time.mktime(this_month_end.timetuple()))\n",
|
|
"\n",
|
|
" return this_month_start*1000, this_month_end*1000\n",
|
|
"\n",
|
|
"\n",
|
|
"def read_processes_instances(token, formUuid, page=1, n=100, searchField={}):\n",
|
|
" \"\"\" 函数功能：读取流程表单的所有数据 \n",
|
|
" 获取表单的数据实例数据,返回表单的所有实例 -- 应用：F6客户服务\n",
|
|
" \"\"\"\n",
|
|
"\n",
|
|
" api = f'https://api.dingtalk.com/v1.0/yida/processes/instances?pageNumber={page}&pageSize={n}'\n",
|
|
"\n",
|
|
" headers = {\n",
|
|
" \"Content-Type\": \"application/json\",\n",
|
|
" \"x-acs-dingtalk-access-token\": token\n",
|
|
" }\n",
|
|
"\n",
|
|
" formData = {\n",
|
|
" 'currentPage': page,\n",
|
|
" 'pageSize': n,\n",
|
|
" \"appType\": \"APP_UYZ0KG6L0CCNV80GZ66O\",\n",
|
|
" \"systemToken\": \"XA966F81JAJOFCVVVKO64E9MIIZV1EWE5SFMKJ2\",\n",
|
|
" \"userId\": \"yida_pub_account\", # 超级管理员账号\n",
|
|
" \"language\": \"zh_CN\",\n",
|
|
" \"formUuid\": formUuid,\n",
|
|
" \"searchFieldJson\": json.dumps(searchField)\n",
|
|
" }\n",
|
|
"\n",
|
|
" res = requests.post(api, headers=headers, json=formData)\n",
|
|
" return res.json()\n",
|
|
"\n",
|
|
"\n",
|
|
"def read_form_instances(token, formUuid, page=1, n=100, searchField={},createFromTimeGMT='',createToTimeGMT=''):\n",
|
|
" \"\"\" 函数功能：读取普通表单的所有数据 -- 应用：F6客户服务 \"\"\"\n",
|
|
"\n",
|
|
" api = f'https://api.dingtalk.com/v1.0/yida/forms/instances/search'\n",
|
|
"\n",
|
|
" headers = {\n",
|
|
" \"Content-Type\": \"application/json\",\n",
|
|
" \"x-acs-dingtalk-access-token\": token\n",
|
|
" }\n",
|
|
"\n",
|
|
" formData = {\n",
|
|
" 'currentPage': page,\n",
|
|
" 'pageSize': n,\n",
|
|
" \"appType\": \"APP_UYZ0KG6L0CCNV80GZ66O\",\n",
|
|
" \"systemToken\": \"XA966F81JAJOFCVVVKO64E9MIIZV1EWE5SFMKJ2\",\n",
|
|
" \"userId\": \"yida_pub_account\",\n",
|
|
" \"language\": \"zh_CN\",\n",
|
|
" \"formUuid\": formUuid,\n",
|
|
" 'createFromTimeGMT':createFromTimeGMT,\n",
|
|
" 'createToTimeGMT':createToTimeGMT,\n",
|
|
" \"searchFieldJson\": json.dumps(searchField)\n",
|
|
" }\n",
|
|
"\n",
|
|
" res = requests.post(api, headers=headers, json=formData)\n",
|
|
" return res.json()\n",
|
|
"\n",
|
|
"\n",
|
|
"def create_form_instances(token, formUuid, formData={}):\n",
|
|
" \"\"\" 函数功能：创建普通表单实例 -- 应用：F6客户服务 \"\"\"\n",
|
|
"\n",
|
|
" api = 'https://api.dingtalk.com/v1.0/yida/forms/instances'\n",
|
|
"\n",
|
|
" headers = {\n",
|
|
" \"Content-Type\": \"application/json\",\n",
|
|
" \"x-acs-dingtalk-access-token\": token\n",
|
|
" }\n",
|
|
"\n",
|
|
" payload = {\n",
|
|
" \"appType\": \"APP_UYZ0KG6L0CCNV80GZ66O\",\n",
|
|
" \"systemToken\": \"XA966F81JAJOFCVVVKO64E9MIIZV1EWE5SFMKJ2\",\n",
|
|
" \"userId\": \"yida_pub_account\",\n",
|
|
" \"language\": \"zh_CN\",\n",
|
|
" \"formUuid\": formUuid,\n",
|
|
" \"formDataJson\": json.dumps(formData)\n",
|
|
" }\n",
|
|
"\n",
|
|
" res = requests.post(api, headers=headers, json=payload)\n",
|
|
" return res\n",
|
|
"\n",
|
|
"\n",
|
|
"def update_form_instances(token, formInstId, formData={}):\n",
|
|
" \"\"\" 函数功能：更新普通表单实例 -- 应用：F6客户服务 \"\"\"\n",
|
|
"\n",
|
|
" api = 'https://api.dingtalk.com/v1.0/yida/forms/instances'\n",
|
|
"\n",
|
|
" headers = {\n",
|
|
" \"Content-Type\": \"application/json\",\n",
|
|
" \"x-acs-dingtalk-access-token\": token\n",
|
|
" }\n",
|
|
"\n",
|
|
" payload = {\n",
|
|
" \"appType\": \"APP_UYZ0KG6L0CCNV80GZ66O\",\n",
|
|
" \"systemToken\": \"XA966F81JAJOFCVVVKO64E9MIIZV1EWE5SFMKJ2\",\n",
|
|
" \"userId\": \"yida_pub_account\",\n",
|
|
" \"language\": \"zh_CN\",\n",
|
|
" \"formInstanceId\": formInstId,\n",
|
|
" \"updateFormDataJson\": json.dumps(formData)\n",
|
|
" }\n",
|
|
"\n",
|
|
" res = requests.put(api, headers=headers, json=payload)\n",
|
|
" return res\n",
|
|
"def get_approval_records(token: str, processInstanceId: str):\n",
|
|
" \"\"\" 函数功能：获取流程表单的审批记录 --F6客户服务 应用 \"\"\"\n",
|
|
" appType = \"APP_UYZ0KG6L0CCNV80GZ66O\"\n",
|
|
" systemToken = \"XA966F81JAJOFCVVVKO64E9MIIZV1EWE5SFMKJ2\"\n",
|
|
" userId = \"yida_pub_account\"\n",
|
|
"\n",
|
|
" api = f'https://api.dingtalk.com/v1.0/yida/processes/operationRecords?appType={appType}&systemToken={systemToken}&userId={userId}&language=zh_CN&processInstanceId={processInstanceId}'\n",
|
|
"\n",
|
|
" headers = {\n",
|
|
" \"Content-Type\": \"application/json\",\n",
|
|
" \"x-acs-dingtalk-access-token\": token\n",
|
|
" }\n",
|
|
"\n",
|
|
" res = requests.get(api, headers=headers)\n",
|
|
" # print('获取流程表单的审批记录')\n",
|
|
" return res.json()\n",
|
|
"\n",
|
|
"def get_approval_records_Batch(token, processInstanceId):\n",
|
|
" \"\"\" 函数功能：批量获取流程表单的审批记录 --F6客户服务 应用 \"\"\"\n",
|
|
" appType = \"APP_UYZ0KG6L0CCNV80GZ66O\"\n",
|
|
" systemToken = \"XA966F81JAJOFCVVVKO64E9MIIZV1EWE5SFMKJ2\"\n",
|
|
" userId = \"yida_pub_account\"\n",
|
|
" language = \"zh_CN\"\n",
|
|
"\n",
|
|
" api = f'https://api.dingtalk.com/v1.0/yida/processes/instances/searchWithIds?appType={appType}&systemToken={systemToken}&userId={userId}&language={language}&processInstanceIds={processInstanceId}'\n",
|
|
" print(api)\n",
|
|
" headers = {\n",
|
|
" # \"Content-Type\": \"application/json\",\n",
|
|
" \"x-acs-dingtalk-access-token\": token\n",
|
|
" }\n",
|
|
"\n",
|
|
" res = requests.get(api, headers=headers)\n",
|
|
" return res.json()\n",
|
|
"\n",
|
|
"def timeStamp(timeNum):\n",
|
|
" \"\"\" 函数功能：将时间戳(毫秒) 转化为时间日期格式\"\"\"\n",
|
|
" timeStamp = float(timeNum/1000)\n",
|
|
" timeArray = time.localtime(timeStamp)\n",
|
|
" otherStyleTime = time.strftime(\"%Y-%m-%d %H:%M:%S\", timeArray)\n",
|
|
" return otherStyleTime\n",
|
|
"\n",
|
|
"# 指定日期和时间\n",
|
|
"date_str = '2023/08/17 00:00:00'\n",
|
|
"\n",
|
|
"# 将字符串转换为datetime对象\n",
|
|
"date_obj = datetime.datetime.strptime(date_str, '%Y/%m/%d %H:%M:%S')\n",
|
|
"\n",
|
|
"# 获取当前时间戳（单位为秒）\n",
|
|
"current_timestamp = int(date_obj.timestamp())\n",
|
|
"\n",
|
|
"data_new = []\n",
|
|
"FormDatas = []\n",
|
|
"\n",
|
|
"TOKEN = generateToken()\n",
|
|
"for i in range(0,300):\n",
|
|
" try:\n",
|
|
" a = 86400 * i\n",
|
|
" b = 86400 * (i + 1)\n",
|
|
" # 获取-1天的时间戳（单位为秒）\n",
|
|
" yesterday_timestamp_1 = current_timestamp - a\n",
|
|
"\n",
|
|
" # 获取-2天的时间戳（单位为秒）\n",
|
|
" yesterday_timestamp_2 = current_timestamp - b\n",
|
|
"\n",
|
|
" # 将时间戳转换为 datetime 对象\n",
|
|
" dt_1 = datetime.datetime.fromtimestamp(yesterday_timestamp_1)\n",
|
|
"\n",
|
|
" # 格式化日期字符串\n",
|
|
" date_str_1 = dt_1.strftime(\"%Y-%m-%d\")\n",
|
|
"\n",
|
|
" # 将时间戳转换为 datetime 对象\n",
|
|
" dt_2 = datetime.datetime.fromtimestamp(yesterday_timestamp_2)\n",
|
|
"\n",
|
|
" # 格式化日期字符串\n",
|
|
" date_str_2 = dt_2.strftime(\"%Y-%m-%d\")\n",
|
|
"\n",
|
|
" print(\"时间戳：\", date_str_1, date_str_2)\n",
|
|
"\n",
|
|
" res = read_form_instances(token=TOKEN, formUuid='FORM-GP666M71TNE9GFK57V2O85NLM04I34CSG1TFLC',createFromTimeGMT=date_str_2,createToTimeGMT=date_str_1)\n",
|
|
" totalCount = res.get('totalCount')\n",
|
|
" PAGES = totalCount//100 + 1\n",
|
|
" for page in range(1,PAGES+1):\n",
|
|
" res = read_form_instances(token=TOKEN, formUuid='FORM-GP666M71TNE9GFK57V2O85NLM04I34CSG1TFLC',page=page,n=100,createFromTimeGMT=date_str_2,createToTimeGMT=date_str_1)\n",
|
|
" FormDatas.extend(res.get('data'))\n",
|
|
" for v in range(0,len(res.get('data'))):\n",
|
|
" abc = res['data'][v]['formInstanceId']+\",\"+res['data'][v]['formData']['textField_xgaye7b']+\",\"+res['data'][v]['formData']['textField_6glka3o']\n",
|
|
" data_new.append(abc)\n",
|
|
" except:\n",
|
|
" pass\n",
|
|
"\n",
|
|
"print(f\"表单数据共{len(data_new)}条!\") #耗时1分钟左右\n",
|
|
"# 创建DataFrame对象\n",
|
|
"df = pd.DataFrame({'ID': data_new})\n",
|
|
"# 将DataFrame对象写入Excel文件中\n",
|
|
"df.to_excel(r'C:\\Users\\admin\\Desktop\\data0816.xlsx', index=False)\n"
|
|
]
|
|
}
|
|
],
|
|
"metadata": {
|
|
"language_info": {
|
|
"name": "python"
|
|
},
|
|
"orig_nbformat": 4
|
|
},
|
|
"nbformat": 4,
|
|
"nbformat_minor": 2
|
|
}
|