open-standards/aac/aac-rs/tools/pagesets/phrases/review.py

#!/usr/bin/env python3
"""The two-reader translation check for the phrase page sets.

A text is used in a language only when two independent readers agree on it:

  1. draft   A drafter translates the English (seeing it, with where each text
             is used and what it is for).
  2. blind   A second reader, who never sees the English, back-translates the
             drafter's words into English and says how they read (grammar,
             register, whose gender they mark).
  3. judge   The English, the translation and the back-translation are
             compared: "agree" only when the back-translation means what the
             English means, with nothing added or lost, in natural words.
  4. merge   Agreed texts are recorded as "agreed" (used by generate.py);
             the rest as "disputed" (left in English, visibly) and offered to
             the next round, where the drafter revises them and a fresh blind
             reader checks again.

Each round's record is kept under translations/review/<lang>/round<N>/, so
how each text was obtained is part of the data: draft-input.json (what the
drafter saw), draft.json, blind-input.json (what the blind reader saw, with
no English), back.json and verdict.json. The judge's input is derived from
those (`review.py judge` writes it again), and the <stage>-<n>.json parts
the readers wrote are joined into the files above.

  python3 review.py prepare <lang> <round>   write draft-input.json (texts
                                             not yet agreed, with English)
  python3 review.py blind <lang> <round>     write blind-input.json from
                                             draft.json (no English in it)
  python3 review.py judge <lang> <round>     write judge-input.json
  python3 review.py join <lang> <round> <stage>   join <stage>-<n>.json parts
                                             (draft, back, verdict)
  python3 review.py merge <lang> <round>     fold verdict.json into
                                             translations/<lang>.json
  python3 review.py prune                    drop translations of retired texts
  python3 review.py status                   one line per language

The readers were AI models; translations/<lang>.json says so, and that a
native speaker's review is still wanted for every text.
"""

from __future__ import annotations

import hashlib
import json
import sys
from pathlib import Path

from build import AGREED, LANGUAGES, STATUS_TEXT, TRANSLATIONS, WORK

REVIEW = TRANSLATIONS / "review"

# What each language's words should sound like. Shared by every round.
LANGUAGE_NOTES = {
    "es": "Spanish understood in Spain and Latin America: prefer words common to both; where they differ, use the most widely understood and note it.",
    "fr": "French (France). Use a narrow no-break space (U+202F) before ! ? ; and a no-break space (U+00A0) before :, so large print never wraps them alone.",
    "de": "German (Germany).",
    "pt": "Brazilian Portuguese.",
    "it": "Italian (Italy).",
    "zh": "Simplified Chinese (Mainland), everyday spoken Mandarin.",
    "ja": "Japanese. Polite です/ます for the adult set; natural, friendly speech elsewhere.",
    "ar": "Modern Standard Arabic, simple and natural, as a speech voice reads it. Avoid addressing one gender: use the plural or a neutral construction for 'you'.",
    "hi": "Hindi in Devanagari, everyday spoken Hindi (not Sanskritised). Use आप for 'you' in requests.",
    "ru": "Russian. Use Вы for 'you' in the adult set; ты is natural in the emergent set.",
}


def sid(english: str) -> str:
    """A stable id for an English text."""
    return hashlib.sha1(english.encode()).hexdigest()[:10]


def strings() -> dict[str, dict]:
    path = WORK / "strings.json"
    if not path.exists():
        sys.exit("run `python3 generate.py strings` first")
    return json.loads(path.read_text())


def family(uses: list[str]) -> list[str]:
    fams = []
    for u in uses:
        f = u.split(" ")[0].rstrip(":")
        if f in ("adult", "intent", "themes", "emergent") and f not in fams:
            fams.append(f)
    return fams or ["all"]


def lang_file(lang: str) -> Path:
    return TRANSLATIONS / f"{lang}.json"


def load_lang(lang: str) -> dict:
    path = lang_file(lang)
    if path.exists():
        return json.loads(path.read_text())
    name, locale = LANGUAGES[lang]
    return {
        "language": name,
        "locale": locale,
        "about": (
            "Translations of the phrase page sets. Each text was drafted, then back-translated by a second "
            "reader who never saw the English, and the two compared. A text is used only where they agreed "
            f"(status \"agreed\": {STATUS_TEXT}); \"disputed\" texts stay in English. The readers were AI "
            "models (2026-10-05), not native speakers: a native speaker's review is still wanted for every text."
        ),
        "strings": {},
    }


PART_SIZE = 280


def write_parts(d: Path, stage: str, out: dict) -> int:
    """Write <stage>-input.json and <stage>-input-<n>.json parts of at most
    PART_SIZE items, so a reader can take them one at a time. Returns the
    number of parts."""
    (d / f"{stage}-input.json").write_text(json.dumps(out, ensure_ascii=False, indent=1) + "\n")
    for old in d.glob(f"{stage}-input-*.json"):
        old.unlink()
    items = out["items"]
    n = 0
    for n, i in enumerate(range(0, len(items), PART_SIZE), start=1):
        part = dict(out, items=items[i:i + PART_SIZE], part=f"{n}")
        (d / f"{stage}-input-{n}.json").write_text(json.dumps(part, ensure_ascii=False, indent=1) + "\n")
    return n


def join(lang: str, rnd: int, stage: str) -> None:
    """Join <stage>-<n>.json parts into <stage>.json."""
    d = round_dir(lang, rnd)
    merged: dict = {}
    parts = sorted(d.glob(f"{stage}-[0-9]*.json"), key=lambda p: int(p.stem.split("-")[-1]))
    for part in parts:
        merged.update(json.loads(part.read_text()))
    (d / f"{stage}.json").write_text(json.dumps(merged, ensure_ascii=False, indent=1) + "\n")
    source = {"draft": "draft", "back": "blind", "verdict": "judge"}[stage]
    wanted = {i["id"] for i in json.loads((d / f"{source}-input.json").read_text())["items"]}
    missing = sorted(wanted - set(merged))
    bad = sorted(k for k, v in merged.items() if stage == "draft" and not (
        isinstance(v.get("text"), str) or (isinstance(v.get("forms"), dict) and set(v["forms"]) == {"f", "m"})))
    print(f"{lang} round {rnd}: {stage}.json has {len(merged)} entries from {len(parts)} parts;"
          f" {len(missing)} missing{(': ' + ', '.join(missing[:10])) if missing else ''};"
          f" {len(bad)} malformed{(': ' + ', '.join(bad[:10])) if bad else ''}")


def round_dir(lang: str, rnd: int) -> Path:
    d = REVIEW / lang / f"round{rnd}"
    d.mkdir(parents=True, exist_ok=True)
    return d


def prepare(lang: str, rnd: int) -> None:
    done = load_lang(lang)["strings"]
    previous = {}
    if rnd > 1:
        prev = REVIEW / lang / f"round{rnd - 1}"
        verdicts = json.loads((prev / "verdict.json").read_text())
        drafts = json.loads((prev / "draft.json").read_text())
        previous = {k: {"previous": drafts.get(k), "why_not_agreed": v} for k, v in verdicts.items()}
    items = []
    for en, meta in strings().items():
        k = sid(en)
        if done.get(en, {}).get("status") == AGREED:
            continue
        if rnd > 1 and k not in previous and en in done:
            # Agreed or still waiting in a round of its own: only texts that
            # were disputed, or that no round has seen yet, go to this one.
            continue
        item = {"id": k, "en": en, "role": meta["roles"], "sets": family(meta["uses"]), "used_on": meta["uses"]}
        if meta["notes"]:
            item["notes"] = meta["notes"]
        if k in previous:
            item.update(previous[k])
        items.append(item)
    out = {"language": LANGUAGES[lang][0], "locale": LANGUAGES[lang][1], "guidance": LANGUAGE_NOTES[lang],
           "items": items}
    n = write_parts(round_dir(lang, rnd), "draft", out)
    print(f"{lang} round {rnd}: {len(items)} texts to draft in {n} parts")


def blind(lang: str, rnd: int) -> None:
    d = round_dir(lang, rnd)
    draft = json.loads((d / "draft.json").read_text())
    by_id = {sid(en): meta for en, meta in strings().items()}
    items = []
    for k, t in draft.items():
        meta = by_id.get(k)
        if meta is None:
            continue  # a text the sources no longer use
        item = {"id": k, "role": [r for r in meta["roles"]], "sets": family(meta["uses"])}
        if "forms" in t:
            item["forms"] = t["forms"]
        else:
            item["text"] = t["text"]
        items.append(item)
    out = {"language": LANGUAGES[lang][0], "items": items}
    n = write_parts(d, "blind", out)
    print(f"{lang} round {rnd}: {len(items)} texts for the blind reader in {n} parts")


def judge(lang: str, rnd: int) -> None:
    d = round_dir(lang, rnd)
    draft = json.loads((d / "draft.json").read_text())
    back = json.loads((d / "back.json").read_text())
    by_id = {sid(en): (en, meta) for en, meta in strings().items()}
    items = []
    for k, t in draft.items():
        if k not in by_id:
            continue  # a text the sources no longer use
        en, meta = by_id[k]
        item = {"id": k, "en": en, "role": meta["roles"], "sets": family(meta["uses"])}
        item.update({kk: vv for kk, vv in t.items() if kk in ("text", "forms", "note")})
        item["back_translation"] = back.get(k, {"missing": True})
        items.append(item)
    out = {"language": LANGUAGES[lang][0], "items": items}
    n = write_parts(d, "judge", out)
    print(f"{lang} round {rnd}: {len(items)} texts to judge in {n} parts")


def merge(lang: str, rnd: int) -> None:
    d = round_dir(lang, rnd)
    draft = json.loads((d / "draft.json").read_text())
    back = json.loads((d / "back.json").read_text())
    verdict = json.loads((d / "verdict.json").read_text())
    by_id = {sid(en): en for en in strings()}
    data = load_lang(lang)
    agreed = disputed = 0
    for k, t in draft.items():
        en = by_id.get(k)
        if en is None:
            continue
        v = verdict.get(k, {"verdict": "missing"})
        entry = {"status": AGREED if v.get("verdict") == "agree" else "disputed", "round": rnd}
        if "forms" in t:
            entry["forms"] = t["forms"]
        else:
            entry["text"] = t["text"]
        b = back.get(k, {})
        entry["back_translation"] = b.get("back") or {g: b.get(f"back_{g}") for g in ("f", "m") if b.get(f"back_{g}")}
        notes = [x for x in (t.get("note"), b.get("notes"), v.get("reason")) if x]
        entry["notes"] = " / ".join(notes)
        data["strings"][en] = entry
        if entry["status"] == AGREED:
            agreed += 1
        else:
            disputed += 1
    data["strings"] = dict(sorted(data["strings"].items()))
    lang_file(lang).write_text(json.dumps(data, ensure_ascii=False, indent=1) + "\n")
    print(f"{lang} round {rnd}: {agreed} agreed, {disputed} disputed")


def prune() -> None:
    """Drop translations of English texts the sources no longer use (their
    review records stay under translations/review/)."""
    current = strings()
    for lang in LANGUAGES:
        data = load_lang(lang)
        gone = [en for en in data["strings"] if en not in current]
        for en in gone:
            del data["strings"][en]
        lang_file(lang).write_text(json.dumps(data, ensure_ascii=False, indent=1) + "\n")
        print(f"{lang}: dropped {len(gone)} retired texts{(': ' + '; '.join(gone)) if gone else ''}")


def status() -> None:
    all_s = strings()
    for lang in LANGUAGES:
        st = load_lang(lang)["strings"]
        a = sum(1 for en in all_s if st.get(en, {}).get("status") == AGREED)
        dsp = sum(1 for en in all_s if st.get(en, {}).get("status") == "disputed")
        print(f"{lang}: {a} agreed, {dsp} disputed, {len(all_s) - a - dsp} not yet translated, of {len(all_s)}")


def main() -> None:
    cmd = sys.argv[1]
    if cmd == "status":
        status()
        return
    if cmd == "prune":
        prune()
        return
    lang, rnd = sys.argv[2], int(sys.argv[3])
    if cmd == "join":
        join(lang, rnd, sys.argv[4])
        return
    {"prepare": prepare, "blind": blind, "judge": judge, "merge": merge}[cmd](lang, rnd)


if __name__ == "__main__":
    main()