umbragit/UmbraTrans
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Align the translated paragraphs back onto the verified source utterances

Browse Source 
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.
This commit is contained in:
Umbra Tools
2026-11-06 19:13:30 -08:00
parent 91a1c13307
commit 996fe37f4f
3 changed files with 1125 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

60
tools/translation/README.md
View File

@@ -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/<chapter>/`. 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.

733
tools/translation/align.py Normal file
View File

@@ -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())

335
tools/translation/tests/test_align.py Normal file
View File

@@ -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()