vent_ai_platform/app/db/seed_video_platform.py

"""工业视频平台模拟演示数据。

视频与图像取自 ``settings.video_media_root``（默认 D:/workspace/js/gate_analysis_system，
即风门/局扇/主扇三个识别演示系统的公开资源目录），数据库只保存磁盘路径与业务属性。
"""

from __future__ import annotations

from datetime import datetime, time, timedelta, timezone

from sqlalchemy import func, select
from sqlalchemy.orm import Session

from app.core.config import settings
from app.db.base import new_id
from app.db.models.mine import Mine
from app.db.models.monitoring import DataSource
from app.db.models.video import (
    VIDEO_MODULE_DOOR,
    VIDEO_MODULE_LOCAL_FAN,
    VIDEO_MODULE_MAIN_FAN,
    VideoAnalysisEvent,
    VideoChannel,
    VideoRecord,
    VideoSnapshot,
)


def _media_path(*parts: str) -> str:
    """拼出媒体文件的磁盘绝对路径（不要求文件真实存在）。"""

    return str(settings.video_media_path.joinpath(*parts))


def _file_size(path: str) -> int:
    try:
        from pathlib import Path

        file_path = Path(path)
        if file_path.is_file():
            return file_path.stat().st_size
    except OSError:
        return 0
    return 0


# (code, name, module, area_group, location, device_type, status,
#  video_relative_path, resolution, preview_relative_path, extra_properties)
CHANNEL_SPECS: list[tuple] = [
    (
        "VC-DOOR-01", "五盘区402辅运大巷风门", VIDEO_MODULE_DOOR, "风门区域",
        "五盘区402辅运大巷", "door", "online",
        ("public", "videos", "v1.mp4"), "1920x1080",
        ("public", "images", "open.jpg"), {"door_side": "前门"},
    ),
    (
        "VC-DOOR-02", "六盘区501胶带运输巷风门", VIDEO_MODULE_DOOR, "风门区域",
        "六盘区501胶带运输巷", "door", "online",
        ("public", "videos", "v2.mp4"), "1920x1080",
        ("public", "images", "open1.jpg"), {"door_side": "后门"},
    ),
    (
        "VC-DOOR-03", "七盘区602回风联络巷风门", VIDEO_MODULE_DOOR, "风门区域",
        "七盘区602回风联络巷", "door", "online",
        ("public", "videos", "v1.mp4"), "1920x1080",
        ("public", "images", "gap1.jpg"), {"door_side": "前门"},
    ),
    (
        "VC-DOOR-04", "中央变电所进风通道风门", VIDEO_MODULE_DOOR, "风门区域",
        "中央变电所", "door", "online",
        ("public", "videos", "v2.mp4"), "1920x1080",
        ("public", "images", "person1.jpg"), {"door_side": "后门"},
    ),
    (
        "VC-DOOR-05", "采区水泵房专用通道风门", VIDEO_MODULE_DOOR, "风门区域",
        "采区水泵房", "door", "online",
        ("public", "videos", "si.mp4"), "2560x1440",
        ("public", "images", "car1.jpg"), {"door_side": "前门", "ai_anti_clip": True},
    ),
    (
        "VC-DOOR-06", "南风井绕道处风门", VIDEO_MODULE_DOOR, "风门区域",
        "南风井绕道", "door", "online",
        ("public", "videos", "b1.mp4"), "1920x1080",
        ("public", "images", "20260315_192656.jpg"), {"door_side": "前门"},
    ),
    (
        "VC-DOOR-07", "12煤四盘区回风入口风门", VIDEO_MODULE_DOOR, "风门区域",
        "12煤四盘区", "door", "fault",
        ("public", "videos", "b2.mp4"), "1920x1080",
        ("public", "images", "gap17.jpg"), {"fault_reason": "门体闭合反馈异常"},
    ),
    (
        "VC-DOOR-08", "22107胶运绕道行车风门", VIDEO_MODULE_DOOR, "风门区域",
        "22107胶运绕道", "door", "offline",
        ("public", "videos", "4.mp4"), "1920x1080",
        ("public", "images", "open15.jpg"), {"offline_reason": "网络中断"},
    ),
    (
        "VC-LFAN-01", "1号掘进工作面局部风机-变频器区", VIDEO_MODULE_LOCAL_FAN, "局扇区域",
        "1号掘进工作面", "local_fan", "online",
        ("public", "videos", "fan-main.mp4"), "1920x1080",
        ("public", "images", "fan1.png"), {"area_name": "变频器区"},
    ),
    (
        "VC-LFAN-02", "1号掘进工作面局部风机-配电柜区", VIDEO_MODULE_LOCAL_FAN, "局扇区域",
        "1号掘进工作面", "local_fan", "online",
        ("public", "videos", "fan-main.mp4"), "1920x1080",
        ("public", "images", "fan2.png"), {"area_name": "配电柜区"},
    ),
    (
        "VC-LFAN-03", "2号掘进工作面局部风机", VIDEO_MODULE_LOCAL_FAN, "局扇区域",
        "2号掘进工作面", "local_fan", "online",
        ("public", "videos", "fan-main.mp4"), "1920x1080",
        ("public", "images", "fan3.png"), {"area_name": "风机本体区"},
    ),
    (
        "VC-LFAN-04", "3号掘进工作面局部风机", VIDEO_MODULE_LOCAL_FAN, "局扇区域",
        "3号掘进工作面", "local_fan", "online",
        ("public", "videos", "fan-main.mp4"), "1920x1080",
        ("public", "images", "fan4.png"), {"area_name": "接线盒区"},
    ),
    (
        "VC-LFAN-05", "4号掘进工作面局部风机", VIDEO_MODULE_LOCAL_FAN, "局扇区域",
        "4号掘进工作面", "local_fan", "fault",
        ("public", "videos", "fan-main.mp4"), "1920x1080",
        ("public", "images", "fan5.png"), {"area_name": "散热区", "fault_reason": "风机振动超限"},
    ),
    (
        "VC-MFAN-01", "一号主风机-风机主体", VIDEO_MODULE_MAIN_FAN, "主扇区域",
        "一号主通风机", "main_fan", "online",
        ("public", "videos", "f.mp4"), "1920x1080",
        ("public", "images", "f1.png"), {"run_state": "running", "air_volume": 8600},
    ),
    (
        "VC-MFAN-02", "一号主风机-配电室", VIDEO_MODULE_MAIN_FAN, "主扇区域",
        "一号主风机配电室", "distribution_room", "online",
        ("public", "videos", "p.mp4"), "1920x1080",
        ("public", "images", "20260315_211358.jpg"), {"key_device": "PLC 配电柜"},
    ),
    (
        "VC-MFAN-03", "二号主风机-风机主体", VIDEO_MODULE_MAIN_FAN, "主扇区域",
        "二号主通风机", "main_fan", "online",
        ("public", "videos", "f.mp4"), "1920x1080",
        ("public", "images", "f2.png"), {"run_state": "stopped", "air_volume": 0},
    ),
    (
        "VC-MFAN-04", "主扇配电室", VIDEO_MODULE_MAIN_FAN, "主扇区域",
        "主通风机配电室", "distribution_room", "online",
        ("public", "videos", "p.mp4"), "1920x1080",
        ("public", "images", "20260315_213411.jpg"), {"key_device": "主扇变频柜"},
    ),
    (
        "VC-MFAN-05", "主扇风硐监测点", VIDEO_MODULE_MAIN_FAN, "主扇区域",
        "主通风机风硐", "main_fan", "online",
        ("public", "videos", "s.mp4"), "1920x1080",
        ("public", "images", "20260315_220400.jpg"), {"monitor_point": "风硐"},
    ),
]


def _spot(now: datetime, minutes: int) -> datetime:
    """now 往前 minutes 分钟，并保证落在“本地今天”之内（凌晨播种时自动收敛）。"""

    local_now = now.astimezone()
    local_start = datetime.combine(local_now.date(), time.min, tzinfo=local_now.tzinfo)
    since_midnight = int((local_now - local_start).total_seconds() // 60)
    capped = min(minutes, max(since_midnight - 2, 0))
    return now - timedelta(minutes=capped)


def seed_video_platform_demo(db: Session) -> dict:
    """写入工业视频平台演示数据，已存在通道时跳过。"""

    mine = db.scalar(
        select(Mine).where(Mine.code == settings.default_mine_code)
    ) or db.scalar(select(Mine).order_by(Mine.created_at).limit(1))
    if mine is None:
        return {
            "created": False,
            "reason": "mine_not_found",
            "channels": 0,
            "snapshots": 0,
            "records": 0,
            "events": 0,
        }

    existing = int(
        db.scalar(
            select(func.count(VideoChannel.id)).where(VideoChannel.mine_id == mine.id)
        )
        or 0
    )
    if existing > 0:
        return {
            "created": False,
            "reason": "already_seeded",
            "channels": existing,
            "snapshots": int(
                db.scalar(
                    select(func.count(VideoSnapshot.id)).where(
                        VideoSnapshot.mine_id == mine.id
                    )
                )
                or 0
            ),
            "records": int(
                db.scalar(
                    select(func.count(VideoRecord.id)).where(
                        VideoRecord.mine_id == mine.id
                    )
                )
                or 0
            ),
            "events": int(
                db.scalar(
                    select(func.count(VideoAnalysisEvent.id)).where(
                        VideoAnalysisEvent.mine_id == mine.id
                    )
                )
                or 0
            ),
        }

    now = datetime.now(timezone.utc)
    data_source = _ensure_video_source(db, mine)
    channels = _seed_channels(db, mine, data_source, now)
    snapshots = _seed_snapshots(db, mine, channels, now)
    records = _seed_records(db, mine, channels, snapshots, now)
    events = _seed_events(db, mine, channels, snapshots, now)
    db.commit()
    return {
        "created": True,
        "mine_code": mine.code,
        "channels": len(channels),
        "snapshots": len(snapshots),
        "records": len(records),
        "events": len(events),
        "media_root": str(settings.video_media_path),
    }


def _ensure_video_source(db: Session, mine: Mine) -> DataSource | None:
    source = db.scalar(
        select(DataSource).where(
            DataSource.mine_id == mine.id, DataSource.source_type == "video"
        )
    )
    if source is not None:
        return source
    source = db.scalar(
        select(DataSource).where(
            DataSource.mine_id == mine.id, DataSource.code == "VIDEO"
        )
    )
    if source is not None:
        return source
    source = DataSource(
        id=new_id(),
        mine_id=mine.id,
        code="VIDEO",
        name="工业视频平台",
        source_type="video",
        protocol="GB/T 28181",
        endpoint="10.20.18.10:5060",
        status="online",
        last_sync_at=datetime.now(timezone.utc),
        point_count=218,
        latency_ms=820,
        config={"channels": 218, "media_root": str(settings.video_media_path)},
    )
    db.add(source)
    db.flush()
    return source


def _seed_channels(
    db: Session,
    mine: Mine,
    data_source: DataSource | None,
    now: datetime,
) -> dict[str, VideoChannel]:
    channels: dict[str, VideoChannel] = {}
    for index, spec in enumerate(CHANNEL_SPECS, start=1):
        (
            code, name, module, area_group, location, device_type, status,
            video_parts, resolution, preview_parts, extra,
        ) = spec
        file_path = _media_path(*video_parts)
        channel = VideoChannel(
            id=new_id(),
            mine_id=mine.id,
            data_source_id=data_source.id if data_source else None,
            code=code,
            name=name,
            module=module,
            area_group=area_group,
            location=location,
            device_type=device_type,
            point_code=f"3402000000132002{index:04d}",
            protocol="GB/T 28181",
            stream_url=f"rtsp://10.20.18.10:554/Streaming/Channels/{index:02d}01",
            status=status,
            file_path=file_path,
            file_name=video_parts[-1],
            file_size=_file_size(file_path),
            duration_seconds=None,
            resolution=resolution,
            frame_rate=25,
            preview_image_path=_media_path(*preview_parts),
            is_enabled=True,
            sort_order=index,
            last_online_at=now - timedelta(seconds=30 * index)
            if status == "online"
            else now - timedelta(days=2),
            properties=dict(extra),
        )
        db.add(channel)
        channels[code] = channel
    db.flush()
    return channels


def _snapshot(
    *,
    mine_id: str,
    channel: VideoChannel,
    snapshot_type: str,
    event_type: str,
    level: str,
    image_parts: tuple[str, ...],
    captured_at: datetime,
    title: str = "",
    plate_no: str = "",
    person_name: str = "",
    person_count: int = 0,
    team_name: str = "",
    gap_mm: float | None = None,
    confidence: float | None = None,
    properties: dict | None = None,
) -> VideoSnapshot:
    file_path = _media_path(*image_parts)
    return VideoSnapshot(
        id=new_id(),
        mine_id=mine_id,
        channel_id=channel.id,
        module=channel.module,
        snapshot_type=snapshot_type,
        event_type=event_type,
        level=level,
        title=title or event_type,
        file_path=file_path,
        file_name=image_parts[-1],
        file_size=_file_size(file_path),
        width=None,
        height=None,
        thumbnail_path="",
        plate_no=plate_no,
        person_name=person_name,
        person_count=person_count,
        team_name=team_name,
        gap_mm=gap_mm,
        confidence=confidence,
        captured_at=captured_at,
        properties=properties or {},
    )


def _seed_snapshots(
    db: Session,
    mine: Mine,
    channels: dict[str, VideoChannel],
    now: datetime,
) -> dict[str, VideoSnapshot]:
    rows: list[VideoSnapshot] = []
    by_key: dict[str, VideoSnapshot] = {}

    door_codes = [f"VC-DOOR-{index:02d}" for index in range(1, 6)]
    local_fan_codes = [f"VC-LFAN-{index:02d}" for index in range(1, 6)]
    main_fan_codes = ["VC-MFAN-01", "VC-MFAN-02", "VC-MFAN-03", "VC-MFAN-04"]

    # 1. 风门：车辆通行抓拍
    vehicle_samples = [
        ("car1.jpg", "蒙K3571", 12), ("car2.jpg", "蒙K6829", 29),
        ("car3.jpg", "蒙K1122", 46), ("car4.jpg", "蒙K3344", 63),
        ("car5.jpg", "蒙K5566", 80), ("car6.jpg", "蒙K7719", 97),
        ("car7.jpg", "蒙K2288", 114), ("car8.jpg", "蒙K9966", 131),
        ("car9.jpg", "蒙K4408", 148), ("car10.jpg", "蒙K5533", 165),
        ("car11.jpg", "蒙K8842", 182), ("car12.jpg", "蒙K6317", 199),
    ]
    for index, (image, plate, minutes) in enumerate(vehicle_samples):
        channel = channels[door_codes[index % len(door_codes)]]
        snapshot = _snapshot(
            mine_id=mine.id,
            channel=channel,
            snapshot_type="vehicle_passage",
            event_type="车辆通行",
            level="info",
            image_parts=("public", "images", image),
            captured_at=_spot(now, minutes),
            title=f"{channel.location} 车辆通行 {plate}",
            plate_no=plate,
            confidence=0.96,
        )
        rows.append(snapshot)
        by_key[f"vehicle-{index + 1}"] = snapshot

    # 2. 风门：人员通行抓拍
    person_samples = [
        ("person1.jpg", "张三、李四", 2, "掘进一队", 20),
        ("person2.jpg", "王五", 1, "机电队", 38),
        ("person3.jpg", "赵六、陈七", 2, "回采一队", 55),
        ("open1.jpg", "周八", 1, "机电队", 72),
        ("open11.jpg", "吴九", 1, "维修队", 90),
        ("open12.jpg", "郑十、刘一", 2, "掘进二队", 108),
        ("open13.jpg", "孙十一", 1, "通风队", 126),
        ("open14.jpg", "马十二", 1, "维修队", 145),
    ]
    for index, (image, person, count, team, minutes) in enumerate(person_samples):
        channel = channels[door_codes[index % len(door_codes)]]
        snapshot = _snapshot(
            mine_id=mine.id,
            channel=channel,
            snapshot_type="person_passage",
            event_type="人员通行",
            level="info",
            image_parts=("public", "images", image),
            captured_at=_spot(now, minutes),
            title=f"{channel.location} 人员通行 {person}",
            person_name=person,
            person_count=count,
            team_name=team,
            confidence=0.94,
        )
        rows.append(snapshot)
        by_key[f"person-{index + 1}"] = snapshot

    # 3. 风门：未关严抓拍
    gap_samples = [
        ("gap1.jpg", 15, 5), ("gap2.jpg", 10, 24), ("gap3.jpg", 12, 43),
        ("gap4.jpg", 8, 60), ("gap5.jpg", 9, 78), ("gap6.jpg", 11, 96),
        ("gap7.jpg", 14, 115), ("gap8.jpg", 7, 133), ("gap10.jpg", 16, 152),
    ]
    for index, (image, gap, minutes) in enumerate(gap_samples):
        channel = channels[door_codes[index % len(door_codes)]]
        snapshot = _snapshot(
            mine_id=mine.id,
            channel=channel,
            snapshot_type="door_unclosed",
            event_type="风门未关严",
            level="danger" if gap >= 15 else "warning",
            image_parts=("public", "images", image),
            captured_at=_spot(now, minutes),
            title=f"{channel.location} 门板闭合缝隙{gap}mm",
            gap_mm=float(gap),
            confidence=0.92,
            properties={"threshold_mm": 10, "over_mm": max(gap - 10, 0)},
        )
        rows.append(snapshot)
        by_key[f"unclosed-{index + 1}"] = snapshot

    # 4. 风门：AI 分析类抓拍（来自 20260315 现场图像与门体照片）
    analysis_samples = [
        ("20260315_192656.jpg", "anti_clip", "AI防夹人预警", "danger", 8, {"area": "危险区域"}),
        ("20260315_193111.jpg", "anti_clip", "AI防夹人预警", "danger", 34, {"area": "门体闭合区"}),
        ("20260315_194758.jpg", "foreign_object", "异物卡堵预警", "warning", 50, {"object": "石块"}),
        ("20260315_195803.jpg", "foreign_object", "异物卡堵预警", "warning", 86, {"object": "轨道杂物"}),
        ("20260315_202803.jpg", "seal_failure", "密封失效故障", "warning", 104, {"pressure": "异常"}),
        ("20260315_210425.jpg", "seal_failure", "密封失效故障", "warning", 128, {"pressure": "偏低"}),
        ("20260315_211358.jpg", "illegal_stay", "违规停留预警", "info", 150, {"duration_minutes": 3}),
        ("20260315_212805.jpg", "illegal_stay", "违规停留预警", "info", 170, {"duration_minutes": 2}),
        ("20260315_213411.jpg", "vehicle_crash", "车辆撞击故障", "warning", 188, {"deform": "轻微"}),
        ("20260315_220400.jpg", "patrol", "门体状态巡检", "info", 210, {"result": "门体正常"}),
        ("open2.jpg", "patrol", "门体状态巡检", "info", 232, {"result": "闭锁装置正常"}),
    ]
    for index, (image, snapshot_type, event_type, level, minutes, extra) in enumerate(
        analysis_samples
    ):
        channel = channels[door_codes[index % len(door_codes)]]
        snapshot = _snapshot(
            mine_id=mine.id,
            channel=channel,
            snapshot_type=snapshot_type,
            event_type=event_type,
            level=level,
            image_parts=("public", "images", image),
            captured_at=_spot(now, minutes),
            title=f"{channel.location} {event_type}",
            confidence=0.9,
            properties=dict(extra),
        )
        rows.append(snapshot)
        by_key[f"analysis-{index + 1}"] = snapshot

    # 5. 局部风机：人员轨迹抓拍
    track_samples = [
        ("fan1.png", "张三", "机电队", 1, 12),
        ("fan2.png", "李四", "维修队", 2, 30),
        ("fan3.png", "王五", "掘进一队", 3, 48),
        ("fan4.png", "赵六", "掘进二队", 4, 66),
        ("fan5.png", "陈七", "回采一队", 5, 84),
        ("fan6.png", "周八", "机电队", 1, 102),
    ]
    for index, (image, person, team, channel_index, minutes) in enumerate(track_samples):
        channel = channels[local_fan_codes[channel_index - 1]]
        snapshot = _snapshot(
            mine_id=mine.id,
            channel=channel,
            snapshot_type="person_track",
            event_type="人员检测",
            level="info",
            image_parts=("public", "images", image),
            captured_at=_spot(now, minutes),
            title=f"{channel.name} 人员检测 {person}",
            person_name=person,
            person_count=1,
            team_name=team,
            confidence=0.95,
        )
        rows.append(snapshot)
        by_key[f"track-{index + 1}"] = snapshot

    # 6. 局部风机：预警抓拍
    warn_samples = [
        ("fan7.png", "人员逗留", "张三", 4, {"area": "变频器区", "minutes": 9}),
        ("fan3.png", "非法闯入", "外来人员", 26, {"area": "配电柜区"}),
    ]
    for index, (image, warn_type, person, minutes, extra) in enumerate(warn_samples):
        channel = channels[local_fan_codes[index]]
        snapshot = _snapshot(
            mine_id=mine.id,
            channel=channel,
            snapshot_type="person_track",
            event_type=warn_type,
            level="danger" if warn_type == "非法闯入" else "warning",
            image_parts=("public", "images", image),
            captured_at=_spot(now, minutes),
            title=f"{channel.name} {warn_type} {person}",
            person_name=person,
            person_count=1,
            confidence=0.93,
            properties=dict(extra),
        )
        rows.append(snapshot)
        by_key[f"fan-warn-{index + 1}"] = snapshot

    # 7. 局部风机：设备操作抓拍
    fan_operations = [
        ("风机调频", "张三", "人员准入匹配"),
        ("切换主备", "未知人员", "未检测到人员"),
        ("参数调整", "李四", "人员准入匹配"),
        ("复位操作", "王五", "人员准入匹配"),
        ("风机调频", "赵六", "人员准入匹配"),
        ("参数调整", "外来人员", "未检测到人员"),
        ("切换主备", "陈七", "人员准入匹配"),
        ("风机调频", "周八", "人员准入匹配"),
    ]
    for index, (operation, operator, compare) in enumerate(fan_operations):
        channel = channels[local_fan_codes[index % len(local_fan_codes)]]
        snapshot = _snapshot(
            mine_id=mine.id,
            channel=channel,
            snapshot_type="device_operation",
            event_type=operation,
            level="info" if compare == "人员准入匹配" else "warning",
            image_parts=("public", "images", f"fan{index % 7 + 1}.png"),
            captured_at=_spot(now, 16 + index * 15),
            title=f"{channel.name} {operation} {operator}",
            person_name=operator if compare == "人员准入匹配" else "",
            confidence=0.97,
            properties={"operator": operator, "compare_result": compare},
        )
        rows.append(snapshot)
        by_key[f"fan-op-{index + 1}"] = snapshot

    # 8. 主风机：人员检测抓拍
    detect_samples = [
        ("f1.png", "VC-MFAN-01", "风机主体", "检测到人员", "正常", 10),
        ("f2.png", "VC-MFAN-02", "设备操作区", "检测到人员", "正常", 18),
        ("f3.png", "VC-MFAN-03", "风机主体", "检测到人员", "异常", 35),
        ("f1.png", "VC-MFAN-01", "风机主体", "检测到人员", "正常", 52),
        ("f2.png", "VC-MFAN-02", "设备操作区", "检测到人员", "正常", 70),
        ("f3.png", "VC-MFAN-04", "配电柜", "未检测到人员", "异常", 88),
    ]
    for index, (image, code, spot, result, status, minutes) in enumerate(detect_samples):
        channel = channels[code]
        snapshot = _snapshot(
            mine_id=mine.id,
            channel=channel,
            snapshot_type="person_detect",
            event_type="人员检测",
            level="info" if status == "正常" else "warning",
            image_parts=("public", "images", image),
            captured_at=_spot(now, minutes),
            title=f"{channel.name} {result}",
            confidence=0.94,
            properties={"location": spot, "detect_result": result, "device_status": status},
        )
        rows.append(snapshot)
        by_key[f"detect-{index + 1}"] = snapshot

    # 9. 主风机：设备操作与预警抓拍
    main_operations = [
        ("启动风机", "一号主风机", "PLC操作匹配", 22),
        ("参数调整", "主扇配电室", "PLC操作匹配", 44),
        ("停止风机", "二号主风机", "未检测到人员", 76),
        ("巡检操作", "主扇风硐监测点", "PLC操作匹配", 98),
    ]
    for index, (operation, device, compare, minutes) in enumerate(main_operations):
        channel = channels[main_fan_codes[index % len(main_fan_codes)]]
        snapshot = _snapshot(
            mine_id=mine.id,
            channel=channel,
            snapshot_type="device_operation",
            event_type=operation,
            level="info" if compare == "PLC操作匹配" else "warning",
            image_parts=("public", "images", f"f{index % 3 + 1}.png"),
            captured_at=_spot(now, minutes),
            title=f"{device} {operation}",
            confidence=0.96,
            properties={"device": device, "compare_result": compare},
        )
        rows.append(snapshot)
        by_key[f"main-op-{index + 1}"] = snapshot

    main_warns = [
        ("f2.png", "VC-MFAN-01", "设备断网", 40),
        ("f3.png", "VC-MFAN-02", "设备断网", 92),
    ]
    for index, (image, code, warn_type, minutes) in enumerate(main_warns):
        channel = channels[code]
        snapshot = _snapshot(
            mine_id=mine.id,
            channel=channel,
            snapshot_type="device_warn",
            event_type=warn_type,
            level="warning",
            image_parts=("public", "images", image),
            captured_at=_spot(now, minutes),
            title=f"{channel.name} {warn_type}",
            confidence=0.9,
            properties={"warn_type": warn_type},
        )
        rows.append(snapshot)
        by_key[f"main-warn-{index + 1}"] = snapshot

    db.add_all(rows)
    db.flush()
    return by_key


def _seed_records(
    db: Session,
    mine: Mine,
    channels: dict[str, VideoChannel],
    snapshots: dict[str, VideoSnapshot],
    now: datetime,
) -> list[VideoRecord]:
    rows: list[VideoRecord] = []
    door_codes = [f"VC-DOOR-{index:02d}" for index in range(1, 6)]

    # 车辆通行记录
    vehicle_plates = [
        ("蒙K3571", 12), ("蒙K6829", 29), ("蒙K1122", 46), ("蒙K3344", 63),
        ("蒙K5566", 80), ("蒙K7719", 97), ("蒙K2288", 114), ("蒙K9966", 131),
        ("蒙K4408", 148), ("蒙K5533", 165), ("蒙K8842", 182), ("蒙K6317", 199),
    ]
    for index, (plate, minutes) in enumerate(vehicle_plates):
        channel = channels[door_codes[index % len(door_codes)]]
        snapshot = snapshots.get(f"vehicle-{index + 1}")
        rows.append(
            VideoRecord(
                id=new_id(),
                mine_id=mine.id,
                channel_id=channel.id,
                snapshot_id=snapshot.id if snapshot else None,
                module=VIDEO_MODULE_DOOR,
                record_type="vehicle_passage",
                device_name=channel.name,
                location=channel.location,
                plate_no=plate,
                person_count=0,
                result="通行成功",
                device_status="正常",
                occurred_at=_spot(now, minutes),
                properties={},
            )
        )

    # 人员通行记录
    person_rows = [
        ("张三、李四", 2, "掘进一队", 20, "正常通行"),
        ("王五", 1, "机电队", 38, "正常通行"),
        ("赵六、陈七", 2, "回采一队", 55, "正常通行"),
        ("周八", 1, "机电队", 72, "正常通行"),
        ("吴九", 1, "维修队", 90, "正常通行"),
        ("郑十、刘一", 2, "掘进二队", 108, "正常通行"),
        ("孙十一", 1, "通风队", 126, "正常通行"),
        ("马十二", 1, "维修队", 145, "正常通行"),
    ]
    for index, (person, count, team, minutes, result) in enumerate(person_rows):
        channel = channels[door_codes[index % len(door_codes)]]
        snapshot = snapshots.get(f"person-{index + 1}")
        rows.append(
            VideoRecord(
                id=new_id(),
                mine_id=mine.id,
                channel_id=channel.id,
                snapshot_id=snapshot.id if snapshot else None,
                module=VIDEO_MODULE_DOOR,
                record_type="person_passage",
                device_name=channel.name,
                location=channel.location,
                person_name=person,
                person_count=count,
                team_name=team,
                result=result,
                device_status="正常",
                occurred_at=_spot(now, minutes),
                properties={},
            )
        )

    # 风门未关严记录
    gap_rows = [
        (15, 5), (10, 24), (12, 43), (8, 60), (9, 78), (11, 96), (14, 115), (7, 133), (16, 152)
    ]
    for index, (gap, minutes) in enumerate(gap_rows):
        channel = channels[door_codes[index % len(door_codes)]]
        snapshot = snapshots.get(f"unclosed-{index + 1}")
        rows.append(
            VideoRecord(
                id=new_id(),
                mine_id=mine.id,
                channel_id=channel.id,
                snapshot_id=snapshot.id if snapshot else None,
                module=VIDEO_MODULE_DOOR,
                record_type="door_unclosed",
                device_name=channel.name,
                location=channel.location,
                gap_mm=float(gap),
                level="严重" if gap >= 15 else "一般",
                device_status="异常",
                occurred_at=_spot(now, minutes),
                properties={"threshold_mm": 10},
            )
        )

    # 局部风机：人员轨迹（最后两条未离开，用于“实时人数”）
    local_fan_codes = [f"VC-LFAN-{index:02d}" for index in range(1, 6)]
    track_rows = [
        ("张三", "机电队", "变频器区", 96, 96 - 8, 500, None),
        ("李四", "维修队", "配电柜区", 84, 84 - 5, 310, None),
        ("王五", "掘进一队", "风机本体区", 210, 200, 600, None),
        ("赵六", "掘进二队", "接线盒区", 260, 250, 480, None),
        ("陈七", "回采一队", "散热区", 330, 322, 420, None),
        ("周八", "机电队", "变频器区", 380, 372, 360, None),
        ("洪七", "维修队", "配电柜区", 24, None, None, "onsite"),
        ("孙九", "机电队", "风机本体区", 14, None, None, "onsite"),
    ]
    for index, (person, team, area, enter_minutes, leave_minutes, stay, onsite) in enumerate(
        track_rows
    ):
        channel = channels[local_fan_codes[index % len(local_fan_codes)]]
        snapshot = snapshots.get(f"track-{index + 1}")
        rows.append(
            VideoRecord(
                id=new_id(),
                mine_id=mine.id,
                channel_id=channel.id,
                snapshot_id=snapshot.id if snapshot else None,
                module=VIDEO_MODULE_LOCAL_FAN,
                record_type="person_track",
                device_name=channel.name,
                area_name=area,
                location=channel.location,
                person_name=person,
                person_count=1,
                team_name=team,
                enter_time=_spot(now, enter_minutes),
                leave_time=_spot(now, leave_minutes) if leave_minutes else None,
                stay_seconds=stay,
                device_status="在场" if onsite else "已离开",
                occurred_at=_spot(now, enter_minutes),
                properties={"stay_text": _stay_text(stay)},
            )
        )

    # 局部风机：区域停留统计
    area_rows = [
        ("变频器区", 2, "8分20秒", "3分45秒", "6分02秒", "12分05秒"),
        ("配电柜区", 1, "5分10秒", "5分10秒", "5分10秒", "5分10秒"),
        ("风机本体区", 0, "0秒", "0秒", "0秒", "0秒"),
        ("接线盒区", 0, "0秒", "0秒", "0秒", "0秒"),
        ("散热区", 0, "0秒", "0秒", "0秒", "0秒"),
    ]
    for index, (area, count, max_s, min_s, avg_s, total_s) in enumerate(area_rows):
        channel = channels[local_fan_codes[index % len(local_fan_codes)]]
        rows.append(
            VideoRecord(
                id=new_id(),
                mine_id=mine.id,
                channel_id=channel.id,
                snapshot_id=None,
                module=VIDEO_MODULE_LOCAL_FAN,
                record_type="area_stay",
                device_name=channel.name,
                area_name=area,
                location=channel.location,
                person_count=count,
                occurred_at=_spot(now, 60),
                properties={
                    "max_stay": max_s,
                    "min_stay": min_s,
                    "avg_stay": avg_s,
                    "total_stay": total_s,
                },
            )
        )

    # 局部风机：预警记录
    fan_warn_rows = [
        ("person_warn", "张三", "变频器区", "人员逗留", 30, "处理中"),
        ("person_warn", "外来人员", "配电柜区", "非法闯入", 26, "未处理"),
        ("person_warn", "李四", "配电柜区", "人员逗留", 120, "已处理"),
    ]
    for index, (record_type, person, area, warn_type, minutes, handle) in enumerate(
        fan_warn_rows
    ):
        channel = channels[local_fan_codes[index % len(local_fan_codes)]]
        snapshot = snapshots.get(f"fan-warn-{index + 1}")
        rows.append(
            VideoRecord(
                id=new_id(),
                mine_id=mine.id,
                channel_id=channel.id,
                snapshot_id=snapshot.id if snapshot else None,
                module=VIDEO_MODULE_LOCAL_FAN,
                record_type=record_type,
                device_name=channel.name,
                area_name=area,
                location=channel.location,
                person_name=person,
                person_count=1,
                warn_type=warn_type,
                handle_status=handle,
                occurred_at=_spot(now, minutes),
                properties={"region": area},
            )
        )

    # 主风机：人员检测记录
    detect_rows = [
        ("VC-MFAN-01", "风机主体", "检测到人员", "正常", 10),
        ("VC-MFAN-02", "设备操作区", "检测到人员", "正常", 18),
        ("VC-MFAN-03", "风机主体", "检测到人员", "异常", 35),
        ("VC-MFAN-01", "风机主体", "检测到人员", "正常", 52),
        ("VC-MFAN-02", "设备操作区", "检测到人员", "正常", 70),
        ("VC-MFAN-04", "配电柜", "未检测到人员", "异常", 88),
    ]
    for index, (code, spot, result, device_status, minutes) in enumerate(detect_rows):
        channel = channels[code]
        snapshot = snapshots.get(f"detect-{index + 1}")
        rows.append(
            VideoRecord(
                id=new_id(),
                mine_id=mine.id,
                channel_id=channel.id,
                snapshot_id=snapshot.id if snapshot else None,
                module=VIDEO_MODULE_MAIN_FAN,
                record_type="person_detect",
                device_name=channel.name,
                location=spot,
                result=result,
                device_status=device_status,
                occurred_at=_spot(now, minutes),
                properties={},
            )
        )

    # 主风机：设备操作日志
    main_op_rows = [
        ("VC-MFAN-01", "一号主风机", "启动风机", "PLC操作匹配", "正常", 22),
        ("VC-MFAN-02", "主扇配电室", "参数调整", "PLC操作匹配", "正常", 44),
        ("VC-MFAN-03", "二号主风机", "停止风机", "未检测到人员", "异常", 76),
        ("VC-MFAN-05", "主扇风硐监测点", "巡检操作", "PLC操作匹配", "正常", 98),
        ("VC-MFAN-01", "一号主风机", "参数调整", "PLC操作匹配", "正常", 150),
        ("VC-MFAN-04", "主扇配电室", "巡检操作", "PLC操作匹配", "正常", 180),
    ]
    for index, (code, device, operation, compare, status, minutes) in enumerate(
        main_op_rows
    ):
        channel = channels[code]
        snapshot = snapshots.get(f"main-op-{index + 1}")
        rows.append(
            VideoRecord(
                id=new_id(),
                mine_id=mine.id,
                channel_id=channel.id,
                snapshot_id=snapshot.id if snapshot else None,
                module=VIDEO_MODULE_MAIN_FAN,
                record_type="device_operation",
                device_name=device,
                location=channel.location,
                operation_type=operation,
                compare_result=compare,
                device_status=status,
                occurred_at=_spot(now, minutes),
                properties={},
            )
        )

    # 主风机：设备预警记录
    main_warn_rows = [
        ("VC-MFAN-01", "风机主体", "设备断网", 40, "已处理"),
        ("VC-MFAN-02", "配电柜", "设备断网", 92, "处理中"),
        ("VC-MFAN-01", "风机主体", "轴温异常", 130, "未处理"),
        ("VC-MFAN-05", "风硐", "巡检缺失", 200, "已处理"),
    ]
    for index, (code, spot, warn_type, minutes, handle) in enumerate(main_warn_rows):
        channel = channels[code]
        snapshot = snapshots.get(f"main-warn-{index + 1}") if index < 2 else None
        rows.append(
            VideoRecord(
                id=new_id(),
                mine_id=mine.id,
                channel_id=channel.id,
                snapshot_id=snapshot.id if snapshot else None,
                module=VIDEO_MODULE_MAIN_FAN,
                record_type="device_warn",
                device_name=channel.name,
                location=spot,
                warn_type=warn_type,
                handle_status=handle,
                occurred_at=_spot(now, minutes),
                properties={},
            )
        )

    db.add_all(rows)
    db.flush()
    return rows


def _seed_events(
    db: Session,
    mine: Mine,
    channels: dict[str, VideoChannel],
    snapshots: dict[str, VideoSnapshot],
    now: datetime,
) -> list[VideoAnalysisEvent]:
    rows: list[VideoAnalysisEvent] = []

    door_events = [
        ("VC-DOOR-02", "AI防夹人预警", "danger", "危险区域有人闯入，已触发PLC停止", 6, "pending", "analysis-1"),
        ("VC-DOOR-04", "风门未关严", "warning", "门板闭合缝隙15mm，超标10mm", 22, "processing", "unclosed-1"),
        ("VC-DOOR-01", "违规停留预警", "info", "禁停区有车辆滞留3分钟，未及时驶离", 41, "handled", "analysis-7"),
        ("VC-DOOR-03", "密封失效故障", "warning", "疑似风门密封胶条脱落，存在漏风风险", 58, "pending", "analysis-5"),
        ("VC-DOOR-05", "车辆撞击故障", "warning", "风门门体疑似受到轻微撞击，需检查变形", 75, "processing", "analysis-9"),
        ("VC-DOOR-06", "异物卡堵预警", "warning", "门体通道检测到石块异物，影响闭合", 94, "pending", "analysis-3"),
        ("VC-DOOR-07", "AI防夹人预警", "danger", "风门闭合区检测到人员肢体，紧急止停", 118, "handled", "analysis-2"),
        ("VC-DOOR-03", "风门未关严", "warning", "门板闭合缝隙16mm，超标11mm", 140, "pending", "unclosed-9"),
        ("VC-DOOR-08", "违规停留预警", "info", "禁停区有人员逗留，已语音提醒", 165, "ignored", "analysis-8"),
        ("VC-DOOR-06", "异物卡堵预警", "warning", "风门轨道有杂物卡滞", 190, "handled", "analysis-4"),
    ]
    for code, event_type, level, description, minutes, status, snap_key in door_events:
        channel = channels[code]
        snapshot = snapshots.get(snap_key)
        rows.append(
            _event(
                mine=mine,
                channel=channel,
                snapshot=snapshot,
                event_type=event_type,
                category="风门",
                level=level,
                title=f"{channel.name} {event_type}",
                description=description,
                status=status,
                minutes=minutes,
                now=now,
                confidence=0.93,
            )
        )

    fan_events = [
        ("VC-LFAN-01", "人员逗留预警", "warning", "张三维修队变频器区停留超8分钟", 30, "processing", "fan-warn-1"),
        ("VC-LFAN-02", "非法闯入预警", "danger", "外来人员进入配电柜区", 26, "pending", "fan-warn-2"),
        ("VC-LFAN-04", "未检测到人员", "danger", "3号风机切换主备未检测到操作人员", 70, "pending", "fan-op-2"),
        ("VC-LFAN-05", "风机离线预警", "info", "4号局部风机通讯中断", 210, "handled", "fan-op-6"),
        ("VC-LFAN-03", "人员超时逗留", "warning", "人员配电柜区停留时间超限", 122, "handled", "track-3"),
    ]
    for code, event_type, level, description, minutes, status, snap_key in fan_events:
        channel = channels[code]
        snapshot = snapshots.get(snap_key)
        rows.append(
            _event(
                mine=mine,
                channel=channel,
                snapshot=snapshot,
                event_type=event_type,
                category="局部通风",
                level=level,
                title=f"{channel.name} {event_type}",
                description=description,
                status=status,
                minutes=minutes,
                now=now,
                confidence=0.91,
            )
        )

    main_events = [
        ("VC-MFAN-01", "设备断网", "warning", "一号主风机通讯中断5分钟", 40, "handled", "main-warn-1"),
        ("VC-MFAN-03", "未检测到人员", "danger", "二号主风机停止风机操作未检测到人员", 76, "pending", "main-op-3"),
        ("VC-MFAN-04", "设备断网", "warning", "主扇配电室网络抖动", 92, "processing", "main-warn-2"),
        ("VC-MFAN-01", "轴温异常", "warning", "一号主风机轴承温度 78℃ 超限", 130, "pending", "detect-3"),
        ("VC-MFAN-05", "巡检缺失", "info", "风硐监测点巡检记录缺失", 200, "ignored", None),
    ]
    for code, event_type, level, description, minutes, status, snap_key in main_events:
        channel = channels[code]
        snapshot = snapshots.get(snap_key or "")
        rows.append(
            _event(
                mine=mine,
                channel=channel,
                snapshot=snapshot,
                event_type=event_type,
                category="主通风机",
                level=level,
                title=f"{channel.name} {event_type}",
                description=description,
                status=status,
                minutes=minutes,
                now=now,
                confidence=0.92,
            )
        )

    db.add_all(rows)
    db.flush()
    return rows


def _event(
    *,
    mine: Mine,
    channel: VideoChannel,
    snapshot: VideoSnapshot | None,
    event_type: str,
    category: str,
    level: str,
    title: str,
    description: str,
    status: str,
    minutes: int,
    now: datetime,
    confidence: float,
) -> VideoAnalysisEvent:
    occurred_at = _spot(now, minutes)
    handled_at = None
    handled_by = ""
    if status in {"handled", "ignored"}:
        handled_at = min(occurred_at + timedelta(minutes=8), now)
        handled_by = "通防科值班员"
    return VideoAnalysisEvent(
        id=new_id(),
        mine_id=mine.id,
        channel_id=channel.id,
        snapshot_id=snapshot.id if snapshot else None,
        module=channel.module,
        event_type=event_type,
        category=category,
        level=level,
        title=title,
        description=description,
        confidence=confidence,
        status=status,
        occurred_at=occurred_at,
        handled_at=handled_at,
        handled_by=handled_by,
        handle_remark="已通知现场复核" if handled_at else "",
        properties={"source": "video_ai"},
    )


def _stay_text(stay_seconds: int | None) -> str:
    if not stay_seconds:
        return ""
    minutes, seconds = divmod(int(stay_seconds), 60)
    return f"{minutes}分{seconds:02d}秒"