yingshiyun/analysis/video_scanner.py

"""扫描 recordings/{设备号}/{YYYYMMDD}/*.mp4，登记进 videos 表。

start_time 只精确到"天"：文件名时间戳实测证实记录的是下载/导出时刻而非真实录制
时刻（容器是 MPEG-PS，没有 creation_time 元数据可交叉验证），因此放弃追求绝对
精确的录制时间，start_time 直接取自文件所在日期文件夹（capture_date）。

automate.py 的 ADB 自动化下载在滚动/点击时序上有已知问题，会把同一段录像重复
下载成多份文件名不同的拷贝（详见 dedup_finder.py）。这里在入库时按 MD5 查重：
新文件如果和已入库的某个视频内容完全相同，标记为 duplicate 状态，不再进入
Phase2~5 的分析流程，避免继续产生虚高的事件统计。
"""
import json
import subprocess
from datetime import datetime

from analysis.config import RECORDINGS_DIR, PROJECT_ROOT
from analysis.db import (
    insert_video, init_db, count_videos, video_exists, get_video_id_by_md5,
    get_videos_missing_md5, update_video_md5,
)
from analysis.hashing import compute_md5


def backfill_md5():
    """给这个字段上线之前已经入库、还没有 file_md5 的老记录补算一次，
    否则新文件查重时比对不到旧文件，查重形同虚设。"""
    for video in get_videos_missing_md5():
        file_path = PROJECT_ROOT / video["file_path"]
        if not file_path.exists():
            continue  # 文件已被删除（例如清理过的重复文件），跳过
        update_video_md5(video["id"], compute_md5(file_path))


def probe_video(file_path):
    """用 ffprobe 取时长/帧率/分辨率。文件损坏或探测失败时返回 None。"""
    try:
        out = subprocess.run(
            ["ffprobe", "-v", "quiet", "-print_format", "json", "-show_format", "-show_streams", str(file_path)],
            capture_output=True, text=True, timeout=30,
        )
        data = json.loads(out.stdout)
    except (subprocess.TimeoutExpired, json.JSONDecodeError, OSError):
        return None

    fmt = data.get("format")
    if not fmt or "duration" not in fmt:
        return None

    video_stream = next((s for s in data.get("streams", []) if s.get("codec_type") == "video"), None)
    if not video_stream:
        return None

    fps = None
    if video_stream.get("r_frame_rate"):
        num, _, den = video_stream["r_frame_rate"].partition("/")
        if num and den and int(den) != 0:
            fps = int(num) / int(den)

    return {
        "duration_sec": float(fmt["duration"]),
        "fps": fps,
        "width": video_stream.get("width"),
        "height": video_stream.get("height"),
    }


def capture_date_to_start_time(capture_date):
    return f"{capture_date[0:4]}-{capture_date[4:6]}-{capture_date[6:8]}"


def scan_all():
    """扫描全部设备目录下的全部录像文件，登记入库。返回 {已处理, 新增损坏, 新增重复} 统计。"""
    init_db()
    backfill_md5()
    processed = 0
    corrupt = 0
    duplicates = 0

    if not RECORDINGS_DIR.exists():
        return {"processed": 0, "corrupt": 0, "duplicates": 0}

    for device_dir in sorted(RECORDINGS_DIR.iterdir()):
        if not device_dir.is_dir():
            continue
        device_serial = device_dir.name

        for date_dir in sorted(device_dir.iterdir()):
            if not date_dir.is_dir():
                continue
            capture_date = date_dir.name

            for file_path in sorted(date_dir.glob("*.mp4")):
                rel_path = str(file_path.relative_to(PROJECT_ROOT))
                if video_exists(rel_path):
                    continue  # 已登记过，跳过重复的探测/哈希计算

                file_size_bytes = file_path.stat().st_size
                file_md5 = compute_md5(file_path)
                duplicate_of = get_video_id_by_md5(file_md5)

                if duplicate_of is not None:
                    # 与已入库视频内容完全相同，标记为 duplicate，不进入后续分析流程
                    video = {
                        "device_serial": device_serial,
                        "capture_date": capture_date,
                        "file_path": rel_path,
                        "file_name": file_path.name,
                        "file_size_bytes": file_size_bytes,
                        "duration_sec": None,
                        "fps": None,
                        "width": None,
                        "height": None,
                        "start_time": capture_date_to_start_time(capture_date),
                        "status": "duplicate",
                        "error_message": f"与 video_id={duplicate_of} 内容完全相同（MD5一致），跳过分析",
                        "file_md5": file_md5,
                        "duplicate_of": duplicate_of,
                    }
                    insert_video(video)
                    duplicates += 1
                    processed += 1
                    continue

                probe = probe_video(file_path)

                if probe is None:
                    status = "corrupt"
                    error_message = "ffprobe 探测失败或文件损坏（可能是 0 字节空文件）"
                    duration_sec = fps = width = height = None
                    corrupt += 1
                else:
                    status = "pending"
                    error_message = None
                    duration_sec = probe["duration_sec"]
                    fps = probe["fps"]
                    width = probe["width"]
                    height = probe["height"]

                video = {
                    "device_serial": device_serial,
                    "capture_date": capture_date,
                    "file_path": rel_path,
                    "file_name": file_path.name,
                    "file_size_bytes": file_size_bytes,
                    "duration_sec": duration_sec,
                    "fps": fps,
                    "width": width,
                    "height": height,
                    "start_time": capture_date_to_start_time(capture_date),
                    "status": status,
                    "error_message": error_message,
                    "file_md5": file_md5,
                    "duplicate_of": None,
                }
                insert_video(video)
                processed += 1

    return {"processed": processed, "corrupt": corrupt, "duplicates": duplicates}


if __name__ == "__main__":
    print("===== 开始扫描录像文件 =====")
    result = scan_all()
    print(f"→ 本次处理 {result['processed']} 个文件，其中损坏/空文件 {result['corrupt']} 个，"
          f"重复文件 {result['duplicates']} 个（已跳过分析）")
    print(f"→ 数据库当前共登记 {count_videos()} 条记录")
    print("===== 扫描完成 =====")