diff --git a/tools/translation/README.md b/tools/translation/README.md index c385ae7..9020f56 100644 --- a/tools/translation/README.md +++ b/tools/translation/README.md @@ -33,6 +33,7 @@ Its sha256 is recorded in `Status.txt` and in `source/book.json`. | 2 | Context-aware draft translation with glossary and analysis | `qwen3.5-9b` | | 3 | Chapter voice/term consistency (optional) | critic model | | 4 | Bilingual critic: meaning, pronoun, irony, term, omission | `deepseek-r1-8b`, fallback `qwen3.5-9b` | +| 5 | Align the translation back onto the verified source utterances | drafter model | Every analysis, draft and critique is a file under `runs//`. A run can be stopped and resumed; existing artefacts are reused unless `--force` is @@ -44,6 +45,59 @@ character voices, the scene analysis, and the previous chunk's summary. The critic uses a different model and returns a severity-sorted issue list; it does not rewrite the text. +## Pass 5 - aligning back to the utterances + +Passes 0-4 work on *paragraphs*, because epub and pdf are paragraph media. +The audiobook needs *utterances*: UmbraAudio has already verified who speaks +every one of them, and every clip is voiced separately by that speaker's +voice. So the last pass divides each translated paragraph between the source +utterances it was built from. + +The division is by meaning, not by position. An intent-based translation +merges and splits sentences for the target language's sake, so the sentence +counts do not line up on their own - that is the whole point of Passes 1-2. +A model is given the source utterances and the translation and asked which +part of the translation renders which source utterance. + +Two rules the model cannot break: + +- a source utterance may only be absorbed into another utterance **by the same + speaker** - a line must not be handed to a different character; +- a scene break is silence, and silence has no translation. + +The partition is checked against the translated text: the segments must +reproduce it exactly once whitespace is removed. A model answer that drops, +duplicates or invents text is rejected and the paragraph falls back to a +mechanical split, which is reported as needing review. **The alignment never +rewrites the translation** - it only decides where the boundaries fall. + +When the target absorbed a source utterance, that utterance is recorded as +*not voiced* with the reason. UmbraAudio's `utterance_translations.voiced` +flag is what carries it: the row is kept so the gap is visible and +reconstructable, and no clip is generated for it. + +```bash +python3 tools/translation/align.py \ + --book productions/deal-hunter-zh/source/book.json \ + --run-dir productions/deal-hunter-zh/runs \ + --bundle /home/umbra/audio-read/bundles/dh-bundle/manifest.jsonl \ + --out productions/deal-hunter-zh/aligned/delivery.json \ + --chapters all +``` + +The output is a delivery payload in the shape UmbraAudio's +`translations.import_translations` accepts, so the translation can be +ingested without either side inventing a second contract: + +```bash +python scripts/import_translations.py --project "DealHunterShip_4.0e_whole-to-audio" \ + --delivery productions/deal-hunter-zh/aligned/delivery.json --write +``` + +`--no-llm` runs the mechanical split only. It never loses text, but it cannot +tell which clause renders which English sentence, so everything it touches is +reported as needing review. + ## Commands Extract the source: @@ -121,6 +175,6 @@ Chinese numerals (`第一章　打捞`). Body text is 11 pt Noto Serif CJK SC wi catch mechanical defects; a human editor is still required. - It does not recover the original DOCX paragraphing. The bundle is sentence-level, so paragraphs are reconstructed and can be reviewed. -- It does not produce a 1:1 utterance alignment for TTS. Source ranges are - preserved on every paragraph; splitting or aligning back to the 4,319 - utterances is a separate pass. +- It does not produce the audio. The alignment pass hands UmbraAudio one row + per utterance; voicing, loudness and the audiobook masters are UmbraAudio's + job. diff --git a/tools/translation/align.py b/tools/translation/align.py new file mode 100644 index 0000000..8978a46 --- /dev/null +++ b/tools/translation/align.py @@ -0,0 +1,733 @@ +# Copyright 2026 Umbra Tools (umbratools.com) +# SPDX-License-Identifier: Apache-2.0 + +"""Map the translated paragraphs back onto the verified source utterances. + +The gap this closes +------------------- +The translation pipeline works on *paragraphs*. Pass 0 groups consecutive +narration sentences into a readable paragraph and Passes 1-4 translate that +paragraph. The reading edition needs paragraphs: epub and pdf are paragraph +media, and a paragraph is what a reader sees. + +The audiobook needs *utterances*. UmbraAudio has already verified who speaks +every one of the book's utterances, and every clip is voiced separately by +that speaker's own voice. A paragraph-level translation therefore cannot be +spoken: it has to be cut back up, or it gets spoken by the wrong voice. + +``tools/translation/README.md`` said so plainly: *"It does not produce a 1:1 +utterance alignment for TTS. Source ranges are preserved on every paragraph; +splitting or aligning back to the 4,319 utterances is a separate pass."* + +This is that pass. + +What it does +------------ +For every source paragraph it takes the translated paragraph and assigns the +translated text back to the individual source utterances the paragraph was +built from: + +* one source utterance -> the translation is copied across unchanged; +* several source utterances -> the translation is divided between them. + +The division is by meaning, not by position. A sentence-level translator +would have produced one target sentence per source sentence and the counts +would line up on their own. An intent-based translator does not: it merges +two English sentences into one Chinese sentence, or splits one into two, +because that is what the target language wants. So the division is done by a +model that is given the source utterances and the translation and asked which +part of the translation renders which source utterance. + +Every source utterance gets exactly one segment, so the alignment can never +silently lose a line. When the target absorbed a source utterance, the +segment is recorded as *not voiced* with the reason, which is what +UmbraAudio's ``utterance_translations.voiced`` flag is for: the row is kept +so the gap is visible and reconstructable, and no clip is generated for it. + +Two rules the model cannot break: + +* a source utterance may only be absorbed into another utterance by the same + speaker - a line must not be handed to a different character; +* a scene break is silence, and silence has no translation. + +Nothing here rewrites the translation. The partition is checked against the +translated text: the segments must reproduce it exactly once whitespace is +removed. A model answer that drops, duplicates or invents text is rejected +and the paragraph falls back to a mechanical split, which is reported as +needing review. + +The output is a delivery payload in the shape UmbraAudio's +``translations.import_translations`` accepts, so neither side has to invent a +second contract. +""" + +from __future__ import annotations + +import argparse +import json +import re +import sys +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any + +from llm import LLM, LLMError + +# The aligner compares English source against translated text, which is the +# critic's job, but it has to *rewrite* the partition rather than report on +# it. The drafter model is the one that produced the target text and knows +# its sentence boundaries best, so it is the default. +DEFAULT_ALIGN_MODEL = "qwen/qwen3.5-9b" + +SYSTEM_ALIGNER = """你是一位双语对齐编辑，负责把已经译好的中文段落还原到英文原文的每一条话语上。 +你的任务不是翻译，也不是改写，而是判断：中文译文里的哪一部分对应英文原文里的哪一条话语。 +译文为了中文的自然可能合并或拆分了句子，所以不能按句子序号机械对应，要按意思对应。 +每一条英文话语都必须恰好出现一次；被合并进别条话语的，用 merged_into 指明并把 target_text 留空。 +只输出 JSON，不要解释，不要 Markdown 代码围栏。""" + +_SENTENCE_TERMINATORS = "。！？!?；;…" +_CLOSING_MARKS = "”’」』）)】》\"'" + + +def normalise_target_language(value: str) -> str: + """``zh-CN`` -> ``zh``. + + UmbraAudio stores the dropdown code, not the locale, so the payload has + to carry the base code or the delivery lands in a language the review UI + does not offer. + """ + return value.strip().split("-")[0].split("_")[0].lower() + + +def parse_row_id(source_id: str) -> int | None: + """``utt-83578`` -> ``83578``. + + The bundle names an utterance by its UmbraAudio database row, so the + numeric suffix is the id UmbraAudio stores the translation against. + """ + match = re.search(r"(\d+)\s*$", source_id or "") + return int(match.group(1)) if match else None + + +def _compact(text: str) -> tuple[str, list[int]]: + """Strip whitespace, remembering where each kept character came from.""" + chars: list[str] = [] + positions: list[int] = [] + for index, ch in enumerate(text): + if not ch.isspace(): + chars.append(ch) + positions.append(index) + return "".join(chars), positions + + +def split_target_sentences(text: str) -> list[str]: + """Split translated text into sentences, keeping the terminator. + + Chinese terminates with a full stop, and a closing quote may follow it, + so the closing marks are pulled into the sentence they close. + """ + sentences: list[str] = [] + buffer: list[str] = [] + index = 0 + while index < len(text): + ch = text[index] + buffer.append(ch) + index += 1 + if ch in _SENTENCE_TERMINATORS: + while index < len(text) and text[index] in _CLOSING_MARKS: + buffer.append(text[index]) + index += 1 + piece = "".join(buffer).strip() + if piece: + sentences.append(piece) + buffer = [] + tail = "".join(buffer).strip() + if tail: + sentences.append(tail) + return sentences + + +def _split_at_comma(text: str) -> list[str] | None: + """Split one sentence in two at its last clause boundary. + + An intent-based translation merges English sentences with a comma, a + colon or a dash rather than with a full stop, because that is what reads + naturally in Chinese. So when the target has fewer sentences than the + source has utterances, the boundary is a clause boundary. + """ + for mark in ("——", "—", "：", "，", "、", "；", ",", ";"): + positions = [m.start() for m in re.finditer(re.escape(mark), text)] + for position in reversed(positions): + if position <= 0: + continue + end = position + len(mark) + if end >= len(text): + # A mark at the very end has nothing after it to split off. + continue + head = text[:end].strip() + tail = text[end:].strip() + if head and tail: + return [head, tail] + return None + + +def mechanical_split(text: str, count: int) -> list[str] | None: + """Divide *text* into exactly *count* pieces, or return None. + + This is the fallback when the model cannot be trusted. It is not good + enough to publish - a mechanical split cannot tell which Chinese clause + renders which English sentence - but it never loses or invents text, and + every paragraph it touches is reported as needing review. + """ + if count <= 0: + return None + if count == 1: + return [text.strip()] if text.strip() else None + + pieces = split_target_sentences(text) + if not pieces: + return None + + while len(pieces) > count: + # Merge the adjacent pair with the smallest combined length, so the + # result stays as close to the original sentence boundaries as it can. + best = 0 + best_length = None + for i in range(len(pieces) - 1): + length = len(pieces[i]) + len(pieces[i + 1]) + if best_length is None or length < best_length: + best = i + best_length = length + pieces[best : best + 2] = [(pieces[best] + pieces[best + 1]).strip()] + pieces = [p for p in pieces if p] + + while len(pieces) < count: + # Split the longest piece that can actually be split. The longest + # piece is not always the splittable one: a Chinese sentence with no + # clause boundary can be longer than one with three commas in it. + ordered = sorted(range(len(pieces)), key=lambda i: len(pieces[i]), reverse=True) + target_index = None + replacement = None + for index in ordered: + candidate = _split_at_comma(pieces[index]) + if candidate is not None: + target_index = index + replacement = candidate + break + if target_index is None: + return None + pieces[target_index : target_index + 1] = replacement + + if len(pieces) != count or not all(p.strip() for p in pieces): + return None + return pieces + + +def repartition(translation: str, anchors: list[str]) -> list[str] | None: + """Cut *translation* into exactly ``len(anchors)`` parts using *anchors*. + + The anchors are the model's proposed segment texts. They are located in + the translation and the gaps between them are given to the preceding + part, so the parts concatenate back to the translation exactly. That is + what makes the alignment checkable: the translation is partitioned, never + rewritten. + + Returns None when an anchor cannot be found, which means the model + invented or dropped text. + """ + if not anchors or len(anchors) == 0: + return None + compact_text, positions = _compact(translation) + if not compact_text: + return None + + compact_anchors: list[str] = [] + for anchor in anchors: + compact_anchor, _ = _compact(anchor) + if not compact_anchor: + return None + compact_anchors.append(compact_anchor) + + cursor = 0 + ends: list[int] = [] + for anchor in compact_anchors: + found = compact_text.find(anchor, cursor) + if found < 0: + return None + cursor = found + len(anchor) + ends.append(cursor) + if cursor != len(compact_text): + # The model left content behind. The tail belongs to the last part. + ends[-1] = len(compact_text) + + # Turn the compact-space boundaries back into offsets in the original. + boundaries: list[int] = [] + for end in ends: + if end <= 0: + boundaries.append(0) + elif end >= len(positions): + boundaries.append(len(translation)) + else: + boundaries.append(positions[end - 1] + 1) + + parts: list[str] = [] + start = 0 + for boundary in boundaries: + parts.append(translation[start:boundary]) + start = boundary + parts = [p.strip() for p in parts] + if len(parts) != len(anchors) or not all(parts): + return None + return parts + + +@dataclass +class SourceUtterance: + """One verified source utterance from the UmbraAudio bundle.""" + + source_id: str + row_id: int | None + index: int + speaker: str + type: str + text: str + + @property + def voice(self) -> str: + return (self.speaker or "narrator").strip().lower() + + +@dataclass +class Segment: + """The part of the translation that renders one source utterance.""" + + source_id: str + text: str = "" + voiced: bool = True + merged_into: str = "" + reason: str = "" + + +@dataclass +class ParagraphAlignment: + """How one translated paragraph was divided between its source utterances.""" + + paragraph_id: str + method: str + segments: list[Segment] = field(default_factory=list) + problems: list[str] = field(default_factory=list) + + @property + def needs_review(self) -> bool: + return bool(self.problems) or self.method in {"mechanical", "failed"} + + +def source_utterances(paragraph: dict, bundle: dict[str, dict]) -> list[SourceUtterance]: + """The verified utterances a paragraph was built from. + + The bundle is the authority: it carries the text, the speaker UmbraAudio + resolved, and the database row the translation will be stored against. + """ + out: list[SourceUtterance] = [] + source_ids = paragraph.get("source_ids") or [] + source_indices = paragraph.get("source_indices") or [] + for position, source_id in enumerate(source_ids): + record = bundle.get(source_id) or {} + index = source_indices[position] if position < len(source_indices) else position + row_id = record.get("database_row_id") + if row_id is None: + row_id = parse_row_id(source_id) + out.append( + SourceUtterance( + source_id=source_id, + row_id=int(row_id) if row_id is not None else None, + index=int(index), + speaker=record.get("speaker_name_resolved") or paragraph.get("speaker") or "narrator", + type=record.get("utterance_type") or paragraph.get("type") or "narration", + text=(record.get("text") or "").strip(), + ) + ) + return out + + +def alignment_prompt(paragraph_id: str, sources: list[SourceUtterance], translation: str) -> str: + """The prompt that asks for the partition of a translated paragraph.""" + lines = [] + for source in sources: + lines.append(f"[{source.source_id}] (说话者：{source.speaker}) {source.text}") + return f"""段落 id：{paragraph_id} + +英文原文（每一条都是一条已经确认说话者的话语）： +{chr(10).join(lines)} + +中文译文： +{translation} + +请把中文译文分配回上面每一条英文话语。输出 JSON： +{{ + "segments": [ + {{"source_id": "utt-xxxxx", "target_text": "对应的中文片段"}}, + {{"source_id": "utt-yyyyy", "target_text": "", "merged_into": "utt-xxxxx"}} + ] +}} + +要求： +- segments 必须为原文的每一条话语各给一项，source_id 一一对应，不多不少。 +- target_text 必须是中文译文里**原样出现**的连续片段，按原文顺序拼接起来必须等于整段译文，不得删改一个字。 +- 如果译文把两条原文合并成了一句，被合并的那条 target_text 留空，用 merged_into 指明它并入了哪一条。 +- merged_into 只能指向同一条段落里、同一说话者、且有 target_text 的话语。 +- 不要翻译，不要润色，不要补充，只做对应。""" + + +def parse_segments( + data: Any, + sources: list[SourceUtterance], + translation: str, +) -> tuple[list[Segment], list[str]]: + """Validate a model answer and turn it into segments. + + Returns ``(segments, problems)``. A non-empty problem list means the + answer was rejected; the caller falls back to a mechanical split. + """ + problems: list[str] = [] + raw = data.get("segments") if isinstance(data, dict) else None + if not isinstance(raw, list): + return [], ["model answer has no segments list"] + + by_id = {source.source_id: source for source in sources} + seen: dict[str, dict] = {} + for item in raw: + if not isinstance(item, dict): + problems.append("segment is not an object") + continue + source_id = str(item.get("source_id") or "") + if source_id not in by_id: + problems.append(f"unknown source_id {source_id!r}") + continue + if source_id in seen: + problems.append(f"source_id {source_id!r} appears twice") + continue + seen[source_id] = item + + missing = [source.source_id for source in sources if source.source_id not in seen] + if missing: + problems.append(f"missing segments for {', '.join(missing)}") + if problems: + return [], problems + + merged: dict[str, str] = {} + for source_id, item in seen.items(): + target = str(item.get("merged_into") or "").strip() + if target: + if target not in by_id: + problems.append(f"{source_id} merged into unknown {target!r}") + continue + if target == source_id: + problems.append(f"{source_id} merged into itself") + continue + if by_id[target].voice != by_id[source_id].voice: + problems.append( + f"{source_id} ({by_id[source_id].speaker}) may not be merged into " + f"{target} ({by_id[target].speaker}): a different speaker's line" + ) + continue + merged[source_id] = target + + for source_id, target in merged.items(): + if target in merged: + problems.append(f"{source_id} merged into {target}, which is itself merged") + if problems: + return [], problems + + segments: list[Segment] = [] + anchors: list[str] = [] + for source in sources: + item = seen[source.source_id] + if source.source_id in merged: + target = merged[source.source_id] + segments.append( + Segment( + source_id=source.source_id, + text="", + voiced=False, + merged_into=target, + reason=f"absorbed into {target} in the translated paragraph", + ) + ) + continue + text = str(item.get("target_text") or "").strip() + if not text: + problems.append(f"{source.source_id} has no target text and no merge target") + continue + anchors.append(text) + segments.append(Segment(source_id=source.source_id, text=text)) + + if problems: + return [], problems + + parts = repartition(translation, anchors) + if parts is None: + return [], ["model answer does not reproduce the translated text"] + + # Hand the exact slices back to the segments, in source order. + position = 0 + for segment in segments: + if not segment.voiced: + continue + segment.text = parts[position] + position += 1 + return segments, [] + + +def align_paragraph( + paragraph: dict, + translation: str, + bundle: dict[str, dict], + llm: LLM | None = None, + model: str = DEFAULT_ALIGN_MODEL, +) -> ParagraphAlignment: + """Divide one translated paragraph between its source utterances.""" + paragraph_id = str(paragraph.get("id") or "") + sources = source_utterances(paragraph, bundle) + + if paragraph.get("type") == "scene_break": + return ParagraphAlignment(paragraph_id=paragraph_id, method="scene_break") + + if not sources: + return ParagraphAlignment( + paragraph_id=paragraph_id, + method="failed", + problems=["paragraph has no source utterances"], + ) + + if len(sources) == 1: + return ParagraphAlignment( + paragraph_id=paragraph_id, + method="one_to_one", + segments=[Segment(source_id=sources[0].source_id, text=translation.strip())], + ) + + if llm is not None: + try: + data, _usage = llm.chat_json( + SYSTEM_ALIGNER, + alignment_prompt(paragraph_id, sources, translation), + model=model, + temperature=0.0, + cache_key=f"align-{paragraph_id}", + ) + segments, problems = parse_segments(data, sources, translation) + if not problems: + return ParagraphAlignment(paragraph_id=paragraph_id, method="llm", segments=segments) + except LLMError as exc: + problem = f"alignment model failed: {exc}" + else: + problem = "; ".join(problems) + mechanical = mechanical_split(translation, len(sources)) + if mechanical is None: + return ParagraphAlignment(paragraph_id=paragraph_id, method="failed", problems=[problem]) + return ParagraphAlignment( + paragraph_id=paragraph_id, + method="mechanical", + segments=[ + Segment(source_id=source.source_id, text=text) for source, text in zip(sources, mechanical, strict=True) + ], + problems=[problem], + ) + + mechanical = mechanical_split(translation, len(sources)) + if mechanical is None: + return ParagraphAlignment( + paragraph_id=paragraph_id, + method="failed", + problems=["cannot divide the translated paragraph between its source utterances"], + ) + return ParagraphAlignment( + paragraph_id=paragraph_id, + method="mechanical", + segments=[ + Segment(source_id=source.source_id, text=text) for source, text in zip(sources, mechanical, strict=True) + ], + ) + + +def align_chapter( + chapter: dict, + translated: dict[str, str], + bundle: dict[str, dict], + llm: LLM | None = None, + model: str = DEFAULT_ALIGN_MODEL, +) -> tuple[list[ParagraphAlignment], dict[str, Any]]: + """Align every translated paragraph of one chapter.""" + alignments: list[ParagraphAlignment] = [] + for paragraph in chapter.get("paragraphs") or []: + paragraph_id = str(paragraph.get("id") or "") + if paragraph.get("type") == "scene_break": + continue + translation = translated.get(paragraph_id) + if translation is None or not translation.strip(): + alignments.append( + ParagraphAlignment( + paragraph_id=paragraph_id, + method="failed", + problems=["paragraph has no translation"], + ) + ) + continue + alignments.append(align_paragraph(paragraph, translation, bundle, llm=llm, model=model)) + + methods: dict[str, int] = {} + problems: list[str] = [] + for alignment in alignments: + methods[alignment.method] = methods.get(alignment.method, 0) + 1 + for problem in alignment.problems: + problems.append(f"{alignment.paragraph_id}: {problem}") + + report = { + "chapter_id": chapter.get("id"), + "paragraphs": len(alignments), + "methods": methods, + "needs_review": sum(1 for a in alignments if a.needs_review), + "problems": problems, + } + return alignments, report + + +def build_delivery( + book: dict, + chapters: list[tuple[dict, list[ParagraphAlignment]]], + bundle: dict[str, dict], + *, + source_ref: str = "", +) -> dict[str, Any]: + """The payload UmbraAudio's ``import_translations`` accepts.""" + target = normalise_target_language(str(book.get("target_language") or "")) + utterances: list[dict[str, Any]] = [] + unmatched: list[str] = [] + + for chapter, alignments in chapters: + bundle_ids = {str(paragraph.get("id") or ""): paragraph for paragraph in chapter.get("paragraphs") or []} + for alignment in alignments: + paragraph = bundle_ids.get(alignment.paragraph_id) or {} + sources = {source.source_id: source for source in source_utterances(paragraph, bundle)} + for segment in alignment.segments: + source = sources.get(segment.source_id) + if source is None or source.row_id is None: + unmatched.append(f"{alignment.paragraph_id}:{segment.source_id}") + continue + if segment.voiced: + utterances.append( + { + "utterance_id": source.row_id, + "text": segment.text, + "voiced": True, + "source_ref": source_ref or alignment.paragraph_id, + } + ) + else: + utterances.append( + { + "utterance_id": source.row_id, + "voiced": False, + "reason": segment.reason, + "source_ref": source_ref or alignment.paragraph_id, + } + ) + + return { + "target_language": target, + "source_system": "umbratrans", + "utterances": utterances, + "unmatched": unmatched, + } + + +def load_bundle(path: Path) -> dict[str, dict]: + """Read the UmbraAudio controller bundle, keyed by utterance id.""" + bundle: dict[str, dict] = {} + with path.open("r", encoding="utf-8") as handle: + for line in handle: + line = line.strip() + if not line: + continue + record = json.loads(line) + if record.get("kind") == "utterance": + bundle[str(record.get("id"))] = record + return bundle + + +def load_translations(run_dir: Path, chapter_id: str) -> dict[str, str]: + """The translated paragraphs of one chapter.""" + path = run_dir / chapter_id / "translation.json" + if not path.exists(): + return {} + data = json.loads(path.read_text(encoding="utf-8")) + paragraphs = data.get("paragraphs") or {} + return {str(key): str(value) for key, value in paragraphs.items()} + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser( + description="Map the translated paragraphs back onto the verified source utterances." + ) + parser.add_argument("--book", type=Path, required=True) + parser.add_argument("--run-dir", type=Path, required=True) + parser.add_argument("--out", type=Path, required=True) + parser.add_argument("--bundle", type=Path, default=None) + parser.add_argument("--chapters", default="all") + parser.add_argument("--model", default=DEFAULT_ALIGN_MODEL) + parser.add_argument("--base-url", default=None) + parser.add_argument("--no-llm", action="store_true", help="use the mechanical split only") + parser.add_argument("--source-ref", default="") + args = parser.parse_args(argv) + + from pipeline import select_chapters # noqa: PLC0415 - keeps the tools importable standalone + + book = json.loads(args.book.read_text(encoding="utf-8")) + bundle_path = args.bundle or Path(str(book.get("bundle") or "")) + if not bundle_path.exists(): + print(f"bundle not found: {bundle_path}", file=sys.stderr) + return 2 + bundle = load_bundle(bundle_path) + + llm = None + if not args.no_llm: + llm = LLM(base_url=args.base_url) if args.base_url else LLM() + llm.cache_dir = args.run_dir / "cache" + + selected = select_chapters(book, args.chapters) + chapters: list[tuple[dict, list[ParagraphAlignment]]] = [] + reports: list[dict[str, Any]] = [] + + for chapter in selected: + chapter_id = str(chapter.get("id") or "") + translated = load_translations(args.run_dir, chapter_id) + alignments, report = align_chapter(chapter, translated, bundle, llm=llm, model=args.model) + chapters.append((chapter, alignments)) + reports.append(report) + print( + f"{chapter_id}: {report['paragraphs']} paragraphs, " + f"methods={report['methods']}, needs review={report['needs_review']}", + flush=True, + ) + + delivery = build_delivery(book, chapters, bundle, source_ref=args.source_ref) + args.out.parent.mkdir(parents=True, exist_ok=True) + args.out.write_text( + json.dumps(delivery, ensure_ascii=False, indent=2) + "\n", + encoding="utf-8", + ) + + voiced = sum(1 for u in delivery["utterances"] if u.get("voiced")) + collapsed = len(delivery["utterances"]) - voiced + print( + f"delivery: {len(delivery['utterances'])} utterances " + f"({voiced} voiced, {collapsed} collapsed), {len(delivery['unmatched'])} unmatched" + ) + print(f"wrote {args.out}") + print(json.dumps({"chapters": reports}, ensure_ascii=False, indent=2)) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tools/translation/tests/test_align.py b/tools/translation/tests/test_align.py new file mode 100644 index 0000000..9573ffd --- /dev/null +++ b/tools/translation/tests/test_align.py @@ -0,0 +1,335 @@ +# Copyright 2026 Umbra Tools (umbratools.com) +# SPDX-License-Identifier: Apache-2.0 + +"""Tests for the utterance alignment pass. + +The alignment model is an LLM, so the tests cover the deterministic parts: +the mechanical split, the repartition that guarantees the translation is +never rewritten, the validation rules that stop a line being handed to a +different speaker, and the delivery payload UmbraAudio ingests. +""" + +from __future__ import annotations + +import sys +import unittest +from pathlib import Path + +HERE = Path(__file__).resolve().parent +sys.path.insert(0, str(HERE.parent)) + +from align import ( # noqa: E402 + Segment, + align_paragraph, + build_delivery, + mechanical_split, + normalise_target_language, + parse_row_id, + parse_segments, + repartition, + source_utterances, + split_target_sentences, +) + + +def bundle_fixture() -> dict[str, dict]: + """Three narration utterances and one dialogue, as UmbraAudio has them.""" + return { + "utt-100": { + "id": "utt-100", + "database_row_id": 100, + "utterance_index": 0, + "speaker_name_resolved": "narrator", + "utterance_type": "narration", + "text": "The ship was large.", + }, + "utt-101": { + "id": "utt-101", + "database_row_id": 101, + "utterance_index": 1, + "speaker_name_resolved": "narrator", + "utterance_type": "narration", + "text": "The designer had laid it out carefully.", + }, + "utt-102": { + "id": "utt-102", + "database_row_id": 102, + "utterance_index": 2, + "speaker_name_resolved": "Kainda Walters Foster Murphy", + "utterance_type": "dialogue", + "text": "I want to see the bridge.", + }, + } + + +def paragraph_fixture() -> dict: + return { + "id": "ch-1-p0000", + "speaker": "narrator", + "type": "narration", + "text": "The ship was large. The designer had laid it out carefully.", + "source_indices": [0, 1], + "source_ids": ["utt-100", "utt-101"], + } + + +class LanguageTests(unittest.TestCase): + def test_locale_becomes_base_code(self) -> None: + self.assertEqual(normalise_target_language("zh-CN"), "zh") + self.assertEqual(normalise_target_language("es-ES"), "es") + + def test_row_id_is_parsed_from_the_bundle_id(self) -> None: + self.assertEqual(parse_row_id("utt-83578"), 83578) + self.assertIsNone(parse_row_id("utt-abc")) + + +class SentenceTests(unittest.TestCase): + def test_chinese_sentences_keep_their_terminator(self) -> None: + self.assertEqual( + split_target_sentences("这艘船很大。设计师很用心。"), + ["这艘船很大。", "设计师很用心。"], + ) + + def test_closing_quote_stays_with_its_sentence(self) -> None: + self.assertEqual(split_target_sentences("“你好。”然后她走了。"), ["“你好。”", "然后她走了。"]) + + def test_text_without_a_terminator_is_one_sentence(self) -> None: + self.assertEqual(split_target_sentences("没有句号"), ["没有句号"]) + + +class MechanicalSplitTests(unittest.TestCase): + def test_exact_sentence_count(self) -> None: + text = "第一句。第二句。" + self.assertEqual(mechanical_split(text, 2), ["第一句。", "第二句。"]) + + def test_too_many_sentences_are_merged(self) -> None: + text = "第一句。第二句。第三句。" + parts = mechanical_split(text, 2) + self.assertIsNotNone(parts) + assert parts is not None + self.assertEqual(len(parts), 2) + self.assertEqual("".join(parts), text) + + def test_too_few_sentences_are_split_at_a_comma(self) -> None: + text = "第一句很长，包含很多细节。" + parts = mechanical_split(text, 2) + self.assertIsNotNone(parts) + assert parts is not None + self.assertEqual(len(parts), 2) + self.assertEqual("".join(parts), text) + + def test_nothing_is_lost(self) -> None: + text = "第一句。第二句。第三句。第四句。" + parts = mechanical_split(text, 3) + assert parts is not None + self.assertEqual("".join(parts), text) + + +class RepartitionTests(unittest.TestCase): + def test_gaps_go_to_the_preceding_part(self) -> None: + translation = "第一句。第二句。" + parts = repartition(translation, ["第一句。", "第二句。"]) + assert parts is not None + self.assertEqual("".join(parts), translation) + + def test_the_result_reproduces_the_translation(self) -> None: + translation = "这艘船很大。设计师很用心。" + parts = repartition(translation, ["这艘船很大。", "设计师很用心。"]) + assert parts is not None + self.assertEqual("".join(parts), translation) + + def test_a_missing_anchor_is_rejected(self) -> None: + self.assertIsNone(repartition("这艘船很大。", ["完全不同的句子。"])) + + def test_whitespace_differences_are_tolerated(self) -> None: + translation = "第一句。 第二句。" + parts = repartition(translation, ["第一句。", "第二句。"]) + assert parts is not None + # Each part is stripped, so the whitespace between them is dropped. + # The invariant is that nothing but whitespace is lost. + self.assertEqual("".join(parts).replace(" ", ""), translation.replace(" ", "")) + + +class ParseSegmentsTests(unittest.TestCase): + def setUp(self) -> None: + self.sources = source_utterances(paragraph_fixture(), bundle_fixture()) + self.translation = "这艘船很大。设计师很用心。" + + def test_a_valid_answer_is_accepted(self) -> None: + data = { + "segments": [ + {"source_id": "utt-100", "target_text": "这艘船很大。"}, + {"source_id": "utt-101", "target_text": "设计师很用心。"}, + ] + } + segments, problems = parse_segments(data, self.sources, self.translation) + self.assertEqual(problems, []) + self.assertEqual([s.text for s in segments], ["这艘船很大。", "设计师很用心。"]) + + def test_a_missing_segment_is_rejected(self) -> None: + data = {"segments": [{"source_id": "utt-100", "target_text": "这艘船很大。"}]} + _, problems = parse_segments(data, self.sources, self.translation) + self.assertTrue(problems) + + def test_a_duplicated_source_id_is_rejected(self) -> None: + data = { + "segments": [ + {"source_id": "utt-100", "target_text": "这艘船很大。"}, + {"source_id": "utt-100", "target_text": "设计师很用心。"}, + ] + } + _, problems = parse_segments(data, self.sources, self.translation) + self.assertTrue(problems) + + def test_text_that_is_not_in_the_translation_is_rejected(self) -> None: + data = { + "segments": [ + {"source_id": "utt-100", "target_text": "这艘船很大。"}, + {"source_id": "utt-101", "target_text": "凭空多出来的一句。"}, + ] + } + _, problems = parse_segments(data, self.sources, self.translation) + self.assertTrue(problems) + + def test_a_merge_is_accepted_and_marked_not_voiced(self) -> None: + data = { + "segments": [ + {"source_id": "utt-100", "target_text": "这艘船很大。设计师很用心。"}, + {"source_id": "utt-101", "target_text": "", "merged_into": "utt-100"}, + ] + } + segments, problems = parse_segments(data, self.sources, self.translation) + self.assertEqual(problems, []) + self.assertTrue(segments[0].voiced) + self.assertFalse(segments[1].voiced) + self.assertEqual(segments[1].merged_into, "utt-100") + + def test_a_line_cannot_be_handed_to_another_speaker(self) -> None: + paragraph = { + "id": "ch-1-p0001", + "speaker": "narrator", + "type": "narration", + "text": "The ship was large. I want to see the bridge.", + "source_indices": [0, 2], + "source_ids": ["utt-100", "utt-102"], + } + sources = source_utterances(paragraph, bundle_fixture()) + data = { + "segments": [ + {"source_id": "utt-100", "target_text": "这艘船很大。我想看看舰桥。"}, + {"source_id": "utt-102", "target_text": "", "merged_into": "utt-100"}, + ] + } + _, problems = parse_segments(data, sources, "这艘船很大。我想看看舰桥。") + self.assertTrue(problems) + self.assertTrue(any("different speaker" in p for p in problems)) + + def test_a_merge_chain_is_rejected(self) -> None: + data = { + "segments": [ + {"source_id": "utt-100", "target_text": "这艘船很大。"}, + {"source_id": "utt-101", "target_text": "", "merged_into": "utt-101"}, + ] + } + _, problems = parse_segments(data, self.sources, self.translation) + self.assertTrue(problems) + + +class AlignParagraphTests(unittest.TestCase): + def test_a_single_source_utterance_is_copied(self) -> None: + paragraph = { + "id": "ch-1-p0002", + "speaker": "Kainda Walters Foster Murphy", + "type": "dialogue", + "text": "I want to see the bridge.", + "source_indices": [2], + "source_ids": ["utt-102"], + } + alignment = align_paragraph(paragraph, "我想看看舰桥。", bundle_fixture()) + self.assertEqual(alignment.method, "one_to_one") + self.assertEqual(alignment.segments[0].text, "我想看看舰桥。") + + def test_a_scene_break_has_no_translation(self) -> None: + paragraph = { + "id": "ch-1-break1", + "speaker": "", + "type": "scene_break", + "text": "* * *", + "source_indices": [3], + "source_ids": ["utt-103"], + } + alignment = align_paragraph(paragraph, "* * *", bundle_fixture()) + self.assertEqual(alignment.method, "scene_break") + self.assertEqual(alignment.segments, []) + + def test_without_a_model_the_split_is_mechanical_and_flagged(self) -> None: + alignment = align_paragraph(paragraph_fixture(), "这艘船很大。设计师很用心。", bundle_fixture()) + self.assertEqual(alignment.method, "mechanical") + self.assertTrue(alignment.needs_review) + self.assertEqual(len(alignment.segments), 2) + + def test_a_model_partition_is_used_and_verified(self) -> None: + class FakeLLM: + def __init__(self) -> None: + self.calls = 0 + + def chat_json(self, system: str, user: str, **kwargs: object) -> tuple[dict, object]: + self.calls += 1 + return ( + { + "segments": [ + {"source_id": "utt-100", "target_text": "这艘船很大。"}, + {"source_id": "utt-101", "target_text": "设计师很用心。"}, + ] + }, + None, + ) + + llm = FakeLLM() + alignment = align_paragraph(paragraph_fixture(), "这艘船很大。设计师很用心。", bundle_fixture(), llm=llm) + self.assertEqual(alignment.method, "llm") + self.assertFalse(alignment.needs_review) + self.assertEqual("".join(s.text for s in alignment.segments), "这艘船很大。设计师很用心。") + self.assertEqual(llm.calls, 1) + + +class DeliveryTests(unittest.TestCase): + def test_the_payload_carries_the_database_row_id(self) -> None: + alignment = align_paragraph(paragraph_fixture(), "这艘船很大。设计师很用心。", bundle_fixture()) + book = {"target_language": "zh-CN"} + delivery = build_delivery( + book, + [({"id": "ch-1", "paragraphs": [paragraph_fixture()]}, [alignment])], + bundle_fixture(), + ) + self.assertEqual(delivery["target_language"], "zh") + self.assertEqual([u["utterance_id"] for u in delivery["utterances"]], [100, 101]) + self.assertTrue(all(u["voiced"] for u in delivery["utterances"])) + self.assertEqual(delivery["unmatched"], []) + + def test_a_collapsed_utterance_is_delivered_as_not_voiced(self) -> None: + alignment = align_paragraph(paragraph_fixture(), "这艘船很大。设计师很用心。", bundle_fixture()) + alignment.segments = [ + Segment(source_id="utt-100", text="这艘船很大。设计师很用心。"), + Segment( + source_id="utt-101", + text="", + voiced=False, + merged_into="utt-100", + reason="absorbed into utt-100 in the translated paragraph", + ), + ] + delivery = build_delivery( + {"target_language": "zh"}, + [({"id": "ch-1", "paragraphs": [paragraph_fixture()]}, [alignment])], + bundle_fixture(), + ) + collapsed = [u for u in delivery["utterances"] if not u["voiced"]] + self.assertEqual(len(collapsed), 1) + self.assertEqual(collapsed[0]["utterance_id"], 101) + self.assertIn("absorbed", collapsed[0]["reason"]) + self.assertNotIn("text", collapsed[0]) + + +if __name__ == "__main__": + unittest.main()