16337
5b00fd0464
- 新增 H5 告警详情页面（截图+信息+3个操作按钮） - 企微卡片"查看详情"跳转到 H5 页面 - 操作按钮改为：前往处理/已处理/误报忽略 - 新增 alarm_detail API 供 H5 页面获取告警+VLM分析数据 - 挂载 /static 目录提供 H5 页面访问
120 lines
3.7 KiB
Python
120 lines
3.7 KiB
Python
"""
|
|
企微回调路由
|
|
|
|
处理安保人员在企微卡片上的操作（前往处理/已处理/误报忽略）。
|
|
提供告警详情接口供 H5 页面使用。
|
|
"""
|
|
|
|
from datetime import datetime
|
|
from fastapi import APIRouter, Depends, Query
|
|
from pydantic import BaseModel
|
|
from typing import Optional
|
|
|
|
from app.yudao_compat import YudaoResponse
|
|
from app.models import get_session, AlarmEvent, AlarmLlmAnalysis
|
|
from app.services.alarm_event_service import get_alarm_event_service, AlarmEventService
|
|
from app.utils.logger import logger
|
|
|
|
router = APIRouter(prefix="/api/wechat", tags=["企微回调"])
|
|
|
|
|
|
class AlarmActionRequest(BaseModel):
|
|
"""企微卡片操作请求"""
|
|
alarm_id: str
|
|
action: str # confirm / complete / ignore
|
|
operator_uid: str # 企微 userid
|
|
remark: Optional[str] = None
|
|
|
|
|
|
@router.get("/alarm_detail")
|
|
async def get_alarm_detail(alarm_id: str = Query(..., description="告警ID")):
|
|
"""
|
|
告警详情接口（供 H5 页面使用，无认证）
|
|
|
|
返回告警基本信息 + VLM 分析描述
|
|
"""
|
|
db = get_session()
|
|
try:
|
|
alarm = db.query(AlarmEvent).filter(AlarmEvent.alarm_id == alarm_id).first()
|
|
if not alarm:
|
|
return YudaoResponse.error(404, "告警不存在")
|
|
|
|
# 查 VLM 分析结果
|
|
vlm_desc = ""
|
|
analysis = db.query(AlarmLlmAnalysis).filter(
|
|
AlarmLlmAnalysis.alarm_id == alarm_id
|
|
).order_by(AlarmLlmAnalysis.id.desc()).first()
|
|
if analysis:
|
|
vlm_desc = analysis.summary or ""
|
|
|
|
return YudaoResponse.success({
|
|
"alarm_id": alarm.alarm_id,
|
|
"alarm_type": alarm.alarm_type,
|
|
"device_id": alarm.device_id,
|
|
"scene_id": alarm.scene_id,
|
|
"event_time": alarm.event_time.strftime('%Y-%m-%d %H:%M:%S') if alarm.event_time else "",
|
|
"alarm_level": alarm.alarm_level,
|
|
"alarm_status": alarm.alarm_status,
|
|
"handle_status": alarm.handle_status,
|
|
"snapshot_url": alarm.snapshot_url or "",
|
|
"handler": alarm.handler or "",
|
|
"handle_remark": alarm.handle_remark or "",
|
|
"vlm_description": vlm_desc,
|
|
})
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.post("/callback/alarm_action")
|
|
async def alarm_action_callback(
|
|
req: AlarmActionRequest,
|
|
service: AlarmEventService = Depends(get_alarm_event_service),
|
|
):
|
|
"""
|
|
企微告警操作回调（无认证，由 H5 页面调用）
|
|
|
|
action:
|
|
- confirm: 前往处理 → handle_status=HANDLING
|
|
- complete: 已处理完成 → handle_status=DONE, alarm_status=CLOSED
|
|
- ignore: 误报忽略 → alarm_status=FALSE, handle_status=DONE
|
|
"""
|
|
action_map = {
|
|
"confirm": {
|
|
"alarm_status": "CONFIRMED",
|
|
"handle_status": "HANDLING",
|
|
"remark": "前往处理",
|
|
},
|
|
"complete": {
|
|
"alarm_status": "CLOSED",
|
|
"handle_status": "DONE",
|
|
"remark": "手动结单",
|
|
},
|
|
"ignore": {
|
|
"alarm_status": "FALSE",
|
|
"handle_status": "DONE",
|
|
"remark": "标记误报",
|
|
},
|
|
}
|
|
|
|
action_cfg = action_map.get(req.action)
|
|
if not action_cfg:
|
|
return YudaoResponse.error(400, f"无效操作: {req.action}")
|
|
|
|
result = service.handle_alarm(
|
|
alarm_id=req.alarm_id,
|
|
alarm_status=action_cfg["alarm_status"],
|
|
handle_status=action_cfg["handle_status"],
|
|
handler=req.operator_uid,
|
|
remark=req.remark or action_cfg["remark"],
|
|
)
|
|
|
|
if not result:
|
|
return YudaoResponse.error(404, "告警不存在")
|
|
|
|
logger.info(
|
|
f"企微操作: alarm={req.alarm_id}, action={req.action}, "
|
|
f"operator={req.operator_uid}"
|
|
)
|
|
|
|
return YudaoResponse.success(True)
|