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

Ingest the translation delivery, and stop mis-reporting locale codes

Browse Source 
UmbraTrans delivers one row per utterance per target language, in the shape
translations.import_translations already accepts. There was no script to run
that ingest, so the delivery side of the contract existed only as a function.

scripts/import_translations.py runs it: report-only by default, --write to
store, --confirm to store as confirmed rather than draft. It prints the
coverage before and after, so a delivery can be checked without writing it.

Two bugs in the ingest, both found by running a real delivery:

* import_translations folded the target language with lang.normalize, which
  does not know script subtags. A delivery labelled "zh-Hans" - which is
  what UmbraTrans writes - matched nothing, so every row was dropped and
  reported as a problem. It now folds with canonical_code, and an unknown
  language is still stored: the text is reviewable even when the render
  will be refused.

* coverage and not_voiced_rows compared the stored code to the requested
  code as strings, so a row stored as "zh-Hans" read as "nothing delivered"
  when the reviewer asked about "zh". The coverage report was wrong in the
  direction that makes a reviewer distrust it.

Measured on the real Deal Hunter Chinese delivery: 4,300 rows ingested, the
review UI's resolver then renders Chinese for 12/12 utterances of chapter 1
and 415/416 of chapter 6 - the missing one being the spoken chapter
announcement, which is deliberately not part of the reading edition.

spec 27 records what the shipped samples actually are: single-pass drafts
made against a hand-cleaned paraphrase of the English, matching the
speaker-verified database text in 0 of 12 utterances. The production Chinese
edition on oldbox does use the intent-based process; what it lacked was the
alignment back onto the utterances, which is now implemented in UmbraTrans.

7 new tests. Suite: 890 passed, 8 failed (the same 8 pre-existing failures).
This commit is contained in:
Umbra Tools
2026-11-06 19:14:38 -08:00
parent 6bb69729b9
commit c476cfea1d
4 changed files with 415 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files

116
scripts/import_translations.py Normal file
View File

@@ -0,0 +1,116 @@
#!/usr/bin/env python3
# Copyright 2026 Umbra Tools (umbratools.com)
# SPDX-License-Identifier: Apache-2.0
"""Ingest a translation delivery from UmbraTrans (spec 31 section 8).
UmbraTrans delivers one row per utterance per target language. The payload
shape is the one ``umbraaudio.translations.import_translations`` accepts, so
the delivery contract exists on both sides and neither side has to invent a
second one.
# see what would be written (read-only)
python scripts/import_translations.py --project "DealHunterShip_4.0e_whole-to-audio" \
--delivery /tmp/delivery.json
# write it
python scripts/import_translations.py --project "..." --delivery /tmp/delivery.json --write
# write it and mark the rows confirmed rather than draft
python scripts/import_translations.py --project "..." \
--delivery /tmp/delivery.json --write --confirm
A delivery is a *draft* by default. A row that says the target language does
not voice an utterance is kept, with the reason, so the gap stays visible:
deleting it would lose the alignment between the source and the translation.
"""
from __future__ import annotations
import argparse
import json
import sqlite3
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).parent.parent))
from umbraaudio.config import load_config
from umbraaudio.translations import coverage, import_translations, source_language
def _project_id(conn: object) -> int:
row = conn.execute("SELECT id FROM project LIMIT 1").fetchone() # type: ignore[attr-defined]
if row is None:
raise SystemExit("this database has no project row")
return int(row[0])
def _describe(payload: dict) -> dict:
utterances = payload.get("utterances") or []
voiced = sum(1 for row in utterances if row.get("voiced", True))
return {
"target_language": payload.get("target_language"),
"utterances": len(utterances),
"voiced": voiced,
"not_voiced": len(utterances) - voiced,
}
def main() -> int:
parser = argparse.ArgumentParser(description="Ingest a translation delivery from UmbraTrans.")
parser.add_argument("--project", required=True, help="project directory name")
parser.add_argument("--delivery", type=Path, required=True, help="delivery payload JSON")
parser.add_argument("--write", action="store_true", help="write the rows (default: report)")
parser.add_argument("--confirm", action="store_true", help="store as confirmed, not draft")
args = parser.parse_args()
config = load_config(args.project)
db_path = config.db_path
if not db_path.exists():
print(f"database not found: {db_path}", file=sys.stderr)
return 2
payload = json.loads(args.delivery.read_text(encoding="utf-8"))
if not isinstance(payload, dict):
print("delivery must be a JSON object", file=sys.stderr)
return 2
print("DELIVERY")
for key, value in _describe(payload).items():
print(f" {key:16s} {value}")
conn = sqlite3.connect(db_path)
try:
project_id = _project_id(conn)
code = payload.get("target_language") or ""
before = coverage(conn, code)
print("COVERAGE BEFORE")
for key, value in before.items():
print(f" {key:16s} {value}")
if not args.write:
print("\nNothing written. Add --write to ingest the delivery.")
return 0
report = import_translations(conn, project_id, payload, confirm=args.confirm)
after = coverage(conn, code)
print("\nIMPORTED")
for key, value in report.as_dict().items():
print(f" {key:20s} {value}")
print("COVERAGE AFTER")
for key, value in after.items():
print(f" {key:16s} {value}")
print(f"\nproject source language: {source_language(conn)}")
if report.problems:
print("\nproblems:")
for problem in report.problems:
print(f" - {problem}")
return 1
return 0
finally:
conn.close()
if __name__ == "__main__":
raise SystemExit(main())

70
specs/27-translation-pipeline-multipass.md
View File

@@ -115,6 +115,76 @@ They are useful as a **bad-case corpus**: Pass 4 run over them should find real
errors. If the critic cannot find errors in known-imperfect text, it is not
doing its job.
### Measured (2026-11-06): the sample files are not the book
The sample files under `samples/translations/` were measured against the
speaker-verified English in the project's database. **0 of 12 utterances match
the database text exactly**; the sample English is a hand-cleaned paraphrase
of it, with the source's typos silently repaired ("the AI the runs the ship"
-> "the AI that runs the ship", "Kianda's aunt" -> "Kainda's aunt", "I trade
salvage" -> "I trade salvage"). The sample files also carry `Kainda` where the
database carries the canonical `Kainda Walters Foster Murphy`, and `Narrator`
where the database carries `NULL`.
So those files are wrong twice over: they are single-pass, and they were
translated against a different English base than the one the audiobook speaks.
They must not be treated as a translation of anything. The only other
translation in the system is one placeholder row whose `text` is the English
source under a `zh-Hans` code, which the resolver correctly ignores.
### The real Deal Hunter Chinese edition
The production translation is not in this repository. It is built by UmbraTrans
on oldbox (`productions/deal-hunter-zh`), and it **does** use the intent-based
process:
| Pass | What it does | Model |
|---|---|---|
| 0 | Paragraph/scene/chunk assembly, source indices kept | none |
| 1 | Scene intent: summary, register, tone, participants, risks | `lfm2.5-2.6b` |
| 2 | Context-aware draft, given glossary, character voices, scene analysis, prior chunk summary | `qwen3.5-9b` |
| 3 | Chapter voice/term consistency (optional) | critic model |
| 4 | Bilingual critic: meaning, pronoun, irony, term, omission | `deepseek-r1-8b` |
Its Pass 1 output carries per-paragraph risks with guidance, which is intent
analysis, not word mapping. Its critic runs on a *different* model from the
drafter, which is the point of Pass 4.
### The missing pass, and why it matters
That pipeline works on **paragraphs**, because epub and pdf are paragraph
media. The audiobook needs **utterances**: this project 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.
Measured on the real book: 3,047 paragraphs cover 4,317 source utterances.
2,322 paragraphs map to one utterance, 180 to two and 545 to three. All 725
multi-utterance paragraphs are narration; dialogue is already one paragraph
per speech turn.
So the missing piece is an alignment pass that divides each translated
paragraph between the utterances it was built from. It is implemented in
UmbraTrans as `tools/translation/align.py` (Pass 5), and its output is a
delivery payload in the shape this project's `translations.import_translations`
already accepts.
Two rules the alignment must not 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.
A source utterance the target does not voice is delivered as
`voiced = false` with the reason. The row is kept, so the gap stays visible;
it is not deleted.
Measured on the real book (2026-11-06), 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
and the paragraph was flagged rather than accepted.
## Open questions
1. Which model runs Passes 1-4? This is where the local LLM question returns,

164
tests/test_translation_delivery.py Normal file
View File

@@ -0,0 +1,164 @@
# Copyright 2026 Umbra Tools (umbratools.com)
# SPDX-License-Identifier: Apache-2.0
"""Tests for the translation delivery ingest (spec 31 section 8).
UmbraTrans delivers one row per utterance per target language. The two
behaviours pinned here are the ones that would make a reviewer distrust the
delivery:
* a delivery stored under a locale code (``zh-Hans``) must still be found
when the reviewer asks about the base code (``zh``);
* a source utterance the target does not voice must be *kept*, with its
reason, so the gap between the English and the translation stays visible
instead of silently disappearing.
"""
from __future__ import annotations
import sqlite3
import pytest
from umbraaudio import translations as tr
def _seed(conn: sqlite3.Connection) -> None:
conn.executescript(
"""
CREATE TABLE project (id INTEGER PRIMARY KEY, name TEXT);
CREATE TABLE chapters (id INTEGER PRIMARY KEY, chapter_index INTEGER);
CREATE TABLE utterances (
id INTEGER PRIMARY KEY,
chapter_id INTEGER,
utterance_index INTEGER,
speaker_name TEXT,
text TEXT,
context_before TEXT,
context_after TEXT,
language_code TEXT,
language_origin TEXT
);
INSERT INTO project (id, name) VALUES (1, 'Test');
INSERT INTO chapters (id, chapter_index) VALUES (1, 1), (2, 2);
INSERT INTO utterances (id, chapter_id, utterance_index, speaker_name, text) VALUES
(1, 1, 0, 'narrator', 'The ship landed.'),
(2, 1, 1, 'Kainda', 'Welcome to the ship.'),
(3, 1, 2, 'Kainda', 'It is very large.'),
(4, 2, 0, 'narrator', 'Later that day.'),
(5, 2, 1, 'Nanny', 'Danger.');
"""
)
conn.commit()
@pytest.fixture
def conn() -> sqlite3.Connection:
c = sqlite3.connect(":memory:")
c.row_factory = sqlite3.Row
_seed(c)
tr.ensure_tables(c)
yield c
c.close()
def _delivery(**overrides: object) -> dict:
payload = {
"target_language": "zh",
"utterances": [
{"utterance_id": 1, "text": "飞船降落了。", "voiced": True},
{"utterance_id": 2, "text": "欢迎登船。", "voiced": True},
{"utterance_id": 3, "text": "", "voiced": False, "reason": "absorbed into utt-2"},
],
}
payload.update(overrides)
return payload
# --- ingest ---------------------------------------------------------------
def test_delivery_writes_one_row_per_utterance(conn: sqlite3.Connection) -> None:
report = tr.import_translations(conn, 1, _delivery())
assert report.accepted == 3
assert report.not_voiced == 1
rows = conn.execute(
"SELECT utterance_id, text, voiced, status FROM utterance_translations ORDER BY utterance_id"
).fetchall()
assert len(rows) == 3
assert rows[0]["text"] == "飞船降落了。"
assert rows[0]["status"] == tr.STATUS_DRAFT
def test_a_collapsed_utterance_is_kept_with_its_reason(conn: sqlite3.Connection) -> None:
tr.import_translations(conn, 1, _delivery())
row = conn.execute(
"SELECT text, voiced, review_reason FROM utterance_translations WHERE utterance_id=3"
).fetchone()
assert row is not None
assert row["voiced"] == 0
assert "absorbed" in (row["review_reason"] or "")
# The row exists, so the gap is visible; it was not deleted.
assert conn.execute("SELECT COUNT(*) FROM utterances WHERE id=3").fetchone()[0] == 1
def test_confirmed_rows_are_not_overwritten(conn: sqlite3.Connection) -> None:
tr.import_translations(conn, 1, _delivery(), confirm=True)
report = tr.import_translations(conn, 1, _delivery())
assert report.skipped_confirmed == 3
row = conn.execute(
"SELECT text FROM utterance_translations WHERE utterance_id=1"
).fetchone()
assert row["text"] == "飞船降落了。"
# --- the locale-code bug --------------------------------------------------
def test_coverage_finds_a_delivery_stored_under_a_locale_code(
conn: sqlite3.Connection,
) -> None:
"""UmbraTrans writes ``zh-Hans``; the reviewer asks about ``zh``."""
tr.import_translations(conn, 1, _delivery(target_language="zh-Hans"))
assert conn.execute(
"SELECT DISTINCT language_code FROM utterance_translations"
).fetchone()[0] == "zh"
report = tr.coverage(conn, "zh")
assert report["delivered"] == 3
assert report["voiced"] == 2
assert report["not_voiced"] == 1
def test_coverage_finds_a_row_written_with_a_script_subtag(
conn: sqlite3.Connection,
) -> None:
"""A row stored directly as ``zh-Hans`` must not read as "nothing"."""
conn.execute(
"INSERT INTO utterance_translations "
"(project_id, utterance_id, language_code, language_name, text, status, origin) "
"VALUES (1, 1, 'zh-Hans', 'Chinese', '飞船降落了。', 'draft', 'umbratrans')"
)
conn.commit()
report = tr.coverage(conn, "zh")
assert report["delivered"] == 1
assert report["missing"] == 4
def test_not_voiced_rows_are_found_under_a_locale_code(conn: sqlite3.Connection) -> None:
conn.execute(
"INSERT INTO utterance_translations "
"(project_id, utterance_id, language_code, language_name, text, status, origin, "
"voiced, review_reason) "
"VALUES (1, 3, 'zh-Hans', 'Chinese', '', 'draft', 'umbratrans', 0, 'absorbed into utt-2')"
)
conn.commit()
rows = tr.not_voiced_rows(conn, "zh")
assert len(rows) == 1
assert rows[0]["utterance_id"] == 3
assert rows[0]["chapter_id"] == 1
def test_coverage_of_an_undelivered_language_is_zero(conn: sqlite3.Connection) -> None:
report = tr.coverage(conn, "de")
assert report["delivered"] == 0
assert report["missing"] == 5

85
umbraaudio/translations.py
View File

@@ -267,7 +267,16 @@ def import_translations(
ensure_tables(conn)
report = ImportReport()
target = lang.normalize(payload.get("target_language"))
# The target may arrive as a locale code (``zh-Hans``, ``pt-BR``), which
# is what UmbraTrans writes. It has to be folded to the base code the
# review UI offers, or the delivery is stored under a code nothing
# selects and the coverage report says "nothing delivered".
raw_target = payload.get("target_language")
target = canonical_code(raw_target) if raw_target else None
if target is None and raw_target:
# An unknown language is still worth storing: the text is reviewable
# even when the render will be refused.
target = str(raw_target).strip().lower()
target_name = lang.language_name(target) if target else None
accepted_status = STATUS_CONFIRMED if confirm else STATUS_DRAFT
now = _now()
@@ -366,6 +375,27 @@ def import_translations(
return report
def _matching_language_codes(conn: sqlite3.Connection, code: str) -> list[str]:
"""The stored language codes that mean *code*.
UmbraTrans writes locale codes with script subtags (``zh-Hans``,
``pt-BR``); the review UI works in the base code (``zh``, ``pt``). A
report that compared the strings would say "nothing delivered" while the
delivery was sitting in the table, which is exactly the kind of false
negative that makes a reviewer distrust the number.
"""
target = canonical_code(code) or str(code).strip().lower()
rows = conn.execute("SELECT DISTINCT language_code FROM utterance_translations").fetchall()
out: list[str] = []
for row in rows:
stored = _col(row, "language_code", 0)
if stored is None:
continue
if (canonical_code(stored) or str(stored).strip().lower()) == target:
out.append(str(stored))
return out
def coverage(conn: sqlite3.Connection, language_code: str) -> dict[str, int]:
"""How much of the book exists in a target language.
@@ -376,30 +406,41 @@ def coverage(conn: sqlite3.Connection, language_code: str) -> dict[str, int]:
than raising: that is a real state (a book that has never been sent), not an
error.
"""
code = lang.normalize(language_code) or language_code
code = canonical_code(language_code) or str(language_code).strip().lower()
def scalar(sql: str, args: tuple = ()) -> int:
def scalar(sql: str, args: tuple[str, ...] = ()) -> int:
return conn.execute(sql, args).fetchone()[0]
total = scalar("SELECT COUNT(*) FROM utterances")
empty = {
"total": total,
"delivered": 0,
"voiced": 0,
"not_voiced": 0,
"confirmed": 0,
"missing": total,
}
if not _translations_table_exists(conn):
return {
"total": total,
"delivered": 0,
"voiced": 0,
"not_voiced": 0,
"confirmed": 0,
"missing": total,
}
return empty
delivered = scalar("SELECT COUNT(*) FROM utterance_translations WHERE language_code=?", (code,))
matching = _matching_language_codes(conn, code)
if not matching:
return empty
placeholders = ",".join("?" for _ in matching)
delivered = scalar(
f"SELECT COUNT(*) FROM utterance_translations WHERE language_code IN ({placeholders})",
tuple(matching),
)
voiced = scalar(
"SELECT COUNT(*) FROM utterance_translations WHERE language_code=? AND voiced=1",
(code,),
f"SELECT COUNT(*) FROM utterance_translations "
f"WHERE language_code IN ({placeholders}) AND voiced=1",
tuple(matching),
)
confirmed = scalar(
"SELECT COUNT(*) FROM utterance_translations WHERE language_code=? AND status=?",
(code, STATUS_CONFIRMED),
f"SELECT COUNT(*) FROM utterance_translations "
f"WHERE language_code IN ({placeholders}) AND status=?",
(*matching, STATUS_CONFIRMED),
)
return {
"total": total,
@@ -418,19 +459,23 @@ def not_voiced_rows(conn: sqlite3.Connection, language_code: str) -> list[dict[s
so a source line has no counterpart. They are *kept*, with a reason, rather
than deleted: deleting loses the alignment (spec 31 section 6.2).
"""
code = lang.normalize(language_code) or language_code
code = canonical_code(language_code) or str(language_code).strip().lower()
if not _translations_table_exists(conn):
return []
matching = _matching_language_codes(conn, code)
if not matching:
return []
placeholders = ",".join("?" for _ in matching)
rows = conn.execute(
"""
f"""
SELECT t.utterance_id, t.review_reason, u.chapter_id, u.utterance_index,
u.speaker_name, u.text
FROM utterance_translations t
JOIN utterances u ON u.id = t.utterance_id
WHERE t.language_code = ? AND t.voiced = 0
WHERE t.language_code IN ({placeholders}) AND t.voiced = 0
ORDER BY u.chapter_id, u.utterance_index
""",
(code,),
tuple(matching),
).fetchall()
return [
{