UmbraTrans/tools/translation/backtranslate.py

# Copyright 2026 Umbra Tools (umbratools.com)
# SPDX-License-Identifier: Apache-2.0

"""Back-translation checker for verifying Chinese translation intent.

Uses a *different* LLM model than the drafter to translate Chinese back to
English, then compares against the original English to verify that meaning,
tone, and intent are preserved. This is an independent quality gate -- if the
back-translation captures the same story beats, the Chinese is faithful.
"""

from __future__ import annotations

import json
import argparse
import json
import re
import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent))

from llm import DEFAULT_MODEL, LLM, LLMError, Usage  # noqa: E402

# A model *different* from the drafter (qwen/qwen3.5-9b) for independent
# verification. gemma-4-12b-qat is a good choice: capable but architecturally
# distinct from Qwen.
BACKTRANSLATE_MODEL = "google/gemma-4-12b-qat"

SYSTEM_BACKTRANSLATE = """你是一位严格的双语审校。你的任务是把中文译文译回英文，然后与英文原文对比，判断原意是否被保留。

只输出 JSON，不要解释，不要 Markdown 代码围栏。输出结构：
{
  "back_translation": "完整的英文回译",
  "intent_preserved": true/false,
  "confidence": 0.0-1.0,
  "issues": [
    {"id": "段落id", "problem": "具体问题描述", "severity": "block|major|minor"}
  ],
  "notes": "整体评估（可选）
}

只有当读者会因此理解错、人物声音被抹平、或事实被改变时才报告问题。不要因为“措辞不同”而报错。
"""


def backtrans_prompt(
    *,
    chapter_title: str,
    paragraphs: list[dict],
    drafts: dict[str, str],
) -> str:
    pairs = []
    for p in paragraphs:
        if p.get("type") == "scene_break":
            continue
        pairs.append(
            {
                "id": p["id"],
                "speaker": p.get("speaker"),
                "type": p.get("type"),
                "en": p["text"],
                "zh": drafts.get(p["id"], ""),
            }
        )
    pairs_json = json.dumps(pairs, ensure_ascii=False, indent=2)
    return f"""书名：Deal Hunter
本章：{chapter_title}

请把下面每段中文译文译回英文，并与英文原文对比，判断原意是否保留。
输出 JSON：{{"back_translation": "...", "intent_preserved": true/false, "confidence": 0.0-1.0, "issues": [...], "notes": "..."}}

英文原文与中文译文如下：
{pairs_json}"""


def check_chapter(
    llm: LLM,
    chapter: dict,
    drafts: dict[str, str],
    sample_fraction: float = 0.3,
) -> dict:
    """Back-translate a sample of paragraphs from one chapter."""
    paras = [p for p in chapter["paragraphs"] if p.get("type") != "scene_break"]
    if not paras:
        return {"chapter_id": chapter["id"], "error": "no paragraphs"}

    # Sample a subset for back-translation (full book back-translation is expensive)
    import random

    random.seed(42)  # deterministic sampling
    n_sample = max(1, int(len(paras) * sample_fraction))
    sample = random.sample(paras, min(n_sample, len(paras)))

    prompt = backtrans_prompt(
        chapter_title=f"{chapter['title']} / {chapter['title_zh']}",
        paragraphs=sample,
        drafts=drafts,
    )

    try:
        value, usage = llm.chat(
            SYSTEM_BACKTRANSLATE,
            prompt,
            model=BACKTRANSLATE_MODEL,
            temperature=0.1,
            max_tokens=8000,
            cache_key=f"{chapter['id']}.backtranslate.{BACKTRANSLATE_MODEL}",
        )
    except LLMError as exc:
        return {
            "chapter_id": chapter["id"],
            "chapter_title": chapter["title"],
            "error": str(exc),
            "sampled": len(sample),
            "total": len(paras),
        }

    try:
        result = json.loads(value)
    except json.JSONDecodeError:
        # Try to extract JSON from markdown fence or surrounding prose
        text = value.strip()
        if text.startswith("```"):
            text = re.sub(r"^```[a-zA-Z0-9_-]*\s*", "", text)
            text = re.sub(r"\s*```$", "", text).strip()
        try:
            result = json.loads(text)
        except json.JSONDecodeError:
            return {
                "chapter_id": chapter["id"],
                "chapter_title": chapter["title"],
                "error": f"could not parse back-translation response: {value[:200]}",
                "sampled": len(sample),
                "total": len(paras),
                "raw": value,
            }

    result["chapter_id"] = chapter["id"]
    result["chapter_title"] = chapter["title"]
    result["sampled"] = len(sample)
    result["total"] = len(paras)
    result["model"] = BACKTRANSLATE_MODEL
    return result


def main() -> None:
    ap = argparse.ArgumentParser(description="Back-translate Chinese to English for intent verification")
    ap.add_argument("book", help="Path to book.json")
    ap.add_argument("run_dir", help="Path to runs directory containing translation.json per chapter")
    ap.add_argument("chapters", nargs="*", help="Chapter IDs to check (default: all translated)")
    ap.add_argument("--sample", type=float, default=0.3, help="Fraction of paragraphs to back-translate (default 0.3)")
    ap.add_argument("--model", default=BACKTRANSLATE_MODEL, help="Back-translation model")
    args = ap.parse_args()

    from llm import DEFAULT_MODEL, LLM, LLMError, Usage  # noqa: E402

    book = json.loads(Path(args.book).read_text(encoding="utf-8"))
    run_dir = Path(args.run_dir)
    llm = LLM(cache_dir=run_dir / "cache", usage_log=run_dir / "usage.jsonl")

    wanted = set(args.chapters) if args.chapters else None
    chapters = [c for c in book["chapters"] if not wanted or c["id"] in wanted or str(c["index"]) in wanted]

    results = []
    for ch in chapters:
        draft_path = run_dir / ch["id"] / "translation.json"
        if not draft_path.exists():
            print(f"SKIP {ch['id']}: no translation.json", flush=True)
            continue
        draft_data = json.loads(draft_path.read_text(encoding="utf-8"))
        drafts = draft_data.get("paragraphs", {})
        print(f"Checking {ch['id']} ({ch['title']})...", flush=True)
        result = check_chapter(llm, ch, drafts, sample_fraction=args.sample)
        results.append(result)
        print(f"  verdict: {result.get('intent_preserved')}, confidence: {result.get('confidence')}, issues: {len(result.get('issues', []))}", flush=True)

    # Summary
    total_issues = sum(len(r.get("issues", [])) for r in results)
    blocked = [r for r in results if any(i.get("severity") == "block" for i in r.get("issues", []))]
    print(f"\n=== BACK-TRANSLATION SUMMARY ===")
    print(f"Chapters checked: {len(results)}")
    print(f"Total issues: {total_issues}")
    print(f"Chapters with block-level issues: {len(blocked)}")
    for r in blocked:
        print(f"  BLOCK: {r['chapter_id']} - {r['chapter_title']}")
        for i in r.get("issues", []):
            if i.get("severity") == "block":
                print(f"    {i['id']}: {i['problem']}")


if __name__ == "__main__":
    main()