chore: 加入 episodes.json 校验与修正脚本

- scripts/validate_episodes.py：结构/字段/时间戳/音频对账等校验 - scripts/fix_episodes.py：重算统计字段、引号归一、去空白（自动备份） - scripts/README.md：用法与已知数据缺口说明 Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-21 15:16:59 +08:00
parent 060daf02a1
commit 6091c441cd
3 changed files with 307 additions and 0 deletions
@@ -0,0 +1,31 @@
+# scripts/
+
+维护内容数据 `family-album-usa/episodes.json` 的工具脚本（纯 Python 标准库，无需安装依赖）。
+
+## validate_episodes.py — 数据校验（只读）
+
+把人工体检固化为可重复检查：结构/字段一致性、`lineCount` 自洽、统计字段是否准确、时间戳单调性、音频对账（引用 vs 磁盘）、说话人前缀与引号风格提示。
+
+```bash
+python3 scripts/validate_episodes.py            # 默认数据路径
+python3 scripts/validate_episodes.py 某/episodes.json
+```
+
+发现 `ERROR` 退出码为 1，可接入 CI / pre-commit。
+
+## fix_episodes.py — 格式修正（会先备份）
+
+只做安全可逆的修正：重算统计字段、弯引号统一为直引号、去 `en` 首尾空白。
+**不**剥离说话人前缀、**不**改时间戳数值、**不**为无音频集补造数据（这些需补料后另行处理，详见 `docs/04-architecture.md`）。
+
+```bash
+python3 scripts/fix_episodes.py --dry-run        # 预览将改什么
+python3 scripts/fix_episodes.py                  # 就地修正，自动备份 .bak
+```
+
+## 已知数据缺口（背景）
+
+- ep18–26 共 9 集**无音频**，ep18/19 转写为剧本格式（含说话人前缀）——需补原始音频后再处理。
+- ep1/ep12 各缺 1 幕音频；`audio/episode12_act2.mp3` 为游离文件。
+
+MVP 建议先锁定 ep1–17。
@@ -0,0 +1,116 @@
+#!/usr/bin/env python3
+"""修正 episodes.json 的廉价格式问题（会先备份原文件）。
+
+只做安全、可逆性强的就地修正，**不臆造数据、不改正文语义**：
+  1. 重算统计字段，使其与实际一致：
+       meta.totalEpisodes / meta.totalActs / meta.audioActs（新增）
+       episode.totalLines / episode.hasAudio
+  2. 弯引号统一为直引号（“ ”→"，‘ ’→'），便于点词/分词匹配
+  3. 去除 en 首尾空白
+
+**刻意不做**（需人工/补料后另行处理）：
+  - 不剥离 ep18/19 的说话人前缀（来源不一，待补音频后统一）
+  - 不删除任何 act/line，不修改时间戳数值，不为无音频集补造音频
+
+用法：
+    python3 scripts/fix_episodes.py                 # 就地修正，自动备份
+    python3 scripts/fix_episodes.py --dry-run       # 只报告将改什么，不写
+    python3 scripts/fix_episodes.py --in X --out Y  # 指定输入/输出
+    python3 scripts/fix_episodes.py --no-backup     # 跳过备份
+"""
+import argparse
+import datetime
+import json
+import os
+import shutil
+
+HERE = os.path.dirname(os.path.abspath(__file__))
+REPO = os.path.dirname(HERE)
+DEFAULT_JSON = os.path.join(REPO, "family-album-usa", "episodes.json")
+
+QUOTE_MAP = {"“": '"', "”": '"', "‘": "'", "’": "'"}
+
+
+def normalize_quotes(s):
+    for a, b in QUOTE_MAP.items():
+        s = s.replace(a, b)
+    return s
+
+
+def fix(data):
+    """就地修正 data，返回改动计数字典。"""
+    changes = {"quotes": 0, "stripped": 0, "stats": 0}
+    eps = data["episodes"]
+
+    # 1. 正文：引号与空白
+    for e in eps:
+        for a in e.get("acts", []):
+            for l in a.get("lines", []):
+                en = l.get("en", "")
+                new = normalize_quotes(en)
+                if new != en:
+                    changes["quotes"] += 1
+                    en = new
+                stripped = en.strip()
+                if stripped != en:
+                    changes["stripped"] += 1
+                    en = stripped
+                l["en"] = en
+
+    # 2. 统计字段重算
+    acts = [a for e in eps for a in e.get("acts", [])]
+    meta = data.setdefault("meta", {})
+
+    def set_if_diff(d, k, v):
+        if d.get(k) != v:
+            d[k] = v
+            changes["stats"] += 1
+
+    set_if_diff(meta, "totalEpisodes", len(eps))
+    set_if_diff(meta, "totalActs", len(acts))
+    set_if_diff(meta, "audioActs", sum(1 for a in acts if a.get("hasAudio")))
+    for e in eps:
+        set_if_diff(e, "totalLines", sum(len(a.get("lines", [])) for a in e.get("acts", [])))
+        set_if_diff(e, "hasAudio", any(a.get("hasAudio") for a in e.get("acts", [])))
+
+    return changes
+
+
+def main():
+    p = argparse.ArgumentParser(description="修正 episodes.json 的格式问题")
+    p.add_argument("--in", dest="src", default=DEFAULT_JSON, help="输入文件")
+    p.add_argument("--out", dest="dst", default=None, help="输出文件（默认覆盖输入）")
+    p.add_argument("--dry-run", action="store_true", help="只报告，不写文件")
+    p.add_argument("--no-backup", action="store_true", help="不生成 .bak 备份")
+    args = p.parse_args()
+
+    dst = args.dst or args.src
+    with open(args.src, encoding="utf-8") as f:
+        data = json.load(f)
+
+    changes = fix(data)
+    print("将进行的改动：")
+    print(f"  弯引号 -> 直引号 ：{changes['quotes']} 处")
+    print(f"  去首尾空白       ：{changes['stripped']} 处")
+    print(f"  重算统计字段     ：{changes['stats']} 个")
+
+    if args.dry_run:
+        print("\n[dry-run] 未写入任何文件。")
+        return
+
+    # 备份（仅覆盖写时）
+    if dst == args.src and not args.no_backup:
+        stamp = datetime.datetime.now().strftime("%Y%m%d-%H%M%S")
+        bak = f"{args.src}.{stamp}.bak"
+        shutil.copy2(args.src, bak)
+        print(f"\n已备份原文件 -> {bak}")
+
+    with open(dst, "w", encoding="utf-8") as f:
+        json.dump(data, f, ensure_ascii=False, indent=2)
+        f.write("\n")
+    print(f"已写入 -> {dst}")
+    print("建议接着运行：python3 scripts/validate_episodes.py")
+
+
+if __name__ == "__main__":
+    main()
@@ -0,0 +1,160 @@
+#!/usr/bin/env python3
+"""校验 family-album-usa/episodes.json 的结构与数据质量。
+
+把人工体检固化为可重复运行的检查。每次改动内容数据后跑一遍，防止数据漂移。
+
+用法：
+    python3 scripts/validate_episodes.py                # 用默认路径
+    python3 scripts/validate_episodes.py path/to/episodes.json
+
+退出码：发现 ERROR 返回 1，仅 WARN/INFO 返回 0。
+"""
+import json
+import os
+import re
+import sys
+
+# 默认数据路径：脚本所在目录的上一级 / family-album-usa
+HERE = os.path.dirname(os.path.abspath(__file__))
+REPO = os.path.dirname(HERE)
+DEFAULT_JSON = os.path.join(REPO, "family-album-usa", "episodes.json")
+AUDIO_DIR = os.path.join(REPO, "family-album-usa", "audio")
+
+# 服务路径前缀 -> 磁盘目录的映射（audio 字段写的是对外路径）
+SERVE_PREFIX = "/family-album/audio/"
+SPEAKER_RE = re.compile(r"^[A-Z][A-Za-z .]{1,20}:\s")
+CURLY = "“”‘’"
+
+errors = []
+warns = []
+infos = []
+
+
+def err(msg):
+    errors.append(msg)
+
+
+def warn(msg):
+    warns.append(msg)
+
+
+def info(msg):
+    infos.append(msg)
+
+
+def validate(path):
+    with open(path, encoding="utf-8") as f:
+        data = json.load(f)
+
+    # --- 顶层结构 ---
+    if not isinstance(data, dict) or "episodes" not in data:
+        err("顶层缺少 'episodes' 键")
+        return data
+    meta = data.get("meta", {})
+    eps = data["episodes"]
+    acts = [(e, a) for e in eps for a in e.get("acts", [])]
+    lines = [l for _, a in acts for l in a.get("lines", [])]
+
+    # --- episode id 连续性 ---
+    ids = [e.get("id") for e in eps]
+    if ids != list(range(1, len(eps) + 1)):
+        warn(f"episode id 非 1..{len(eps)} 连续：{ids}")
+
+    # --- 字段一致性 ---
+    line_keys = {tuple(sorted(l.keys())) for l in lines}
+    if line_keys - {("en", "t")}:
+        warn(f"line 字段不统一，出现：{line_keys}")
+    for l in lines:
+        if not isinstance(l.get("t"), (int, float)):
+            err(f"line.t 非数字：{l!r}")
+        elif l["t"] < 0:
+            err(f"line.t 为负：{l!r}")
+        if not str(l.get("en", "")).strip():
+            err(f"line.en 为空：{l!r}")
+        elif l["en"] != l["en"].strip():
+            warn(f"line.en 首尾有空白：{l['en']!r}")
+
+    # --- lineCount 与实际行数 ---
+    for e, a in acts:
+        n = len(a.get("lines", []))
+        if a.get("lineCount") != n:
+            err(f"ep{e['id']} act{a.get('act')} lineCount={a.get('lineCount')} != 实际 {n}")
+
+    # --- meta / episode 统计字段自洽 ---
+    if meta.get("totalEpisodes") != len(eps):
+        warn(f"meta.totalEpisodes={meta.get('totalEpisodes')} != 实际 {len(eps)}")
+    audio_acts = sum(1 for _, a in acts if a.get("hasAudio"))
+    if meta.get("totalActs") not in (len(acts), audio_acts):
+        warn(f"meta.totalActs={meta.get('totalActs')}（实际 act={len(acts)}，有音频={audio_acts}）口径不明")
+    for e in eps:
+        real = sum(len(a.get("lines", [])) for a in e.get("acts", []))
+        if "totalLines" in e and e["totalLines"] != real:
+            warn(f"ep{e['id']} totalLines={e['totalLines']} != 实际 {real}")
+        real_has = any(a.get("hasAudio") for a in e.get("acts", []))
+        if "hasAudio" in e and bool(e["hasAudio"]) != real_has:
+            warn(f"ep{e['id']} hasAudio={e['hasAudio']} != 实际 {real_has}")
+
+    # --- 时间戳单调性 ---
+    for e, a in acts:
+        ts = [l["t"] for l in a.get("lines", []) if isinstance(l.get("t"), (int, float))]
+        for i in range(1, len(ts)):
+            if ts[i] < ts[i - 1]:
+                warn(f"ep{e['id']} act{a['act']} 第{i}行时间戳倒退：{ts[i]} < {ts[i-1]}")
+                break
+
+    # --- 音频对账 ---
+    disk = set(os.listdir(AUDIO_DIR)) if os.path.isdir(AUDIO_DIR) else set()
+    if not disk:
+        warn(f"音频目录不存在或为空：{AUDIO_DIR}")
+    referenced = set()
+    for e, a in acts:
+        au = a.get("audio")
+        if a.get("hasAudio") and not au:
+            err(f"ep{e['id']} act{a['act']} hasAudio=true 但缺 audio 字段")
+        if au:
+            if not au.startswith(SERVE_PREFIX):
+                warn(f"ep{e['id']} act{a['act']} audio 路径前缀异常：{au}")
+            fn = au.rsplit("/", 1)[-1]
+            referenced.add(fn)
+            if disk and fn not in disk:
+                err(f"引用的音频文件磁盘缺失：{fn}")
+    if disk:
+        for fn in sorted(disk - referenced):
+            warn(f"游离音频（磁盘有但未被引用）：{fn}")
+
+    # --- 内容缺口与格式提示（不阻断） ---
+    no_audio = [e["id"] for e in eps if not any(a.get("hasAudio") for a in e["acts"])]
+    if no_audio:
+        info(f"完全无音频的 episode（{len(no_audio)} 集）：{no_audio}")
+    spk = sum(1 for l in lines if SPEAKER_RE.match(l["en"]))
+    if spk:
+        per = sorted({e["id"] for e, a in acts for l in a["lines"] if SPEAKER_RE.match(l["en"])})
+        info(f"含说话人前缀的行：{spk} 行，分布于 episode {per}")
+    curly = sum(1 for l in lines if any(c in l["en"] for c in CURLY))
+    if curly:
+        info(f"含弯引号的行：{curly}（建议统一为直引号）")
+    info(f"规模：{len(eps)} 集 / {len(acts)} act（{audio_acts} 有音频）/ {len(lines)} 行")
+    return data
+
+
+def main():
+    path = sys.argv[1] if len(sys.argv) > 1 else DEFAULT_JSON
+    print(f"校验：{path}\n")
+    try:
+        validate(path)
+    except (OSError, json.JSONDecodeError) as ex:
+        print(f"[FATAL] 无法读取/解析：{ex}")
+        sys.exit(2)
+
+    for m in infos:
+        print(f"[INFO]  {m}")
+    for m in warns:
+        print(f"[WARN]  {m}")
+    for m in errors:
+        print(f"[ERROR] {m}")
+    print(f"\n小结：{len(errors)} 个 ERROR，{len(warns)} 个 WARN，{len(infos)} 条 INFO")
+    sys.exit(1 if errors else 0)
+
+
+if __name__ == "__main__":
+    main()