""" AI 告警路由 - 芋道规范 提供与芋道主平台一致的接口规范。 API 路径规范： - /admin-api/ai-alert/alert/page - 分页查询 - /admin-api/ai-alert/alert/get - 获取详情 - /admin-api/ai-alert/alert/handle - 处理告警 - /admin-api/ai-alert/alert/delete - 删除告警 - /admin-api/ai-alert/alert/statistics - 获取统计 - /admin-api/ai-alert/camera-summary/page - 摄像头汇总 分页参数规范： - pageNo: 页码（从1开始） - pageSize: 每页大小 响应格式规范： { "code": 0, "data": { "list": [...], "total": 100 }, "msg": "success" } """ from fastapi import APIRouter, Query, Depends, HTTPException from typing import Optional from datetime import datetime from app.yudao_compat import YudaoResponse, get_current_user from app.services.alert_service import get_alert_service, AlertService from app.schemas import AlertHandleRequest router = APIRouter(prefix="/admin-api/ai-alert", tags=["AI告警"]) # ==================== 告警管理 ==================== @router.get("/alert/page") async def get_alert_page( pageNo: int = Query(1, ge=1, description="页码"), pageSize: int = Query(20, ge=1, le=100, description="每页大小"), cameraId: Optional[str] = Query(None, description="摄像头ID"), deviceId: Optional[str] = Query(None, description="设备ID"), alertType: Optional[str] = Query(None, description="告警类型"), status: Optional[str] = Query(None, description="状态: pending/confirmed/ignored/resolved/dispatched"), level: Optional[str] = Query(None, description="级别: low/medium/high/critical"), startTime: Optional[datetime] = Query(None, description="开始时间"), endTime: Optional[datetime] = Query(None, description="结束时间"), service: AlertService = Depends(get_alert_service), current_user: dict = Depends(get_current_user) ): """ 分页查询告警列表 权限标识：ai-alert:alert:query（测试阶段放行） """ alerts, total = service.get_alerts( camera_id=cameraId, device_id=deviceId, alert_type=alertType, status=status, level=level, start_time=startTime, end_time=endTime, page=pageNo, page_size=pageSize, ) # 转换为前端需要的格式 alert_list = [] for alert in alerts: alert_dict = alert.to_dict() # 字段名映射：snake_case -> camelCase alert_list.append({ "id": alert_dict.get("id"), "alertNo": alert_dict.get("alert_no"), "cameraId": alert_dict.get("camera_id"), "cameraName": alert_dict.get("camera_id"), # TODO: 从设备服务获取 "roiId": alert_dict.get("roi_id"), "bindId": alert_dict.get("bind_id"), "deviceId": alert_dict.get("device_id"), "alertType": alert_dict.get("alert_type"), "alertTypeName": _get_alert_type_name(alert_dict.get("alert_type")), "algorithm": alert_dict.get("algorithm"), "confidence": alert_dict.get("confidence"), "durationMinutes": alert_dict.get("duration_minutes"), "triggerTime": alert_dict.get("trigger_time"), "message": alert_dict.get("message"), "bbox": alert_dict.get("bbox"), "snapshotUrl": alert_dict.get("snapshot_url"), "ossUrl": alert_dict.get("oss_url"), "status": alert_dict.get("status"), "level": alert_dict.get("level"), "handleRemark": alert_dict.get("handle_remark"), "handledBy": alert_dict.get("handled_by"), "handledAt": alert_dict.get("handled_at"), "workOrderId": alert_dict.get("work_order_id"), "aiAnalysis": alert_dict.get("ai_analysis"), "createdAt": alert_dict.get("created_at"), "updatedAt": alert_dict.get("updated_at"), }) return YudaoResponse.page( list_data=alert_list, total=total, page_no=pageNo, page_size=pageSize ) @router.get("/alert/get") async def get_alert( id: int = Query(..., description="告警ID"), service: AlertService = Depends(get_alert_service), current_user: dict = Depends(get_current_user) ): """ 获取告警详情 权限标识：ai-alert:alert:query """ alert = service.get_alert(id) if not alert: raise HTTPException(status_code=404, detail="告警不存在") alert_dict = alert.to_dict() return YudaoResponse.success({ "id": alert_dict.get("id"), "alertNo": alert_dict.get("alert_no"), "cameraId": alert_dict.get("camera_id"), "cameraName": alert_dict.get("camera_id"), "roiId": alert_dict.get("roi_id"), "bindId": alert_dict.get("bind_id"), "deviceId": alert_dict.get("device_id"), "alertType": alert_dict.get("alert_type"), "alertTypeName": _get_alert_type_name(alert_dict.get("alert_type")), "algorithm": alert_dict.get("algorithm"), "confidence": alert_dict.get("confidence"), "durationMinutes": alert_dict.get("duration_minutes"), "triggerTime": alert_dict.get("trigger_time"), "message": alert_dict.get("message"), "bbox": alert_dict.get("bbox"), "snapshotUrl": alert_dict.get("snapshot_url"), "ossUrl": alert_dict.get("oss_url"), "status": alert_dict.get("status"), "level": alert_dict.get("level"), "handleRemark": alert_dict.get("handle_remark"), "handledBy": alert_dict.get("handled_by"), "handledAt": alert_dict.get("handled_at"), "workOrderId": alert_dict.get("work_order_id"), "aiAnalysis": alert_dict.get("ai_analysis"), "createdAt": alert_dict.get("created_at"), "updatedAt": alert_dict.get("updated_at"), }) @router.put("/alert/handle") async def handle_alert( id: int = Query(..., description="告警ID"), status: str = Query(..., description="处理状态: confirmed/ignored/resolved"), remark: Optional[str] = Query(None, description="处理备注"), service: AlertService = Depends(get_alert_service), current_user: dict = Depends(get_current_user) ): """ 处理告警 权限标识：ai-alert:alert:handle """ handle_data = AlertHandleRequest(status=status, remark=remark) handled_by = current_user.get("username", "admin") alert = service.handle_alert(id, handle_data, handled_by) if not alert: raise HTTPException(status_code=404, detail="告警不存在") return YudaoResponse.success(True) @router.delete("/alert/delete") async def delete_alert( id: int = Query(..., description="告警ID"), service: AlertService = Depends(get_alert_service), current_user: dict = Depends(get_current_user) ): """ 删除告警 权限标识：ai-alert:alert:delete """ success = service.delete_alert(id) if not success: raise HTTPException(status_code=404, detail="告警不存在") return YudaoResponse.success(True) @router.get("/alert/statistics") async def get_statistics( service: AlertService = Depends(get_alert_service), current_user: dict = Depends(get_current_user) ): """ 获取告警统计 权限标识：ai-alert:alert:query """ stats = service.get_statistics() # 转换为驼峰命名 return YudaoResponse.success({ "total": stats.get("total", 0), "pending": stats.get("pending", 0), "confirmed": stats.get("confirmed", 0), "ignored": stats.get("ignored", 0), "resolved": stats.get("resolved", 0), "dispatched": stats.get("dispatched", 0), "byType": stats.get("by_type", {}), "byLevel": stats.get("by_level", {}), }) # ==================== 摄像头告警汇总 ==================== @router.get("/camera-summary/page") async def get_camera_summary_page( pageNo: int = Query(1, ge=1, description="页码"), pageSize: int = Query(20, ge=1, le=100, description="每页大小"), service: AlertService = Depends(get_alert_service), current_user: dict = Depends(get_current_user) ): """ 获取摄像头告警汇总（分页） 以摄像头为维度聚合告警数据 权限标识：ai-alert:camera-summary:query """ result = service.get_camera_alert_summary(page=pageNo, page_size=pageSize) return YudaoResponse.page( list_data=result.get("list", []), total=result.get("total", 0), page_no=pageNo, page_size=pageSize ) # ==================== 辅助函数 ==================== def _get_alert_type_name(alert_type: Optional[str]) -> str: """获取告警类型名称""" type_names = { "leave_post": "离岗检测", "intrusion": "周界入侵", "crowd": "人群聚集", "fire": "火焰检测", "smoke": "烟雾检测", "fall": "跌倒检测", "helmet": "安全帽检测", "unknown": "未知类型", } return type_names.get(alert_type, alert_type or "未知类型")