XW-AIOT/iot-device-management-service
4
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

功能：告警级别体系统一为 0紧急/1重要/2普通/3轻微

Browse Source 
更新 API 接口、Schema 验证、报告生成、企微通知的级别映射，
与前端和边缘端保持一致。

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
16337
2026-03-18 16:39:16 +08:00
parent 44c5df7302
commit dd86da5bcd
4 changed files with 8 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
app/routers/yudao_aiot_alarm.py
View File

@@ -50,7 +50,7 @@ async def _alarm_to_camel(alarm_dict: dict, camera_info_map: dict = None, camera
# alarm_level int → 文本映射 # alarm_level int → 文本映射
alarm_level = alarm_dict.get("alarm_level") alarm_level = alarm_dict.get("alarm_level")
level_map = {1: "low", 2: "medium", 3: "high", 4: "critical"} level_map = {0: "critical", 1: "high", 2: "medium", 3: "low"}
level_str = level_map.get(alarm_level, "medium") if alarm_level else "medium" level_str = level_map.get(alarm_level, "medium") if alarm_level else "medium"
# alarm_status → 前端 status 映射 # alarm_status → 前端 status 映射
@@ -148,7 +148,7 @@ async def get_alert_page(
edgeNodeId: Optional[str] = Query(None, description="边缘节点ID"), edgeNodeId: Optional[str] = Query(None, description="边缘节点ID"),
alarmType: Optional[str] = Query(None, alias="alertType", description="告警类型"), alarmType: Optional[str] = Query(None, alias="alertType", description="告警类型"),
alarmStatus: Optional[str] = Query(None, alias="status", description="告警状态"), alarmStatus: Optional[str] = Query(None, alias="status", description="告警状态"),
alarmLevel: Optional[int] = Query(None, description="告警级别: 1提醒/2一般/3严重/4紧急"), alarmLevel: Optional[int] = Query(None, description="告警级别: 0紧急/1重要/2普通/3轻微"),
startTime: Optional[datetime] = Query(None, description="开始时间"), startTime: Optional[datetime] = Query(None, description="开始时间"),
endTime: Optional[datetime] = Query(None, description="结束时间"), endTime: Optional[datetime] = Query(None, description="结束时间"),
service: AlarmEventService = Depends(get_alarm_event_service), service: AlarmEventService = Depends(get_alarm_event_service),

2
app/schemas.py
View File

@@ -120,7 +120,7 @@ class EdgeAlarmReport(BaseModel):
device_id: str = Field(..., max_length=64, description="摄像头/设备ID") device_id: str = Field(..., max_length=64, description="摄像头/设备ID")
scene_id: Optional[str] = Field(None, max_length=64, description="场景/ROI ID") scene_id: Optional[str] = Field(None, max_length=64, description="场景/ROI ID")
event_time: str = Field(..., description="事件发生时间 ISO8601") event_time: str = Field(..., description="事件发生时间 ISO8601")
alarm_level: int = Field(2, ge=1, le=4, description="告警级别: 1提醒 2一般 3严重 4紧急") alarm_level: int = Field(2, ge=0, le=3, description="告警级别: 0紧急 1重要 2普通 3轻微")
snapshot_url: Optional[str] = Field(None, max_length=512, description="截图 COS object_key") snapshot_url: Optional[str] = Field(None, max_length=512, description="截图 COS object_key")
algorithm_code: Optional[str] = Field(None, max_length=64, description="算法编码") algorithm_code: Optional[str] = Field(None, max_length=64, description="算法编码")
confidence_score: Optional[float] = Field(None, ge=0, le=1, description="置信度 0-1") confidence_score: Optional[float] = Field(None, ge=0, le=1, description="置信度 0-1")

2
app/services/report_generator.py
View File

@@ -17,7 +17,7 @@ from app.utils.timezone import beijing_now
TYPE_NAMES = {"leave_post": "人员离岗", "intrusion": "周界入侵"} TYPE_NAMES = {"leave_post": "人员离岗", "intrusion": "周界入侵"}
LEVEL_NAMES = {1: "提醒", 2: "一般", 3: "严重", 4: "紧急"} LEVEL_NAMES = {0: "紧急", 1: "重要", 2: "普通", 3: "轻微"}
STATUS_NAMES = { STATUS_NAMES = {
"NEW": "待处理", "CONFIRMED": "已确认", "NEW": "待处理", "CONFIRMED": "已确认",
"FALSE": "误报", "CLOSED": "已关闭", "FALSE": "误报", "CLOSED": "已关闭",

8
app/services/wechat_service.py
View File

@@ -23,7 +23,7 @@ ALARM_TYPE_NAMES = {
} }
# 告警级别映射 # 告警级别映射
ALARM_LEVEL_NAMES = {1: "提醒", 2: "一般", 3: "严重", 4: "紧急"} ALARM_LEVEL_NAMES = {0: "紧急", 1: "重要", 2: "普通", 3: "轻微"}
class WeChatService: class WeChatService:
@@ -210,7 +210,7 @@ class WeChatService:
access_token = await self._get_access_token() access_token = await self._get_access_token()
type_name = ALARM_TYPE_NAMES.get(alarm_type, alarm_type) type_name = ALARM_TYPE_NAMES.get(alarm_type, alarm_type)
level_name = ALARM_LEVEL_NAMES.get(alarm_level, "一般") level_name = ALARM_LEVEL_NAMES.get(alarm_level, "普通")
msg = { msg = {
"touser": "|".join(user_ids), "touser": "|".join(user_ids),
@@ -221,7 +221,7 @@ class WeChatService:
"task_id": alarm_id, "task_id": alarm_id,
"source": { "source": {
"desc": "AI安防告警", "desc": "AI安防告警",
"desc_color": 3 if alarm_level >= 3 else 0, "desc_color": 3 if alarm_level <= 1 else 0,
}, },
"main_title": { "main_title": {
"title": f"【{level_name}】{type_name}告警", "title": f"【{level_name}】{type_name}告警",
@@ -595,7 +595,7 @@ class WeChatService:
return False return False
type_name = ALARM_TYPE_NAMES.get(alarm_type, alarm_type) type_name = ALARM_TYPE_NAMES.get(alarm_type, alarm_type)
level_name = ALARM_LEVEL_NAMES.get(alarm_level, "一般") level_name = ALARM_LEVEL_NAMES.get(alarm_level, "普通")
success = True success = True
# ---- 1. 发送告警截图（image 消息） ---- # ---- 1. 发送告警截图（image 消息） ----