Upload 3 files

app.py

@@ -0,0 +1,726 @@
+"""Vernacular - translation review tool for the Riverstone language pack.
+
+Reads the review records produced by pipeline/translate_pack.py
+(translations/<lang>/**), shows each string with its context and proposed
+translation (character-toned for dialogue), and lets a reviewer approve,
+reject, or correct it. Decisions are persisted straight back into the record
+files, which pipeline/export_pack.py turns into the final language pack.
+"""
+
+import json
+import re
+import zipfile
+from pathlib import Path
+from xml.etree import ElementTree as ET
+
+import gradio as gr
+
+import config
+import build_character_wikis
+from converter.docs import readLocalDoc
+from converter.sheets import readLocalSheet
+from converter.utils import toSafeEntityName
+from pipeline import clients
+from pipeline import translate_pack
+from pipeline.build_file_context import normalize_key
+from pipeline.rules import iter_translatable
+from pipeline.dialogue_map import build_dialogue_map, display_name, load_wiki
+
+_context_path = (
+    config.FILE_CONTEXT_PATH
+    if config.FILE_CONTEXT_PATH.exists()
+    else config.FILE_CONTEXT_SAMPLE_PATH
+)
+FILE_CONTEXT = (
+    json.loads(_context_path.read_text(encoding="utf-8")) if _context_path.exists() else {}
+)
+NEIGHBOR_LINES = 2
+UPLOAD_SOURCE_DIR = config.ROOT / "uploaded_wiki_sources"
+UPLOAD_REVIEW_DIR = config.SOURCE_DIR / "Uploaded" / "Filler Chats"
+WIKI_BUNDLE_PATH = config.CHAR_WIKI_DIR / "character_wikis.json"
+
+
+# --- character wiki upload/build/download -----------------------------------
+
+
+def _safe_id(name: str) -> str:
+    stem = Path(name).stem.lower()
+    stem = re.sub(r"[^a-z0-9]+", "_", stem).strip("_")
+    return stem or "uploaded_character"
+
+
+def _unique_path(path: Path) -> Path:
+    if not path.exists():
+        return path
+    for i in range(2, 1000):
+        candidate = path.with_name(f"{path.stem}_{i}{path.suffix}")
+        if not candidate.exists():
+            return candidate
+    raise RuntimeError(f"Could not find a free filename for {path.name}")
+
+
+def _uploaded_path(uploaded) -> Path:
+    return Path(uploaded.name if hasattr(uploaded, "name") else uploaded)
+
+
+def _text_from_docx(path: Path) -> str:
+    with zipfile.ZipFile(path) as archive:
+        xml = archive.read("word/document.xml")
+    root = ET.fromstring(xml)
+    ns = {"w": "http://schemas.openxmlformats.org/wordprocessingml/2006/main"}
+    paragraphs = []
+    for paragraph in root.findall(".//w:p", ns):
+        text = "".join(node.text or "" for node in paragraph.findall(".//w:t", ns))
+        if text.strip():
+            paragraphs.append(text)
+    return "\n".join(paragraphs)
+
+
+def _xlsx_shared_strings(archive: zipfile.ZipFile) -> list[str]:
+    try:
+        root = ET.fromstring(archive.read("xl/sharedStrings.xml"))
+    except KeyError:
+        return []
+    ns = {"x": "http://schemas.openxmlformats.org/spreadsheetml/2006/main"}
+    strings = []
+    for item in root.findall(".//x:si", ns):
+        strings.append("".join(node.text or "" for node in item.findall(".//x:t", ns)))
+    return strings
+
+
+def _text_from_xlsx(path: Path) -> str:
+    ns = {"x": "http://schemas.openxmlformats.org/spreadsheetml/2006/main"}
+    lines = []
+    with zipfile.ZipFile(path) as archive:
+        shared = _xlsx_shared_strings(archive)
+        sheet_names = sorted(
+            name for name in archive.namelist()
+            if name.startswith("xl/worksheets/sheet") and name.endswith(".xml")
+        )
+        for sheet_name in sheet_names:
+            lines.append(f"[{Path(sheet_name).stem}]")
+            root = ET.fromstring(archive.read(sheet_name))
+            for row in root.findall(".//x:row", ns):
+                values = []
+                for cell in row.findall("x:c", ns):
+                    value = cell.find("x:v", ns)
+                    if value is None or value.text is None:
+                        values.append("")
+                    elif cell.attrib.get("t") == "s":
+                        index = int(value.text)
+                        values.append(shared[index] if index < len(shared) else "")
+                    else:
+                        values.append(value.text)
+                if any(v.strip() for v in values):
+                    lines.append(" | ".join(values))
+    return "\n".join(lines)
+
+
+def _text_from_upload(path: Path) -> str:
+    suffix = path.suffix.lower()
+    if suffix == ".docx":
+        return _text_from_docx(path)
+    if suffix == ".xlsx":
+        return _text_from_xlsx(path)
+    if suffix in {".txt", ".csv", ".json"}:
+        return path.read_text(encoding="utf-8", errors="replace")
+    if suffix in {".doc", ".xls"}:
+        raise ValueError(
+            f"{path.name}: old binary {suffix} files are not supported; save as "
+            f"{'.docx' if suffix == '.doc' else '.xlsx'} and upload again."
+        )
+    raise ValueError(f"{path.name}: unsupported file type.")
+
+
+def _converter_file_path_key(path: Path) -> str:
+    parts = [toSafeEntityName(part) for part in path.with_suffix("").parts]
+    return "/".join(parts)
+
+
+def _fallback_chat_payload(text: str) -> list[dict]:
+    return [
+        {"id": i + 1, "type": "0", "text": line}
+        for i, line in enumerate(text.splitlines())
+        if line.strip()
+    ]
+
+
+def _converted_upload_payload(source_path: Path, text: str, review_rel: str):
+    suffix = source_path.suffix.lower()
+    file_path = _converter_file_path_key(Path(review_rel).with_suffix(source_path.suffix))
+    if suffix == ".docx":
+        converted = readLocalDoc(source_path, file_path)
+    elif suffix == ".xlsx":
+        converted = readLocalSheet(source_path, file_path)
+    else:
+        return _fallback_chat_payload(text)
+
+    if any(iter_translatable(converted, review_rel)):
+        return converted
+    return _fallback_chat_payload(text)
+
+
+def _write_upload_source(uploaded) -> tuple[str, dict, str]:
+    source_path = _uploaded_path(uploaded)
+    char_id = _safe_id(source_path.name)
+    text = _text_from_upload(source_path)
+    if not text.strip():
+        raise ValueError(f"{source_path.name}: no readable text found.")
+    wiki_target = _unique_path(UPLOAD_SOURCE_DIR / f"{char_id}.json")
+    wiki_payload = {
+        "source_file": source_path.name,
+        "content": text,
+    }
+    wiki_target.write_text(
+        json.dumps(wiki_payload, indent=2, ensure_ascii=False), encoding="utf-8"
+    )
+    wiki_rel = str(wiki_target.relative_to(config.ROOT)).replace("\\", "/")
+
+    review_target = _unique_path(UPLOAD_REVIEW_DIR / f"{char_id}.json")
+    review_rel = str(review_target.relative_to(config.SOURCE_DIR)).replace("\\", "/")
+    review_payload = _converted_upload_payload(source_path, text, review_rel)
+    review_target.write_text(
+        json.dumps(review_payload, indent=2, ensure_ascii=False), encoding="utf-8"
+    )
+    char_data_path = config.CHAR_DATA_DIR / f"{char_id}.json"
+    char_data = {
+        "id": char_id,
+        "name": [Path(source_path.name).stem],
+        "mentions": [f"English_JSON/{review_rel}"],
+        "wiki_mentions": [wiki_rel],
+        "references": [],
+    }
+    if char_data_path.exists():
+        existing = json.loads(char_data_path.read_text(encoding="utf-8"))
+        mentions = existing.get("mentions", [])
+        wiki_mentions = existing.get("wiki_mentions", [])
+        char_data = {
+            **existing,
+            "id": char_id,
+            "name": existing.get("name") or char_data["name"],
+            "mentions": mentions
+            + ([] if f"English_JSON/{review_rel}" in mentions else [f"English_JSON/{review_rel}"]),
+            "wiki_mentions": wiki_mentions
+            + ([] if wiki_rel in wiki_mentions else [wiki_rel]),
+            "references": existing.get("references", []),
+        }
+    char_data_path.write_text(
+        json.dumps(char_data, indent=2, ensure_ascii=False), encoding="utf-8"
+    )
+    return char_id, char_data, review_rel
+
+
+def _export_wiki_bundle() -> Path:
+    config.CHAR_WIKI_DIR.mkdir(exist_ok=True)
+    bundle = {}
+    for wiki_path in sorted(config.CHAR_WIKI_DIR.glob("*.md")):
+        if wiki_path.name.endswith(".sample.md"):
+            continue
+        char_id = wiki_path.stem
+        bundle[char_id] = {
+            "name": display_name(char_id),
+            "wiki": wiki_path.read_text(encoding="utf-8").strip(),
+        }
+    WIKI_BUNDLE_PATH.write_text(
+        json.dumps(bundle, indent=2, ensure_ascii=False), encoding="utf-8"
+    )
+    return WIKI_BUNDLE_PATH
+
+
+def _import_wiki_bundle(uploaded) -> int:
+    if not uploaded:
+        return 0
+    path = _uploaded_path(uploaded)
+    data = json.loads(path.read_text(encoding="utf-8"))
+    if not isinstance(data, dict):
+        raise ValueError("Uploaded wiki JSON must be an object keyed by character id.")
+    config.CHAR_WIKI_DIR.mkdir(exist_ok=True)
+    imported = 0
+    for raw_id, value in data.items():
+        char_id = _safe_id(raw_id)
+        wiki = value.get("wiki") if isinstance(value, dict) else value
+        name = value.get("name") if isinstance(value, dict) else raw_id
+        if not isinstance(wiki, str) or not wiki.strip():
+            continue
+        (config.CHAR_WIKI_DIR / f"{char_id}.md").write_text(
+            wiki.strip() + "\n", encoding="utf-8"
+        )
+        char_data_path = config.CHAR_DATA_DIR / f"{char_id}.json"
+        if not char_data_path.exists():
+            char_data_path.write_text(
+                json.dumps(
+                    {"id": char_id, "name": [str(name or raw_id)], "mentions": [], "references": []},
+                    indent=2,
+                    ensure_ascii=False,
+                ),
+                encoding="utf-8",
+            )
+        imported += 1
+    _export_wiki_bundle()
+    return imported
+
+
+def _process_uploaded_character(char_data: dict, progress: dict) -> dict:
+    char_id = char_data["id"]
+    names = char_data.get("name") or []
+    name = names[0] if names else char_id
+    aka = f' (also appearing as {", ".join(names[1:])})' if len(names) > 1 else ""
+    mentions = char_data.get("wiki_mentions") or char_data.get("mentions", [])
+    wiki_path = config.CHAR_WIKI_DIR / f"{char_id}.md"
+    done = list(progress.get(char_id, []))
+    wiki = wiki_path.read_text(encoding="utf-8").strip() if wiki_path.exists() else ""
+    pending = [mention for mention in mentions if mention not in done]
+
+    if not pending:
+        return progress
+
+    system_prompt = build_character_wikis.SYSTEM_PROMPT.format(name=name, aka=aka)
+    for rel_path in pending:
+        file_path = config.ROOT / rel_path
+        data = json.loads(file_path.read_text(encoding="utf-8"))
+        user_msg = build_character_wikis.USER_TEMPLATE.format(
+            wiki=wiki or build_character_wikis.EMPTY_WIKI_PLACEHOLDER,
+            path=rel_path.removeprefix("uploaded_wiki_sources/"),
+            hint=build_character_wikis.attribution_hint(rel_path, name),
+            content=json.dumps(data, indent=2, ensure_ascii=False),
+        )
+        reply = build_character_wikis.clean_reply(
+            clients.update_character_wiki(system_prompt, user_msg)
+        )
+        if reply.startswith("#"):
+            wiki = reply
+            wiki_path.write_text(wiki + "\n", encoding="utf-8")
+        else:
+            raise RuntimeError(
+                f"Model returned a malformed wiki for {name}; previous wiki was kept."
+            )
+        done = done + [rel_path]
+        progress = {**progress, char_id: done}
+        build_character_wikis.save_progress(progress)
+    return progress
+
+
+def build_uploaded_wikis(uploaded_files, existing_wiki_json):
+    uploaded_files = uploaded_files or []
+    try:
+        UPLOAD_SOURCE_DIR.mkdir(exist_ok=True)
+        UPLOAD_REVIEW_DIR.mkdir(parents=True, exist_ok=True)
+        config.CHAR_DATA_DIR.mkdir(exist_ok=True)
+        config.CHAR_WIKI_DIR.mkdir(exist_ok=True)
+
+        imported = _import_wiki_bundle(existing_wiki_json)
+        progress = build_character_wikis.load_progress()
+        built = []
+        review_files = []
+        for uploaded in uploaded_files:
+            char_id, char_data, review_rel = _write_upload_source(uploaded)
+            progress = _process_uploaded_character(char_data, progress)
+            built.append(char_id)
+            review_files.append(review_rel)
+        bundle_path = _export_wiki_bundle()
+    except Exception as exc:
+        return (
+            f"Wiki update failed: {type(exc).__name__}: {exc}",
+            gr.update(value=str(WIKI_BUNDLE_PATH) if WIKI_BUNDLE_PATH.exists() else None),
+            [],
+        )
+
+    parts = []
+    if imported:
+        parts.append(f"imported {imported} existing wiki entr{'y' if imported == 1 else 'ies'}")
+    if built:
+        parts.append(f"updated {len(built)} wiki entr{'y' if len(built) == 1 else 'ies'}")
+    if not parts:
+        parts.append("no files selected; refreshed the downloadable wiki JSON")
+    return (
+        "Wiki JSON ready: " + ", ".join(parts) + ".",
+        gr.update(value=str(bundle_path)),
+        sorted(set(review_files)),
+    )
+
+
+# --- record access -----------------------------------------------------------
+
+
+def record_path(rel: str) -> Path:
+    return config.TRANSLATIONS_DIR / rel
+
+
+def load_record(rel: str) -> dict:
+    return json.loads(record_path(rel).read_text(encoding="utf-8"))
+
+
+def save_record(rel: str, record: dict) -> None:
+    record_path(rel).write_text(
+        json.dumps(record, indent=2, ensure_ascii=False), encoding="utf-8"
+    )
+
+
+def list_record_files() -> list[str]:
+    if not config.TRANSLATIONS_DIR.exists():
+        return []
+    return sorted(
+        str(p.relative_to(config.TRANSLATIONS_DIR))
+        for p in config.TRANSLATIONS_DIR.rglob("*.json")
+        if not p.name.startswith(".")
+    )
+
+
+def proposed_text(item: dict) -> str:
+    return item.get("toned") or item.get("mt") or ""
+
+
+def reviewed(item: dict) -> bool:
+    return item["status"] in ("approved", "edited")
+
+
+def file_progress(rel: str) -> tuple[int, int]:
+    items = load_record(rel)["items"]
+    return sum(1 for i in items if reviewed(i)), len(items)
+
+
+def file_choices() -> list[tuple[str, str]]:
+    choices = []
+    for rel in list_record_files():
+        done, total = file_progress(rel)
+        marker = "✅" if done == total else f"{done}/{total} reviewed"
+        choices.append((f"{rel}  ·  {marker}", rel))
+    return choices
+
+
+def overall_stats() -> str:
+    counts = {"approved": 0, "edited": 0, "rejected": 0, "pending": 0}
+    total = 0
+    for rel in list_record_files():
+        for item in load_record(rel)["items"]:
+            counts[item["status"]] = counts.get(item["status"], 0) + 1
+            total += 1
+    if not total:
+        return "No review records found - run `python -m pipeline.translate_pack` first."
+    done = counts["approved"] + counts["edited"]
+    return (
+        f"**{config.SOURCE_LANG_NAME} → {config.TARGET_LANG_NAME}** · "
+        f"{total} strings · ✅ {counts['approved']} approved · "
+        f"✏️ {counts['edited']} edited · ❌ {counts['rejected']} rejected · "
+        f"⏳ {counts['pending']} pending · **{done / total:.0%} reviewed**"
+    )
+
+
+# --- rendering ---------------------------------------------------------------
+
+
+def context_panel(record: dict, index: int) -> str:
+    rel = record["file"]
+    item = record["items"][index]
+    meta = FILE_CONTEXT.get(normalize_key(rel), {})
+    lines = [f"**File:** `{rel}`  ·  **Key:** `{item['key']}`"]
+
+    if meta:
+        lines.append(f"**Content type:** {meta.get('label', '?')} — {meta.get('role', '')}")
+    if item["character"]:
+        lines.append(
+            f"**Speaker:** 🎭 {display_name(item['character'])} "
+            f"(dialogue - translation was adjusted to the character's voice)"
+        )
+    elif item["kind"] == "player":
+        lines.append("**Speaker:** 🕵️ player / phone owner (plain translation)")
+    if meta.get("summary"):
+        lines.append(f"<small>{meta['summary']}</small>")
+
+    neighbors = []
+    start = max(0, index - NEIGHBOR_LINES)
+    for i in range(start, min(len(record["items"]), index + NEIGHBOR_LINES + 1)):
+        other = record["items"][i]
+        marker = "→" if i == index else "&nbsp;&nbsp;"
+        who = display_name(other["character"]) if other["character"] else other["kind"]
+        neighbors.append(f"{marker} **{who}:** {other['source']}")
+    lines.append("**Surrounding lines:**<br>" + "<br>".join(neighbors))
+    return "\n\n".join(lines)
+
+
+def wiki_markdown(item: dict) -> str:
+    if not item["character"]:
+        return "*No character wiki for this line.*"
+    # Full wiki locally; trimmed public sample (.sample.md) on HF Spaces.
+    for name in (f"{item['character']}.md", f"{item['character']}.sample.md"):
+        wiki_path = config.CHAR_WIKI_DIR / name
+        if wiki_path.exists():
+            return wiki_path.read_text(encoding="utf-8")
+    return "*Wiki missing.*"
+
+
+STATUS_ICONS = {"pending": "⏳", "approved": "✅", "edited": "✏️", "rejected": "❌"}
+
+
+def status_badge(item: dict) -> str:
+    extra = f" — final: *{item['final']}*" if item.get("final") else ""
+    return f"### {STATUS_ICONS[item['status']]} {item['status'].upper()}{extra}"
+
+
+def _truncate(text: str, limit: int = 80) -> str:
+    return text if len(text) <= limit else text[: limit - 1] + "…"
+
+
+def items_table(record: dict) -> list[list]:
+    """Overview of every string in the file, for the jump-to table."""
+    rows = []
+    for i, item in enumerate(record["items"]):
+        speaker = display_name(item["character"]) if item["character"] else item["kind"]
+        rows.append(
+            [
+                i + 1,
+                STATUS_ICONS[item["status"]],
+                speaker,
+                _truncate(item["source"]),
+                _truncate(item.get("final") or proposed_text(item)),
+            ]
+        )
+    return rows
+
+
+def render(rel: str | None, index: int):
+    """Outputs: context, source, translation box, status, item label, wiki, stats, table."""
+    if not rel:
+        return ("*Select a file to start reviewing.*", "", "", "", "0 / 0", "", overall_stats(), [])
+    if not record_path(rel).exists():
+        msg = (
+            f"*This file has not been translated yet. Run:*\n\n"
+            f'```\npython -m pipeline.translate_pack --filter "{rel}"\n```'
+        )
+        return (msg, "", "", "", "0 / 0", "", overall_stats(), [])
+    record = load_record(rel)
+    items = record["items"]
+    index = max(0, min(index, len(items) - 1))
+    item = items[index]
+    translation = item["final"] if item.get("final") else proposed_text(item)
+    return (
+        context_panel(record, index),
+        item["source"],
+        translation,
+        status_badge(item),
+        f"{index + 1} / {len(items)}",
+        wiki_markdown(item),
+        overall_stats(),
+        items_table(record),
+    )
+
+
+# --- actions -----------------------------------------------------------------
+
+
+def next_pending(items: list[dict], after: int) -> int:
+    order = list(range(after + 1, len(items))) + list(range(0, after + 1))
+    for i in order:
+        if items[i]["status"] == "pending":
+            return i
+    return min(after + 1, len(items) - 1)
+
+
+def select_file(rel: str):
+    index = 0
+    if rel and record_path(rel).exists():
+        index = next_pending(load_record(rel)["items"], -1)
+    return (rel, index, *render(rel, index))
+
+
+def jump_to(rel: str, evt: gr.SelectData):
+    index = evt.index[0]
+    return (index, *render(rel, index))
+
+
+def navigate(rel: str, index: int, delta: int):
+    if not rel:
+        return (index, *render(rel, index))
+    new_index = index + delta
+    return (new_index, *render(rel, new_index))
+
+
+def decide(rel: str, index: int, text: str, decision: str):
+    if not rel or not record_path(rel).exists():
+        return (index, *render(rel, index), gr.update())
+    record = load_record(rel)
+    item = record["items"][index]
+    text = text.strip()
+    if decision == "reject":
+        item["status"], item["final"] = "rejected", None
+    elif text and text != proposed_text(item):
+        item["status"], item["final"] = "edited", text
+    else:
+        item["status"], item["final"] = "approved", proposed_text(item)
+    save_record(rel, record)
+    new_index = next_pending(record["items"], index)
+    return (new_index, *render(rel, new_index), gr.update(choices=file_choices(), value=rel))
+
+
+def _translate_uploaded_records(uploaded_review_files: list[str]) -> str:
+    uploaded_review_files = sorted(set(uploaded_review_files or []))
+    if not uploaded_review_files and UPLOAD_REVIEW_DIR.exists():
+        uploaded_review_files = sorted(
+            str(path.relative_to(config.SOURCE_DIR)).replace("\\", "/")
+            for path in UPLOAD_REVIEW_DIR.glob("*.json")
+        )
+    if not uploaded_review_files:
+        return "No uploaded source files are waiting for translation."
+
+    dmap = build_dialogue_map()
+    cache = (
+        json.loads(config.CACHE_PATH.read_text(encoding="utf-8"))
+        if config.CACHE_PATH.exists()
+        else {}
+    )
+    updated = 0
+    translated = 0
+    toned = 0
+    skipped = 0
+    for rel in uploaded_review_files:
+        source_path = config.SOURCE_DIR / rel
+        if not source_path.exists():
+            skipped += 1
+            continue
+        data = json.loads(source_path.read_text(encoding="utf-8"))
+        char_id = dmap.get(rel)
+        record_path_for_rel = config.TRANSLATIONS_DIR / rel
+        existing = translate_pack.load_record(record_path_for_rel)
+        record = {
+            "file": rel,
+            "character": char_id,
+            "items": translate_pack.build_items(data, rel, char_id, existing),
+        }
+        if not record["items"]:
+            skipped += 1
+            continue
+        if translate_pack.needs_work(record, "all"):
+            wiki = load_wiki(char_id) if char_id else None
+            name = display_name(char_id) if char_id else None
+            items, cache, counts = clients.translate_and_tone_items(
+                record["items"],
+                cache,
+                wiki,
+                name,
+                translate_pack.CONTEXT_LINES,
+            )
+            record["items"] = items
+            translated += counts["translated"]
+            toned += counts["toned"]
+            updated += 1
+        translate_pack.save_json(record_path_for_rel, record)
+        translate_pack.save_json(config.CACHE_PATH, cache)
+
+    parts = [
+        f"{updated} uploaded file{'s' if updated != 1 else ''} processed",
+        f"{translated} translation{'s' if translated != 1 else ''}",
+        f"{toned} tone pass{'es' if toned != 1 else ''}",
+    ]
+    if skipped:
+        parts.append(f"{skipped} skipped")
+    return "Refresh complete: " + ", ".join(parts) + "."
+
+
+def refresh(uploaded_review_files):
+    status = _translate_uploaded_records(uploaded_review_files)
+    return gr.update(choices=file_choices()), overall_stats(), status
+
+
+# --- UI ----------------------------------------------------------------------
+
+with gr.Blocks(title="Vernacular - Riverstone Translation Review") as demo:
+    gr.Markdown(f"# 🌍 Vernacular — {config.TARGET_LANG_NAME} translation review")
+    stats_md = gr.Markdown(overall_stats())
+
+    with gr.Accordion("Character wiki builder", open=True):
+        wiki_source_files = gr.File(
+            label="Upload character source files",
+            file_count="multiple",
+            file_types=[".doc", ".docx", ".xls", ".xlsx", ".json", ".txt", ".csv"],
+        )
+        existing_wiki_json = gr.File(
+            label="Upload existing wiki JSON",
+            file_count="single",
+            file_types=[".json"],
+        )
+        with gr.Row():
+            build_wiki_btn = gr.Button("Build / update wiki JSON", variant="primary")
+            wiki_json_file = gr.File(
+                label="Download wiki JSON",
+                value=str(WIKI_BUNDLE_PATH) if WIKI_BUNDLE_PATH.exists() else None,
+                interactive=False,
+            )
+        wiki_build_status = gr.Markdown("")
+
+    with gr.Row():
+        file_dd = gr.Dropdown(
+            choices=file_choices(), label="File", interactive=True, scale=5
+        )
+        refresh_btn = gr.Button("🔄 Refresh", scale=1)
+
+    current_file = gr.State(None)
+    current_index = gr.State(0)
+    uploaded_review_files = gr.State([])
+
+    with gr.Row():
+        with gr.Column(scale=3):
+            context_md = gr.Markdown("*Select a file to start reviewing.*")
+            source_tb = gr.Textbox(
+                label=f"{config.SOURCE_LANG_NAME} original",
+                lines=2,
+                interactive=False,
+            )
+            translation_tb = gr.Textbox(
+                label=f"{config.TARGET_LANG_NAME} translation (edit if needed)",
+                lines=3,
+                interactive=True,
+            )
+            with gr.Row():
+                prev_btn = gr.Button("← Prev")
+                item_label = gr.Markdown("0 / 0")
+                next_btn = gr.Button("Next →")
+            with gr.Row():
+                approve_btn = gr.Button("✅ Approve / Save edit", variant="primary")
+                reject_btn = gr.Button("❌ Reject", variant="stop")
+            status_md = gr.Markdown("")
+        with gr.Column(scale=2):
+            with gr.Accordion("🎭 Character wiki", open=False):
+                wiki_md = gr.Markdown("")
+
+    with gr.Accordion("📋 All strings in this file (click a row to jump)", open=False):
+        items_df = gr.Dataframe(
+            headers=["#", "", "speaker", f"{config.SOURCE_LANG_NAME} original",
+                     f"{config.TARGET_LANG_NAME} translation"],
+            interactive=False,
+            wrap=True,
+        )
+
+    render_outputs = [context_md, source_tb, translation_tb, status_md, item_label,
+                      wiki_md, stats_md, items_df]
+
+    file_dd.change(select_file, [file_dd], [current_file, current_index, *render_outputs])
+    items_df.select(jump_to, [current_file], [current_index, *render_outputs])
+    prev_btn.click(
+        lambda rel, i: navigate(rel, i, -1),
+        [current_file, current_index],
+        [current_index, *render_outputs],
+    )
+    next_btn.click(
+        lambda rel, i: navigate(rel, i, +1),
+        [current_file, current_index],
+        [current_index, *render_outputs],
+    )
+    approve_btn.click(
+        lambda rel, i, t: decide(rel, i, t, "approve"),
+        [current_file, current_index, translation_tb],
+        [current_index, *render_outputs, file_dd],
+    )
+    reject_btn.click(
+        lambda rel, i, t: decide(rel, i, t, "reject"),
+        [current_file, current_index, translation_tb],
+        [current_index, *render_outputs, file_dd],
+    )
+    refresh_btn.click(refresh, [uploaded_review_files], [file_dd, stats_md, wiki_build_status])
+    build_wiki_btn.click(
+        build_uploaded_wikis,
+        [wiki_source_files, existing_wiki_json],
+        [wiki_build_status, wiki_json_file, uploaded_review_files],
+    )
+
+if __name__ == "__main__":
+    demo.launch()

config.py

@@ -0,0 +1,70 @@
+"""Central configuration for the translation pipeline and review tool.
+
+To target a new language, change the three TARGET_* values (and PACK_NAME).
+Everything else derives from them.
+"""
+
+from pathlib import Path
+
+ROOT = Path(__file__).parent
+
+# --- Target language -------------------------------------------------------
+TARGET_LANG_CODE = "de"  # ISO 639-1 code understood by TranslateGemma
+TARGET_LANG_NAME = "German"
+PACK_NAME = "German_JSON"
+
+SOURCE_LANG_CODE = "en"
+SOURCE_LANG_NAME = "English"
+
+# --- Paths ------------------------------------------------------------------
+SOURCE_DIR = ROOT / "English_JSON"
+PACK_DIR = ROOT / PACK_NAME
+TRANSLATIONS_DIR = ROOT / "translations" / TARGET_LANG_CODE
+CACHE_PATH = TRANSLATIONS_DIR / ".cache.json"
+FILE_CONTEXT_PATH = ROOT / "file_context.json"
+# Public fallback used on HF Spaces: context for the demo sample files only.
+FILE_CONTEXT_SAMPLE_PATH = ROOT / "file_context.sample.json"
+FILE_MAPPING_PATH = ROOT / "FILE_MAPPING.md"
+CHAR_DATA_DIR = ROOT / "character_data"
+CHAR_WIKI_DIR = ROOT / "character_wikis"
+
+# --- Hugging Face in-process model inference --------------------------------
+# Both translation stages now run inside Python with Transformers. The model
+# repos below are the canonical Hugging Face checkpoints, not local HTTP
+# servers. TranslateGemma is gated, so the runtime needs an HF token whose
+# account has accepted Google's Gemma terms.
+TRANSLATE_MODEL_ID = "google/translategemma-12b-it"
+TONE_MODEL_ID = "google/gemma-4-12B-it"
+
+# "auto" follows the model card examples and works locally and on GPU Spaces.
+# Set HF_DEVICE_MAP = None to load on CPU first and explicitly move to cuda.
+HF_DEVICE_MAP = "auto"
+HF_DTYPE = "auto"
+HF_ATTN_IMPLEMENTATION = None
+
+TRANSLATE_MAX_NEW_TOKENS = 1024
+TONE_MAX_NEW_TOKENS = 512
+
+# Gemma's officially recommended sampling parameters for creative rewriting.
+TONE_GENERATION_KWARGS = {
+    "do_sample": True,
+    "temperature": 1.0,
+    "top_k": 64,
+    "top_p": 0.95,
+}
+
+# ZeroGPU only exposes the real GPU inside @spaces.GPU functions. The decorator
+# is a no-op outside ZeroGPU, so keeping these values here makes deployment a
+# config change instead of a code fork.
+ZERO_GPU_DURATION_S = 600
+ZERO_GPU_SIZE = "xlarge"
+
+# Keeping both 12B checkpoints resident can exceed the default 48GB ZeroGPU
+# slice. The pipeline unloads the other model before loading the requested one
+# unless this is set to True on a larger GPU.
+HF_KEEP_BOTH_MODELS = False
+HF_PRELOAD_MODELS = ()
+
+REQUEST_TIMEOUT_S = 600
+MAX_ATTEMPTS = 3
+RETRY_BACKOFF_S = 5

requirements.txt

@@ -0,0 +1,9 @@
+gradio==6.16.0
+spaces
+torch
+torchvision
+accelerate
+transformers
+sentencepiece
+protobuf
+huggingface_hub