chongming-plugins/aibox_sophon_adapter/routes.py

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

处理 TOML 无法描述的特殊端点：
  - POST /aibox_sophon_adapter/video_upload  视频上传（FORM-DATA）
  - GET  /aibox_sophon_adapter/video/{id}    视频下载/播放
  - POST /aibox_sophon_adapter/alarm_report  告警上报（JSON，手动解析）

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

import aiofiles
import aiofiles.os
import httpx

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

from chongming_request import Client

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_sophon_adapter", tags=["aibox_sophon_adapter"])


class _FileStream:
    """
    文件流 — 用于流式上传视频，避免一次性将整个文件读入内存。

    注意: httpx 0.28 的 multipart 上传（FileField）要求文件对象提供同步的
    read()/seek() 方法，httpx 内部以 64KB 分块同步读取；AsyncByteStream 只能
    作为整个请求体 content= 使用，无法与 multipart 字段共存。因此这里提供
    同步文件接口，配合同步上下文管理器 with 使用（本地磁盘小分块读取，
    内存占用依然有界）。
    """

    def __init__(self, file_path: str):
        self._file_path = file_path
        self._file = None

    def __enter__(self) -> "_FileStream":
        self._file = open(self._file_path, "rb")
        return self

    def __exit__(self, exc_type, exc, tb) -> None:
        if self._file is not None:
            self._file.close()
            self._file = None

    def read(self, size: int = -1) -> bytes:
        assert self._file is not None, "必须在 with 上下文内使用"
        return self._file.read(size)

    def seek(self, offset: int, whence: int = 0) -> int:
        assert self._file is not None, "必须在 with 上下文内使用"
        return self._file.seek(offset, whence)

# ============================================================
# 辅助函数 - 播放 tts 语音
# ============================================================

async def ip_sound_column_tts_play(zenoh_session, payload: dict) -> dict:
    """
    播放 ip 音柱 tts 语音

    可通过配置 ip_sound_column.enabled 关闭。失败时返回 {"command": ..., "result": -1}。

    Args:
        zenoh_session: Zenoh 会话（用于调用 ip_sound_column Worker）
        payload: 请求体字典，包含 command / volume / play_count / queue / sync_response /
                 play_interval / tts_pitch / tts_speed / tts_voice_name / tts_text
    """
    from .global_var import ip_sound_column_enabled

    if not ip_sound_column_enabled:
        logger.info("[TTS] ip_sound_column 播放已关闭 (ip_sound_column.enabled=false)")
        return {"command": payload.get("command", ""), "result": -1}

    async with Client(session=zenoh_session, logger=logger) as client:
        try:
            result = await client.call(
                worker_type="ip_sound_column",
                subject="ip_sound_column.tts",
                method="POST",
                payload=payload,
                timeout=10.0,
                retries=1,
            )
            return result
        except Exception as e:
            logger.error(f"[TTS] ip_sound_column 调用失败: {type(e).__name__}: {e}")
            return {"command": payload.get("command", ""), "result": -1}

# ============================================================
# 辅助函数 — 告警处理各步骤拆分为独立函数
# ============================================================

async def _analyze_with_doubao(zenoh_session, full_data: dict) -> bool:
    """
    豆包 AI 分析告警图片，判断是否存在违规。

    可通过配置 doubao_analysis.enabled 关闭。判定为 False 时返回 False（忽略告警）。

    Args:
        zenoh_session: Zenoh 会话（用于调用 doubao Worker）
        full_data: 完整告警报文

    Returns:
        True: 继续处理告警；False: 判定为 False，应忽略告警。
    """
    from .global_var import doubao_analysis_enabled

    if not doubao_analysis_enabled:
        logger.info("[AlarmReport] 豆包分析已关闭 (doubao_analysis.enabled=false)")
        return True

    async with Client(session=zenoh_session, logger=logger) as client:
        try:
            result = await client.call(
                worker_type="doubao",
                subject="doubao.analyze",
                payload={
                    "image_base64": full_data.get("ImageData", ""),
                    "prompt": f"判断是否存在{full_data.get("Summary", "")}，直接返回True或False，不要输出其他内容",
                },
                timeout=30.0,
                retries=1,
            )
            analysis = result.get("result", "")
            error = result.get("error", "")
            if error:
                logger.warning(f"Doubao 分析失败: {error}")
            if analysis == "False":
                return False
        except Exception as e:
            logger.error(f"Doubao 分析调用失败: {type(e).__name__}: {e}")

    return True


async def _save_alarm_record(alarm: dict) -> None:
    """
    将告警写入本地数据库（按 AlarmId 去重，超出上限时裁剪旧记录）。

    Args:
        alarm: 完整告警报文字典。
    """
    from .global_var import db, alarm_max_records, upload_root

    if db is None:
        return

    alarm_id = alarm.get("AlarmId") or alarm.get("UniqueId", "")
    media = alarm.get("Media", {})
    media_name = media.get("MediaName", "") if isinstance(media, dict) else ""
    alarm_data_str = orjson.dumps(alarm).decode()

    try:
        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
            existing = (await session.execute(stmt)).scalar_one_or_none()

            if existing is not None:
                logger.warning(f"[AlarmReport] 重复告警，已忽略: AlarmId={alarm_id}")
                return

            record = AlarmRecord(
                board_id=alarm.get("BoardId", ""),
                board_ip=alarm.get("BoardIp", ""),
                alarm_id=alarm_id,
                summary=alarm.get("Summary", ""),
                task_session=alarm.get("TaskSession", ""),
                media_name=media_name,
                video_file=alarm.get("VideoFile", ""),
                alarm_time=alarm.get("Time", ""),
                alarm_timestamp=alarm.get("TimeStamp", 0),
                alarm_data=alarm_data_str,
            )
            session.add(record)
            await session.commit()
            logger.info(
                f"[AlarmReport] 新告警: AlarmId={alarm_id}, "
                f"BoardId={record.board_id}, Summary={record.summary}, "
                f"Media={media_name}, VideoFile={record.video_file}"
            )

            # 裁剪告警记录，保留最新 N 条
            if alarm_max_records > 0:
                from .alarm_trim import trim_alarm_records
                await trim_alarm_records(session, upload_root, alarm_max_records)
    except Exception as e:
        logger.error(f"[AlarmReport] 数据库写入失败: {e}")


async def _upload_video_to_external(
    http_client: httpx.AsyncClient,
    video_url: str,
    board_ip: str,
    board_id: str,
    task_session: str,
    video_file: str,
) -> str:
    """
    上传本地告警视频到外部平台，返回外部 VideoId。

    本地视频存储于 data/uploads/aibox_sophon_adapter/<video_file>/ 目录。
    采用流式上传，避免将整个文件读入内存。失败时返回空字符串。

    Args:
        http_client: httpx 异步客户端
        video_url: 外部视频上传接口地址
        board_ip / board_id / task_session: 上传表单字段
        video_file: 本地 VideoId（目录名）

    Returns:
        外部返回的 VideoId；失败返回空字符串。
    """
    from .global_var import upload_root

    video_dir_path = os.path.join(upload_root, "aibox_sophon_adapter", video_file)
    if not await aiofiles.os.path.isdir(video_dir_path):
        logger.warning(f"[AlarmReport] 视频目录不存在: {video_dir_path}")
        return ""

    video_files = await aiofiles.os.listdir(video_dir_path)
    if not video_files:
        logger.warning(f"[AlarmReport] 视频目录为空: {video_dir_path}")
        return ""

    video_path = os.path.join(video_dir_path, video_files[0])
    try:
        # httpx 0.28 的 multipart 以同步 read() 流式读取文件（64KB 分块），
        # 因此这里用同步 with 打开文件流
        with _FileStream(video_path) as video_stream:
            resp = await http_client.post(
                video_url,
                data={
                    "BoardIp": board_ip,
                    "BoardId": board_id,
                    "TaskSession": task_session,
                },
                files={"Video": (video_files[0], video_stream, "video/mp4")},  # type: ignore[arg-type]
            )
        if resp.is_success:
            video_id = resp.json().get("VideoId", "")
            logger.info(f"[AlarmReport] 视频上传成功: VideoId={video_id}, URL={video_url}")
            return video_id
        else:
            logger.warning(
                f"[AlarmReport] 视频上传失败: Status={resp.status_code}, Body={resp.text[:500]}"
            )
    except Exception as e:
        logger.error(f"[AlarmReport] 视频上传异常: Error={type(e).__name__}: {e}")

    return ""


async def _forward_alarm_to_external(alarm: dict) -> None:
    """
    转发告警到外部平台：先上传视频获取外部 VideoId，替换 VideoFile 后再上报告警。

    可通过配置 alarm_forward.enabled 关闭。所有失败仅记录日志，不影响主流程。

    Args:
        alarm: 完整告警报文字典。
    """
    from .global_var import (
        alarm_forward_enabled,
        alarm_forward_url,
        alarm_forward_video_url,
        alarm_forward_timeout,
    )

    if not (alarm_forward_enabled and alarm_forward_url):
        return

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

    try:
        async with httpx.AsyncClient(timeout=alarm_forward_timeout) as http_client:
            # 1. 先上传视频到外部接口
            forward_video_id = await _upload_video_to_external(
                http_client,
                alarm_forward_video_url,
                board_ip,
                board_id,
                task_session,
                video_file,
            )

            # 2. 替换 VideoFile 为外部返回的 VideoId，然后上报告警
            forward_data = {**alarm}
            if forward_video_id:
                forward_data["VideoFile"] = forward_video_id

            resp = await http_client.post(
                alarm_forward_url,
                json=forward_data,
                headers={"Content-Type": "application/json"},
            )
            if resp.is_success:
                logger.info(
                    f"[AlarmReport] 转发成功: AlarmId={alarm_id}, "
                    f"URL={alarm_forward_url}, Status={resp.status_code}"
                )
            else:
                logger.warning(
                    f"[AlarmReport] 转发失败: AlarmId={alarm_id}, "
                    f"URL={alarm_forward_url}, Status={resp.status_code}, "
                    f"Body={resp.text[:500]}"
                )
    except httpx.TimeoutException:
        logger.error(
            f"[AlarmReport] 转发超时: AlarmId={alarm_id}, "
            f"URL={alarm_forward_url}, Timeout={alarm_forward_timeout}s"
        )
    except Exception as e:
        logger.error(
            f"[AlarmReport] 转发异常: AlarmId={alarm_id}, "
            f"URL={alarm_forward_url}, Error={type(e).__name__}: {e}"
        )


@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)
        await aiofiles.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 = await aiofiles.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 await aiofiles.os.path.isdir(video_dir):
        raise HTTPException(status_code=404, detail=f"视频不存在: {video_id}")

    # 取目录下第一个文件作为视频（通常只有一个）
    files = await aiofiles.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 await aiofiles.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. 豆包 AI 分析（判定为 False 则忽略告警）
      3. 写入本地 alarm_record 表（含完整报文 JSON）
      4. 转发外部平台（先上传视频，替换 VideoFile 后再上报）

    返回格式：{"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"  请求体: {orjson.dumps(payload).decode()[:3000]}"
        )
        raise HTTPException(status_code=422, detail=e.errors())

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

    # 1. 豆包 AI 分析（可通过配置开关关闭），判定为 False 时忽略告警
    if not await _analyze_with_doubao(request.app.state.zenoh_session, full_data):
        return AlarmReportResponse(
            Result=AlarmReportResult(Code=0, Desc="Doubao 分析判定为 False，忽略告警")
        )

    # 2. ip 音柱 TTS 播放（可通过配置开关关闭），失败时仅记录日志，不影响主流程
    from .global_var import ip_sound_column_volume, ip_sound_column_tts
    
    properties = full_data.get("Result", {}).get("Properties", [])
    area = ""
    if len(properties) > 0:
        properties_dict = {prop.get("property", ""): prop.get("value", "") for prop in properties}
        area = f"{properties_dict.get('IntrudeRegionId', '')},"
    
    tts_text = ip_sound_column_tts.get(
        full_data.get("Summary", ""), 
        f"告警: {area}{full_data.get('TaskDesc', '')}"
    )

    tts_payload = {
        "command": "play_local_tts",
        "volume": ip_sound_column_volume,
        "play_count": 1,
        "queue": True,
        "sync_response": False,
        "play_interval": 2,
        "tts_pitch": 50,
        "tts_speed": 50,
        "tts_voice_name": "xiaoyan",
        "tts_text": tts_text,
    }
    
    tts_result = await ip_sound_column_tts_play(request.app.state.zenoh_session, tts_payload)
    if tts_result.get("result") != 0:
        logger.warning(f"[AlarmReport] ip_sound_column TTS 播放失败: {tts_result}")

    # 3. 写入本地数据库（按 AlarmId 去重 + 裁剪旧记录）
    await _save_alarm_record(full_data)

    # 4. 转发外部平台（先上传视频，再上报）
    await _forward_alarm_to_external(full_data)

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