chongming-plugins/aibox_sophon_adapter/routes.py

"""
AI Box Sophon Adapter — 自定义 FastAPI 路由

处理 TOML 无法描述的特殊端点：
  - POST /aibox/video_upload     视频上传（FORM-DATA）
  - POST /aibox/alarm_report     告警上报（JSON，手动解析）

路由前缀: /aibox（网关自动加 /api/v1）
"""
import os
import json
import uuid
import logging

from fastapi import APIRouter, UploadFile, File, Form, Request, Body, HTTPException
from pydantic import ValidationError
from aiofile import AIOFile

from .models.video_models import VideoUploadResponse, VideoUploadResult
from .models.alarm_models import (
    AlarmReportRequest,
    AlarmReportResponse,
    AlarmReportResult,
)

logger = logging.getLogger(__name__)

# 自定义路由器，前缀与 config.toml 中的 router_prefix 保持一致
# 最终路径由网关 register_routes() 加上全局 prefix (/api/v1) 组合而成
router = APIRouter(prefix="/aibox", tags=["aibox_sophon_adapter"])


@router.post(
    "/video_upload",
    summary="视频上传接口",
    description="盒子告警时通过 HTTP FORM-DATA 上报视频文件，返回唯一 VideoId",
    response_model=VideoUploadResponse,
)
async def video_upload(
    request: Request,
    BoardIp: str = Form(..., description="盒子网络地址"),
    BoardId: str = Form(..., description="盒子唯一标识"),
    TaskSession: str = Form(..., description="所属任务标识"),
    GBDeviceId: str = Form(default="", description="盒子国标设备号（不存在则为空字符串）"),
    GBTaskChnId: str = Form(default="", description="所属任务国标通道号（不存在则为空字符串）"),
    Video: UploadFile = File(..., description="视频文件"),
) -> VideoUploadResponse:
    """
    视频上传接口

    接收盒子告警时录制的视频文件，保存到本地存储，返回唯一 VideoId。
    上报告警报文时可携带该 VideoId 关联视频文件。

    复用网关 upload 接口的本地存储机制（data/uploads/ 目录）。
    """
    # 生成唯一 VideoId
    video_id = uuid.uuid4().hex

    try:
        # 复用网关存储路径约定: data/uploads/<worker_type>/<uuid>/<filename>
        upload_root = request.app.state.upload_root
        worker_type = "aibox_sophon_adapter"
        object_key = f"{worker_type}/{video_id}/{Video.filename}"
        dest = os.path.join(upload_root, object_key)
        os.makedirs(os.path.dirname(dest), exist_ok=True)

        # 流式写入文件（与网关 /api/v1/storage/upload 一致）
        async with AIOFile(dest, "wb") as afp:
            while chunk := await Video.read(1024 * 1024):
                await afp.write(chunk)

        file_size = os.path.getsize(dest)

        logger.info(
            f"[VideoUpload] 视频已保存: VideoId={video_id}, "
            f"BoardId={BoardId}, BoardIp={BoardIp}, "
            f"TaskSession={TaskSession}, File={Video.filename}, "
            f"Size={file_size} bytes, Path={object_key}"
        )

        return VideoUploadResponse(
            Result=VideoUploadResult(Code=0, Desc=""),
            VideoId=video_id,
        )

    except Exception as e:
        logger.error(
            f"[VideoUpload] 上传失败: BoardId={BoardId}, "
            f"BoardIp={BoardIp}, Error={e}"
        )
        return VideoUploadResponse(
            Result=VideoUploadResult(Code=-1, Desc=str(e)),
            VideoId="",
        )


@router.post(
    "/alarm_report",
    summary="告警上报接口",
    description="接收盒子告警报文（JSON），所有字段已在 OpenAPI 中声明，未知字段自动保留",
    response_model=AlarmReportResponse,
)
async def alarm_report(
    request: Request,
    payload: dict = Body(..., description="告警报文 JSON 对象"),
) -> AlarmReportResponse:
    """
    告警上报接口

    盒子检测到违规事件后，先上传视频再上报 JSON 告警报文。
    使用手动 Pydantic 验证以便在验证失败时记录完整请求体，便于排查。

    业务处理：
      1. 手动 Pydantic 验证基础字段类型（验证失败时记录原始 JSON）
      2. model_dump() 获取完整 dict（含额外字段）
      3. 提取关键索引字段存入 alarm_record 表
      4. 完整报文 JSON 字符串存入 alarm_data

    返回格式：{"Result": {"Code": 0, "Desc": ""}}
    仅当 HTTP 200 + Code 非 0 时，盒子认为上报失败。
    """
    # 手动 Pydantic 验证，失败时记录完整原始请求体
    try:
        body = AlarmReportRequest.model_validate(payload)
    except ValidationError as e:
        logger.error(
            f"[AlarmReport] Pydantic 验证失败:\n"
            f"  错误: {e.errors()}\n"
            f"  请求体: {json.dumps(payload, ensure_ascii=False)[:3000]}"
        )
        raise HTTPException(status_code=422, detail=e.errors())

    # 获取完整报文（含 extra 字段，排除 None 值以生成整洁 JSON）
    full_data = body.model_dump(exclude_none=True)

    board_id = full_data.get("BoardId", "")
    board_ip = full_data.get("BoardIp", "")
    alarm_id = full_data.get("AlarmId") or full_data.get("UniqueId", "")
    summary = full_data.get("Summary", "")
    task_session = full_data.get("TaskSession", "")
    video_file = full_data.get("VideoFile", "")

    media = full_data.get("Media", {})
    media_name = media.get("MediaName", "") if isinstance(media, dict) else ""

    alarm_time = full_data.get("Time", "")
    alarm_timestamp = full_data.get("TimeStamp", 0)

    alarm_data_str = json.dumps(full_data, ensure_ascii=False)

    try:
        from .global_var import db
        if db is not None:
            async with db.get_session() as session:
                from .models.orm import AlarmRecord
                from sqlalchemy import select

                stmt = select(AlarmRecord).where(AlarmRecord.alarm_id == alarm_id) # type: ignore
                result = await session.execute(stmt)
                existing = result.scalar_one_or_none()

                if existing is None:
                    record = AlarmRecord(
                        board_id=board_id,
                        board_ip=board_ip,
                        alarm_id=alarm_id,
                        summary=summary,
                        task_session=task_session,
                        media_name=media_name,
                        video_file=video_file,
                        alarm_time=alarm_time,
                        alarm_timestamp=alarm_timestamp,
                        alarm_data=alarm_data_str,
                    )
                    session.add(record)
                    await session.commit()
                    logger.info(
                        f"[AlarmReport] 新告警: AlarmId={alarm_id}, "
                        f"BoardId={board_id}, Summary={summary}, "
                        f"Media={media_name}, VideoFile={video_file}"
                    )
                else:
                    logger.warning(
                        f"[AlarmReport] 重复告警，已忽略: AlarmId={alarm_id}"
                    )
    except Exception as e:
        logger.error(f"[AlarmReport] 数据库写入失败: {e}")

    return AlarmReportResponse(
        Result=AlarmReportResult(Code=0, Desc="")
    )