lingo/scripts/validate_episodes.py

#!/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()