diff --git a/aibox_sophon_adapter/__init__.py b/aibox_sophon_adapter/__init__.py index 6cfc2e7..0e3b152 100644 --- a/aibox_sophon_adapter/__init__.py +++ b/aibox_sophon_adapter/__init__.py @@ -113,6 +113,7 @@ async def create_workers(session, config, count=1, worker_manager=None): alarm_fwd_cfg = config.get("alarm_forward", {}) alarm_fwd_enabled = alarm_fwd_cfg.get("enabled", False) alarm_fwd_url = alarm_fwd_cfg.get("url", "http://dg.datagoo.cn:60014/open-api/event-alarm/receive") + alarm_fwd_video_url = alarm_fwd_cfg.get("video_upload_url", "http://dg.datagoo.cn:60014/open-api/event-alarm/receiveVideoUpload") alarm_fwd_timeout = alarm_fwd_cfg.get("timeout", 10.0) # 初始化数据库 @@ -163,6 +164,7 @@ async def create_workers(session, config, count=1, worker_manager=None): set_global_var("doubao_analysis_enabled", doubao_enabled) set_global_var("alarm_forward_enabled", alarm_fwd_enabled) set_global_var("alarm_forward_url", alarm_fwd_url) + set_global_var("alarm_forward_video_url", alarm_fwd_video_url) set_global_var("alarm_forward_timeout", alarm_fwd_timeout) logger.info( diff --git a/aibox_sophon_adapter/global_var.py b/aibox_sophon_adapter/global_var.py index a42419d..6310adf 100644 --- a/aibox_sophon_adapter/global_var.py +++ b/aibox_sophon_adapter/global_var.py @@ -34,6 +34,7 @@ doubao_analysis_enabled: bool = True # 告警转发配置 alarm_forward_enabled: bool = False alarm_forward_url: str = "http://dg.datagoo.cn:60014/open-api/event-alarm/receive" +alarm_forward_video_url: str = "http://dg.datagoo.cn:60014/open-api/event-alarm/receiveVideoUpload" alarm_forward_timeout: float = 10.0 # 心跳超时配置（秒） @@ -64,6 +65,7 @@ __all__ = [ "doubao_analysis_enabled", "alarm_forward_enabled", "alarm_forward_url", + "alarm_forward_video_url", "alarm_forward_timeout", "set_global_var", ] diff --git a/aibox_sophon_adapter/routes.py b/aibox_sophon_adapter/routes.py index 40f2a4c..662f081 100644 --- a/aibox_sophon_adapter/routes.py +++ b/aibox_sophon_adapter/routes.py @@ -2,10 +2,11 @@ AI Box Sophon Adapter — 自定义 FastAPI 路由 处理 TOML 无法描述的特殊端点： - - POST /aibox/video_upload 视频上传（FORM-DATA） - - POST /aibox/alarm_report 告警上报（JSON，手动解析） + - POST /aibox_sophon_adapter/video_upload 视频上传（FORM-DATA） + - GET /aibox_sophon_adapter/video/{id} 视频下载/播放 + - POST /aibox_sophon_adapter/alarm_report 告警上报（JSON，手动解析） -路由前缀: /aibox（网关自动加 /api/v1） +路由前缀: /aibox_sophon_adapter（网关自动加 /api/v1） """ import os import json @@ -14,6 +15,7 @@ import hashlib import logging import aiofiles +import aiofiles.os import httpx from fastapi import APIRouter, UploadFile, File, Form, Request, Body, HTTPException @@ -36,6 +38,268 @@ logger = logging.getLogger(__name__) router = APIRouter(prefix="/aibox_sophon_adapter", tags=["aibox_sophon_adapter"]) +class _FileStream(httpx.AsyncByteStream): + """ + 异步文件流 — 用于流式上传视频，避免一次性将整个文件读入内存。 + + 以 1MB 分块异步读取文件，供 httpx multipart 上传使用。 + """ + + _CHUNK_SIZE = 1024 * 1024 + + def __init__(self, file_path: str): + self._file_path = file_path + self._file = None + + async def __aenter__(self) -> "_FileStream": + self._file = await aiofiles.open(self._file_path, "rb") + return self + + async def __aexit__(self, exc_type, exc, tb) -> None: + if self._file is not None: + await self._file.close() + + async def __aiter__(self): + assert self._file is not None, "必须在 async with 上下文内使用" + while chunk := await self._file.read(self._CHUNK_SIZE): + yield chunk + + +# ============================================================ +# 辅助函数 — 告警处理各步骤拆分为独立函数 +# ============================================================ + +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) 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 = json.dumps(alarm, ensure_ascii=False) + + 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// 目录。 + 采用流式上传，避免将整个文件读入内存。失败时返回空字符串。 + + 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: + async 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="视频上传接口", @@ -70,7 +334,7 @@ async def video_upload( 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) + await aiofiles.os.makedirs(os.path.dirname(dest), exist_ok=True) # 流式写入文件，同时计算 MD5 和总字节数 total_bytes = 0 @@ -81,7 +345,7 @@ async def video_upload( total_bytes += len(chunk) md5_hash.update(chunk) - file_size = os.path.getsize(dest) + file_size = await aiofiles.os.path.getsize(dest) file_md5 = md5_hash.hexdigest() logger.info( @@ -127,16 +391,16 @@ async def video_download( worker_type = "aibox_sophon_adapter" # 在 uploads 目录下查找匹配的视频文件 video_dir = os.path.join(upload_root, worker_type, video_id) - if not os.path.isdir(video_dir): + if not await aiofiles.os.path.isdir(video_dir): raise HTTPException(status_code=404, detail=f"视频不存在: {video_id}") # 取目录下第一个文件作为视频（通常只有一个） - files = os.listdir(video_dir) + 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 os.path.isfile(video_path): + if not await aiofiles.os.path.isfile(video_path): raise HTTPException(status_code=404, detail=f"视频文件不存在: {files[0]}") return FileResponse( @@ -162,17 +426,15 @@ async def alarm_report( 盒子检测到违规事件后，先上传视频再上报 JSON 告警报文。 使用手动 Pydantic 验证以便在验证失败时记录完整请求体，便于排查。 - 业务处理： - 1. 手动 Pydantic 验证基础字段类型（验证失败时记录原始 JSON） - 2. model_dump() 获取完整 dict（含额外字段） - 3. 提取关键索引字段存入 alarm_record 表 - 4. 完整报文 JSON 字符串存入 alarm_data + 处理流程： + 1. 手动 Pydantic 验证基础字段类型（失败时记录原始 JSON） + 2. 豆包 AI 分析（判定为 False 则忽略告警） + 3. 写入本地 alarm_record 表（含完整报文 JSON） + 4. 转发外部平台（先上传视频，替换 VideoFile 后再上报） 返回格式：{"Result": {"Code": 0, "Desc": ""}} 仅当 HTTP 200 + Code 非 0 时，盒子认为上报失败。 """ - zenoh_session = request.app.state.zenoh_session - # 手动 Pydantic 验证，失败时记录完整原始请求体 try: body = AlarmReportRequest.model_validate(payload) @@ -187,123 +449,17 @@ async def alarm_report( # 获取完整报文（含 extra 字段，排除 None 值以生成整洁 JSON） full_data = body.model_dump(exclude_none=True) - # 豆包 AI 分析（可通过配置开关关闭） - from .global_var import doubao_analysis_enabled - if doubao_analysis_enabled: - async with Client(session=zenoh_session) 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 AlarmReportResponse( - Result=AlarmReportResult(Code=0, Desc="Doubao 分析判定为 False，忽略告警") - ) - except Exception as e: - logger.error(f"Doubao 分析调用失败: {type(e).__name__}: {e}") - else: - logger.info("[AlarmReport] 豆包分析已关闭 (doubao_analysis.enabled=false)") + # 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，忽略告警") + ) - 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", "") + # 2. 写入本地数据库（按 AlarmId 去重 + 裁剪旧记录） + await _save_alarm_record(full_data) - 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}" - ) - # 裁剪告警记录，保留最新 N 条 - from .global_var import alarm_max_records, upload_root as _upload_root - if alarm_max_records > 0: - from .alarm_trim import trim_alarm_records - await trim_alarm_records(session, _upload_root, alarm_max_records) - else: - logger.warning( - f"[AlarmReport] 重复告警，已忽略: AlarmId={alarm_id}" - ) - except Exception as e: - logger.error(f"[AlarmReport] 数据库写入失败: {e}") - - # 告警转发到外部平台 - from .global_var import alarm_forward_enabled, alarm_forward_url, alarm_forward_timeout - if alarm_forward_enabled and alarm_forward_url: - try: - async with httpx.AsyncClient(timeout=alarm_forward_timeout) as http_client: - forward_resp = await http_client.post( - alarm_forward_url, - json=full_data, - headers={"Content-Type": "application/json"}, - ) - if forward_resp.is_success: - logger.info( - f"[AlarmReport] 转发成功: AlarmId={alarm_id}, " - f"URL={alarm_forward_url}, Status={forward_resp.status_code}" - ) - else: - logger.warning( - f"[AlarmReport] 转发失败: AlarmId={alarm_id}, " - f"URL={alarm_forward_url}, Status={forward_resp.status_code}, " - f"Body={forward_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}" - ) + # 3. 转发外部平台（先上传视频，再上报） + await _forward_alarm_to_external(full_data) return AlarmReportResponse( Result=AlarmReportResult(Code=0, Desc="")