From 320f6c19a2dd963b50efea12ff369aea094a505a Mon Sep 17 00:00:00 2001 From: chakcy <947105045@qq.com> Date: Tue, 28 Jul 2026 14:16:00 +0800 Subject: [PATCH] =?UTF-8?q?modify:=20sophon=20=E8=A7=86=E9=A2=91=E6=96=87?= =?UTF-8?q?=E4=BB=B6=E4=B8=8A=E4=BC=A0=E3=80=81=E5=91=8A=E8=AD=A6=E6=8E=A5?= =?UTF-8?q?=E5=8F=A3?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- aibox_sophon_adapter/__init__.py | 14 +- aibox_sophon_adapter/global_var.py | 1 + aibox_sophon_adapter/models/__init__.py | 51 ++--- aibox_sophon_adapter/models/alarm_models.py | 139 ++++++++++++++ aibox_sophon_adapter/models/board_models.py | 2 +- aibox_sophon_adapter/models/orm.py | 20 +- aibox_sophon_adapter/models/pydantic.py | 38 +++- aibox_sophon_adapter/models/video_models.py | 16 ++ aibox_sophon_adapter/router/__init__.py | 5 + aibox_sophon_adapter/router/board_ping.py | 4 +- aibox_sophon_adapter/router/get_alarm.py | 57 ++++++ aibox_sophon_adapter/router/get_device.py | 4 +- .../router/get_offline_devices.py | 6 +- aibox_sophon_adapter/router/list_alarms.py | 85 +++++++++ aibox_sophon_adapter/router/list_devices.py | 6 +- aibox_sophon_adapter/routes.py | 177 ++++++++++++++++++ auth/router/__init__.py | 2 +- 17 files changed, 575 insertions(+), 52 deletions(-) create mode 100644 aibox_sophon_adapter/models/alarm_models.py create mode 100644 aibox_sophon_adapter/models/video_models.py create mode 100644 aibox_sophon_adapter/router/get_alarm.py create mode 100644 aibox_sophon_adapter/router/list_alarms.py create mode 100644 aibox_sophon_adapter/routes.py diff --git a/aibox_sophon_adapter/__init__.py b/aibox_sophon_adapter/__init__.py index 03857ba..7d4a832 100644 --- a/aibox_sophon_adapter/__init__.py +++ b/aibox_sophon_adapter/__init__.py @@ -7,6 +7,9 @@ AI Box Sophon Adapter 插件 3. 持久化心跳数据到 SQLite 数据库 4. 提供 HTTP API 查询设备在线状态和详情 5. 后台定时检查离线设备，触发异常告警 + 6. 视频上传接口 — 盒子告警时录制视频上传（FORM-DATA） + 7. 告警上报接口 — 违规事件 JSON 报文字动解析入库 + 8. 告警查询接口 — 告警列表/详情查询 心跳超时机制: - 设备按配置间隔（默认 5 秒）上报心跳 @@ -22,6 +25,7 @@ from chongming_worker import auto_bind_models from chongming_worker.runtime import create_workers as _create_workers from .router import * +from .routes import router # 自定义 FastAPI 路由（网关自动发现） logger = logging.getLogger(__name__) @@ -51,7 +55,7 @@ async def _check_offline_devices(interval: float): from .models.orm import BoardHeartbeat from sqlalchemy import select - stmt = select(BoardHeartbeat).where(BoardHeartbeat.is_online == True) + stmt = select(BoardHeartbeat).where(BoardHeartbeat.is_online == True) # type: ignore result = await session.execute(stmt) records = result.scalars().all() @@ -97,18 +101,12 @@ async def create_workers(session, config, count=1, worker_manager=None): # 初始化数据库 data_dir = Path(config.get("data_dir", "./data")) if isinstance(config, dict) else Path("./data") - db_path = data_dir / "storage" / "chongming" / "aibox_sophon_adapter.db" + db_path = data_dir / "aibox_sophon_adapter.db" db_path.parent.mkdir(parents=True, exist_ok=True) db = Database(db_path) await db.initialize() - # 创建表 - from .models.orm import BoardHeartbeat - async with db.get_session() as sess: - from sqlmodel import SQLModel - await sess.run_sync(lambda s: SQLModel.metadata.create_all(s.bind)) - set_global_var("db", db) set_global_var("heartbeat_timeout", hb_timeout) set_global_var("heartbeat_interval", hb_interval) diff --git a/aibox_sophon_adapter/global_var.py b/aibox_sophon_adapter/global_var.py index 9f195da..057459e 100644 --- a/aibox_sophon_adapter/global_var.py +++ b/aibox_sophon_adapter/global_var.py @@ -6,6 +6,7 @@ AI Box Sophon Adapter — 全局变量 from typing import Any from chongming_database import Database +from .models.orm import * # 数据库实例 db: Database | None = None diff --git a/aibox_sophon_adapter/models/__init__.py b/aibox_sophon_adapter/models/__init__.py index 2b0478a..e1448a4 100644 --- a/aibox_sophon_adapter/models/__init__.py +++ b/aibox_sophon_adapter/models/__init__.py @@ -1,45 +1,28 @@ """ 模型导出 — aibox_sophon_adapter - -包含 gen_models 自动生成的模型和手动定义的 board_models。 +自动生成，请勿手动修改 """ -from .pydantic import ( - AiboxSophonAdapterResponse, - GetDeviceGetRequest, - GetDeviceGetResponse, - GetOfflineDevicesGetResponse, - ListDevicesGetResponse, -) -from .board_models import ( - BoardPingRequest, - HostDiskInfo, - MediaInfo, - MediaStatusInfo, - AlgItemInfo, - AlgTaskStatusInfo, - TaskInfo, - TpuInfo, - DeviceSummary, -) -from .orm import BoardHeartbeat + +from .pydantic import AiboxSophonAdapterResponse +from .pydantic import AlarmReportPostResponse +from .pydantic import GetAlarmGetRequest +from .pydantic import GetAlarmGetResponse +from .pydantic import GetDeviceGetRequest +from .pydantic import GetDeviceGetResponse +from .pydantic import GetOfflineDevicesGetResponse +from .pydantic import ListAlarmsGetRequest +from .pydantic import ListAlarmsGetResponse +from .pydantic import ListDevicesGetResponse __all__ = [ - # 自动生成 "AiboxSophonAdapterResponse", + "AlarmReportPostResponse", + "GetAlarmGetRequest", + "GetAlarmGetResponse", "GetDeviceGetRequest", "GetDeviceGetResponse", "GetOfflineDevicesGetResponse", + "ListAlarmsGetRequest", + "ListAlarmsGetResponse", "ListDevicesGetResponse", - # 手动定义 - "BoardPingRequest", - "HostDiskInfo", - "MediaInfo", - "MediaStatusInfo", - "AlgItemInfo", - "AlgTaskStatusInfo", - "TaskInfo", - "TpuInfo", - "DeviceSummary", - # ORM - "BoardHeartbeat", ] diff --git a/aibox_sophon_adapter/models/alarm_models.py b/aibox_sophon_adapter/models/alarm_models.py new file mode 100644 index 0000000..cc46986 --- /dev/null +++ b/aibox_sophon_adapter/models/alarm_models.py @@ -0,0 +1,139 @@ +""" +告警上报 — Pydantic 数据模型 + +包含请求体和响应体。请求体声明了规范中所有已知字段， +未声明的额外字段通过 model_config extra=allow 保留。 +""" +from typing import Any, Optional +from pydantic import BaseModel, Field + + +# ============================================================ +# 告警报文子结构 +# ============================================================ + +class GpsInfo(BaseModel): + """GPS 位置信息""" + available: bool = Field(default=False, description="是否有 GPS 功能") + kSpeed: float = Field(default=0.0, description="对地速度 (km/h)") + nSpeed: float = Field(default=0.0, description="对地速率 (节)") + latitude: float = Field(default=0.0, description="纬度") + longitude: float = Field(default=0.0, description="经度") + angleCourse: float = Field(default=0.0, description="GPS 角度") + + +class MediaInfo(BaseModel): + """视频通道信息""" + MediaName: str = Field(default="", description="通道标识") + MediaDesc: str = Field(default="", description="通道描述") + MediaUrl: str = Field(default="", description="通道视频流地址") + MediaWidth: int = Field(default=0, description="通道视频宽度") + MediaHeight: int = Field(default=0, description="通道视频高度") + SubId: str = Field(default="", description="关联 GB28181 通道号") + GBTransport: bool = Field(default=False, description="是否接入 GB28181") + RtspTransport: bool = Field(default=False, description="是否开启 RTSP 转发") + Params: Optional[dict] = Field(default=None, description="预留其他参数") + + +class ResultProperty(BaseModel): + """告警 Result 中的属性项""" + property: str = Field(default="", description="属性标识") + desc: str = Field(default="", description="属性描述") + value: Any = Field(default=None, description="属性值（String/Int/Float）") + display: str = Field(default="", description="属性详细信息") + type: str = Field(default="", description="属性值类型") + + +class AlarmResult(BaseModel): + """违规检测结果（结构随算法变化，extra=allow 兜底）""" + model_config = {"extra": "allow"} + + Type: str = Field(default="", description="违规类型标识（NoHelmet / Intrude / PeopleCross 等）") + Description: str = Field(default="", description="违规描述") + Cropped: bool = Field(default=False, description="是否裁剪") + RegType: str = Field(default="Rectangle", description="上报点集类型（Rectangle / Line / Polygon）") + RelativeBox: list[float] = Field(default_factory=list, description="矩形目标位置 [X,Y,Width,Height] 百分比") + RelativeRegion: list[dict] = Field(default_factory=list, description="多边形目标位置") + Properties: list[ResultProperty] = Field(default_factory=list, description="算法特殊属性") + + +# ============================================================ +# 告警上报请求体 +# ============================================================ + +class AlarmReportRequest(BaseModel): + """ + 告警上报请求体 — 声明规范中所有已知字段 + + 未声明的额外字段通过 extra=allow 保留，完整数据可通过 model_dump() 获取。 + 盒子检测到违规事件后，先上传视频再上报此 JSON 告警报文。 + """ + model_config = {"extra": "allow"} + + # 设备信息 + BoardIp: str = Field(default="", description="盒子网络地址") + BoardId: str = Field(default="", description="盒子唯一标识") + + # 告警标识 + AlarmId: str = Field(default="", description="告警标识") + UniqueId: str = Field(default="", description="告警唯一标识（与 AlarmId 等价)") + + # 透传字段 + Addition: Optional[dict] = Field(default=None, description="透传字段") + + # 截图数据 + ImageData: str = Field(default="", description="无标注告警图 JPEG Base64 编码") + ImageDataLabeled: str = Field(default="", description="有标注告警图 JPEG Base64 编码") + + # 截图本地路径 + LocalRawPath: str = Field(default="", description="无标注告警图在盒子中的相对路径") + LocalLabeledPath: str = Field(default="", description="有标注告警图在盒子中的相对路径") + + # 通道信息 + Media: MediaInfo = Field(default_factory=MediaInfo, description="配置的媒体通道") + + # 违规描述 + Summary: str = Field(default="", description="违规类型（NoHelmet / Intrude / PeopleCross 等）") + + # 任务信息 + TaskSession: str = Field(default="", description="任务编号") + GBDeviceId: str = Field(default="", description="国标设备编号，不存在则为空字符串") + GBTaskChnId: str = Field(default="", description="任务对应的国标通道编号，不存在则为空字符串") + TaskDesc: str = Field(default="", description="任务描述") + + # 时间 + Time: str = Field(default="", description="告警时间戳字符串") + TimeStamp: int = Field(default=0, description="告警时间戳（微秒）") + + # 视频 + VideoFile: str = Field(default="", description="关联的视频标识（视频上传接口返回）") + + # GPS + GPS: GpsInfo = Field(default_factory=GpsInfo, description="当前位置信息") + + # 违规结果 + Result: AlarmResult = Field(default_factory=AlarmResult, description="违规检测结果") + + # 预留字段 + Type: int = Field(default=0, description="预留字段") + DataId: int = Field(default=0, description="数据 ID") + Id: int = Field(default=0, description="记录 ID") + + +# ============================================================ +# 告警上报响应体 +# ============================================================ + +class AlarmReportResult(BaseModel): + """告警上报接口返回状态""" + Code: int = Field(default=0, description="接口状态错误码，0 表示成功，非 0 表示失败") + Desc: str = Field(default="", description="错误码描述") + + +class AlarmReportResponse(BaseModel): + """告警上报接口完整响应 + + 仅当 HTTP 200 + Code 非 0 时，盒子认为上报失败。 + """ + Result: AlarmReportResult = Field(default_factory=AlarmReportResult, description="接口状态返回") + diff --git a/aibox_sophon_adapter/models/board_models.py b/aibox_sophon_adapter/models/board_models.py index dc58375..7d27579 100644 --- a/aibox_sophon_adapter/models/board_models.py +++ b/aibox_sophon_adapter/models/board_models.py @@ -45,7 +45,7 @@ class AlgItemInfo(BaseModel): lineRequired: bool = Field(default=False, description="是否需要线") majorId: int = Field(..., description="主 ID") majorOnly: bool = Field(default=False, description="仅主算法") - minorId: int = Field(..., description="副 ID") + minorId: int = Field(default=0, description="副 ID（majorOnly=true 时可能不传）") name: str = Field(..., description="算法名称") zoneDesc: str = Field(default="", description="区域描述") zoneRequired: bool = Field(default=False, description="是否需要区域") diff --git a/aibox_sophon_adapter/models/orm.py b/aibox_sophon_adapter/models/orm.py index dd59506..d68dfd6 100644 --- a/aibox_sophon_adapter/models/orm.py +++ b/aibox_sophon_adapter/models/orm.py @@ -11,7 +11,7 @@ from sqlmodel import Field, SQLModel class BoardHeartbeat(SQLModel, table=True): """设备心跳记录表""" - __tablename__ = "board_heartbeat" + __tablename__ = "board_heartbeat" # type: ignore id: Optional[int] = Field(default=None, primary_key=True) board_id: str = Field(index=True, unique=True, description="设备唯一标识 (BoardId)") @@ -29,3 +29,21 @@ class BoardHeartbeat(SQLModel, table=True): is_online: bool = Field(default=True, description="是否在线") created_at: datetime = Field(default_factory=datetime.now, description="首次上报时间") updated_at: datetime = Field(default_factory=datetime.now, description="最后更新时间") + + +class AlarmRecord(SQLModel, table=True): + """告警记录表 — 存储盒子上报的完整告警报文""" + __tablename__ = "alarm_record" # type: ignore + + id: Optional[int] = Field(default=None, primary_key=True) + board_id: str = Field(index=True, description="设备唯一标识 (BoardId)") + board_ip: str = Field(default="", description="设备 IP") + alarm_id: str = Field(index=True, unique=True, description="告警唯一标识 (AlarmId/UniqueId)") + summary: str = Field(default="", description="违规类型 (Summary)") + task_session: str = Field(default="", description="任务编号 (TaskSession)") + media_name: str = Field(default="", description="通道名称") + video_file: str = Field(default="", description="关联视频标识 (VideoFile)") + alarm_time: str = Field(default="", description="告警时间 (Time)") + alarm_timestamp: int = Field(default=0, description="告警时间戳-微秒 (TimeStamp)") + alarm_data: str = Field(default="{}", description="完整告警报文 JSON 字符串") + created_at: datetime = Field(default_factory=datetime.now, description="上报时间") diff --git a/aibox_sophon_adapter/models/pydantic.py b/aibox_sophon_adapter/models/pydantic.py index 50dfd38..9350435 100644 --- a/aibox_sophon_adapter/models/pydantic.py +++ b/aibox_sophon_adapter/models/pydantic.py @@ -8,6 +8,28 @@ from __future__ import annotations from typing import Any, Optional from pydantic import BaseModel, Field +class AlarmReportPostResponse(BaseModel): + """告警上报接口 — 响应体""" + Result: dict[str, Any] = Field(...) + +class GetAlarmGetRequest(BaseModel): + """查询单条告警详情 — 请求体""" + alarm_id: str = Field(...) + +class GetAlarmGetResponse(BaseModel): + """查询单条告警详情 — 响应体""" + alarm_id: str = Field(...) + board_id: str = Field(...) + board_ip: str = Field(...) + summary: str = Field(...) + task_session: str = Field(...) + media_name: str = Field(...) + video_file: str = Field(...) + alarm_time: str = Field(...) + alarm_timestamp: int = Field(...) + alarm_data: dict[str, Any] = Field(...) + created_at: str = Field(...) + class GetDeviceGetRequest(BaseModel): """查询单个设备详情 — 请求体""" board_id: str = Field(...) @@ -24,6 +46,20 @@ class GetOfflineDevicesGetResponse(BaseModel): offline_devices: list[dict] = Field(...) total: int = Field(...) +class ListAlarmsGetRequest(BaseModel): + """查询告警记录列表 — 请求体""" + board_id: str = Field(...) + summary: str = Field(...) + page: int = Field(...) + page_size: int = Field(...) + +class ListAlarmsGetResponse(BaseModel): + """查询告警记录列表 — 响应体""" + alarms: list[dict] = Field(...) + total: int = Field(...) + page: int = Field(...) + page_size: int = Field(...) + class ListDevicesGetResponse(BaseModel): """查询所有设备状态 — 响应体""" devices: list[dict] = Field(...) @@ -31,4 +67,4 @@ class ListDevicesGetResponse(BaseModel): # 所有响应模型的联合类型 -AiboxSophonAdapterResponse = ListDevicesGetResponse | GetOfflineDevicesGetResponse | GetDeviceGetResponse +AiboxSophonAdapterResponse = GetOfflineDevicesGetResponse | AlarmReportPostResponse | GetAlarmGetResponse | GetDeviceGetResponse | ListAlarmsGetResponse | ListDevicesGetResponse diff --git a/aibox_sophon_adapter/models/video_models.py b/aibox_sophon_adapter/models/video_models.py new file mode 100644 index 0000000..ce52b1a --- /dev/null +++ b/aibox_sophon_adapter/models/video_models.py @@ -0,0 +1,16 @@ +""" +视频上传 — Pydantic 数据模型 +""" +from pydantic import BaseModel, Field + + +class VideoUploadResult(BaseModel): + """视频上传接口返回状态""" + Code: int = Field(default=0, description="接口状态错误码，0 表示成功") + Desc: str = Field(default="", description="错误码描述") + + +class VideoUploadResponse(BaseModel): + """视频上传接口完整响应""" + Result: VideoUploadResult = Field(default_factory=VideoUploadResult, description="接口状态返回") + VideoId: str = Field(..., description="视频唯一标识") diff --git a/aibox_sophon_adapter/router/__init__.py b/aibox_sophon_adapter/router/__init__.py index d8cd927..36b66ed 100644 --- a/aibox_sophon_adapter/router/__init__.py +++ b/aibox_sophon_adapter/router/__init__.py @@ -5,10 +5,15 @@ from .board_ping import board_ping from .list_devices import list_devices from .get_device import get_device from .get_offline_devices import get_offline_devices +from .list_alarms import list_alarms +from .get_alarm import get_alarm + __all__ = [ "board_ping", "list_devices", "get_device", "get_offline_devices", + "list_alarms", + "get_alarm", ] diff --git a/aibox_sophon_adapter/router/board_ping.py b/aibox_sophon_adapter/router/board_ping.py index f51b557..305b16f 100644 --- a/aibox_sophon_adapter/router/board_ping.py +++ b/aibox_sophon_adapter/router/board_ping.py @@ -32,11 +32,13 @@ async def board_ping(request: BoardPingRequest, worker_id: str = "") -> dict: board_id = request.BoardId try: + if db is None: + raise RuntimeError("Database not initialized") async with db.get_session() as session: from ..models.orm import BoardHeartbeat from sqlalchemy import select - stmt = select(BoardHeartbeat).where(BoardHeartbeat.board_id == board_id) + stmt = select(BoardHeartbeat).where(BoardHeartbeat.board_id == board_id) # type: ignore result = await session.execute(stmt) record: BoardHeartbeat | None = result.scalar_one_or_none() diff --git a/aibox_sophon_adapter/router/get_alarm.py b/aibox_sophon_adapter/router/get_alarm.py new file mode 100644 index 0000000..3d500de --- /dev/null +++ b/aibox_sophon_adapter/router/get_alarm.py @@ -0,0 +1,57 @@ +""" +查询单条告警详情 — GET /aibox/get_alarm?alarm_id=xxx +""" +import json +import logging + +from chongming_worker import rpc_method + +logger = logging.getLogger(__name__) + + +@rpc_method( + subject="aibox_sophon_adapter.get_alarm", + method="GET", + summary="查询单条告警详情", +) +async def get_alarm(alarm_id: str, worker_id: str = "") -> dict: + """根据 AlarmId 查询完整的告警报文数据""" + from ..global_var import db + + try: + if db is None: + raise RuntimeError("Database not initialized") + 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) + record = result.scalar_one_or_none() + + if record is None: + return {"error": f"告警不存在: {alarm_id}"} + + alarm_data = {} + try: + alarm_data = json.loads(record.alarm_data) if record.alarm_data else {} + except json.JSONDecodeError: + pass + + return { + "alarm_id": record.alarm_id, + "board_id": record.board_id, + "board_ip": record.board_ip, + "summary": record.summary, + "task_session": record.task_session, + "media_name": record.media_name, + "video_file": record.video_file, + "alarm_time": record.alarm_time, + "alarm_timestamp": record.alarm_timestamp, + "alarm_data": alarm_data, + "created_at": record.created_at.isoformat() if record.created_at else None, + } + + except Exception as e: + logger.error(f"[get_alarm] 查询失败 (alarm_id={alarm_id}): {e}") + return {"error": str(e)} diff --git a/aibox_sophon_adapter/router/get_device.py b/aibox_sophon_adapter/router/get_device.py index 5179a4f..7aad058 100644 --- a/aibox_sophon_adapter/router/get_device.py +++ b/aibox_sophon_adapter/router/get_device.py @@ -20,11 +20,13 @@ async def get_device(board_id: str, worker_id: str = "") -> dict: from ..global_var import db, heartbeat_timeout try: + if db is None: + raise RuntimeError("Database not initialized") async with db.get_session() as session: from ..models.orm import BoardHeartbeat from sqlalchemy import select - stmt = select(BoardHeartbeat).where(BoardHeartbeat.board_id == board_id) + stmt = select(BoardHeartbeat).where(BoardHeartbeat.board_id == board_id) # type: ignore result = await session.execute(stmt) record: BoardHeartbeat | None = result.scalar_one_or_none() diff --git a/aibox_sophon_adapter/router/get_offline_devices.py b/aibox_sophon_adapter/router/get_offline_devices.py index 2abb18c..0ddfbf2 100644 --- a/aibox_sophon_adapter/router/get_offline_devices.py +++ b/aibox_sophon_adapter/router/get_offline_devices.py @@ -21,11 +21,13 @@ async def get_offline_devices(worker_id: str = "") -> dict: offline_devices = [] try: + if db is None: + raise RuntimeError("Database not initialized") async with db.get_session() as session: from ..models.orm import BoardHeartbeat - from sqlalchemy import select + from sqlalchemy import desc, select - stmt = select(BoardHeartbeat).order_by(BoardHeartbeat.last_ping_time.desc()) + stmt = select(BoardHeartbeat).order_by(desc(BoardHeartbeat.last_ping_time)) # type: ignore result = await session.execute(stmt) records = result.scalars().all() diff --git a/aibox_sophon_adapter/router/list_alarms.py b/aibox_sophon_adapter/router/list_alarms.py new file mode 100644 index 0000000..440ed21 --- /dev/null +++ b/aibox_sophon_adapter/router/list_alarms.py @@ -0,0 +1,85 @@ +""" +查询告警记录列表 — GET /aibox/list_alarms +""" +import json +import logging + +from chongming_worker import rpc_method + +logger = logging.getLogger(__name__) + + +@rpc_method( + subject="aibox_sophon_adapter.list_alarms", + method="GET", + summary="查询告警记录列表", +) +async def list_alarms( + board_id: str = "", + summary: str = "", + page: int = 1, + page_size: int = 20, + worker_id: str = "", +) -> dict: + """分页查询告警记录，支持按设备ID和告警类型筛选""" + from ..global_var import db + + alarms = [] + total = 0 + try: + if db is None: + raise RuntimeError("Database not initialized") + async with db.get_session() as session: + from ..models.orm import AlarmRecord + from sqlalchemy import select, func, desc + + # 构建查询 + conditions = [] + if board_id: + conditions.append(AlarmRecord.board_id == board_id) # type: ignore + if summary: + conditions.append(AlarmRecord.summary == summary) # type: ignore + + # 总数 + count_stmt = select(func.count()).select_from(AlarmRecord) + if conditions: + from sqlalchemy import and_ + count_stmt = count_stmt.where(and_(*conditions)) + count_result = await session.execute(count_stmt) + total = count_result.scalar() or 0 + + # 分页数据 + stmt = select(AlarmRecord).order_by(desc(AlarmRecord.created_at)) + if conditions: + from sqlalchemy import and_ + stmt = stmt.where(and_(*conditions)) + stmt = stmt.offset((page - 1) * page_size).limit(page_size) + result = await session.execute(stmt) + records = result.scalars().all() + + for record in records: + alarm_data = {} + try: + alarm_data = json.loads(record.alarm_data) if record.alarm_data else {} + except json.JSONDecodeError: + pass + + alarms.append({ + "alarm_id": record.alarm_id, + "board_id": record.board_id, + "board_ip": record.board_ip, + "summary": record.summary, + "task_session": record.task_session, + "media_name": record.media_name, + "video_file": record.video_file, + "alarm_time": record.alarm_time, + "alarm_timestamp": record.alarm_timestamp, + "result_type": alarm_data.get("Result", {}).get("Type", ""), + "created_at": record.created_at.isoformat() if record.created_at else None, + }) + + except Exception as e: + logger.error(f"[list_alarms] 查询失败: {e}") + return {"alarms": [], "total": 0, "page": page, "page_size": page_size, "error": str(e)} + + return {"alarms": alarms, "total": total, "page": page, "page_size": page_size} diff --git a/aibox_sophon_adapter/router/list_devices.py b/aibox_sophon_adapter/router/list_devices.py index fee48fa..374c67c 100644 --- a/aibox_sophon_adapter/router/list_devices.py +++ b/aibox_sophon_adapter/router/list_devices.py @@ -21,11 +21,13 @@ async def list_devices(worker_id: str = "") -> dict: devices = [] try: + if db is None: + raise RuntimeError("Database not initialized") async with db.get_session() as session: from ..models.orm import BoardHeartbeat - from sqlalchemy import select + from sqlalchemy import select, desc - stmt = select(BoardHeartbeat).order_by(BoardHeartbeat.updated_at.desc()) + stmt = select(BoardHeartbeat).order_by(desc(BoardHeartbeat.updated_at)) # type: ignore result = await session.execute(stmt) records = result.scalars().all() diff --git a/aibox_sophon_adapter/routes.py b/aibox_sophon_adapter/routes.py new file mode 100644 index 0000000..ec90334 --- /dev/null +++ b/aibox_sophon_adapter/routes.py @@ -0,0 +1,177 @@ +""" +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="") + ) diff --git a/auth/router/__init__.py b/auth/router/__init__.py index 0eae3fb..f3c9629 100644 --- a/auth/router/__init__.py +++ b/auth/router/__init__.py @@ -5,4 +5,4 @@ from .update_profile import update_profile from .change_password import change_password -__all__ = ["login", "register", "me", "update_profile", "change_password"] \ No newline at end of file +__all__ = ["login", "register", "me", "update_profile", "change_password"]