vent_ai_platform/app/db/models/video.py

from __future__ import annotations

from datetime import datetime

from sqlalchemy import (
    JSON,
    BigInteger,
    Boolean,
    DateTime,
    Float,
    ForeignKey,
    Integer,
    String,
)
from sqlalchemy.orm import Mapped, mapped_column

from app.db.base import Base, TimestampMixin, UUIDPrimaryKeyMixin


# 工业视频平台业务模块：风门 / 局部风机 / 主风机及配电室
VIDEO_MODULE_DOOR = "door"
VIDEO_MODULE_LOCAL_FAN = "local_fan"
VIDEO_MODULE_MAIN_FAN = "main_fan"

VIDEO_MODULES = (VIDEO_MODULE_DOOR, VIDEO_MODULE_LOCAL_FAN, VIDEO_MODULE_MAIN_FAN)


class VideoChannel(UUIDPrimaryKeyMixin, TimestampMixin, Base):
    """视频通道：设备树、分屏预览和实时视频对应的摄像机/视频流。

    实时流地址保存在 ``stream_url``，演示视频以磁盘路径保存在 ``file_path``。
    """

    __tablename__ = "video_channels"

    mine_id: Mapped[str] = mapped_column(
        String(32), ForeignKey("mines.id", ondelete="CASCADE"), index=True
    )
    data_source_id: Mapped[str | None] = mapped_column(
        String(32),
        ForeignKey("data_sources.id", ondelete="SET NULL"),
        nullable=True,
        index=True,
    )
    code: Mapped[str] = mapped_column(String(64), unique=True, index=True)
    name: Mapped[str] = mapped_column(String(128))
    module: Mapped[str] = mapped_column(String(32), index=True)
    area_group: Mapped[str] = mapped_column(String(64), default="")
    location: Mapped[str] = mapped_column(String(255), default="")
    device_type: Mapped[str] = mapped_column(String(32), default="")
    point_code: Mapped[str] = mapped_column(String(64), default="")
    protocol: Mapped[str] = mapped_column(String(64), default="GB/T 28181")
    stream_url: Mapped[str] = mapped_column(String(512), default="")
    status: Mapped[str] = mapped_column(String(32), default="online", index=True)
    file_path: Mapped[str] = mapped_column(String(512), default="")
    file_name: Mapped[str] = mapped_column(String(255), default="")
    file_size: Mapped[int] = mapped_column(BigInteger, default=0)
    duration_seconds: Mapped[int | None] = mapped_column(Integer, nullable=True)
    resolution: Mapped[str] = mapped_column(String(32), default="")
    frame_rate: Mapped[int | None] = mapped_column(Integer, nullable=True)
    preview_image_path: Mapped[str] = mapped_column(String(512), default="")
    is_enabled: Mapped[bool] = mapped_column(Boolean, default=True)
    sort_order: Mapped[int] = mapped_column(Integer, default=0)
    last_online_at: Mapped[datetime | None] = mapped_column(
        DateTime(timezone=True), nullable=True
    )
    properties: Mapped[dict] = mapped_column(JSON, default=dict)


class VideoSnapshot(UUIDPrimaryKeyMixin, TimestampMixin, Base):
    """AI 识别抓拍图像：车辆/人员通行、风门未关严、人员检测等。

    ``file_path`` 保存图像在磁盘上的绝对路径，接口按路径读取原图。
    """

    __tablename__ = "video_snapshots"

    mine_id: Mapped[str] = mapped_column(
        String(32), ForeignKey("mines.id", ondelete="CASCADE"), index=True
    )
    channel_id: Mapped[str] = mapped_column(
        String(32),
        ForeignKey("video_channels.id", ondelete="CASCADE"),
        index=True,
    )
    module: Mapped[str] = mapped_column(String(32), index=True)
    snapshot_type: Mapped[str] = mapped_column(String(32), index=True)
    event_type: Mapped[str] = mapped_column(String(64), default="", index=True)
    level: Mapped[str] = mapped_column(String(16), default="info")
    title: Mapped[str] = mapped_column(String(128), default="")
    file_path: Mapped[str] = mapped_column(String(512), default="")
    file_name: Mapped[str] = mapped_column(String(255), default="")
    file_size: Mapped[int] = mapped_column(BigInteger, default=0)
    width: Mapped[int | None] = mapped_column(Integer, nullable=True)
    height: Mapped[int | None] = mapped_column(Integer, nullable=True)
    thumbnail_path: Mapped[str] = mapped_column(String(512), default="")
    plate_no: Mapped[str] = mapped_column(String(32), default="")
    person_name: Mapped[str] = mapped_column(String(64), default="")
    person_count: Mapped[int] = mapped_column(Integer, default=0)
    team_name: Mapped[str] = mapped_column(String(64), default="")
    gap_mm: Mapped[float | None] = mapped_column(Float, nullable=True)
    confidence: Mapped[float | None] = mapped_column(Float, nullable=True)
    captured_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), index=True)
    properties: Mapped[dict] = mapped_column(JSON, default=dict)


class VideoAnalysisEvent(UUIDPrimaryKeyMixin, TimestampMixin, Base):
    """视频智能分析事件与实时告警（防夹人、卡堵、密封失效、违规停留等）。"""

    __tablename__ = "video_analysis_events"

    mine_id: Mapped[str] = mapped_column(
        String(32), ForeignKey("mines.id", ondelete="CASCADE"), index=True
    )
    channel_id: Mapped[str | None] = mapped_column(
        String(32),
        ForeignKey("video_channels.id", ondelete="SET NULL"),
        nullable=True,
        index=True,
    )
    snapshot_id: Mapped[str | None] = mapped_column(
        String(32),
        ForeignKey("video_snapshots.id", ondelete="SET NULL"),
        nullable=True,
        index=True,
    )
    module: Mapped[str] = mapped_column(String(32), index=True)
    event_type: Mapped[str] = mapped_column(String(64), index=True)
    category: Mapped[str] = mapped_column(String(32), default="")
    level: Mapped[str] = mapped_column(String(16), default="info", index=True)
    title: Mapped[str] = mapped_column(String(128))
    description: Mapped[str] = mapped_column(String(512), default="")
    confidence: Mapped[float | None] = mapped_column(Float, nullable=True)
    status: Mapped[str] = mapped_column(String(32), default="pending", index=True)
    occurred_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), index=True)
    handled_at: Mapped[datetime | None] = mapped_column(
        DateTime(timezone=True), nullable=True
    )
    handled_by: Mapped[str] = mapped_column(String(64), default="")
    handle_remark: Mapped[str] = mapped_column(String(255), default="")
    properties: Mapped[dict] = mapped_column(JSON, default=dict)


class VideoRecord(UUIDPrimaryKeyMixin, TimestampMixin, Base):
    """通行与操作记录：三大模块底部记录面板的统一宽表。

    ``record_type`` 取值：
    ``vehicle_passage`` 车辆通行、``person_passage`` 人员通行、``door_unclosed`` 风门未关严、
    ``person_track`` 人员轨迹、``area_stay`` 区域统计、``person_warn`` 预警记录、
    ``person_detect`` 人员检测、``device_operation`` 设备操作、``device_warn`` 设备预警。
    """

    __tablename__ = "video_records"

    mine_id: Mapped[str] = mapped_column(
        String(32), ForeignKey("mines.id", ondelete="CASCADE"), index=True
    )
    channel_id: Mapped[str] = mapped_column(
        String(32),
        ForeignKey("video_channels.id", ondelete="CASCADE"),
        index=True,
    )
    snapshot_id: Mapped[str | None] = mapped_column(
        String(32),
        ForeignKey("video_snapshots.id", ondelete="SET NULL"),
        nullable=True,
        index=True,
    )
    module: Mapped[str] = mapped_column(String(32), index=True)
    record_type: Mapped[str] = mapped_column(String(32), index=True)
    device_name: Mapped[str] = mapped_column(String(128), default="")
    area_name: Mapped[str] = mapped_column(String(64), default="")
    location: Mapped[str] = mapped_column(String(255), default="")
    plate_no: Mapped[str] = mapped_column(String(32), default="")
    person_name: Mapped[str] = mapped_column(String(64), default="")
    person_count: Mapped[int] = mapped_column(Integer, default=0)
    team_name: Mapped[str] = mapped_column(String(64), default="")
    operation_type: Mapped[str] = mapped_column(String(64), default="")
    compare_result: Mapped[str] = mapped_column(String(64), default="")
    warn_type: Mapped[str] = mapped_column(String(64), default="")
    result: Mapped[str] = mapped_column(String(64), default="")
    handle_status: Mapped[str] = mapped_column(String(32), default="")
    device_status: Mapped[str] = mapped_column(String(32), default="")
    level: Mapped[str] = mapped_column(String(16), default="")
    gap_mm: Mapped[float | None] = mapped_column(Float, nullable=True)
    stay_seconds: Mapped[int | None] = mapped_column(Integer, nullable=True)
    enter_time: Mapped[datetime | None] = mapped_column(
        DateTime(timezone=True), nullable=True
    )
    leave_time: Mapped[datetime | None] = mapped_column(
        DateTime(timezone=True), nullable=True
    )
    occurred_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), index=True)
    properties: Mapped[dict] = mapped_column(JSON, default=dict)