chongming-plugins/aibox_sophon_adapter/router/get_alarm.py

"""
查询单条告警详情 — GET /aibox/get_alarm?alarm_id=xxx
"""
import orjson
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 = orjson.loads(record.alarm_data.encode("utf-8")) if record.alarm_data else {}
            except orjson.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)}