"""Vernacular - translation review tool for the Riverstone language pack. Reads the review records produced by pipeline/translate_pack.py (translations//**), 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 docs import readLocalDoc from sheets import readLocalSheet from 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" EXAMPLE_UPLOAD_DIR = config.ROOT / "examples" / "uploads" EXAMPLE_UPLOAD_FILES = [ EXAMPLE_UPLOAD_DIR / "Unknown.xlsx", EXAMPLE_UPLOAD_DIR / "Sarah.xlsx", EXAMPLE_UPLOAD_DIR / "About Riverstone.docx", EXAMPLE_UPLOAD_DIR / "Matt.xlsx", ] EXAMPLE_UPLOAD_CHOICES = [ (path.name, str(path)) for path in EXAMPLE_UPLOAD_FILES if path.exists() ] # --- 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)), ) def load_selected_examples(selected_examples): selected_examples = selected_examples or [] return gr.update(value=selected_examples) # --- 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"{meta['summary']}") 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 "  " who = display_name(other["character"]) if other["character"] else other["kind"] neighbors.append(f"{marker} **{who}:** {other['source']}") lines.append("**Surrounding lines:**
" + "
".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"], ) with gr.Row(): example_files = gr.CheckboxGroup( choices=EXAMPLE_UPLOAD_CHOICES, label="Example source files", interactive=True, ) load_examples_btn = gr.Button("Use selected examples", scale=1) 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], ) load_examples_btn.click( load_selected_examples, [example_files], [wiki_source_files], ) if __name__ == "__main__": demo.launch()