From 6091c441cd82cea5f71638f76bb69b0d036b00db Mon Sep 17 00:00:00 2001 From: ila Date: Sun, 21 Jun 2026 15:16:59 +0800 Subject: [PATCH] =?UTF-8?q?chore:=20=E5=8A=A0=E5=85=A5=20episodes.json=20?= =?UTF-8?q?=E6=A0=A1=E9=AA=8C=E4=B8=8E=E4=BF=AE=E6=AD=A3=E8=84=9A=E6=9C=AC?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - scripts/validate_episodes.py：结构/字段/时间戳/音频对账等校验 - scripts/fix_episodes.py：重算统计字段、引号归一、去空白（自动备份） - scripts/README.md：用法与已知数据缺口说明 Co-Authored-By: Claude Opus 4.8 --- scripts/README.md | 31 +++++++ scripts/fix_episodes.py | 116 +++++++++++++++++++++++++ scripts/validate_episodes.py | 160 +++++++++++++++++++++++++++++++++++ 3 files changed, 307 insertions(+) create mode 100644 scripts/README.md create mode 100644 scripts/fix_episodes.py create mode 100644 scripts/validate_episodes.py diff --git a/scripts/README.md b/scripts/README.md new file mode 100644 index 0000000..d22838a --- /dev/null +++ b/scripts/README.md @@ -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。 diff --git a/scripts/fix_episodes.py b/scripts/fix_episodes.py new file mode 100644 index 0000000..bbecd5e --- /dev/null +++ b/scripts/fix_episodes.py @@ -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() diff --git a/scripts/validate_episodes.py b/scripts/validate_episodes.py new file mode 100644 index 0000000..c34b0ae --- /dev/null +++ b/scripts/validate_episodes.py @@ -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()