feat: 媒体库字幕重生成统计脚本（数量/时长/旧字幕备份）

模型与管线切换后需要把媒体库里旧管线生成的字幕整批重跑。批量引擎按"视频旁已有字幕即 SKIPPED"判定，重跑前必须先统计范围、再把旧字幕改名备份，否则建出来的任务会把所有视频全部跳过。 - `scripts/plan_regenerate_subtitles.py`：扫描媒体库，按 CN 产物的最早 mtime 判定生成时间，统计待重生成的数量/时长/体积并输出 JSON 计划；`--backup-old [--apply] [--select]` 把旧字幕原地改名为 `<原名>.old-<日期>` 供批量重跑。脚本不依赖仓库，可拷到媒体库主机直接跑（CIFS 挂载下逐文件 ffprobe 太慢）。 - 测试覆盖真实 10 秒视频与真实字幕文件，含 ffprobe 时长探测、分类边界、备份幂等与选择清单解析。
2026-09-18 22:23:19 +08:00
parent dcdc5e8604
commit f656ec98c5
6 changed files with 673 additions and 0 deletions
@@ -0,0 +1,317 @@
+"""plan_regenerate_subtitles.py 的模块级测试（数据 → 测试过程 → 验证结果）。
+
+被测模块：字幕重生成计划统计脚本。用例在临时目录里放**真实视频文件**与
+真实字幕文件，调用脚本的真实函数；只有 ffprobe 子进程这一 I/O 边界在需要
+确定性时长时注入固定值，其余用例跑真实 ffprobe（缺可执行文件时跳过）。
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import shutil
+from datetime import datetime
+from pathlib import Path
+
+import pytest
+
+from scripts.plan_regenerate_subtitles import (
+    CURRENT,
+    MISSING,
+    PENDING,
+    _agg,
+    backup_old_subtitles,
+    build_items,
+    classify,
+    cn_products,
+    format_summary,
+    main,
+    probe_duration,
+    read_select_list,
+    sidecar_subtitles,
+)
+
+DATA = Path(__file__).parent / "data"
+CLIP = DATA / "clip_10s.mp4"  # 真实 10 秒 mp4，供扫描与时长探测使用
+BEFORE = datetime(2026, 9, 1)
+
+
+def _set_mtime(path: Path, when: str) -> None:
+    """把文件 mtime 设为指定日期（本地时区），模拟不同时期生成的字幕。"""
+    ts = datetime.fromisoformat(when).timestamp()
+    os.utime(path, (ts, ts))
+
+
+def _make_video(tmp_path: Path, name: str) -> Path:
+    """在临时目录放一份真实视频文件，返回其路径。"""
+    video = tmp_path / f"{name}.mp4"
+    shutil.copyfile(CLIP, video)
+    return video
+
+
+def _make_subtitle(video: Path, suffix: str, when: str) -> Path:
+    """在视频旁写一个真实字幕文件（内容为合法 SRT/ASS 片段）并设置 mtime。"""
+    sub = video.with_name(video.name[: -len(video.suffix)] + suffix)
+    if suffix.endswith(".srt"):
+        sub.write_text("1\n00:00:01,000 --> 00:00:02,000\n你好\n", encoding="utf-8")
+    else:
+        sub.write_text("[Script Info]\nTitle: t\n", encoding="utf-8")
+    _set_mtime(sub, when)
+    return sub
+
+
+def test_products_pick_earliest_mtime_so_style_rewrite_stays_old(tmp_path: Path) -> None:
+    """数据：中文字幕生成于 2025-12，双目 .ass 被样式脚本在 2026-09 原地改写。
+
+    过程：扫描该视频并读取 CN 产物生成时间。
+
+    验证：只认本流水线产物，生成时间取最早值（2025-12），视频判为待重生成。
+    """
+    video = _make_video(tmp_path, "movie")
+    _make_subtitle(video, ".srt", "2025-11-01")  # 片源自带字幕，不参与判定
+    _make_subtitle(video, ".CN.srt", "2025-12-01")
+    _make_subtitle(video, ".CN_dual_eye.ass", "2026-09-06")
+
+    items = build_items(tmp_path, workers=2, probe=False)
+
+    assert [p.name for p in cn_products(video)] == ["movie.CN.srt", "movie.CN_dual_eye.ass"]
+    assert items[0]["generated_at"].startswith("2025-12-01")
+    assert [i["video"] for i in classify(items, BEFORE)[PENDING]] == [str(video)]
+
+
+def test_classify_groups_by_generation_time(tmp_path: Path) -> None:
+    """数据：三个视频——旧字幕、新字幕、完全没有 CN 字幕。
+
+    过程：扫描后按 2026-09-01 分界线分类。
+
+    验证：分别落入待重生成 / 已最新 / 无 CN 字幕三组，边界时刻本身算已最新。
+    """
+    old = _make_video(tmp_path, "old")
+    _make_subtitle(old, ".CN.srt", "2026-04-05")
+    fresh = _make_video(tmp_path, "fresh")
+    _make_subtitle(fresh, ".CN.srt", "2026-09-01")
+    _make_subtitle(fresh, ".CN_dual_eye.ass", "2026-09-06")
+    never = _make_video(tmp_path, "never")
+
+    groups = classify(build_items(tmp_path, workers=2, probe=False), BEFORE)
+
+    assert [i["video"] for i in groups[PENDING]] == [str(old)]
+    assert [i["video"] for i in groups[CURRENT]] == [str(fresh)]
+    assert [i["video"] for i in groups[MISSING]] == [str(never)]
+
+
+def test_name_with_extra_suffix_still_matches_product(tmp_path: Path) -> None:
+    """数据：手工改过名的产物 `<视频名>666.CN.srt`（媒体库里真实存在这种命名）。
+
+    过程：扫描该视频的 CN 产物。
+
+    验证：按「文件名含视频主名 + CN 产物后缀」命中，与批量引擎的旁挂判定一致。
+    """
+    video = _make_video(tmp_path, "clip")
+    _make_subtitle(video, "666.CN.srt", "2025-12-14")
+
+    items = build_items(tmp_path, workers=2, probe=False)
+
+    assert items[0]["generated_at"].startswith("2025-12-14")
+    assert classify(items, BEFORE)[PENDING]
+
+
+def test_aggregate_sums_duration_and_size(tmp_path: Path) -> None:
+    """数据：两个待重生成视频，时长各有值、其中一个探测失败。
+
+    过程：汇总该组的数量、时长、体积。
+
+    验证：时长合计跳过探测失败项并单独计数，体积按全部文件累计。
+    """
+    video = _make_video(tmp_path, "a")
+    _make_subtitle(video, ".CN.srt", "2025-12-01")
+    items = [
+        {"size": 100, "duration": 12.5},
+        {"size": 50, "duration": None},
+    ]
+
+    agg = _agg(items)
+
+    assert agg == {"count": 2, "seconds": 12.5, "bytes": 150, "no_duration": 1}
+
+
+def test_main_writes_plan_json(tmp_path: Path) -> None:
+    """数据：真实 10 秒视频 + 旧 CN 字幕 + 新 CN 字幕各一个。
+
+    过程：调用 main() 全流程并写出计划 JSON。
+
+    验证：统计报告含待重生成/已最新行数，JSON 里两条明细状态与时长正确。
+    """
+    if shutil.which("ffprobe") is None:
+        pytest.skip("环境缺少 ffprobe，无法探测真实视频时长")
+    old = _make_video(tmp_path, "old")
+    _make_subtitle(old, ".CN.srt", "2025-12-01")
+    fresh = _make_video(tmp_path, "fresh")
+    _make_subtitle(fresh, ".CN.srt", "2026-09-10")
+    out = tmp_path / "plan.json"
+
+    code = main([str(tmp_path), "--before", "2026-09-01", "--out", str(out), "--workers", "2"])
+
+    assert code == 0
+    payload = json.loads(out.read_text(encoding="utf-8"))
+    by_name = {Path(i["video"]).name: i for i in payload["items"]}
+    assert by_name["old.mp4"]["status"] == PENDING
+    assert by_name["old.mp4"]["duration"] == pytest.approx(10.0, abs=0.5)
+    assert by_name["fresh.mp4"]["status"] == CURRENT
+    assert payload["totals"][PENDING]["count"] == 1
+    assert payload["totals"][PENDING]["seconds"] == pytest.approx(10.0, abs=0.5)
+
+
+def test_probe_duration_reads_real_video() -> None:
+    """数据：真实 10 秒 mp4（模块目录内素材）。
+
+    过程：调用真实 ffprobe 读取时长。
+
+    验证：返回约 10 秒，说明统计脚本的时长口径来自容器真实时长。
+    """
+    if shutil.which("ffprobe") is None:
+        pytest.skip("环境缺少 ffprobe，无法探测真实视频时长")
+
+    assert probe_duration(CLIP) == pytest.approx(10.0, abs=0.1)
+
+
+def test_format_summary_reports_all_groups() -> None:
+    """数据：三组各一条明细（旧/新/无字幕），时长与体积已知。
+
+    过程：格式化统计报告。
+
+    验证：报告包含总量、待重生成、已最新、无 CN 字幕四段关键信息。
+    """
+    root = Path("/media")
+    items = [
+        {"duration": 3600.0, "size": 0, "generated_at": "2025-12-01T00:00:00"},
+        {"duration": 1800.0, "size": 0, "generated_at": "2026-09-10T00:00:00"},
+        {"duration": None, "size": 0, "generated_at": None},
+    ]
+
+    text = format_summary(root, items, classify(items, BEFORE), BEFORE)
+
+    assert "视频总数: 3 个" in text
+    assert "待重新生成（CN 字幕早于 2026-09-01）" in text
+    assert "已是最新" in text
+    assert "无 CN 字幕（首次生成）" in text
+    assert "警告: 1 个视频未能读出时长" in text
+
+
+def test_sidecar_subtitles_matches_batch_engine_rule(tmp_path: Path) -> None:
+    """数据：视频旁有片源 .srt、CN 产物与无关文件各一份。
+
+    过程：列出批量引擎会认定为"已有字幕"的旁挂文件。
+
+    验证：只收文件名含视频主名的字幕，不含无关图片与别人的字幕。
+    """
+    video = _make_video(tmp_path, "movie")
+    source = _make_subtitle(video, ".srt", "2025-11-01")
+    cn_srt = _make_subtitle(video, ".CN.srt", "2025-12-01")
+    cn_ass = _make_subtitle(video, ".CN_dual_eye.ass", "2026-09-06")
+    (tmp_path / "movie-poster.jpg").write_bytes(b"poster")
+    (tmp_path / "other.srt").write_text("1\n", encoding="utf-8")
+
+    found = sidecar_subtitles(video)
+
+    assert found == [cn_srt, cn_ass, source]
+
+
+def test_backup_renames_every_sidecar_and_unsets_skip(tmp_path: Path) -> None:
+    """数据：待重生成视频带片源 .srt + CN 产物 + 既有备份目标冲突。
+
+    过程：执行改名备份。
+
+    验证：全部旁挂字幕改名成 .old 后缀、原文件消失；改名后批量引擎再也
+    不会把该视频判为"已有字幕"。
+    """
+    video = _make_video(tmp_path, "movie")
+    source = _make_subtitle(video, ".srt", "2025-11-01")
+    cn_srt = _make_subtitle(video, ".CN.srt", "2025-12-01")
+    cn_ass = _make_subtitle(video, ".CN_dual_eye.ass", "2026-09-06")
+
+    changed = backup_old_subtitles([{"video": str(video)}], suffix=".old", apply=True)
+
+    assert len(changed) == 3
+    assert sidecar_subtitles(video) == []
+    for path in (source, cn_srt, cn_ass):
+        assert not path.exists()
+        assert (path.parent / (path.name + ".old")).read_text(encoding="utf-8")
+
+
+def test_backup_dry_run_only_reports(tmp_path: Path) -> None:
+    """数据：一个带旧字幕的视频。
+
+    过程：不传 apply 走预览。
+
+    验证：返回改名计划，磁盘文件保持不变。
+    """
+    video = _make_video(tmp_path, "movie")
+    sub = _make_subtitle(video, ".CN.srt", "2025-12-01")
+
+    changed = backup_old_subtitles([{"video": str(video)}], suffix=".old", apply=False)
+
+    assert changed == [(sub, sub.with_name(sub.name + ".old"))]
+    assert sub.exists()
+
+
+def test_backup_does_not_overwrite_existing_backup(tmp_path: Path) -> None:
+    """数据：视频旁已有同名备份文件（同一批重复执行）。
+
+    过程：再次执行改名备份。
+
+    验证：已存在的备份不被覆盖，原文件也不被删除，返回空变更。
+    """
+    video = _make_video(tmp_path, "movie")
+    sub = _make_subtitle(video, ".CN.srt", "2025-12-01")
+    backup = sub.with_name(sub.name + ".old")
+    backup.write_text("first\n", encoding="utf-8")
+
+    changed = backup_old_subtitles([{"video": str(video)}], suffix=".old", apply=True)
+
+    assert changed == []
+    assert backup.read_text(encoding="utf-8") == "first\n"
+    assert sub.exists()
+
+
+def test_backup_respects_select_list(tmp_path: Path) -> None:
+    """数据：两个待重生成视频，只选中其中一个（选择文件带行尾注释）。
+
+    过程：执行改名备份。
+
+    验证：只改选中视频的旁挂字幕，另一个保持原样。
+    """
+    chosen = _make_video(tmp_path, "chosen")
+    chosen_sub = _make_subtitle(chosen, ".CN.srt", "2025-12-01")
+    kept = _make_video(tmp_path, "kept")
+    kept_sub = _make_subtitle(kept, ".CN.srt", "2025-12-02")
+    select = tmp_path / "select.txt"
+    select.write_text(f"# 只处理这一个\n{chosen}  # 40.3min\n\n", encoding="utf-8")
+
+    changed = backup_old_subtitles(
+        [{"video": str(chosen)}, {"video": str(kept)}],
+        suffix=".old", apply=True, select=read_select_list(select),
+    )
+
+    assert changed == [(chosen_sub, chosen_sub.with_name(chosen_sub.name + ".old"))]
+    assert kept_sub.exists()
+
+
+def test_format_summary_marks_missing_durations(tmp_path: Path) -> None:
+    """数据：只做分类统计、未探测时长的明细（--backup-old 场景）。
+
+    过程：格式化统计报告。
+
+    验证：报告不显示 0 小时，而是标注未探测时长，也不出现时长缺失警告。
+    """
+    root = Path("/media")
+    items = [
+        {"duration": None, "size": 10, "generated_at": "2025-12-01T00:00:00"},
+        {"duration": None, "size": 20, "generated_at": "2026-09-10T00:00:00"},
+    ]
+
+    text = format_summary(root, items, classify(items, BEFORE), BEFORE)
+
+    assert "未探测时长" in text
+    assert "0.00 小时" not in text
+    assert "警告" not in text