102 lines
3.4 KiB
Python
Executable File
102 lines
3.4 KiB
Python
Executable File
# utils\lzmx.py
|
|
import requests
|
|
import json
|
|
from datetime import datetime
|
|
|
|
# 推送系统工单处理信息到量子密信
|
|
def send_lzmx_message(webhook_url, handler, device_id, device_type, distance_from_branch, restore_time):
|
|
headers = {'Content-Type': 'application/json'}
|
|
|
|
message = {
|
|
"type": "text",
|
|
"textMsg": {
|
|
"content":
|
|
f"📢 故障处理通知\n"
|
|
f"故障设备：{device_id}\n"
|
|
f"处理时间：{restore_time.strftime('%Y-%m-%d %H:%M:%S')}\n"
|
|
f"故障类型：{device_type}\n"
|
|
f"距离支局：{distance_from_branch}公里\n"
|
|
f"处理人：{handler}"
|
|
}
|
|
}
|
|
|
|
try:
|
|
response = requests.post(
|
|
webhook_url,
|
|
headers=headers,
|
|
data=json.dumps(message)
|
|
)
|
|
response.raise_for_status()
|
|
return True
|
|
except Exception as e:
|
|
print(f"发送量子密信消息失败: {str(e)}")
|
|
return False
|
|
|
|
# 推送手工申报工单信息到量子密信
|
|
def send_manual_lzmx_message(webhook_url, handler, device_id, fault_type, distance_from_branch, fault_time, project_type, device_location):
|
|
headers = {'Content-Type': 'application/json'}
|
|
|
|
message = {
|
|
"type": "text",
|
|
"textMsg": {
|
|
"content":
|
|
f"📝 故障处理通知\n"
|
|
f"项目类型：{project_type}\n"
|
|
f"故障设备：{device_id}\n"
|
|
f"设备位置：{device_location}\n"
|
|
f"故障时间：{fault_time.strftime('%Y-%m-%d %H:%M:%S')}\n"
|
|
f"故障类型：{fault_type}\n"
|
|
f"距离支局：{distance_from_branch}公里\n"
|
|
f"处理人：{handler}"
|
|
}
|
|
}
|
|
|
|
try:
|
|
response = requests.post(
|
|
webhook_url,
|
|
headers=headers,
|
|
data=json.dumps(message)
|
|
)
|
|
response.raise_for_status()
|
|
return True
|
|
except Exception as e:
|
|
print(f"发送量子密信消息失败: {str(e)}")
|
|
return False
|
|
|
|
# 每周五统计本周情况推送量子密信
|
|
def send_weekly_stats_message(webhook_url, stats):
|
|
headers = {'Content-Type': 'application/json'}
|
|
|
|
# 格式化前三名处理人信息
|
|
top_handlers = []
|
|
for handler, amount in stats['top_handlers']:
|
|
top_handlers.append(f"👤 {handler}: {amount}元")
|
|
|
|
message = {
|
|
"type": "text",
|
|
"textMsg": {
|
|
"content":
|
|
f"📊 本周工单统计报告\n"
|
|
f"—————————————————\n"
|
|
f" 本周派单：{stats['total_orders']}单\n"
|
|
f" 已处理数：{stats['completed_orders']}单\n"
|
|
f" 未处理数：{stats['pending_orders']}单\n"
|
|
f" 手工故障：{stats['manual_orders']}个\n\n"
|
|
f"🏆 前三名预计奖励：\n"
|
|
f"{chr(10).join(top_handlers)}\n"
|
|
f"—————————————————\n"
|
|
f"统计时间：{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"
|
|
}
|
|
}
|
|
|
|
try:
|
|
response = requests.post(
|
|
webhook_url,
|
|
headers=headers,
|
|
data=json.dumps(message)
|
|
)
|
|
response.raise_for_status()
|
|
return True
|
|
except Exception as e:
|
|
print(f"发送量子密信消息失败: {str(e)}")
|
|
return False |