chongming/chongming-plugins
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

modify: sophon 视频文件上传、告警接口

Browse Source
This commit is contained in:
chakcy
2026-07-28 14:16:00 +08:00
parent 79eb7fe5cc
commit 320f6c19a2
17 changed files with 575 additions and 52 deletions
Download Patch File Download Diff File Expand all files Collapse all files
aibox_sophon_adapter/__init__.py
+6 -8
View File
@@ -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)
aibox_sophon_adapter/global_var.py
+1
View File
@@ -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
aibox_sophon_adapter/models/__init__.py
+17 -34
View File
@@ -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",
]
aibox_sophon_adapter/models/alarm_models.py
+139
View File
@@ -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="接口状态返回")
aibox_sophon_adapter/models/board_models.py
+1 -1
View File
@@ -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="是否需要区域")
aibox_sophon_adapter/models/orm.py
+19 -1
View File
@@ -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="上报时间")
aibox_sophon_adapter/models/pydantic.py
+37 -1
View File
@@ -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
aibox_sophon_adapter/models/video_models.py
+16
View File
@@ -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="视频唯一标识")
aibox_sophon_adapter/router/__init__.py
+5
View File
@@ -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",
]
aibox_sophon_adapter/router/board_ping.py
+3 -1
View File
@@ -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()
aibox_sophon_adapter/router/get_alarm.py
+57
View File
@@ -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)}
aibox_sophon_adapter/router/get_device.py
+3 -1
View File
@@ -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()
aibox_sophon_adapter/router/get_offline_devices.py
+4 -2
View File
@@ -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()
aibox_sophon_adapter/router/list_alarms.py
+85
View File
@@ -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}
aibox_sophon_adapter/router/list_devices.py
+4 -2
View File
@@ -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()
aibox_sophon_adapter/routes.py
+177
View File
@@ -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/<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)
# 流式写入文件（与网关 /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="")
)