From 6ffcc7927742a165e42919aa569faf39b8dac8b3 Mon Sep 17 00:00:00 2001 From: 16337 <1633794139@qq.com> Date: Tue, 7 Apr 2026 14:05:37 +0800 Subject: [PATCH] =?UTF-8?q?=E9=87=8D=E6=9E=84(test):=20test=5Fwork=5Forder?= =?UTF-8?q?=5Freal=20=E6=94=B9=E7=94=A8=20constants=20=E7=BB=9F=E4=B8=80?= =?UTF-8?q?=E5=AE=9A=E4=B9=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- test_work_order_real.py | 175 ++++++++++++++++++++++++++++++++++++++++ 1 file changed, 175 insertions(+) create mode 100644 test_work_order_real.py diff --git a/test_work_order_real.py b/test_work_order_real.py new file mode 100644 index 0000000..f800e9d --- /dev/null +++ b/test_work_order_real.py @@ -0,0 +1,175 @@ +""" +工单创建测试 — 使用真实告警数据 + 真实截图 + +1. 从 MySQL 查询最近一条带截图的告警 +2. 用该告警数据创建工单到 192.168.0.104:48080 +""" + +import asyncio +import hashlib +import json +import time +import uuid + +import httpx +import pymysql + +from app.constants import ALARM_TYPE_NAMES, ALARM_LEVEL_NAMES + +# ===== 数据库配置 ===== +DB_CONFIG = { + "host": "124.221.55.225", + "port": 3306, + "user": "vsp_user", + "password": "VspPass2024!", + "database": "aiot_alarm", + "charset": "utf8mb4", +} + +# ===== 工单 API 配置 ===== +BASE_URL = "http://192.168.0.104:48080" +APP_ID = "alarm-system" +APP_SECRET = "tQ3v5q1z2ZLu7hrU1yseaHwg1wJUcmF1" +TENANT_ID = "1" + + +def query_recent_alarm_with_snapshot(): + """从数据库查询最近一条带截图的告警""" + conn = pymysql.connect(**DB_CONFIG) + try: + with conn.cursor(pymysql.cursors.DictCursor) as cur: + sql = """ + SELECT alarm_id, alarm_type, device_id, scene_id, + event_time, alarm_level, confidence_score, + snapshot_url, edge_node_id, area_id + FROM alarm_event + WHERE snapshot_url IS NOT NULL + AND snapshot_url != '' + ORDER BY event_time DESC + LIMIT 5 + """ + cur.execute(sql) + rows = cur.fetchall() + return rows + finally: + conn.close() + + +def build_sign(query_str: str, body_json: str, nonce: str, timestamp: str) -> str: + """构建 SHA256 签名""" + header_str = f"appId={APP_ID}&nonce={nonce}×tamp={timestamp}" + sign_string = f"{query_str}{body_json}{header_str}{APP_SECRET}" + return hashlib.sha256(sign_string.encode("utf-8")).hexdigest() + + +def build_headers(body_json: str) -> dict: + """构造完整请求头""" + nonce = uuid.uuid4().hex[:16] + timestamp = str(int(time.time() * 1000)) + sign = build_sign("", body_json, nonce, timestamp) + return { + "Content-Type": "application/json", + "tenant-id": TENANT_ID, + "appId": APP_ID, + "timestamp": timestamp, + "nonce": nonce, + "sign": sign, + } + + +async def create_order(alarm: dict) -> str: + """用真实告警数据创建工单""" + alarm_type = alarm["alarm_type"] + alarm_level = alarm.get("alarm_level", 2) + type_name = ALARM_TYPE_NAMES.get(alarm_type, alarm_type) + level_name = ALARM_LEVEL_NAMES.get(alarm_level, "一般") + device_id = alarm["device_id"] + event_time = str(alarm["event_time"])[:19] if alarm["event_time"] else "" + + body = { + "title": f"【{level_name}】{type_name}告警 - {device_id}", + "areaId": alarm.get("area_id") or 1317, + "alarmId": alarm["alarm_id"], + "alarmType": alarm_type, + "description": f"设备 {device_id} 于 {event_time} 检测到{type_name}，置信度 {alarm.get('confidence_score', 0):.2f}", + "priority": alarm_level, + "triggerSource": "自动上报", + "cameraId": device_id, + "imageUrl": alarm["snapshot_url"], + } + + body_json = json.dumps(body, ensure_ascii=False, separators=(",", ":")) + headers = build_headers(body_json) + url = f"{BASE_URL}/open-api/ops/security/order/create" + + print(f"\n{'='*60}") + print(f"创建工单 — 使用真实告警数据") + print(f"{'='*60}") + print(f" 告警ID: {alarm['alarm_id']}") + print(f" 告警类型: {type_name}") + print(f" 设备ID: {device_id}") + print(f" 事件时间: {event_time}") + print(f" 置信度: {alarm.get('confidence_score', 'N/A')}") + print(f" 截图URL: {alarm['snapshot_url'][:80]}...") + print(f" 区域ID: {body['areaId']}") + print(f" 工单标题: {body['title']}") + print() + print(f" 请求URL: {url}") + print(f" 请求Body: {body_json[:200]}...") + print() + + async with httpx.AsyncClient(timeout=15) as client: + resp = await client.post(url, content=body_json, headers=headers) + + print(f" HTTP Status: {resp.status_code}") + print(f" Response Body: {resp.text[:500]}") + + if resp.status_code == 200: + data = resp.json() + if data.get("code") == 0: + order_id = str(data.get("data", "")) + print(f"\n ✅ 工单创建成功! orderId = {order_id}") + return order_id + else: + print(f"\n ❌ API 返回错误: code={data.get('code')}, msg={data.get('msg')}") + else: + print(f"\n ❌ HTTP 请求失败: {resp.status_code}") + return "" + + +async def main(): + print(f"工单系统地址: {BASE_URL}") + print(f"数据库: {DB_CONFIG['host']}:{DB_CONFIG['port']}/{DB_CONFIG['database']}") + print() + + # 1. 查询最近带截图的告警 + print("正在查询最近带截图的告警...") + alarms = query_recent_alarm_with_snapshot() + + if not alarms: + print("❌ 未找到带截图的告警，无法测试") + return + + print(f"\n找到 {len(alarms)} 条带截图的告警：") + print(f"{'─'*80}") + for i, a in enumerate(alarms): + type_name = ALARM_TYPE_NAMES.get(a["alarm_type"], a["alarm_type"]) + print(f" [{i+1}] {a['alarm_id'][:40]}...") + print(f" 类型={type_name} 设备={a['device_id']} 时间={str(a['event_time'])[:19]}") + print(f" 截图={a['snapshot_url'][:70]}...") + print() + + # 2. 用第一条创建工单 + alarm = alarms[0] + print(f"将使用第 1 条告警创建工单...") + order_id = await create_order(alarm) + + if order_id: + print(f"\n{'='*60}") + print(f"测试完成 — 工单 {order_id} 已创建") + print(f"请在 AIoT 平台确认工单内容和截图是否正确") + print(f"{'='*60}") + + +if __name__ == "__main__": + asyncio.run(main())