176 lines
5.7 KiB
Python
176 lines
5.7 KiB
Python
"""
|
|
工单创建测试 — 使用真实告警数据 + 真实截图
|
|
|
|
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())
|