11fd8d8060
A paragraph whose translation could not be verified produced no rows at all, which is indistinguishable from one that was never translated: the review lane showed a missing utterance with no explanation. The project's own rule for a gap is "the row is kept, the gap is visible, no clip is generated", and it already applies to the collapse orphans. An unverifiable paragraph is the same kind of gap, so it now gets the same treatment: a voiced=false row carrying the reason. Measured on the real ch-15 window: the ten drifted paragraphs now produce thirteen explicit not-voiced rows instead of vanishing, and the reviewer can see that the translation was dropped because it renders another paragraph rather than because nobody translated it. 60 tests, ruff clean.
1044 lines
41 KiB
Python
1044 lines
41 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 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", "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) -> 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)
|
||
],
|
||
)
|
||
|
||
|
||
# --- 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 = "",
|
||
) -> 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.
|
||
"""
|
||
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}
|
||
【本段中文译文】
|
||
{translation}
|
||
|
||
请判断这段中文译文到底对应哪一段原文，三选一：
|
||
|
||
- "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 = "",
|
||
) -> 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)
|
||
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,
|
||
) -> 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.
|
||
"""
|
||
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,
|
||
)
|
||
|
||
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,
|
||
) -> 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`.
|
||
"""
|
||
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))
|
||
|
||
if verify and llm is not None:
|
||
_apply_drift_verification(alignments, chapter, translated, bundle, llm, 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)}
|
||
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,
|
||
"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="")
|
||
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:
|
||
# 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,
|
||
)
|
||
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())
|