vent_ai_platform/app/services/video_platform_service.py

from __future__ import annotations

import mimetypes
from collections.abc import Iterator
from datetime import datetime, time, timedelta, timezone
from pathlib import Path

from sqlalchemy import delete, update
from sqlalchemy.orm import Session

from app.core.config import settings
from app.core.errors import APIError
from app.db.models.video import (
    VIDEO_MODULE_DOOR,
    VIDEO_MODULE_LOCAL_FAN,
    VIDEO_MODULE_MAIN_FAN,
    VideoAnalysisEvent,
    VideoChannel,
    VideoRecord,
    VideoSnapshot,
)
from app.db.repositories import video_platform as video_repo
from app.schemas.video import (
    VideoChannelCreate,
    VideoChannelUpdate,
    VideoEventUpdate,
)


OPEN_EVENT_STATUSES = ("pending", "processing")

# 前端页面每个模块的固定文案与记录页签定义（列名与页面表格一致）。
# type_filter_options / scope_filter_options / snapshot_types 由数据库数据动态生成。
MODULE_DEFINITIONS: dict[str, dict] = {
    VIDEO_MODULE_DOOR: {
        "title": "风门视频图像综合解析系统",
        "status_title": "风门状态总览",
        "selector_label": "风门选择",
        "type_filter_label": "告警类型",
        "record_tabs": [
            ("vehicle_passage", "车辆通行记录", ["序号", "风门名称", "车牌号", "通行时间", "结果"]),
            ("person_passage", "人员通行记录", ["序号", "风门名称", "人员信息", "通行时间", "结果"]),
            ("door_unclosed", "风门未关严记录", ["序号", "风门名称", "检测时间", "缝隙(mm)", "等级"]),
        ],
    },
    VIDEO_MODULE_LOCAL_FAN: {
        "title": "局部风机无人值守演示系统",
        "status_title": "风机状态总览",
        "selector_label": "风机选择",
        "type_filter_label": "操作类型",
        "record_tabs": [
            ("person_track", "人员轨迹", ["序号", "风机", "区域", "姓名", "区队", "进入", "离开", "时长"]),
            ("area_stay", "区域统计", ["序号", "区域", "人数", "最长停留", "最短停留", "平均停留", "累计停留"]),
            ("person_warn", "预警记录", ["序号", "风机", "区域", "姓名", "预警类型", "预警时间", "处理状态"]),
        ],
    },
    VIDEO_MODULE_MAIN_FAN: {
        "title": "主通风机无人巡检演示系统",
        "status_title": "主风机运行状态",
        "selector_label": "设备选择",
        "type_filter_label": "操作类型",
        "record_tabs": [
            ("person_detect", "人员检测记录", ["序号", "设备", "位置", "检测时间", "检测结果", "状态"]),
            ("device_operation", "设备操作日志", ["序号", "设备", "操作类型", "操作时间", "人员比对", "状态"]),
            ("device_warn", "设备预警记录", ["序号", "设备", "位置", "预警类型", "预警时间", "处理状态"]),
        ],
    },
}

RECORD_TYPE_LABELS: dict[str, str] = {
    record_type: label
    for definition in MODULE_DEFINITIONS.values()
    for record_type, label, _ in definition["record_tabs"]
}

MODULE_TITLES: dict[str, str] = {
    key: str(value["title"]) for key, value in MODULE_DEFINITIONS.items()
}


# ---------------------------------------------------------------------------
# 通用工具
# ---------------------------------------------------------------------------


def _now() -> datetime:
    return datetime.now(timezone.utc)


def utc_day_bounds(day: datetime | None = None) -> tuple[datetime, datetime]:
    """返回本地自然日对应的 UTC 起止时间，用于“今日车辆/今日人员”等统计。"""

    local_now = (day or _now()).astimezone()
    local_start = datetime.combine(local_now.date(), time.min, tzinfo=local_now.tzinfo)
    local_end = local_start + timedelta(days=1)
    return local_start.astimezone(timezone.utc), local_end.astimezone(timezone.utc)


def _file_meta(path: str) -> tuple[bool, int]:
    if not path:
        return False, 0
    try:
        file_path = Path(path)
        if file_path.is_file():
            return True, file_path.stat().st_size
    except OSError:
        return False, 0
    return False, 0


def _api_path(path: str) -> str:
    return f"{settings.api_v1_prefix}{path}"


def parse_range_header(range_header: str | None, file_size: int) -> tuple[int, int]:
    """解析 HTTP Range 请求头，返回闭区间 ``(start, end)``。

    仅支持单段 ``bytes=start-end`` 形式；非法或越界时返回完整区间。
    """

    if not range_header or not range_header.startswith("bytes=") or file_size <= 0:
        return 0, max(file_size - 1, 0)
    spec = range_header.split("=", 1)[1].split(",", 1)[0].strip()
    if "-" not in spec:
        return 0, max(file_size - 1, 0)
    start_text, _, end_text = spec.partition("-")
    try:
        if start_text:
            start = int(start_text)
            end = int(end_text) if end_text else file_size - 1
        else:  # bytes=-N 表示最后 N 字节
            length = int(end_text or "0")
            if length <= 0:
                return 0, max(file_size - 1, 0)
            start = max(file_size - length, 0)
            end = file_size - 1
    except ValueError:
        return 0, max(file_size - 1, 0)
    start = max(start, 0)
    end = min(end, file_size - 1)
    if start > end:
        return 0, max(file_size - 1, 0)
    return start, end


def iter_file_range(path: Path, start: int, end: int) -> Iterator[bytes]:
    """按块读取磁盘文件的闭区间内容，供视频流式响应使用。"""

    chunk_size = max(int(settings.video_stream_chunk_size), 64 * 1024)
    remaining = end - start + 1
    with path.open("rb") as handle:
        handle.seek(start)
        while remaining > 0:
            chunk = handle.read(min(chunk_size, remaining))
            if not chunk:
                break
            remaining -= len(chunk)
            yield chunk


def guess_content_type(path: str) -> str:
    content_type, _ = mimetypes.guess_type(path)
    return content_type or "application/octet-stream"


# ---------------------------------------------------------------------------
# 序列化
# ---------------------------------------------------------------------------


def _channel_payload(channel: VideoChannel) -> dict:
    exists, size = _file_meta(channel.file_path or channel.stream_url)
    payload = {
        column.name: getattr(channel, column.name)
        for column in VideoChannel.__table__.columns
    }
    payload.update(
        {
            "file_size": channel.file_size or size,
            "has_video": bool(channel.stream_url or channel.file_path),
            "media_available": exists,
            "playback_url": _api_path(f"/video-platform/channels/{channel.id}/stream"),
            "snapshot_url": _api_path(f"/video-platform/channels/{channel.id}/snapshot"),
        }
    )
    return payload


def _snapshot_payload(
    snapshot: VideoSnapshot,
    *,
    channel_name: str = "",
) -> dict:
    exists, size = _file_meta(snapshot.file_path)
    payload = {
        column.name: getattr(snapshot, column.name)
        for column in VideoSnapshot.__table__.columns
    }
    payload.update(
        {
            "file_size": snapshot.file_size or size,
            "channel_name": channel_name,
            "media_available": exists,
            "image_url": _api_path(f"/video-platform/snapshots/{snapshot.id}/image"),
            "thumbnail_url": _api_path(
                f"/video-platform/snapshots/{snapshot.id}/thumbnail"
            ),
        }
    )
    return payload


def _event_payload(event: VideoAnalysisEvent, *, channel_name: str = "") -> dict:
    payload = {
        column.name: getattr(event, column.name)
        for column in VideoAnalysisEvent.__table__.columns
    }
    payload.update(
        {
            "channel_name": channel_name,
            "snapshot_image_url": (
                _api_path(f"/video-platform/snapshots/{event.snapshot_id}/image")
                if event.snapshot_id
                else ""
            ),
        }
    )
    return payload


def _record_payload(record: VideoRecord) -> dict:
    payload = {
        column.name: getattr(record, column.name)
        for column in VideoRecord.__table__.columns
    }
    payload["record_type_label"] = RECORD_TYPE_LABELS.get(record.record_type, "")
    payload["snapshot_image_url"] = (
        _api_path(f"/video-platform/snapshots/{record.snapshot_id}/image")
        if record.snapshot_id
        else ""
    )
    return payload


def _channel_names(db: Session, channels: list[VideoChannel]) -> dict[str, str]:
    return {channel.id: channel.name for channel in channels}


# ---------------------------------------------------------------------------
# 页面配置 / 总览 / 状态指标
# ---------------------------------------------------------------------------


def module_configs(db: Session, mine_id: str) -> list[dict]:
    channels = video_repo.list_channels(db, mine_id)
    configs: list[dict] = []
    for module, definition in MODULE_DEFINITIONS.items():
        module_channels = [item for item in channels if item.module == module]
        event_types = video_repo.distinct_snapshot_event_types(db, mine_id, module)
        configs.append(
            {
                "key": module,
                "title": definition["title"],
                "status_title": definition["status_title"],
                "selector_label": definition["selector_label"],
                "type_filter_label": definition["type_filter_label"],
                "type_filter_options": ["全部类型", *event_types],
                "scope_filter_label": definition["selector_label"],
                "scope_filter_options": [
                    "全部设备",
                    *[item.name for item in module_channels],
                ],
                "snapshot_types": event_types,
                "record_tabs": [
                    {
                        "record_type": record_type,
                        "label": label,
                        "columns": columns,
                    }
                    for record_type, label, columns in definition["record_tabs"]
                ],
            }
        )
    return configs


def platform_overview(db: Session, mine_id: str) -> dict:
    channel_total = video_repo.count_channels(db, mine_id)
    channel_online = video_repo.count_channels(db, mine_id, status="online")
    channel_offline = video_repo.count_channels(db, mine_id, status="offline")
    channel_fault = video_repo.count_channels(db, mine_id, status="fault")
    event_open_count = video_repo.count_events(
        db, mine_id, statuses=OPEN_EVENT_STATUSES
    )
    snapshot_count = video_repo.count_snapshots(db, mine_id)
    summary = [
        {
            "key": "channels",
            "label": "视频通道",
            "value": channel_total,
            "unit": "路",
            "hint": "视频平台接入总量",
            "tone": "default",
        },
        {
            "key": "configured",
            "label": "已配置通道",
            "value": channel_total,
            "unit": "路",
            "hint": "通道管理可维护通道",
            "tone": "online",
        },
        {
            "key": "online",
            "label": "在线通道",
            "value": channel_online,
            "unit": "路",
            "hint": "视频连接状态正常",
            "tone": "warning",
        },
        {
            "key": "events",
            "label": "识别事件",
            "value": event_open_count,
            "unit": "条",
            "hint": "未闭环智能分析事件",
            "tone": "ai",
        },
    ]
    return {
        "mine_id": mine_id,
        "channel_total": channel_total,
        "channel_configured": channel_total,
        "channel_online": channel_online,
        "channel_offline": channel_offline,
        "channel_fault": channel_fault,
        "event_open_count": event_open_count,
        "snapshot_count": snapshot_count,
        "modules": module_configs(db, mine_id),
        "summary": summary,
    }


def module_summary(db: Session, mine_id: str, module: str) -> dict:
    _require_module(module)
    today_start, today_end = utc_day_bounds()
    open_events = video_repo.count_events(
        db, mine_id, module=module, statuses=OPEN_EVENT_STATUSES
    )

    if module == VIDEO_MODULE_DOOR:
        normal = video_repo.count_channels(db, mine_id, module=module, status="online")
        offline = video_repo.count_channels(db, mine_id, module=module, status="offline")
        fault = video_repo.count_channels(db, mine_id, module=module, status="fault")
        car_today = video_repo.count_records(
            db,
            mine_id,
            module=module,
            record_type="vehicle_passage",
            start=today_start,
            end=today_end,
        )
        person_today = video_repo.sum_person_count(
            db,
            mine_id,
            module=module,
            record_type="person_passage",
            start=today_start,
            end=today_end,
        )
        metrics = [
            _metric("normal_door", "正常风门", normal, "扇", "normal"),
            _metric("offline_device", "离线设备", offline, "台", "offline"),
            _metric("fault_device", "设备故障", fault, "台", "fault"),
            _metric("open_warning", "实时预警", open_events, "条", "warning"),
            _metric("car_today", "今日车辆", car_today, "辆", "accent"),
            _metric("person_today", "今日人员", person_today, "人", "accent"),
        ]
        summary = [
            _card("normal", "正常风门", normal, "扇", tone="normal"),
            _card("offline", "离线设备", offline, "台", tone="offline"),
            _card("fault", "设备故障", fault, "台", tone="fault"),
            _card("warning", "实时预警", open_events, "条", tone="warning"),
            _card("car", "今日车辆", car_today, "辆", tone="accent"),
            _card("person", "今日人员", person_today, "人", tone="accent"),
        ]
    elif module == VIDEO_MODULE_LOCAL_FAN:
        normal = video_repo.count_channels(db, mine_id, module=module, status="online")
        offline = video_repo.count_channels(db, mine_id, module=module, status="offline")
        fault = video_repo.count_channels(db, mine_id, module=module, status="fault")
        onsite_person = video_repo.count_records(
            db, mine_id, module=module, record_type="person_track", in_progress_only=True
        )
        enter_today = video_repo.count_records(
            db,
            mine_id,
            module=module,
            record_type="person_track",
            start=today_start,
            end=today_end,
        )
        metrics = [
            _metric("normal_fan", "风机正常", normal, "台", "normal"),
            _metric("offline_fan", "风机离线", offline, "台", "offline"),
            _metric("fault_fan", "风机故障", fault, "台", "fault"),
            _metric("open_warning", "实时预警", open_events, "条", "warning"),
            _metric("onsite_person", "实时人数", onsite_person, "人", "accent"),
            _metric("enter_today", "今日进入", enter_today, "人", "accent"),
        ]
        summary = [
            _card("normal", "风机正常", normal, "台", tone="normal"),
            _card("offline", "风机离线", offline, "台", tone="offline"),
            _card("fault", "风机故障", fault, "台", tone="fault"),
            _card("warning", "实时预警", open_events, "条", tone="warning"),
            _card("onsite", "实时人数", onsite_person, "人", tone="accent"),
            _card("enter", "今日进入", enter_today, "人", tone="accent"),
        ]
    else:  # VIDEO_MODULE_MAIN_FAN
        channels = video_repo.list_channels(db, mine_id, module=module)
        running = sum(
            1
            for item in channels
            if str(item.properties.get("run_state", "")) == "running"
        )
        stopped = sum(
            1
            for item in channels
            if str(item.properties.get("run_state", "")) == "stopped"
        )
        fault = video_repo.count_channels(db, mine_id, module=module, status="fault")
        detect_today = video_repo.count_records(
            db,
            mine_id,
            module=module,
            record_type="person_detect",
            start=today_start,
            end=today_end,
        )
        records_today = video_repo.count_records(
            db, mine_id, module=module, start=today_start, end=today_end
        )
        metrics = [
            _metric("running_fan", "运行风机", running, "台", "normal"),
            _metric("stopped_fan", "停止风机", stopped, "台", "offline"),
            _metric("fault_device", "设备故障", fault, "台", "fault"),
            _metric("open_warning", "预警总数", open_events, "条", "warning"),
            _metric("person_detect", "人员检测", detect_today, "次", "accent"),
            _metric("records_today", "今日记录", records_today, "条", "accent"),
        ]
        summary = [
            _card("running", "运行风机", running, "台", tone="normal"),
            _card("stopped", "停止风机", stopped, "台", tone="offline"),
            _card("fault", "设备故障", fault, "台", tone="fault"),
            _card("warning", "预警总数", open_events, "条", tone="warning"),
            _card("detect", "人员检测", detect_today, "次", tone="accent"),
            _card("records", "今日记录", records_today, "条", tone="accent"),
        ]

    return {
        "mine_id": mine_id,
        "module": module,
        "title": MODULE_TITLES[module],
        "status_title": MODULE_DEFINITIONS[module]["status_title"],
        "metrics": metrics,
        "summary": summary,
    }


def _metric(key: str, label: str, value: int | float, unit: str, tone: str) -> dict:
    return {"key": key, "label": label, "value": value, "unit": unit, "tone": tone}


def _card(key: str, label: str, value: int | float, unit: str, tone: str) -> dict:
    return {"key": key, "label": label, "value": value, "unit": unit, "tone": tone}


def _require_module(module: str) -> str:
    if module not in MODULE_DEFINITIONS:
        raise APIError(400, "invalid_module", f"不支持的视频模块：{module}")
    return module


# ---------------------------------------------------------------------------
# 查询接口
# ---------------------------------------------------------------------------


def list_channels(
    db: Session,
    mine_id: str,
    *,
    module: str | None = None,
    status: str | None = None,
    keyword: str | None = None,
    limit: int = 200,
) -> list[dict]:
    channels = video_repo.list_channels(
        db, mine_id, module=module, status=status, keyword=keyword, limit=limit
    )
    return [_channel_payload(channel) for channel in channels]


def get_channel(db: Session, channel_id: str) -> dict:
    channel = video_repo.get_channel(db, channel_id)
    if channel is None:
        raise APIError(404, "video_channel_not_found", "视频通道不存在")
    return _channel_payload(channel)


def list_snapshots(
    db: Session,
    mine_id: str,
    *,
    module: str | None = None,
    snapshot_type: str | None = None,
    event_type: str | None = None,
    channel_id: str | None = None,
    start: datetime | None = None,
    end: datetime | None = None,
    keyword: str | None = None,
    page: int = 1,
    page_size: int = 24,
) -> tuple[list[dict], int]:
    items, total = video_repo.list_snapshots(
        db,
        mine_id,
        module=module,
        snapshot_type=snapshot_type,
        event_type=event_type,
        channel_id=channel_id,
        start=start,
        end=end,
        keyword=keyword,
        page=page,
        page_size=page_size,
    )
    names = _channel_names(db, video_repo.list_channels(db, mine_id))
    return (
        [_snapshot_payload(item, channel_name=names.get(item.channel_id, "")) for item in items],
        total,
    )


def get_snapshot(db: Session, snapshot_id: str) -> dict:
    snapshot = video_repo.get_snapshot(db, snapshot_id)
    if snapshot is None:
        raise APIError(404, "video_snapshot_not_found", "识别抓拍图像不存在")
    channel = video_repo.get_channel(db, snapshot.channel_id)
    return _snapshot_payload(
        snapshot, channel_name=channel.name if channel else ""
    )


def list_events(
    db: Session,
    mine_id: str,
    *,
    module: str | None = None,
    level: str | None = None,
    status: str | None = None,
    event_type: str | None = None,
    start: datetime | None = None,
    end: datetime | None = None,
    limit: int = 100,
) -> list[dict]:
    events = video_repo.list_events(
        db,
        mine_id,
        module=module,
        level=level,
        status=status,
        event_type=event_type,
        start=start,
        end=end,
        limit=limit,
    )
    names = _channel_names(db, video_repo.list_channels(db, mine_id))
    return [
        _event_payload(event, channel_name=names.get(event.channel_id or "", ""))
        for event in events
    ]


def update_event(db: Session, event_id: str, payload: VideoEventUpdate) -> dict:
    event = video_repo.get_event(db, event_id)
    if event is None:
        raise APIError(404, "video_event_not_found", "视频分析事件不存在")
    data = payload.model_dump(exclude_unset=True, exclude_none=True)
    if "status" in data:
        event.status = str(data["status"])
        if event.status in {"handled", "ignored"}:
            event.handled_at = _now()
        else:
            event.handled_at = None
    if "handled_by" in data:
        event.handled_by = str(data["handled_by"])
    if "handle_remark" in data:
        event.handle_remark = str(data["handle_remark"])
    db.add(event)
    db.commit()
    db.refresh(event)
    channel = video_repo.get_channel(db, event.channel_id or "")
    return _event_payload(event, channel_name=channel.name if channel else "")


def list_records(
    db: Session,
    mine_id: str,
    *,
    module: str | None = None,
    record_type: str | None = None,
    channel_id: str | None = None,
    start: datetime | None = None,
    end: datetime | None = None,
    keyword: str | None = None,
    page: int = 1,
    page_size: int = 50,
) -> tuple[list[dict], int]:
    items, total = video_repo.list_records(
        db,
        mine_id,
        module=module,
        record_type=record_type,
        channel_id=channel_id,
        start=start,
        end=end,
        keyword=keyword,
        page=page,
        page_size=page_size,
    )
    return [_record_payload(item) for item in items], total


# ---------------------------------------------------------------------------
# 通道管理
# ---------------------------------------------------------------------------


def create_channel(db: Session, payload: VideoChannelCreate) -> dict:
    _require_module(payload.module)
    if video_repo.get_channel_by_code(db, payload.code) is not None:
        raise APIError(409, "video_channel_code_exists", "视频通道编码已存在")
    channel = VideoChannel(**payload.model_dump())
    channel.file_name = Path(payload.file_path).name if payload.file_path else ""
    exists, size = _file_meta(payload.file_path)
    channel.file_size = size if exists else 0
    channel.last_online_at = _now() if channel.status == "online" else None
    db.add(channel)
    db.commit()
    db.refresh(channel)
    return _channel_payload(channel)


def update_channel(
    db: Session,
    channel_id: str,
    payload: VideoChannelUpdate,
) -> dict:
    channel = video_repo.get_channel(db, channel_id)
    if channel is None:
        raise APIError(404, "video_channel_not_found", "视频通道不存在")
    data = payload.model_dump(exclude_unset=True)
    if "module" in data and data["module"] is not None:
        _require_module(str(data["module"]))
    for key, value in data.items():
        if value is not None:
            setattr(channel, key, value)
    if "file_path" in data and data["file_path"] is not None:
        channel.file_name = Path(str(data["file_path"])).name if data["file_path"] else ""
        exists, size = _file_meta(str(data["file_path"]))
        channel.file_size = size if exists else 0
    if "status" in data and data["status"] == "online":
        channel.last_online_at = _now()
    db.add(channel)
    db.commit()
    db.refresh(channel)
    return _channel_payload(channel)


def delete_channel(db: Session, channel_id: str) -> None:
    channel = video_repo.get_channel(db, channel_id)
    if channel is None:
        raise APIError(404, "video_channel_not_found", "视频通道不存在")
    # 先断开关联引用，再删除通道及其抓拍/记录，避免事件出现悬空外键。
    db.execute(
        update(VideoAnalysisEvent)
        .where(VideoAnalysisEvent.channel_id == channel_id)
        .values(channel_id=None)
    )
    db.execute(
        update(VideoAnalysisEvent)
        .where(
            VideoAnalysisEvent.snapshot_id.in_(
                db.query(VideoSnapshot.id).filter(VideoSnapshot.channel_id == channel_id)
            )
        )
        .values(snapshot_id=None)
    )
    db.execute(delete(VideoRecord).where(VideoRecord.channel_id == channel_id))
    db.execute(delete(VideoSnapshot).where(VideoSnapshot.channel_id == channel_id))
    db.delete(channel)
    db.commit()


# ---------------------------------------------------------------------------
# 媒体文件
# ---------------------------------------------------------------------------


def resolve_channel_media(channel_id: str, db: Session) -> tuple[Path, str]:
    channel = video_repo.get_channel(db, channel_id)
    if channel is None:
        raise APIError(404, "video_channel_not_found", "视频通道不存在")
    path_text = channel.file_path or channel.stream_url
    if not path_text:
        raise APIError(404, "video_media_missing", "该通道未配置可播放的视频文件")
    path = Path(path_text)
    if not path.is_file():
        raise APIError(
            404,
            "video_media_missing",
            f"视频文件不存在：{path_text}",
        )
    return path, guess_content_type(path.name)


def resolve_snapshot_media(snapshot_id: str, db: Session) -> tuple[Path, str]:
    snapshot = video_repo.get_snapshot(db, snapshot_id)
    if snapshot is None:
        raise APIError(404, "video_snapshot_not_found", "识别抓拍图像不存在")
    path_text = snapshot.file_path
    if not path_text:
        raise APIError(404, "snapshot_media_missing", "该抓拍记录未保存图像文件")
    path = Path(path_text)
    if not path.is_file():
        raise APIError(404, "snapshot_media_missing", f"图像文件不存在：{path_text}")
    return path, guess_content_type(path.name)


def resolve_snapshot_thumbnail(snapshot_id: str, db: Session) -> tuple[Path, str]:
    snapshot = video_repo.get_snapshot(db, snapshot_id)
    if snapshot is None:
        raise APIError(404, "video_snapshot_not_found", "识别抓拍图像不存在")
    path_text = snapshot.thumbnail_path or snapshot.file_path
    if not path_text:
        raise APIError(404, "snapshot_media_missing", "该抓拍记录未保存图像文件")
    path = Path(path_text)
    if not path.is_file():
        return resolve_snapshot_media(snapshot_id, db)
    return path, guess_content_type(path.name)


def resolve_channel_latest_snapshot(channel_id: str, db: Session) -> tuple[Path, str]:
    snapshot = video_repo.latest_snapshot(db, channel_id)
    if snapshot is None:
        raise APIError(404, "video_snapshot_not_found", "该通道暂无识别抓拍图像")
    return resolve_snapshot_media(snapshot.id, db)