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 hashlib
import logging

import aiofiles

from fastapi import APIRouter, UploadFile, File, Form, Request, Body, HTTPException
from fastapi.responses import FileResponse
from pydantic import ValidationError

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
    # 获取设备声明的文件大小（若 Starlette 不支持则记 -1）
    declared_size = getattr(Video, 'size', -1) or -1

    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)

        # 流式写入文件，同时计算 MD5 和总字节数
        total_bytes = 0
        md5_hash = hashlib.md5()
        async with aiofiles.open(dest, "wb") as afp:
            while chunk := await Video.read(1024 * 1024):
                await afp.write(chunk)
                total_bytes += len(chunk)
                md5_hash.update(chunk)

        file_size = os.path.getsize(dest)
        file_md5 = md5_hash.hexdigest()

        logger.info(
            f"[VideoUpload] 视频已保存: VideoId={video_id}, "
            f"BoardId={BoardId}, BoardIp={BoardIp}, "
            f"TaskSession={TaskSession}, File={Video.filename}, "
            f"declared={declared_size}, received={total_bytes}, disk={file_size}, "
            f"md5={file_md5}, 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={type(e).__name__}: {e}"
        )
        return VideoUploadResponse(
            Result=VideoUploadResult(Code=-1, Desc=str(e)),
            VideoId="",
        )


@router.get(
    "/video/{video_id}",
    summary="视频下载/播放",
    description="根据 VideoId 下载或流式播放告警关联视频",
)
async def video_download(
    request: Request,
    video_id: str,
) -> FileResponse:
    """
    视频播放端点

    根据 video_upload 返回的 VideoId 查找并返回视频文件。
    支持浏览器内嵌播放（Content-Disposition: inline）。
    """
    upload_root = request.app.state.upload_root
    worker_type = "aibox_sophon_adapter"
    # 在 uploads 目录下查找匹配的视频文件
    video_dir = os.path.join(upload_root, worker_type, video_id)
    if not os.path.isdir(video_dir):
        raise HTTPException(status_code=404, detail=f"视频不存在: {video_id}")

    # 取目录下第一个文件作为视频（通常只有一个）
    files = os.listdir(video_dir)
    if not files:
        raise HTTPException(status_code=404, detail=f"视频目录为空: {video_id}")

    video_path = os.path.join(video_dir, files[0])
    if not os.path.isfile(video_path):
        raise HTTPException(status_code=404, detail=f"视频文件不存在: {files[0]}")

    return FileResponse(
        video_path,
        media_type="video/mp4",
        headers={"Content-Disposition": "inline"},
    )


@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="")
    )