""" 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 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/// 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(body: AlarmReportRequest) -> AlarmReportResponse: """ 告警上报接口 盒子检测到违规事件后,先上传视频再上报 JSON 告警报文。 请求体所有字段已在 Pydantic 模型中声明(含默认值), model_config extra=allow 确保盒子固件升级新增字段不会丢失。 业务处理: 1. Pydantic 验证基础字段类型 2. model_dump() 获取完整 dict(含额外字段) 3. 提取关键索引字段存入 alarm_record 表 4. 完整报文 JSON 字符串存入 alarm_data 返回格式:{"Result": {"Code": 0, "Desc": ""}} 仅当 HTTP 200 + Code 非 0 时,盒子认为上报失败。 """ # 获取完整报文(含 extra 字段) full_data = body.model_dump() 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="") )