Files
F6--/其它系统脚本/数据迁移工具数据崩溃通知机器人.ipynb
2026-01-30 11:28:35 +08:00

342 lines
11 KiB
Plaintext

{
"cells": [
{
"cell_type": "code",
"execution_count": 1,
"metadata": {},
"outputs": [
{
"name": "stderr",
"output_type": "stream",
"text": [
"c:\\Users\\admin\\.conda\\envs\\F6processing\\lib\\site-packages\\numpy\\_distributor_init.py:30: UserWarning: loaded more than 1 DLL from .libs:\n",
"c:\\Users\\admin\\.conda\\envs\\F6processing\\lib\\site-packages\\numpy\\.libs\\libopenblas.EL2C6PLE4ZYW3ECEVIV3OXXGRN2NRFM2.gfortran-win_amd64.dll\n",
"c:\\Users\\admin\\.conda\\envs\\F6processing\\lib\\site-packages\\numpy\\.libs\\libopenblas64__v0.3.21-gcc_10_3_0.dll\n",
" warnings.warn(\"loaded more than 1 DLL from .libs:\"\n"
]
},
{
"name": "stdout",
"output_type": "stream",
"text": [
"读取到失败提醒表单中 0 条数据!\n"
]
}
],
"source": [
"'''\n",
"/v2/department/listsub # 获取部门id\n",
"/v2/user/list # 获取部门用户详情\n",
"'''\n",
"# 基础函数配置\n",
"import pandas as pd\n",
"import requests\n",
"import json\n",
"import time\n",
"import numpy as np\n",
"from pathlib import Path\n",
"import re\n",
"from datetime import datetime, timedelta\n",
"\n",
"ROOT = Path('.').absolute() # 当前工作目录\n",
"\n",
"def generateToken() -> str:\n",
" \"\"\" 生成 token \"\"\"\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",
"def generateToken_xcx() -> str:\n",
" \"\"\" 生成 token \"\"\"\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",
"def message_notification(phone,txt):\n",
" \"\"\" 机器人发送通知 \"\"\"\n",
" api = 'https://oapi.dingtalk.com/robot/send?access_token=3c9641680b7207cd812d53b68142a2fa4d1265404ec687a061f11de2e0ebfb92'\n",
"\n",
" data = {\n",
" \"at\": {\n",
" \"atMobiles\":[phone],\n",
" },\n",
" \"text\": {\n",
" \"content\":txt\n",
" },\n",
" \"msgtype\":\"text\"\n",
"}\n",
" res = requests.post(api, json=data)\n",
"\n",
" return res.json()\n",
"\n",
"def read_instances(token, formUuid, page, n):\n",
" \"\"\" 函数功能:读取普通表单的所有数据 \"\"\"\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",
" \"appType\" : \"APP_TNVBVZ3K8G56HG03Z45Q\",\n",
" \"systemToken\" : \"CH7669818R0WN18TYTYJ42PE6GY22WZN0BYWKD1\",\n",
" \"userId\" : \"yida_pub_account\",\n",
" \"language\" : \"zh_CN\",\n",
" \"formUuid\" : formUuid,\n",
" \"currentPage\" : page,\n",
" \"pageSize\" : n\n",
" }\n",
"\n",
" res = requests.post(api, headers=headers, json=formData)\n",
" return res.json()\n",
"\n",
"def read_instances_time(token, formUuid, page, n,createFromTimeGMT,createToTimeGMT):\n",
" \"\"\" 函数功能:读取普通表单的所有数据 \"\"\"\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",
" \"appType\" : \"APP_TNVBVZ3K8G56HG03Z45Q\",\n",
" \"systemToken\" : \"CH7669818R0WN18TYTYJ42PE6GY22WZN0BYWKD1\",\n",
" \"userId\" : \"yida_pub_account\",\n",
" \"language\" : \"zh_CN\",\n",
" \"formUuid\" : formUuid,\n",
" \"createFromTimeGMT\" : createFromTimeGMT,\n",
" \"createToTimeGMT\" : createToTimeGMT,\n",
" \"currentPage\" : page,\n",
" \"pageSize\" : n\n",
" }\n",
"\n",
" res = requests.post(api, headers=headers, json=formData)\n",
" return res.json()\n",
"\n",
"class NpEncoder(json.JSONEncoder):\n",
" def default(self, obj):\n",
" if isinstance(obj, np.integer):\n",
" return int(obj)\n",
" elif isinstance(obj, np.floating):\n",
" return float(obj)\n",
" elif isinstance(obj, np.ndarray):\n",
" return obj.tolist()\n",
" else:\n",
" return super(NpEncoder, self).default(obj)\n",
" \n",
"TOKEN = generateToken() \n",
"# 获取当前时间\n",
"current_time = datetime.now()\n",
"formatted_date = current_time.strftime(\"%Y-%m-%d %H:%M:%S\")\n",
"# 计算5分钟前的时间\n",
"five_minutes_ago = (current_time - timedelta(minutes=722)).strftime(\"%Y-%m-%d %H:%M:%S\")\n",
"# 计算30分钟前的时间\n",
"thirty_minutes_ago = (current_time - timedelta(minutes=30)).strftime(\"%Y-%m-%d %H:%M:%S\")\n",
"\n",
"''' 读取失败提醒中的数据 设置对应的错误的提醒信息''' \n",
"FORMID = \"FORM-UP96637151Q3XIFU9NNW9BKFUVHU2XWN0EF9L84\" # 失败提醒\n",
"# 读取流程表单数据\n",
"form_data = read_instances_time(token=TOKEN, formUuid=FORMID, page=1, n=100,createFromTimeGMT=five_minutes_ago,createToTimeGMT=formatted_date)\n",
"PAGES = form_data.get('totalCount')//100 + 1\n",
"ALL_DATA = []\n",
"\"\"\" 获取全量数据 \"\"\"\n",
"for i in range(1, PAGES+1):\n",
" form_data = read_instances_time(token=TOKEN, formUuid=FORMID, page=1, n=100,createFromTimeGMT=five_minutes_ago,createToTimeGMT=formatted_date)\n",
" for data in form_data.get('data'):\n",
" ALL_DATA.append(data)\n",
"print(f'读取到失败提醒表单中 {len(ALL_DATA)} 条数据!')\n",
"if len(ALL_DATA) != 0:\n",
" ''' 读取群机器人与人员对应关系表中的数据 获取人员对应的群机器人''' \n",
" FORMID = \"FORM-JK866XA1UGWA0JS5DF7IM4GA5RIG2778UDVHLF\" # 群机器人与人员对应关系表\n",
" # 读取表单数据\n",
" form_data = read_instances(token=TOKEN, formUuid=FORMID, page=1, n=100)\n",
" PAGES = form_data.get('totalCount')//100 + 1\n",
" ALL_qun_renyuan = {}\n",
" \"\"\" 获取全量数据 \"\"\"\n",
" for i in range(1, PAGES+1):\n",
" form_data = read_instances(token=TOKEN, formUuid=FORMID, page=i, n=100)\n",
" for data in form_data.get('data'):\n",
" ALL_qun_renyuan[data['formData']['textField_nkl14xv']]=data['formData']['textField_9btjl3q']\n",
" print(f'读取到群机器人与人员对应关系表中 {len(ALL_qun_renyuan)} 条数据!')\n",
"\n",
" TOKEN_xcx = generateToken_xcx() \n",
"\n",
" # message_notification(15888265981,\"数据支持:测试接口化调用\")"
]
},
{
"cell_type": "code",
"execution_count": 2,
"metadata": {},
"outputs": [],
"source": [
"for data in ALL_DATA:\n",
" try:\n",
" str1 = json.loads(data['instanceValue'])[0]['fieldData']['value']\n",
" re1 = r'姓名:(.*?),\\n'\n",
" reResult = re.findall(re1, str1)[0]\n",
" print(reResult)\n",
" except:\n",
" pass"
]
},
{
"cell_type": "code",
"execution_count": 48,
"metadata": {},
"outputs": [
{
"data": {
"text/plain": [
"'https://oapi.dingtalk.com/robot/send?access_token=e71835a6c79ba5cf60764b068c729f0228aee297e6772ef6edc9be51d049b64b'"
]
},
"execution_count": 48,
"metadata": {},
"output_type": "execute_result"
}
],
"source": [
"ALL_qun_renyuan[reResult]"
]
},
{
"cell_type": "code",
"execution_count": 49,
"metadata": {},
"outputs": [],
"source": [
"api = \"https://oapi.dingtalk.com/topapi/v2/department/listsub?access_token=5e0189fed9d5317bb0f1cec3b3b57dae\"\n",
"\n",
"formData = {\n",
"\t\"dept_id\":574140511\n",
"}\n",
"res = requests.post(api, json=formData).json()\n",
"\n",
"ALL_quyumingxi = {}\n",
"for data in res['result']:\n",
" ALL_quyumingxi[data['name']]=data['dept_id']"
]
},
{
"cell_type": "code",
"execution_count": 51,
"metadata": {},
"outputs": [],
"source": [
"ALL_renyaunmingxi = {}\n",
"for key in ALL_quyumingxi:\n",
" \n",
" api = \"https://oapi.dingtalk.com/topapi/v2/user/list?access_token=5e0189fed9d5317bb0f1cec3b3b57dae\"\n",
" formData = {\n",
"\t\"dept_id\":ALL_quyumingxi[key],\n",
"\t\"cursor\":0,\n",
"\t\"size\":100\n",
" }\n",
" res = requests.post(api, json=formData).json()\n",
" for i in range(0,len(res['result']['list'])):\n",
" ALL_renyaunmingxi[res['result']['list'][i]['name']]=res['result']['list'][i]['mobile']"
]
},
{
"cell_type": "code",
"execution_count": 53,
"metadata": {},
"outputs": [
{
"data": {
"text/plain": [
"'13565971287'"
]
},
"execution_count": 53,
"metadata": {},
"output_type": "execute_result"
}
],
"source": [
"ALL_renyaunmingxi[reResult]"
]
},
{
"attachments": {},
"cell_type": "markdown",
"metadata": {},
"source": [
"# 测试问答机器人\n"
]
},
{
"cell_type": "code",
"execution_count": 4,
"metadata": {},
"outputs": [],
"source": [
"# 导入模块\n",
"import requests\n",
"from urllib import parse\n",
"\n",
"\n",
"# 组装请求\n",
"def test_get(msg):\n",
" url = 'http://api.qingyunke.com/api.php?key=free&appid=0&msg={}'.format(parse.quote(msg))\n",
" html = requests.get(url)\n",
" return html.json()[\"content\"]\n",
"\n",
"\n",
"# 解析返回并打印\n",
"while True:\n",
" msg = input(\"我:\")\n",
" res = test_get(msg)\n",
" print(\"答:\", res)"
]
}
],
"metadata": {
"kernelspec": {
"display_name": "F6processing",
"language": "python",
"name": "python3"
},
"language_info": {
"codemirror_mode": {
"name": "ipython",
"version": 3
},
"file_extension": ".py",
"mimetype": "text/x-python",
"name": "python",
"nbconvert_exporter": "python",
"pygments_lexer": "ipython3",
"version": "3.9.4"
},
"orig_nbformat": 4
},
"nbformat": 4,
"nbformat_minor": 2
}