iot-device-management-service/app/routers/yudao_alert.py

"""
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 "未知类型")