UmbraAudio/scripts/import_translations.py

#!/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

    # replace this language's drafts, then write it
    python scripts/import_translations.py --project "..." \
        --delivery /tmp/delivery.json --write --replace

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.

``--replace`` clears the language's existing *draft* rows first, so a
re-delivery cannot leave a stale row behind for an utterance the new delivery
does not cover. Confirmed rows are never deleted: a confirmed translation is
an approval, and the import skips it.
"""

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 _clear_drafts(conn: sqlite3.Connection, code: str) -> int:
    """Delete the language's draft rows. Confirmed rows are kept.

    A confirmed translation is an approval; deleting it would destroy it.
    The import skips confirmed rows anyway, so they are left alone.
    """
    from umbraaudio.translations import canonical_code

    target = canonical_code(code) or str(code).strip().lower()
    rows = conn.execute("SELECT DISTINCT language_code FROM utterance_translations").fetchall()
    matching = [
        str(row[0])
        for row in rows
        if row[0] is not None and (canonical_code(row[0]) or str(row[0]).strip().lower()) == target
    ]
    if not matching:
        return 0
    placeholders = ",".join("?" for _ in matching)
    cursor = conn.execute(
        f"DELETE FROM utterance_translations "
        f"WHERE language_code IN ({placeholders}) AND status <> ?",
        (*matching, "confirmed"),
    )
    conn.commit()
    return cursor.rowcount


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")
    parser.add_argument(
        "--replace",
        action="store_true",
        help="clear the language's draft rows first (confirmed rows are kept)",
    )
    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

        if args.replace:
            removed = _clear_drafts(conn, code)
            print(f"\ncleared {removed} draft row(s) for {code}")

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