From 996fe37f4ffc55e27304cc569cb8d3615393168c Mon Sep 17 00:00:00 2001 From: Umbra Tools Date: Fri, 6 Nov 2026 19:13:30 -0800 Subject: [PATCH] Align the translated paragraphs back onto the verified source utterances The translation pipeline works on paragraphs, because epub and pdf are paragraph media. The audiobook needs utterances: UmbraAudio has verified who speaks every one of them, and each clip is voiced separately by that speaker's voice. A paragraph-level translation cannot be spoken. This is the pass that was missing. tools/translation/README.md said so plainly: "It does not produce a 1:1 utterance alignment for TTS ... splitting or aligning back to the 4,319 utterances is a separate pass." 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; * the partition must reproduce the translated text exactly, modulo whitespace, so the alignment can never rewrite the translation. An answer that drops, duplicates or invents text is rejected and the paragraph falls back to a mechanical split, which is reported as needing review. A source utterance the target does not voice is delivered as voiced=false with the reason, which is what UmbraAudio's utterance_translations.voiced flag is for: the row is kept so the gap stays visible, and no clip is generated for it. The output is a delivery payload in the shape UmbraAudio's translations.import_translations already accepts, so neither side has to invent a second contract. Measured on the real book, 42 sampled multi-utterance paragraphs across seven chapters: 41 aligned by the model, 1 fell back to the mechanical split, 0 partition mismatches. The one fallback was a model answer that did not reproduce the translated text; the content check caught it. 44 tests, ruff clean. --- tools/translation/README.md | 60 ++- tools/translation/align.py | 733 ++++++++++++++++++++++++++ tools/translation/tests/test_align.py | 335 ++++++++++++ 3 files changed, 1125 insertions(+), 3 deletions(-) create mode 100644 tools/translation/align.py create mode 100644 tools/translation/tests/test_align.py 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()