vrsub/scripts/build_translate_eval.py

"""构建"翻译质量评测集"候选池（真实数据，不造样本）。

背景与目的
----------
评测"更便宜的 LLM 模型能否用于 llm-translate 节点"时，需要一份**可人工核对
的基准**：日语原文 + 该句真实中文含义。本仓库正好有同一部视频的两条真实产物：

- 日语原文：run_d386ccf124f7（learn-translate）的 whisper ASR 产物
  `steps/asr/transcript.srt`（large-v2 + decode_full，1161 条）；
- 中文基准：run_479b411f299d（ocr-subtitle）的烧录字幕 OCR + LLM 过滤产物
  `steps/filter/filtered.srt`（1144 条）——烧录字幕是画面的一部分，其文本
  约等于官方/人工中文字幕，可作为"该句真实含义"的近似 ground truth。

两条产物来自**同一音轨**（已核对前 5 分钟 16k 单声道 PCM md5 一致），因此可按
时间重叠配对。但配对并非一一对应（OCR 会把多句合并、漏识别呻吟段），所以本
脚本只负责**生成候选池**，最终是否入选由人工逐条确认（脚本不保证正确性）。

用法（仓库根目录）：

    uv run python scripts/build_translate_eval.py             # 打印候选池供人工挑选
    uv run python scripts/build_translate_eval.py --emit-pool scripts/data/translate_eval/candidates.jsonl

产物：candidates.jsonl，每行
    {"start","end","ja","zh_ref","type_hint","kana_len"}
"""

from __future__ import annotations

import argparse
import json
import re
from pathlib import Path

PROJECT_ROOT = Path(__file__).resolve().parent.parent

# 默认输入：两条真实产物路径（相对仓库根）。
DEFAULT_JA = PROJECT_ROOT / "data/storage/runs/run_d386ccf124f7/steps/asr/transcript.srt"
DEFAULT_REF = PROJECT_ROOT / "data/storage/runs/run_479b411f299d/steps/filter/filtered.srt"

# 类型提示用到的字符集合（仅用于给候选打标签，方便人工按类型均衡挑选）。
_KANA = re.compile(r"[ぁ-んァ-ヴー]")
_KANJI = re.compile(r"[\u4e00-\u9fff]")
_LATIN = re.compile(r"[A-Za-z]")
# 纯呻吟/喘息：整句只由呻吟字符与标点组成。
_MOAN_ONLY = re.compile(r"^[あいうえおんはぁっアンー…、。！？!?\s]+$")


def _seconds(ts: str) -> float:
    """SRT 时间戳 HH:MM:SS,mmm -> 秒（浮点）。"""
    hours, minutes, rest = ts.split(":")
    seconds, millis = rest.split(",")
    return int(hours) * 3600 + int(minutes) * 60 + int(seconds) + int(millis) / 1000


def _type_hint(text: str) -> str:
    """给候选打类型提示（供人工按类型均衡取样，不代表语义正确性）。

    - 呻吟：整句只有呻吟字符；
    - 短碎片：有效字符 ≤ 4（助词/应答词，最考验"结合上下文独立成行"）；
    - 拟声/外来语：含长音符或拉丁字母（オーラル/パンパン 等）；
    - 长句：字符数 ≥ 15（考验长句切分与语序重组）；
    - 对话：其余。
    """
    stripped = re.sub(r"[\s、。！？!?…]", "", text)
    if _MOAN_ONLY.fullmatch(text):
        return "呻吟"
    if len(stripped) <= 4:
        return "短碎片"
    if _LATIN.search(text) or "ー" in text:
        return "拟声/外来语"
    if len(stripped) >= 15:
        return "长句"
    return "对话"


def build_candidates(ja_path: Path, ref_path: Path) -> list[dict]:
    """按"时间重叠且唯一"把日语 cue 与中文字幕基准配对，返回候选列表。

    只保留**恰好与一条**参考字幕重叠的日语 cue——多条重叠说明参考字幕把多句
    合并（或 OCR 漏句），无法确定该句的真实含义，这类不进入候选池（人工也无法
    可靠核对）。重叠判定用半开区间（start < other.end and other.start < end）。
    """
    from nodes.srt import parse_srt  # 复用生产解析器，避免另写一套 SRT 规则

    ja_cues = parse_srt(ja_path.read_text(encoding="utf-8"))
    ref_cues = parse_srt(ref_path.read_text(encoding="utf-8"))

    candidates: list[dict] = []
    for cue in ja_cues:
        overlaps = [
            ref for ref in ref_cues
            if _seconds(ref.start) < _seconds(cue.end) and _seconds(cue.start) < _seconds(ref.end)
        ]
        if len(overlaps) != 1:
            continue
        candidates.append(
            {
                "start": cue.start,
                "end": cue.end,
                "ja": cue.text,
                "zh_ref": overlaps[0].text,
                "type_hint": _type_hint(cue.text),
                "kana_len": len(_KANA.findall(cue.text)),
                "has_kanji": bool(_KANJI.search(cue.text)),
            }
        )
    return candidates


def main() -> None:
    parser = argparse.ArgumentParser(description="生成翻译质量评测集候选池")
    parser.add_argument("--ja", type=Path, default=DEFAULT_JA, help="日语 ASR srt")
    parser.add_argument("--ref", type=Path, default=DEFAULT_REF, help="中文基准 srt")
    parser.add_argument("--emit-pool", type=Path, default=None, help="写出候选池 jsonl")
    parser.add_argument("--limit", type=int, default=0, help="只打印前 N 条（0=全部）")
    args = parser.parse_args()

    candidates = build_candidates(args.ja, args.ref)
    print(f"候选池共 {len(candidates)} 条（时间唯一配对）")
    # 类型分布：用于人工按类型均衡挑选。
    dist: dict[str, int] = {}
    for item in candidates:
        dist[item["type_hint"]] = dist.get(item["type_hint"], 0) + 1
    print("类型分布:", dist)

    if args.emit_pool:
        args.emit_pool.parent.mkdir(parents=True, exist_ok=True)
        with args.emit_pool.open("w", encoding="utf-8") as fh:
            for item in candidates:
                fh.write(json.dumps(item, ensure_ascii=False) + "\n")
        print(f"候选池已写入 {args.emit_pool}")

    shown = candidates if args.limit <= 0 else candidates[: args.limit]
    for index, item in enumerate(shown, 1):
        print(f"[{index}] {item['start']} ({item['type_hint']})\n    JA: {item['ja']!r}\n    ZH: {item['zh_ref']!r}")


if __name__ == "__main__":
    main()