3f05489e25
The pipeline was hard-coded to Simplified Chinese in three places: the drafter was told to translate into Chinese, the critic was told to compare English with Chinese, and the glossary's translated forms were read from a "zh" key. A production that declared target_language: es-ES was therefore translated into Chinese and filed under a Spanish name - measured on oldbox, deal-hunter-es held 3,038 paragraphs of Chinese and was reported complete. A new registry, tools/translation/languages.py, is the single source of truth for what a target language is: its names, its script, its punctuation, its glossary key, its typography and its chapter-numbering convention. Ten languages are registered. The prompt scaffolding stays in Chinese - that is what the models read best - and the target is named with its Chinese, English and native names so there is no room to misread it. The registry now reaches every stage: - prompts.py: the analyst, drafter, critic and glossary blocks take the target. The hard rules about this book - Deal Hunter is the ship's name, the AI's name and the book's title, so it stays in English - are the language's own table for Chinese and the global rules for everything else, so a language with no table does not lose them. The Chinese rules are byte-for-byte what they were. - pipeline.py: --target-language, a run-directory guard that refuses to reuse a directory belonging to another language (from the manifest, or by measuring the text it already holds), a draft-keyed-by-target reader, and a refusal of a draft that came back in the wrong script. - book.py: per-language chapter-title and subtitle tables. A language with no table falls back to the English title, never to another language's; a language with a table and a missing chapter gets a visible marker in its own language. - qa.py: the target's quote marks, punctuation, script count and forbidden title forms. The Latin-residue check is skipped for a Latin target, where it would flag the whole book. - align.py: the aligner and verifier prompts name the target, and the sentence and clause splitting uses the target's own terminators. - render.py: the fonts, the language and country codes, the writing direction, the chapter headings and the fixed editorial strings around the translation are the target's. The Latin-run span is skipped for a Latin target, where marking the text as English would tell the renderer it is English. - project_status.py: the dashboard measures a sample of a production's text and reports a production whose text is not in the language it declares as "mismatched", not "complete". A chapter count is a claim about the files; the script check is a measurement of the text. Also fixes: the dashboard gated every language's production on the source production existing, so a Spanish production could never be reported even when it was there. The Chinese render is byte-identical to before on the real book, and the Chinese rules are unchanged. Suite: 620 passed, the same 68 pre-existing environment failures, +49 new tests.
1102 lines
43 KiB
Python
1102 lines
43 KiB
Python
# 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 hashlib
|
||
import json
|
||
import re
|
||
import sys
|
||
from dataclasses import dataclass, field
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
from languages import TargetLanguage, by_code, resolve
|
||
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"
|
||
|
||
CHINESE = resolve("zh-Hans")
|
||
|
||
|
||
def system_aligner(target: TargetLanguage) -> str:
|
||
"""The aligner's instructions, in the target language's terms.
|
||
|
||
The task is the same in every language - map a translated paragraph back
|
||
onto the source utterances it was built from - but the sentence merging
|
||
the translation does is the *target's* merging, so the prompt has to name
|
||
the target.
|
||
"""
|
||
return f"""你是一位双语对齐编辑，负责把已经译好的{target.chinese_name}段落还原到英文原文的每一条话语上。
|
||
你的任务不是翻译，也不是改写，而是判断：{target.chinese_name}译文里的哪一部分对应英文原文里的哪一条话语。
|
||
译文为了{target.chinese_name}的自然可能合并或拆分了句子，所以不能按句子序号机械对应，要按意思对应。
|
||
每一条英文话语都必须恰好出现一次；被合并进别条话语的，用 merged_into 指明并把 target_text 留空。
|
||
只输出 JSON，不要解释，不要 Markdown 代码围栏。"""
|
||
|
||
|
||
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.
|
||
|
||
A registered language answers from the registry, which is the single
|
||
source of truth for what a language's base code is; an unregistered one
|
||
falls back to the string surgery.
|
||
"""
|
||
language = by_code(value)
|
||
if language is not None:
|
||
return language.base
|
||
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,
|
||
target: TargetLanguage | None = None,
|
||
) -> list[str]:
|
||
"""Split translated text into sentences, keeping the terminator.
|
||
|
||
The terminators and the closing marks are the target's own: Chinese
|
||
terminates with a full stop and a closing quote may follow it, while a
|
||
Spanish sentence ends with a question mark or a full stop.
|
||
"""
|
||
target = target or CHINESE
|
||
terminators = target.sentence_terminators
|
||
closing = target.closing_marks
|
||
sentences: list[str] = []
|
||
buffer: list[str] = []
|
||
index = 0
|
||
while index < len(text):
|
||
ch = text[index]
|
||
buffer.append(ch)
|
||
index += 1
|
||
if ch in terminators:
|
||
while index < len(text) and text[index] in closing:
|
||
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,
|
||
target: TargetLanguage | None = None,
|
||
) -> 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 the target language. So when the target has fewer sentences
|
||
than the source has utterances, the boundary is a clause boundary - and
|
||
which characters those are depends on the language.
|
||
"""
|
||
target = target or CHINESE
|
||
for mark in target.clause_marks:
|
||
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,
|
||
target: TargetLanguage | None = None,
|
||
) -> 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, target)
|
||
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], target)
|
||
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", "mismatch"}
|
||
|
||
|
||
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,
|
||
target: TargetLanguage | None = None,
|
||
) -> str:
|
||
"""The prompt that asks for the partition of a translated paragraph."""
|
||
target = target or CHINESE
|
||
lines = []
|
||
for source in sources:
|
||
lines.append(f"[{source.source_id}] (说话者：{source.speaker}) {source.text}")
|
||
return f"""段落 id：{paragraph_id}
|
||
|
||
英文原文（每一条都是一条已经确认说话者的话语）：
|
||
{chr(10).join(lines)}
|
||
|
||
{target.chinese_name}译文：
|
||
{translation}
|
||
|
||
请把{target.chinese_name}译文分配回上面每一条英文话语。输出 JSON：
|
||
{{
|
||
"segments": [
|
||
{{"source_id": "utt-xxxxx", "target_text": "对应的{target.chinese_name}片段"}},
|
||
{{"source_id": "utt-yyyyy", "target_text": "", "merged_into": "utt-xxxxx"}}
|
||
]
|
||
}}
|
||
|
||
要求：
|
||
- segments 必须为原文的每一条话语各给一项，source_id 一一对应，不多不少。
|
||
- target_text 必须是{target.chinese_name}译文里**原样出现**的连续片段，按原文顺序拼接起来必须等于整段译文，不得删改一个字。
|
||
- 如果译文把两条原文合并成了一句，被合并的那条 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,
|
||
target: TargetLanguage | None = None,
|
||
) -> ParagraphAlignment:
|
||
"""Divide one translated paragraph between its source utterances."""
|
||
target = target or CHINESE
|
||
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(target),
|
||
alignment_prompt(paragraph_id, sources, translation, target),
|
||
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), target)
|
||
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), target)
|
||
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)
|
||
],
|
||
)
|
||
|
||
|
||
# --- drift verification ----------------------------------------------------
|
||
|
||
SYSTEM_VERIFIER = """你是一位双语校对，负责检查一段译文是否真的对应它的英文原文。
|
||
译文是按段落 id 存储的；如果写译文时把内容写到了相邻段落，id 和内容就会错位。
|
||
只输出 JSON，不要解释，不要 Markdown 代码围栏。"""
|
||
|
||
|
||
# The three verdicts. The distinction matters: only a paragraph whose
|
||
# translation belongs to a *different* paragraph may be dropped from the
|
||
# delivery. A paragraph that merely leaves something out is still the right
|
||
# translation in the right place, and dropping it would delete a good line and
|
||
# leave a hole where the audio should have spoken.
|
||
VERDICT_ALIGNED = "aligned"
|
||
VERDICT_INCOMPLETE = "incomplete"
|
||
VERDICT_DRIFT = "drift"
|
||
VERDICTS = (VERDICT_ALIGNED, VERDICT_INCOMPLETE, VERDICT_DRIFT)
|
||
|
||
# A paragraph whose English does not end on a sentence boundary is a fragment:
|
||
# Pass 0 split the original passage mid-sentence. A fragment is where the
|
||
# drafter is most likely to re-flow the passage into natural target-language
|
||
# prose and lose the id-to-text correspondence, so that is where the check
|
||
# starts.
|
||
_SENTENCE_END = ".!?…"
|
||
|
||
|
||
def is_fragment(text: str) -> bool:
|
||
"""True when *text* does not end on a sentence boundary.
|
||
|
||
Trailing closing quotes and brackets are ignored: a quoted label such as
|
||
``"operations"`` is a complete quotation but not a complete sentence.
|
||
"""
|
||
stripped = text.strip().rstrip("”’\"')]】》")
|
||
if not stripped:
|
||
return False
|
||
return stripped[-1] not in _SENTENCE_END
|
||
|
||
|
||
def verification_prompt(
|
||
source_text: str,
|
||
translation: str,
|
||
previous_text: str = "",
|
||
next_text: str = "",
|
||
target: TargetLanguage | None = None,
|
||
) -> str:
|
||
"""Ask which paragraph *translation* actually renders.
|
||
|
||
The neighbouring paragraphs are given because the question "does this
|
||
translation belong to this paragraph?" cannot be answered without them:
|
||
a drifted paragraph's translation is a perfectly good translation - of
|
||
the paragraph next door.
|
||
"""
|
||
target = target or CHINESE
|
||
previous_block = f"\n【前一段原文】\n{previous_text}\n" if previous_text else ""
|
||
next_block = f"\n【后一段原文】\n{next_text}\n" if next_text else ""
|
||
return f"""【本段原文】
|
||
{source_text}
|
||
{previous_block}{next_block}
|
||
【本段{target.chinese_name}译文】
|
||
{translation}
|
||
|
||
请判断这段{target.chinese_name}译文到底对应哪一段原文，三选一：
|
||
|
||
- "aligned"：译文对应本段原文，哪怕不完美。措辞不同、近义词、漏掉一个副词、
|
||
数字或单复数的小差异，都算 aligned。
|
||
- "incomplete"：译文对应本段原文，但明显少了一部分内容，而且没有混入别段内容。
|
||
- "drift"：译文对应的是前一段或后一段原文的内容（内容错位）。
|
||
|
||
举例（这些都算 aligned，措辞或数量上的小差异不算错位）：
|
||
* 本段 "The hunter robot holds out a hand and says to her"
|
||
译文 "机器人伸出双手对她说" —— 单复数不一致，still aligned。
|
||
* 本段 "Her brow furrowed, then says" 译文 "她眉头紧锁，说道" —— 省略 then，still aligned。
|
||
* 本段 "...You gave me the helmet..." 译文 "...你给我这顶帽子..." —— 头盔写成帽子，still aligned。
|
||
|
||
举例（这些算 drift，因为译文是相邻那一段的内容）：
|
||
* 本段 "“Yes, Captain.”" 译文 "“上一次有人登上这艘舰桥还是三十四年前，那时是西顿船长在任？”"
|
||
—— 这是上一段的译文。
|
||
* 本段 "“The last time someone was on this bridge...?" 译文 "“Captain”字样。她环顾四周……"
|
||
—— 这是上一段的译文。
|
||
|
||
举例（这种算 incomplete：译文是自己这段的内容，但少了后面几句）：
|
||
* 本段有三句话，译文只译了前两句的意思，译文内容没有混入别段。
|
||
|
||
只输出 JSON：{{"kind": "aligned|incomplete|drift", "reason": "一句话理由"}}"""
|
||
|
||
|
||
def verify_translation(
|
||
source_text: str,
|
||
translation: str,
|
||
llm: LLM,
|
||
model: str = DEFAULT_ALIGN_MODEL,
|
||
previous_text: str = "",
|
||
next_text: str = "",
|
||
target: TargetLanguage | None = None,
|
||
) -> tuple[str, str]:
|
||
"""Which paragraph does *translation* actually render?
|
||
|
||
Returns ``(verdict, reason)`` where *verdict* is one of :data:`VERDICTS`.
|
||
A model failure is reported as ``drift``: the delivery must not carry a
|
||
line nobody checked.
|
||
"""
|
||
prompt = verification_prompt(source_text, translation, previous_text, next_text, target)
|
||
try:
|
||
data, _usage = llm.chat_json(
|
||
SYSTEM_VERIFIER,
|
||
prompt,
|
||
model=model,
|
||
temperature=0.0,
|
||
cache_key=f"verify-{hashlib.sha256(prompt.encode('utf-8')).hexdigest()[:16]}",
|
||
)
|
||
except LLMError as exc:
|
||
return VERDICT_DRIFT, f"verification model failed: {exc}"
|
||
|
||
if not isinstance(data, dict):
|
||
return VERDICT_DRIFT, "verification answer was not an object"
|
||
kind = str(data.get("kind") or "").strip().lower()
|
||
reason = str(data.get("reason") or "").strip()
|
||
if kind not in VERDICTS:
|
||
# An unrecognised verdict is not a pass. A model that cannot answer
|
||
# the question has not verified anything.
|
||
return VERDICT_DRIFT, f"verification answer was not understood: {kind or '<empty>'}"
|
||
return kind, reason
|
||
|
||
|
||
def _apply_drift_verification(
|
||
alignments: list[ParagraphAlignment],
|
||
chapter: dict,
|
||
translated: dict[str, str],
|
||
bundle: dict[str, dict],
|
||
llm: LLM,
|
||
model: str,
|
||
target: TargetLanguage | None = None,
|
||
) -> None:
|
||
"""Check the paragraphs whose translation may have drifted across ids.
|
||
|
||
The check starts at a sentence fragment - Pass 0 split the original
|
||
passage mid-sentence there, and a fragment is exactly where an
|
||
intent-based drafter is told to write natural target-language prose and
|
||
lose the id-to-text correspondence.
|
||
|
||
The paragraphs are then judged as a *window* rather than one at a time,
|
||
because a re-flowed passage cannot be judged one paragraph at a time:
|
||
the same paragraph can be called "aligned" and be wrong (the model does
|
||
exactly that on the two paragraphs before a real drift), and the start of
|
||
a drift is not knowable from any single paragraph.
|
||
|
||
A paragraph the model says belongs to another paragraph is marked
|
||
``mismatch`` and its segments are dropped: the delivery must not carry
|
||
text that renders the wrong source utterance, and a visible gap is better
|
||
than a confident wrong line in the audiobook. A paragraph that is merely
|
||
incomplete keeps its segments and is flagged for review, unless the
|
||
window it sits in is already broken.
|
||
|
||
The cost of this choice is explicit: a fragment inside a broken window is
|
||
dropped even when the model called it aligned, so the audiobook has a
|
||
gap there. The alternative - keeping it - puts a line in the audiobook
|
||
that belongs to a different sentence, which is worse.
|
||
"""
|
||
target = target or CHINESE
|
||
by_id = {alignment.paragraph_id: alignment for alignment in alignments}
|
||
paragraphs = list(chapter.get("paragraphs") or [])
|
||
index_of = {str(p.get("id") or ""): i for i, p in enumerate(paragraphs)}
|
||
|
||
# A re-flowed passage is judged as a *window*, not as a series of
|
||
# independent paragraphs. Two measurements forced this shape:
|
||
#
|
||
# * The same paragraph can be judged "aligned" and be wrong. In the real
|
||
# ch-15 window, the model called the two paragraphs before the drift
|
||
# "aligned" when their Chinese was demonstrably the neighbouring
|
||
# paragraph's.
|
||
# * The start of a drift is not knowable from any single paragraph. The
|
||
# drafter re-flowed the passage, so the content of paragraph N may sit
|
||
# under paragraph N+1, and the first symptom can be a long way from the
|
||
# cause.
|
||
#
|
||
# So: judge every candidate, then resolve each window as a whole. A
|
||
# window that contains a confirmed drift is broken, and inside a broken
|
||
# window a per-paragraph "aligned" verdict is not trustworthy.
|
||
window: list[tuple[str, ParagraphAlignment, str, bool, str]] = []
|
||
window_has_drift = False
|
||
|
||
def resolve_window() -> None:
|
||
"""Apply the window's verdict to the paragraphs inside it."""
|
||
nonlocal window_has_drift
|
||
for _paragraph_id, alignment, reason, was_fragment, verdict in window:
|
||
if alignment.method == "mismatch":
|
||
continue
|
||
if window_has_drift and was_fragment and verdict == VERDICT_ALIGNED:
|
||
# A fragment inside a confirmed drift is exactly where the
|
||
# re-flow happened, and the model's "aligned" verdict there is
|
||
# the one that is known to be unreliable. A paragraph the
|
||
# model called "incomplete" is not dropped on this evidence:
|
||
# it demonstrably renders its own leading content.
|
||
alignment.method = "mismatch"
|
||
alignment.problems = [f"translation is part of a drifted passage: {reason}"]
|
||
alignment.segments = []
|
||
window.clear()
|
||
window_has_drift = False
|
||
|
||
for paragraph in paragraphs:
|
||
paragraph_id = str(paragraph.get("id") or "")
|
||
alignment = by_id.get(paragraph_id)
|
||
if alignment is None:
|
||
continue
|
||
source_text = str(paragraph.get("text") or "").strip()
|
||
translation = str(translated.get(paragraph_id) or "").strip()
|
||
if not source_text or not translation:
|
||
continue
|
||
|
||
fragment = is_fragment(source_text)
|
||
if not window and not fragment:
|
||
continue
|
||
|
||
position = index_of.get(paragraph_id, -1)
|
||
previous_text = ""
|
||
next_text = ""
|
||
if 0 <= position - 1 < len(paragraphs):
|
||
previous_text = str(paragraphs[position - 1].get("text") or "").strip()
|
||
if 0 <= position + 1 < len(paragraphs):
|
||
next_text = str(paragraphs[position + 1].get("text") or "").strip()
|
||
|
||
verdict, reason = verify_translation(
|
||
source_text,
|
||
translation,
|
||
llm,
|
||
model,
|
||
previous_text=previous_text,
|
||
next_text=next_text,
|
||
target=target,
|
||
)
|
||
|
||
if verdict == VERDICT_ALIGNED:
|
||
if window and not fragment:
|
||
# The passage has re-synchronised: this paragraph is a
|
||
# complete sentence and the model says it is where it
|
||
# belongs. The window ends here.
|
||
resolve_window()
|
||
else:
|
||
window.append((paragraph_id, alignment, reason, fragment, verdict))
|
||
continue
|
||
|
||
if verdict == VERDICT_DRIFT:
|
||
window_has_drift = True
|
||
alignment.method = "mismatch"
|
||
alignment.problems = [f"translation belongs to another paragraph: {reason}"]
|
||
alignment.segments = []
|
||
elif window_has_drift:
|
||
# Inside a confirmed drift, "my own text, truncated" is part of
|
||
# the same defect: the drafter re-flowed the passage and this
|
||
# paragraph is one of the pieces it was cut into. The English
|
||
# around a drift is often itself a sentence split across
|
||
# paragraph ids, so a truncated lead-in here is the tail of the
|
||
# previous paragraph rather than a translation of this one.
|
||
alignment.method = "mismatch"
|
||
alignment.problems = [f"translation is part of a drifted passage: {reason}"]
|
||
alignment.segments = []
|
||
else:
|
||
# Nothing has been proven wrong yet, so the text is kept and
|
||
# flagged: it is still this paragraph's own translation, just
|
||
# short of the whole of it.
|
||
alignment.problems = [f"translation is incomplete: {reason}"]
|
||
|
||
window.append((paragraph_id, alignment, reason, fragment, verdict))
|
||
|
||
resolve_window()
|
||
|
||
|
||
def align_chapter(
|
||
chapter: dict,
|
||
translated: dict[str, str],
|
||
bundle: dict[str, dict],
|
||
llm: LLM | None = None,
|
||
model: str = DEFAULT_ALIGN_MODEL,
|
||
verify: bool = False,
|
||
target: TargetLanguage | None = None,
|
||
) -> tuple[list[ParagraphAlignment], dict[str, Any]]:
|
||
"""Align every translated paragraph of one chapter.
|
||
|
||
With *verify*, the paragraphs where the drafter could have re-flowed the
|
||
passage across paragraph ids are checked as well - see
|
||
:func:`verify_drift_run`.
|
||
"""
|
||
target = target or CHINESE
|
||
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, target=target))
|
||
|
||
if verify and llm is not None:
|
||
_apply_drift_verification(alignments, chapter, translated, bundle, llm, model, target)
|
||
|
||
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 = "",
|
||
target: TargetLanguage | None = None,
|
||
) -> dict[str, Any]:
|
||
"""The payload UmbraAudio's ``import_translations`` accepts."""
|
||
if target is None:
|
||
target = resolve(book.get("target_language") or "zh-Hans")
|
||
target_code = target.base
|
||
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)}
|
||
if alignment.method == "mismatch":
|
||
# The paragraph's translation could not be trusted - it
|
||
# renders another paragraph. The utterances still get a row,
|
||
# marked as not voiced and carrying the reason, so the gap is
|
||
# visible in the review lane and says why. Silence would be
|
||
# indistinguishable from "never translated".
|
||
reason = "; ".join(alignment.problems) or "the translation could not be verified"
|
||
for source in sources.values():
|
||
if source.row_id is None:
|
||
continue
|
||
utterances.append(
|
||
{
|
||
"utterance_id": source.row_id,
|
||
"voiced": False,
|
||
"reason": reason,
|
||
"source_ref": source_ref or alignment.paragraph_id,
|
||
}
|
||
)
|
||
continue
|
||
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_code,
|
||
"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(
|
||
"--verify",
|
||
action="store_true",
|
||
help=(
|
||
"check the paragraphs where the drafter could have re-flowed the "
|
||
"passage across paragraph ids, and drop the ones that do not "
|
||
"render their own source"
|
||
),
|
||
)
|
||
parser.add_argument("--source-ref", default="")
|
||
parser.add_argument(
|
||
"--target-language",
|
||
default=None,
|
||
help="the language the translations are in. Defaults to the book's own.",
|
||
)
|
||
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"))
|
||
target = resolve(args.target_language or book.get("target_language") or "zh-Hans")
|
||
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:
|
||
# The cache directory has to exist before the first call: the client
|
||
# writes the answer as it arrives, and a missing directory would turn
|
||
# every model answer into an error - which the fallback would then
|
||
# quietly turn into a mechanical split for the whole book.
|
||
cache_dir = args.run_dir / "cache"
|
||
cache_dir.mkdir(parents=True, exist_ok=True)
|
||
llm = LLM(base_url=args.base_url, cache_dir=cache_dir) if args.base_url else LLM(cache_dir=cache_dir)
|
||
|
||
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,
|
||
verify=args.verify,
|
||
target=target,
|
||
)
|
||
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, target=target)
|
||
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())
|