3f05489e25
The pipeline was hard-coded to Simplified Chinese in three places: the drafter was told to translate into Chinese, the critic was told to compare English with Chinese, and the glossary's translated forms were read from a "zh" key. A production that declared target_language: es-ES was therefore translated into Chinese and filed under a Spanish name - measured on oldbox, deal-hunter-es held 3,038 paragraphs of Chinese and was reported complete. A new registry, tools/translation/languages.py, is the single source of truth for what a target language is: its names, its script, its punctuation, its glossary key, its typography and its chapter-numbering convention. Ten languages are registered. The prompt scaffolding stays in Chinese - that is what the models read best - and the target is named with its Chinese, English and native names so there is no room to misread it. The registry now reaches every stage: - prompts.py: the analyst, drafter, critic and glossary blocks take the target. The hard rules about this book - Deal Hunter is the ship's name, the AI's name and the book's title, so it stays in English - are the language's own table for Chinese and the global rules for everything else, so a language with no table does not lose them. The Chinese rules are byte-for-byte what they were. - pipeline.py: --target-language, a run-directory guard that refuses to reuse a directory belonging to another language (from the manifest, or by measuring the text it already holds), a draft-keyed-by-target reader, and a refusal of a draft that came back in the wrong script. - book.py: per-language chapter-title and subtitle tables. A language with no table falls back to the English title, never to another language's; a language with a table and a missing chapter gets a visible marker in its own language. - qa.py: the target's quote marks, punctuation, script count and forbidden title forms. The Latin-residue check is skipped for a Latin target, where it would flag the whole book. - align.py: the aligner and verifier prompts name the target, and the sentence and clause splitting uses the target's own terminators. - render.py: the fonts, the language and country codes, the writing direction, the chapter headings and the fixed editorial strings around the translation are the target's. The Latin-run span is skipped for a Latin target, where marking the text as English would tell the renderer it is English. - project_status.py: the dashboard measures a sample of a production's text and reports a production whose text is not in the language it declares as "mismatched", not "complete". A chapter count is a claim about the files; the script check is a measurement of the text. Also fixes: the dashboard gated every language's production on the source production existing, so a Spanish production could never be reported even when it was there. The Chinese render is byte-identical to before on the real book, and the Chinese rules are unchanged. Suite: 620 passed, the same 68 pre-existing environment failures, +49 new tests.
907 lines
36 KiB
Python
907 lines
36 KiB
Python
# Copyright 2026 Umbra Tools (umbratools.com)
|
||
# SPDX-License-Identifier: Apache-2.0
|
||
|
||
"""Multi-pass translation orchestration.
|
||
|
||
Pass 0 is local: paragraphs are grouped and chunked with source indices kept.
|
||
Pass 1 analyses intent, Pass 2 drafts, Pass 4 checks meaning with a different
|
||
model. Pass 3 (chapter voice consistency) is optional and runs after the
|
||
chapter's chunks are drafted.
|
||
|
||
Every artefact is written to disk and reused on the next run, so a multi-hour
|
||
book run can be stopped and resumed.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import json
|
||
import re
|
||
import sys
|
||
import time
|
||
from dataclasses import dataclass
|
||
from pathlib import Path
|
||
|
||
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
||
|
||
from languages import SCRIPT_LATIN, TargetLanguage, resolve, script_mismatch # noqa: E402
|
||
from llm import DEFAULT_ANALYST, DEFAULT_CRITIC, DEFAULT_MODEL, LLM, LLMError, Usage # noqa: E402
|
||
from prompts import ( # noqa: E402
|
||
analysis_prompt,
|
||
consistency_prompt,
|
||
critic_prompt,
|
||
draft_prompt,
|
||
load_glossary,
|
||
system_analyst,
|
||
system_critic,
|
||
system_drafter,
|
||
)
|
||
|
||
CHARACTER_NOTES = """- 凯因达：聪明、直率、好奇，带一点叛逆；对权威不耐烦，对朋友温暖。不要把她写成娇滴滴的公主。
|
||
- Hunter / Deal Hunter：干练、简洁、有冷幽默；是船，也是生意人；不用敬语堆砌。
|
||
- 南妮：温和、认真、字面化，偶尔因为太认真而好笑；不要写成卖萌。
|
||
- 维多利亚公爵夫人：端庄、克制、关切；对凯因达既有长辈的威严，也有家人的柔软。
|
||
- 罗曼诺：热情、商人腔、爱谈价格，但不油滑。
|
||
- 旁白：克制、清晰、有画面感；不替人物抒情。"""
|
||
|
||
# The default target, used where a caller has not supplied one. The pipeline
|
||
# was Chinese-only for its whole life, so the default keeps every existing
|
||
# call site behaving exactly as it did.
|
||
CHINESE = resolve("zh-Hans")
|
||
|
||
# Proper nouns that may legitimately stay in Latin script. Anything else that
|
||
# survives in the Chinese text is reported, not silently accepted.
|
||
LATIN_ALLOW = {
|
||
"ai",
|
||
"audio",
|
||
"authorbdmurphy",
|
||
"b",
|
||
"blazes",
|
||
"brutus",
|
||
"com",
|
||
"d",
|
||
"deal",
|
||
"dial",
|
||
"eva",
|
||
"felucca",
|
||
"gisfrid",
|
||
"goodreads",
|
||
"hunter",
|
||
"http",
|
||
"https",
|
||
"isbn",
|
||
"kabir",
|
||
"kainda",
|
||
"konrad",
|
||
"llc",
|
||
"l2",
|
||
"maul",
|
||
"murphy",
|
||
"nanny",
|
||
"niñera",
|
||
"romano",
|
||
"sao",
|
||
"umbra",
|
||
"vince",
|
||
"wakam",
|
||
"wake",
|
||
"weston",
|
||
"willamina",
|
||
"witold",
|
||
"worstauthors",
|
||
"worstauthor",
|
||
"www",
|
||
"zheng",
|
||
"yi",
|
||
"adonai",
|
||
"adric",
|
||
"albert",
|
||
}
|
||
|
||
|
||
def _normalise_quotes(text: str, target: TargetLanguage | None = None) -> str:
|
||
"""Balance the target language's quotation marks.
|
||
|
||
The marks are the language's own: a German draft uses „ and “, a French
|
||
one « and ». The straight double quote the model sometimes emits is
|
||
converted to the target's pair.
|
||
"""
|
||
target = target or CHINESE
|
||
open_mark, close_mark = target.quote_open, target.quote_close
|
||
text = re.sub(r'"([^"\n]*)"', lambda m: open_mark + m.group(1) + close_mark, text)
|
||
if '"' in text:
|
||
parts = text.split('"')
|
||
out = parts[0]
|
||
for i, part in enumerate(parts[1:]):
|
||
out += (open_mark if i % 2 == 0 else close_mark) + part
|
||
text = out
|
||
if open_mark not in text and text.endswith(close_mark):
|
||
text = text[:-1]
|
||
opens, closes = text.count(open_mark), text.count(close_mark)
|
||
if opens > closes:
|
||
text += close_mark * (opens - closes)
|
||
elif closes > opens:
|
||
for _ in range(closes - opens):
|
||
idx = text.rfind(close_mark)
|
||
if idx > text.rfind(open_mark):
|
||
text = text[:idx] + text[idx + 1 :]
|
||
else:
|
||
break
|
||
return text
|
||
|
||
|
||
def sanitize(text: str, target: TargetLanguage | None = None) -> str:
|
||
"""Clean up one drafted paragraph.
|
||
|
||
The Chinese replacements repair a set of mistakes the Chinese drafter
|
||
keeps making - a transliterated title, a transliterated ship name. They
|
||
cannot match in another language, so they are harmless there; the
|
||
punctuation rules at the end are the target's own.
|
||
"""
|
||
target = target or CHINESE
|
||
text = text.strip()
|
||
text = re.sub(r"^\[[^\]]{1,40}\]\s*", "", text)
|
||
text = re.sub(r"^（(?:对话|旁白|叙述)[^）]{0,60}）\s*", "", text)
|
||
text = re.sub(r"^\((?:dialogue|narration)[^)]{0,60}\)\s*", "", text, flags=re.I)
|
||
text = text.replace("《交易猎人》", "《Deal Hunter》")
|
||
text = text.replace("交易猎人", "Deal Hunter")
|
||
text = text.replace("“猎人”", "Hunter")
|
||
text = text.replace("猎人号", "Deal Hunter")
|
||
text = text.replace("亨特", "Hunter")
|
||
text = text.replace("B.D. 墨菲", "B·D·墨菲")
|
||
text = text.replace("B. D. 墨菲", "B·D·墨菲")
|
||
text = text.replace("B.D.墨菲", "B·D·墨菲")
|
||
text = text.replace("郑义少", "郑一嫂")
|
||
text = text.replace("乌姆布拉音频系统", "Umbra Audio")
|
||
text = text.replace("乌姆布拉音频", "Umbra Audio")
|
||
text = text.replace("Umbra Audio 星系系统", "Umbra Audio 系统")
|
||
text = text.replace("Umbra Audio星系系统", "Umbra Audio 系统")
|
||
text = text.replace("音频星系系统", "音频系统")
|
||
text = text.replace("在小时前", "一小时前")
|
||
if target.ellipsis == "……":
|
||
text = text.replace("...", "……").replace("--", "——")
|
||
else:
|
||
text = text.replace("...", target.ellipsis).replace("--", target.dash)
|
||
return _normalise_quotes(text, target).strip()
|
||
|
||
|
||
def latin_residue(text: str) -> tuple[list[str], list[str]]:
|
||
words = re.findall(r"[A-Za-zÀ-ÿ][A-Za-zÀ-ÿ'’\-]{1,}", text)
|
||
common: set[str] = set()
|
||
proper: set[str] = set()
|
||
for word in words:
|
||
base = word.strip("'’-")
|
||
if not base or base.lower() in LATIN_ALLOW:
|
||
continue
|
||
if base.isupper() and len(base) <= 5:
|
||
continue
|
||
if word[0].isupper():
|
||
proper.add(word)
|
||
else:
|
||
common.add(word)
|
||
return sorted(common), sorted(proper)
|
||
|
||
|
||
@dataclass
|
||
class Chunk:
|
||
id: str
|
||
scene_index: int
|
||
chunks_in_scene: int
|
||
paragraphs: list[dict]
|
||
|
||
@property
|
||
def words(self) -> int:
|
||
return sum(len(p.get("text", "").split()) for p in self.paragraphs)
|
||
|
||
|
||
def load_book(path: Path) -> dict:
|
||
return json.loads(path.read_text(encoding="utf-8"))
|
||
|
||
|
||
def chapter_title_target(chapter: dict) -> str:
|
||
"""The chapter's title in the target language.
|
||
|
||
The field was called ``title_zh`` while the pipeline was Chinese-only and
|
||
is now called ``title_target``. The old name is still read, so a book file
|
||
written by an earlier version keeps working.
|
||
"""
|
||
return chapter.get("title_target") or chapter.get("title_zh") or chapter.get("title") or ""
|
||
|
||
|
||
def select_chapters(book: dict, selector: str) -> list[dict]:
|
||
chapters = book["chapters"]
|
||
if selector in {"all", "*"}:
|
||
return chapters
|
||
if selector == "front":
|
||
return [c for c in chapters if c["matter"].startswith("front")]
|
||
if selector == "body":
|
||
return [c for c in chapters if c["matter"] == "body"]
|
||
if selector == "back":
|
||
return [c for c in chapters if c["matter"] == "back"]
|
||
wanted = {s.strip() for s in selector.split(",") if s.strip()}
|
||
out = []
|
||
for c in chapters:
|
||
if str(c["index"]) in wanted or c["id"] in wanted or c["title"] in wanted:
|
||
out.append(c)
|
||
missing = wanted - {str(c["index"]) for c in out} - {c["id"] for c in out} - {c["title"] for c in out}
|
||
if missing:
|
||
raise SystemExit(f"chapter selector did not match: {sorted(missing)}")
|
||
return out
|
||
|
||
|
||
def chunk_chapter(chapter: dict, max_words: int = 260, max_paras: int = 14) -> list[Chunk]:
|
||
paras = chapter["paragraphs"]
|
||
# Assign scene numbers.
|
||
scene_of: dict[str, int] = {}
|
||
scene = 0
|
||
for p in paras:
|
||
if p.get("type") == "scene_break":
|
||
scene += 1
|
||
continue
|
||
scene_of[p["id"]] = scene
|
||
|
||
raw: list[list[dict]] = []
|
||
current: list[dict] = []
|
||
current_words = 0
|
||
current_scene = None
|
||
for p in paras:
|
||
if p.get("type") == "scene_break":
|
||
if current:
|
||
raw.append(current)
|
||
current, current_words, current_scene = [], 0, None
|
||
continue
|
||
p_scene = scene_of.get(p["id"], 0)
|
||
if current and (
|
||
p_scene != current_scene or current_words + len(p["text"].split()) > max_words or len(current) >= max_paras
|
||
):
|
||
raw.append(current)
|
||
current, current_words = [], 0
|
||
current.append(p)
|
||
current_words += len(p["text"].split())
|
||
current_scene = p_scene
|
||
if current:
|
||
raw.append(current)
|
||
|
||
# Count chunks per scene.
|
||
per_scene: dict[int, int] = {}
|
||
for group in raw:
|
||
s = scene_of.get(group[0]["id"], 0)
|
||
per_scene[s] = per_scene.get(s, 0) + 1
|
||
|
||
chunks: list[Chunk] = []
|
||
seen: dict[int, int] = {}
|
||
for i, group in enumerate(raw):
|
||
s = scene_of.get(group[0]["id"], 0)
|
||
seen[s] = seen.get(s, 0) + 1
|
||
chunks.append(
|
||
Chunk(
|
||
id=f"{chapter['id']}-c{i:03d}",
|
||
scene_index=s,
|
||
chunks_in_scene=per_scene[s],
|
||
paragraphs=group,
|
||
)
|
||
)
|
||
return chunks
|
||
|
||
|
||
class Runner:
|
||
def __init__(
|
||
self,
|
||
*,
|
||
book_path: Path,
|
||
glossary_path: Path,
|
||
run_dir: Path,
|
||
model: str = DEFAULT_MODEL,
|
||
analyst_model: str = DEFAULT_ANALYST,
|
||
critic_model: str = DEFAULT_CRITIC,
|
||
base_url: str | None = None,
|
||
force: bool = False,
|
||
critic: bool = True,
|
||
consistency: bool = False,
|
||
max_words: int = 260,
|
||
max_paras: int = 14,
|
||
critic_limit: int = 0,
|
||
target_language: str = "zh-Hans",
|
||
) -> None:
|
||
self.book = load_book(book_path)
|
||
self.glossary = load_glossary(glossary_path)
|
||
self.run_dir = run_dir
|
||
self.run_dir.mkdir(parents=True, exist_ok=True)
|
||
self.target: TargetLanguage = resolve(target_language)
|
||
self.book_path = book_path
|
||
self.glossary_path = glossary_path
|
||
self.model = model
|
||
self.analyst_model = analyst_model
|
||
self.critic_model = critic_model
|
||
self.force = force
|
||
self.do_critic = critic
|
||
self.do_consistency = consistency
|
||
self.max_words = max_words
|
||
self.max_paras = max_paras
|
||
self.critic_limit = critic_limit
|
||
kwargs = {}
|
||
if base_url:
|
||
kwargs["base_url"] = base_url
|
||
self.llm = LLM(
|
||
cache_dir=run_dir / "cache",
|
||
usage_log=run_dir / "usage.jsonl",
|
||
**kwargs,
|
||
)
|
||
self._guard_run_directory()
|
||
|
||
# ----------------------------------------------------------------- files
|
||
def chapter_dir(self, chapter: dict) -> Path:
|
||
d = self.run_dir / chapter["id"]
|
||
d.mkdir(parents=True, exist_ok=True)
|
||
return d
|
||
|
||
def _read(self, path: Path) -> dict | None:
|
||
if path.exists() and not self.force:
|
||
return json.loads(path.read_text(encoding="utf-8"))
|
||
return None
|
||
|
||
def _write(self, path: Path, value: object) -> None:
|
||
path.write_text(json.dumps(value, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
||
|
||
# --------------------------------------------------------------- the guard
|
||
def _guard_run_directory(self) -> None:
|
||
"""Refuse to reuse a run directory that belongs to another language.
|
||
|
||
The pipeline caches by file existence, and the run directory is chosen
|
||
by the caller. So a Spanish run pointed at a directory that already
|
||
holds a Chinese run would skip drafting entirely and publish the
|
||
Chinese under a Spanish name - the exact defect the mislabelled
|
||
production had, and the reason it is worth a guard rather than a
|
||
convention.
|
||
|
||
Two checks, one from the claim and one from the measurement:
|
||
|
||
* The manifest records what the directory is for. A mismatch is
|
||
refused.
|
||
* A directory with no manifest (written before this guard existed) is
|
||
measured instead: the script of the text it already holds is
|
||
compared against the target language.
|
||
"""
|
||
manifest_path = self.run_dir / "run.json"
|
||
previous: dict = {}
|
||
if manifest_path.exists() and not self.force:
|
||
try:
|
||
previous = json.loads(manifest_path.read_text(encoding="utf-8"))
|
||
except (OSError, json.JSONDecodeError):
|
||
previous = {}
|
||
declared = previous.get("target_language")
|
||
if declared and declared != self.target.code:
|
||
raise SystemExit(
|
||
f"run directory {self.run_dir} holds a {declared} run, but this "
|
||
f"run is {self.target.code}. Use a different --run-dir, or pass "
|
||
"--force to overwrite it deliberately."
|
||
)
|
||
self._measure_existing_language()
|
||
self._write(
|
||
manifest_path,
|
||
{
|
||
"target_language": self.target.code,
|
||
"target_name": self.target.name,
|
||
"glossary_key": self.target.glossary_key,
|
||
"book": str(self.book_path),
|
||
"glossary": str(self.glossary_path),
|
||
},
|
||
)
|
||
|
||
def _measure_existing_language(self) -> None:
|
||
"""What language does the run directory already hold?"""
|
||
for chapter_dir in sorted(self.run_dir.glob("ch-*")):
|
||
path = chapter_dir / "translation.json"
|
||
if not path.exists():
|
||
continue
|
||
try:
|
||
data = json.loads(path.read_text(encoding="utf-8"))
|
||
except (OSError, json.JSONDecodeError):
|
||
return
|
||
paragraphs = data.get("paragraphs") or {}
|
||
sample = " ".join(str(value) for value in list(paragraphs.values())[:200])
|
||
problem = script_mismatch(self.target.code, sample)
|
||
if problem:
|
||
raise SystemExit(
|
||
f"run directory {self.run_dir} already holds a translation that "
|
||
f"is not {self.target.name}: {problem}. Use a different "
|
||
"--run-dir, or pass --force to overwrite it deliberately."
|
||
)
|
||
return
|
||
|
||
# ----------------------------------------------------------------- passes
|
||
def analyse(self, chapter: dict, chunk: Chunk, previous_summary: str) -> dict:
|
||
path = self.chapter_dir(chapter) / f"{chunk.id}.analysis.json"
|
||
cached = self._read(path)
|
||
if cached is not None:
|
||
return cached
|
||
prompt = analysis_prompt(
|
||
chapter_title=f"{chapter['title']} / {chapter_title_target(chapter)}",
|
||
scene_index=chunk.scene_index,
|
||
chunk_index=int(chunk.id.rsplit("c", 1)[-1]),
|
||
chunks_in_scene=chunk.chunks_in_scene,
|
||
paragraphs=chunk.paragraphs,
|
||
previous_summary=previous_summary,
|
||
glossary=self.glossary,
|
||
target=self.target,
|
||
)
|
||
value, usage = self.llm.chat_json(
|
||
system_analyst(self.target),
|
||
prompt,
|
||
model=self.analyst_model,
|
||
temperature=0.2,
|
||
max_tokens=1400,
|
||
cache_key=f"{chunk.id}.analysis.{self.analyst_model}",
|
||
# The small analyst model rejects LM Studio's JSON-schema grammar;
|
||
# its answer is parsed from the text instead.
|
||
json_mode=False,
|
||
)
|
||
if isinstance(value, list) and value and isinstance(value[0], dict):
|
||
value = value[0]
|
||
# Sometimes wrapped as {"analysis": {...}} or similar
|
||
if isinstance(value, dict):
|
||
for k in ("analysis", "result", "data"):
|
||
if isinstance(value.get(k), dict):
|
||
value = value[k]
|
||
break
|
||
if not isinstance(value, dict):
|
||
# The analyst is a small model asked for a JSON object, and it
|
||
# sometimes answers with a list or a bare string instead. Analysis is
|
||
# guidance for the drafter, not a requirement for a valid translation,
|
||
# so a malformed answer must not abort a chunk that could otherwise be
|
||
# translated. Record that the analysis is degraded and continue with a
|
||
# minimal one; the drafter falls back to the source and glossary.
|
||
print(
|
||
f" {chunk.id}: analyst returned {type(value).__name__}, using a minimal analysis",
|
||
flush=True,
|
||
)
|
||
value = {"degraded": True, "note": "analyst returned non-object JSON"}
|
||
value["_usage"] = usage.__dict__
|
||
self._write(path, value)
|
||
return value
|
||
|
||
def draft(self, chapter: dict, chunk: Chunk, analysis: dict, previous_summary: str) -> dict:
|
||
path = self.chapter_dir(chapter) / f"{chunk.id}.draft.json"
|
||
cached = self._read(path)
|
||
if cached is not None:
|
||
return cached
|
||
expected = [p["id"] for p in chunk.paragraphs if p.get("type") != "scene_break"]
|
||
prompt = draft_prompt(
|
||
chapter_title=f"{chapter['title']} / {chapter_title_target(chapter)}",
|
||
scene_index=chunk.scene_index,
|
||
chunk_index=int(chunk.id.rsplit("c", 1)[-1]),
|
||
chunks_in_scene=chunk.chunks_in_scene,
|
||
paragraphs=chunk.paragraphs,
|
||
analysis={k: v for k, v in analysis.items() if k != "_usage"},
|
||
previous_summary=previous_summary,
|
||
glossary=self.glossary,
|
||
target=self.target,
|
||
)
|
||
value, usage = self.llm.chat_json(
|
||
system_drafter(self.target),
|
||
prompt,
|
||
model=self.model,
|
||
temperature=0.45,
|
||
max_tokens=6000,
|
||
cache_key=f"{chunk.id}.draft",
|
||
)
|
||
out = self._normalise_draft(value, expected)
|
||
missing = [i for i in expected if not out.get(i)]
|
||
if missing:
|
||
out.update(self._draft_missing(chapter, chunk, analysis, missing))
|
||
out = self._repair_residue(chapter, chunk, out)
|
||
out = self._fix_paragraph_quotes(chunk, out)
|
||
warnings: dict[str, dict[str, list[str]]] = {}
|
||
for pid, zh in out.items():
|
||
common, proper = latin_residue(zh)
|
||
if common or proper:
|
||
warnings[pid] = {"common_english": common, "proper_names": proper}
|
||
result = {
|
||
"paragraphs": out,
|
||
"analysis": analysis,
|
||
"warnings": warnings,
|
||
"_usage": usage.__dict__,
|
||
}
|
||
self._write(path, result)
|
||
return result
|
||
|
||
def _draft_text(self, item: dict) -> str:
|
||
"""The translated text of one drafted paragraph.
|
||
|
||
The key is the target language's own, but the model is not always
|
||
disciplined about it, so a neutral key is accepted too.
|
||
"""
|
||
for key in (self.target.glossary_key, "text", "translation", "zh"):
|
||
value = item.get(key)
|
||
if isinstance(value, str) and value.strip():
|
||
return value
|
||
return ""
|
||
|
||
def _normalise_draft(self, value: object, expected: list[str]) -> dict[str, str]:
|
||
out: dict[str, str] = {}
|
||
if isinstance(value, dict) and isinstance(value.get("paragraphs"), list):
|
||
items = value["paragraphs"]
|
||
elif isinstance(value, list):
|
||
items = value
|
||
elif isinstance(value, dict) and value.get("id") and self._draft_text(value):
|
||
items = [value]
|
||
elif isinstance(value, dict):
|
||
# maybe mapping id->text
|
||
for k, v in value.items():
|
||
if isinstance(v, str) and k.startswith("ch-"):
|
||
out[k] = sanitize(v, self.target)
|
||
return {k: v for k, v in out.items() if k in expected and v}
|
||
else:
|
||
raise LLMError(f"drafter returned unexpected JSON: {str(value)[:200]}")
|
||
for item in items:
|
||
if isinstance(item, dict) and item.get("id"):
|
||
text = self._draft_text(item)
|
||
if text:
|
||
out[str(item["id"])] = sanitize(text, self.target)
|
||
out = {k: v for k, v in out.items() if k in expected and v}
|
||
self._check_draft_language(out)
|
||
return out
|
||
|
||
def _check_draft_language(self, out: dict[str, str]) -> None:
|
||
"""Is the draft actually written in the target language?
|
||
|
||
The drafter is told which language to translate into, and it has been
|
||
known to ignore that and write Chinese anyway. A draft in the wrong
|
||
language is not a translation, so it is refused rather than stored: a
|
||
book that is 32 chapters in the wrong language is a failure that is
|
||
expensive to discover late.
|
||
"""
|
||
if not out:
|
||
return
|
||
sample = " ".join(out.values())
|
||
problem = script_mismatch(self.target.code, sample)
|
||
if problem:
|
||
raise LLMError(f"the drafter answered in the wrong language: {problem}. Expected {self.target.name}.")
|
||
|
||
def _fix_paragraph_quotes(self, chunk: Chunk, drafts: dict[str, str]) -> dict[str, str]:
|
||
"""Remove quotes the model wrapped around unquoted narration."""
|
||
for p in chunk.paragraphs:
|
||
zh = drafts.get(p["id"])
|
||
source = (p.get("text") or "").strip()
|
||
if (
|
||
zh
|
||
and p.get("type") == "narration"
|
||
and not source.startswith("“")
|
||
and zh.startswith("“")
|
||
and zh.endswith("”")
|
||
and zh.count("“") == 1
|
||
and zh.count("”") == 1
|
||
):
|
||
drafts[p["id"]] = zh[1:-1].strip()
|
||
return drafts
|
||
|
||
def _repair_residue(self, chapter: dict, chunk: Chunk, drafts: dict[str, str]) -> dict[str, str]:
|
||
"""Translate stray lowercase English words in place, without touching names.
|
||
|
||
Only meaningful when the target is not written in Latin script: in a
|
||
Spanish or French draft a Latin word is the target language, not
|
||
residue. Running the check there would flag the whole book and
|
||
"repair" it into nonsense.
|
||
"""
|
||
if SCRIPT_LATIN in self.target.scripts:
|
||
return drafts
|
||
by_id = {p["id"]: p for p in chunk.paragraphs}
|
||
for pid, zh in list(drafts.items()):
|
||
common, _proper = latin_residue(zh)
|
||
if not common:
|
||
continue
|
||
source = by_id.get(pid, {}).get("text", "")
|
||
prompt = f"""下面这段{self.target.chinese_name}译文里混入了英文单词：{", ".join(common)}。
|
||
请只把这些英文单词译成{self.target.chinese_name}，其余内容尽量保持不变；不要解释，不要改写剧情。
|
||
输出 JSON：{{"{self.target.glossary_key}": "修正后的{self.target.chinese_name}译文"}}
|
||
|
||
英文原文：
|
||
{source}
|
||
|
||
当前译文：
|
||
{zh}"""
|
||
try:
|
||
value, _ = self.llm.chat_json(
|
||
system_drafter(self.target),
|
||
prompt,
|
||
model=self.model,
|
||
temperature=0.2,
|
||
max_tokens=1500,
|
||
cache_key=f"{chunk.id}.{pid}.residue",
|
||
)
|
||
fixed = self._draft_text(value) if isinstance(value, dict) else None
|
||
if fixed:
|
||
drafts[pid] = sanitize(str(fixed), self.target)
|
||
except LLMError as exc:
|
||
print(f" {chunk.id}/{pid}: residue repair failed: {exc}", flush=True)
|
||
return drafts
|
||
|
||
def _draft_missing(self, chapter: dict, chunk: Chunk, analysis: dict, missing: list[str]) -> dict[str, str]:
|
||
"""Last-resort one-paragraph calls. Keeps a chunk failure from losing the run."""
|
||
result: dict[str, str] = {}
|
||
by_id = {p["id"]: p for p in chunk.paragraphs}
|
||
for pid in missing:
|
||
p = by_id[pid]
|
||
try:
|
||
prompt = draft_prompt(
|
||
chapter_title=f"{chapter['title']} / {chapter_title_target(chapter)}",
|
||
scene_index=chunk.scene_index,
|
||
chunk_index=0,
|
||
chunks_in_scene=1,
|
||
paragraphs=[p],
|
||
analysis=analysis,
|
||
previous_summary="",
|
||
glossary=self.glossary,
|
||
target=self.target,
|
||
)
|
||
value, _ = self.llm.chat_json(
|
||
system_drafter(self.target),
|
||
prompt,
|
||
model=self.model,
|
||
temperature=0.4,
|
||
max_tokens=1500,
|
||
cache_key=f"{chunk.id}.{pid}.draft",
|
||
)
|
||
one = self._normalise_draft(value, [pid])
|
||
if one.get(pid):
|
||
result[pid] = one[pid]
|
||
continue
|
||
except Exception:
|
||
pass
|
||
# Fallback to source text or empty to avoid blocking
|
||
result[pid] = sanitize(str(p.get("text", "")), self.target)
|
||
return result
|
||
|
||
def criticise(self, chapter: dict, chunk: Chunk, draft: dict[str, str]) -> dict:
|
||
path = self.chapter_dir(chapter) / f"{chunk.id}.critic.json"
|
||
cached = self._read(path)
|
||
if cached is not None:
|
||
return cached
|
||
prompt = critic_prompt(
|
||
chapter_title=f"{chapter['title']} / {chapter_title_target(chapter)}",
|
||
paragraphs=chunk.paragraphs,
|
||
draft=draft,
|
||
glossary=self.glossary,
|
||
target=self.target,
|
||
)
|
||
try:
|
||
value, usage = self.llm.chat_json(
|
||
system_critic(self.target),
|
||
prompt,
|
||
model=self.critic_model,
|
||
temperature=0.15,
|
||
max_tokens=6000,
|
||
cache_key=f"{chunk.id}.critic.{self.critic_model}",
|
||
)
|
||
except LLMError as exc:
|
||
# A reasoning-only critic produces empty content. Fall back to the
|
||
# drafter model rather than silently skipping the check.
|
||
print(f" critic {self.critic_model} failed ({exc}); falling back to {self.model}", flush=True)
|
||
try:
|
||
value, usage = self.llm.chat_json(
|
||
system_critic(self.target),
|
||
prompt,
|
||
model=self.model,
|
||
temperature=0.15,
|
||
max_tokens=6000,
|
||
cache_key=f"{chunk.id}.critic.{self.model}",
|
||
)
|
||
except LLMError as exc2:
|
||
# Do not abort a multi-hour run because one critique failed.
|
||
print(f" critic fallback failed ({exc2}); recording a skipped critique", flush=True)
|
||
value = {"issues": [], "verdict": "skipped", "_error": str(exc2)}
|
||
usage = Usage(model=self.model, prompt_tokens=0, completion_tokens=0, elapsed_s=0.0)
|
||
if isinstance(value, list):
|
||
value = {"issues": value, "verdict": "revise" if value else "pass"}
|
||
if not isinstance(value, dict):
|
||
raise LLMError(f"{chunk.id}: critic returned {type(value).__name__}")
|
||
# DeepSeek sometimes invents its own key names. Normalise the common
|
||
# aliases so the report and the review queue stay machine-readable.
|
||
normalised = []
|
||
confirmations = (
|
||
"被正确翻译",
|
||
"翻译正确",
|
||
"已正确",
|
||
"无需修改",
|
||
"没有问题",
|
||
"不存在问题",
|
||
"未改变理解",
|
||
"英文原文存在",
|
||
"原文存在语法错误",
|
||
"建议英文",
|
||
"英文原文表达",
|
||
"原文错误",
|
||
"无需修改中文",
|
||
)
|
||
for raw in value.get("issues", []) or []:
|
||
if not isinstance(raw, dict):
|
||
continue
|
||
problem_text = str(raw.get("problem") or raw.get("explanation") or "")
|
||
if any(marker in problem_text for marker in confirmations):
|
||
continue
|
||
fix_text = str(raw.get("fix") or raw.get("suggestion") or "").strip()
|
||
if not fix_text and str(raw.get("severity") or "major").lower() == "minor":
|
||
continue
|
||
normalised.append(
|
||
{
|
||
"id": raw.get("id") or raw.get("paragraph_id") or raw.get("sentence_id") or "",
|
||
"severity": raw.get("severity") or "major",
|
||
"type": raw.get("type") or "meaning",
|
||
"problem": raw.get("problem") or raw.get("explanation") or "",
|
||
"source_excerpt": raw.get("source_excerpt") or raw.get("english_sentence_part") or "",
|
||
"target_excerpt": raw.get("target_excerpt") or raw.get("chinese_sentence_part") or "",
|
||
"fix": raw.get("fix") or raw.get("suggestion") or "",
|
||
}
|
||
)
|
||
value["issues"] = normalised
|
||
verdict = str(value.get("verdict", "")).lower()
|
||
if verdict == "skipped":
|
||
value["verdict"] = "skipped"
|
||
elif normalised or verdict in {"fail", "revise"}:
|
||
value["verdict"] = "revise"
|
||
else:
|
||
value["verdict"] = "pass"
|
||
value["_usage"] = usage.__dict__
|
||
self._write(path, value)
|
||
return value
|
||
|
||
def consistency(self, chapter: dict, drafts: dict[str, str], paragraphs: list[dict]) -> dict:
|
||
path = self.chapter_dir(chapter) / "consistency.json"
|
||
cached = self._read(path)
|
||
if cached is not None:
|
||
return cached
|
||
prompt = consistency_prompt(
|
||
chapter_title=f"{chapter['title']} / {chapter_title_target(chapter)}",
|
||
paragraphs=paragraphs,
|
||
draft=drafts,
|
||
glossary=self.glossary,
|
||
character_notes=CHARACTER_NOTES,
|
||
target=self.target,
|
||
)
|
||
value, usage = self.llm.chat_json(
|
||
system_critic(self.target),
|
||
prompt,
|
||
model=self.critic_model,
|
||
temperature=0.15,
|
||
max_tokens=6000,
|
||
cache_key=f"{chapter['id']}.consistency",
|
||
)
|
||
if isinstance(value, list):
|
||
value = {"issues": value}
|
||
if not isinstance(value, dict):
|
||
raise LLMError(f"{chapter['id']}: consistency returned {type(value).__name__}")
|
||
value["_usage"] = usage.__dict__
|
||
self._write(path, value)
|
||
return value
|
||
|
||
# ----------------------------------------------------------------- driver
|
||
def run(self, chapters: list[dict]) -> None:
|
||
started = time.time()
|
||
for ci, chapter in enumerate(chapters, 1):
|
||
chunks = chunk_chapter(chapter, self.max_words, self.max_paras)
|
||
print(
|
||
f"[{ci}/{len(chapters)}] {chapter['id']} {chapter['title']} / {chapter_title_target(chapter)}: "
|
||
f"{len(chapter['paragraphs'])} paragraphs, {len(chunks)} chunks",
|
||
flush=True,
|
||
)
|
||
previous_summary = ""
|
||
drafts: dict[str, str] = {}
|
||
issues: list[dict] = []
|
||
for chunk in chunks:
|
||
analysis = self.analyse(chapter, chunk, previous_summary)
|
||
previous_summary = str(analysis.get("summary") or analysis.get("summary_zh") or previous_summary)
|
||
drafted = self.draft(chapter, chunk, analysis, previous_summary)
|
||
drafts.update(drafted["paragraphs"])
|
||
if self.do_critic and (self.critic_limit <= 0 or int(chunk.id.rsplit("c", 1)[-1]) < self.critic_limit):
|
||
crit = self.criticise(chapter, chunk, drafted["paragraphs"])
|
||
for issue in crit.get("issues", []) or []:
|
||
issue = dict(issue)
|
||
issue["chunk"] = chunk.id
|
||
issues.append(issue)
|
||
print(
|
||
f" {chunk.id}: {len(drafted['paragraphs'])}/{len([p for p in chunk.paragraphs if p.get('type') != 'scene_break'])} "
|
||
f"paras, {len(issues)} issue(s) so far",
|
||
flush=True,
|
||
)
|
||
if self.do_consistency:
|
||
cons = self.consistency(chapter, drafts, chapter["paragraphs"])
|
||
for issue in cons.get("issues", []) or []:
|
||
issue = dict(issue)
|
||
issue["type"] = "consistency:" + str(issue.get("type"))
|
||
issues.append(issue)
|
||
summary = {
|
||
"chapter_id": chapter["id"],
|
||
"chapter_index": chapter["index"],
|
||
"title": chapter["title"],
|
||
"title_target": chapter_title_target(chapter),
|
||
"target_language": self.target.code,
|
||
"paragraphs": drafts,
|
||
"issues": issues,
|
||
"chunks": [c.id for c in chunks],
|
||
}
|
||
self._write(self.chapter_dir(chapter) / "translation.json", summary)
|
||
print(f" wrote {self.chapter_dir(chapter) / 'translation.json'}", flush=True)
|
||
print(f"done in {time.time() - started:.0f}s", flush=True)
|
||
|
||
|
||
def report(run_dir: Path) -> str:
|
||
lines = []
|
||
total_paras = 0
|
||
total_issues = 0
|
||
for d in sorted(p for p in run_dir.iterdir() if p.is_dir() and p.name.startswith("ch-")):
|
||
t = d / "translation.json"
|
||
if not t.exists():
|
||
lines.append(f"{d.name}: not started")
|
||
continue
|
||
data = json.loads(t.read_text(encoding="utf-8"))
|
||
n = len(data.get("paragraphs", {}))
|
||
issues = data.get("issues", [])
|
||
total_paras += n
|
||
total_issues += len(issues)
|
||
by_sev: dict[str, int] = {}
|
||
for i in issues:
|
||
by_sev[i.get("severity", "?")] = by_sev.get(i.get("severity", "?"), 0) + 1
|
||
lines.append(f"{d.name}: {n} paragraphs, {len(issues)} issues {by_sev}")
|
||
lines.append(f"TOTAL: {total_paras} paragraphs, {total_issues} issues")
|
||
return "\n".join(lines)
|
||
|
||
|
||
def main() -> None:
|
||
ap = argparse.ArgumentParser(description="Run the Deal Hunter translation pipeline.")
|
||
ap.add_argument("--book", type=Path, required=True)
|
||
ap.add_argument("--glossary", type=Path, required=True)
|
||
ap.add_argument("--run-dir", type=Path, required=True)
|
||
ap.add_argument(
|
||
"--target-language",
|
||
default=None,
|
||
help=(
|
||
"the language to translate into (zh-Hans, es-ES, fr-FR, de-DE, "
|
||
"it-IT, pt-BR, ru-RU, ja-JP, ko-KR, ar-SA). Defaults to the "
|
||
"book's own target_language."
|
||
),
|
||
)
|
||
ap.add_argument("--chapters", default="all", help="all, front, body, back, or comma-separated indices/ids")
|
||
ap.add_argument("--model", default=DEFAULT_MODEL)
|
||
ap.add_argument("--analyst-model", default=DEFAULT_ANALYST)
|
||
ap.add_argument("--critic-model", default=DEFAULT_CRITIC)
|
||
ap.add_argument("--base-url", default=None)
|
||
ap.add_argument("--force", action="store_true")
|
||
ap.add_argument("--no-critic", action="store_true")
|
||
ap.add_argument("--consistency", action="store_true")
|
||
ap.add_argument("--max-words", type=int, default=260)
|
||
ap.add_argument("--max-paras", type=int, default=14)
|
||
ap.add_argument(
|
||
"--critic-limit", type=int, default=0, help="criticise only the first N chunks per chapter (0 = all)"
|
||
)
|
||
ap.add_argument("--report", action="store_true")
|
||
args = ap.parse_args()
|
||
|
||
if args.report:
|
||
print(report(args.run_dir))
|
||
return
|
||
|
||
book = load_book(args.book)
|
||
target_language = args.target_language or book.get("target_language") or "zh-Hans"
|
||
chapters = select_chapters(book, args.chapters)
|
||
runner = Runner(
|
||
book_path=args.book,
|
||
glossary_path=args.glossary,
|
||
run_dir=args.run_dir,
|
||
model=args.model,
|
||
analyst_model=args.analyst_model,
|
||
critic_model=args.critic_model,
|
||
base_url=args.base_url,
|
||
force=args.force,
|
||
critic=not args.no_critic,
|
||
consistency=args.consistency,
|
||
max_words=args.max_words,
|
||
max_paras=args.max_paras,
|
||
critic_limit=args.critic_limit,
|
||
target_language=target_language,
|
||
)
|
||
runner.run(chapters)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|