- Rate-limit failed logins (8 per 15 min per IP or username, 429 + Retry-After) - Add ruff/mypy config and GitHub Actions workflow (ruff, pytest, tsc, tests, build) - Apply ruff autofixes (import order, pyupgrade) - Extract rpy_helpers.py, engine-util.ts, TranslationParts.tsx and translation-types.ts from oversized modules - Add dependency-free React UI tests - Refresh README and ARCHITECTURE (Naninovel/Tomie, status and plans) Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
471 lines
19 KiB
Python
471 lines
19 KiB
Python
"""Server-side work with a version's translations.
|
|
|
|
Everything here reads immutable blobs, so parsed JSON and the text index are
|
|
cached by blob hash and never go stale. Writes produce new blobs and return a
|
|
new manifest; the caller stores it on the version.
|
|
"""
|
|
|
|
import copy
|
|
import csv
|
|
import io
|
|
import json
|
|
import re
|
|
from dataclasses import dataclass, field
|
|
from functools import lru_cache
|
|
from pathlib import PurePosixPath
|
|
from typing import Any
|
|
|
|
from app.blobs import put_bytes, read_bytes
|
|
from app.converter.locales import LANG_NAMES, RENPY_LANGS, short_hash, walk
|
|
|
|
STATES = ("translated", "fuzzy", "machine", "approved")
|
|
|
|
|
|
@lru_cache(maxsize=512)
|
|
def blob_json(blob_id: str) -> Any:
|
|
"""Parsed JSON of a blob. Shared between callers: treat as read-only."""
|
|
return json.loads(read_bytes(blob_id))
|
|
|
|
|
|
def file_json(manifest: dict[str, Any], path: str) -> Any:
|
|
f = manifest["files"].get(path)
|
|
return blob_json(f["hash"]) if f else None
|
|
|
|
|
|
def put_json(files: dict[str, Any], path: str, value: Any) -> None:
|
|
payload = json.dumps(value, ensure_ascii=False, separators=(",", ":")).encode()
|
|
blob_id, size = put_bytes(payload, ".json")
|
|
files[path] = {"hash": blob_id, "size": size}
|
|
|
|
|
|
# --- text index ------------------------------------------------------------------
|
|
|
|
@dataclass
|
|
class LineInfo:
|
|
tid: str
|
|
node_id: str
|
|
module: str # script path in the manifest
|
|
label: str
|
|
who: str
|
|
source: str
|
|
scene: str | None # logical path of the background shown at this line
|
|
prev: str | None = None
|
|
next: str | None = None
|
|
resource: dict[str, Any] | None = None
|
|
|
|
|
|
@dataclass
|
|
class TextIndex:
|
|
lines: list[LineInfo] = field(default_factory=list)
|
|
by_tid: dict[str, LineInfo] = field(default_factory=dict)
|
|
strings: dict[str, str] = field(default_factory=dict) # source text -> usage
|
|
modules: list[tuple[str, int]] = field(default_factory=list)
|
|
|
|
|
|
def text_index(manifest: dict[str, Any]) -> TextIndex:
|
|
game_hash = manifest["files"]["game.json"]["hash"]
|
|
game = blob_json(game_hash)
|
|
scripts = tuple((p, manifest["files"][p]["hash"]) for p in game["scripts"])
|
|
return _build_index(game_hash, scripts)
|
|
|
|
|
|
@lru_cache(maxsize=16)
|
|
def _build_index(game_hash: str, scripts: tuple[tuple[str, str], ...]) -> TextIndex:
|
|
game = blob_json(game_hash)
|
|
chars = game.get("characters", {})
|
|
images = game.get("images", {})
|
|
idx = TextIndex()
|
|
for c in chars.values():
|
|
if isinstance(c.get("name"), str) and c["name"].strip():
|
|
idx.strings.setdefault(c["name"], "name")
|
|
for path, blob in scripts:
|
|
module = blob_json(blob)
|
|
count = 0
|
|
for label in module["labels"]:
|
|
scene: str | None = None
|
|
group = label["name"]
|
|
for node in walk(label["body"]):
|
|
op = node.get("op")
|
|
if op == "label":
|
|
group = node["name"]
|
|
elif op == "scene":
|
|
scene = _image_file(images, node.get("img")) or (None if node.get("img") else scene)
|
|
elif op == "menu":
|
|
for item in node["items"]:
|
|
idx.strings.setdefault(item["text"], "menu")
|
|
elif op == "show" and isinstance((node.get("img") or {}).get("text"), str):
|
|
idx.strings.setdefault(node["img"]["text"], "text")
|
|
elif op in {"say", "nani_add_choice", "nani_command"} and node.get("tid") and "what" in node:
|
|
who = node.get("whoStr") or ""
|
|
if not who and node.get("who"):
|
|
c = chars.get(node["who"], {})
|
|
who = c.get("name") if isinstance(c.get("name"), str) else node["who"]
|
|
info = LineInfo(node["tid"], node.get("id", ""), path, group, who or "",
|
|
node["what"], scene)
|
|
info.resource = node.get("resource")
|
|
if idx.lines and idx.lines[-1].module == path:
|
|
idx.lines[-1].next = info.source
|
|
info.prev = idx.lines[-1].source
|
|
idx.lines.append(info)
|
|
idx.by_tid[info.tid] = info
|
|
count += 1
|
|
idx.modules.append((path, count))
|
|
return idx
|
|
|
|
|
|
def _image_file(images: dict[str, Any], img: dict[str, Any] | None) -> str | None:
|
|
if not img or not img.get("name"):
|
|
return None
|
|
name = img["name"]
|
|
for n in range(len(name), 0, -1):
|
|
d = images.get(" ".join(name[:n]))
|
|
if d:
|
|
return d.get("file") or (d.get("frames") or [{}])[0].get("img", {}).get("file")
|
|
return None
|
|
|
|
|
|
def version_languages(manifest: dict[str, Any] | None) -> dict[str, Any]:
|
|
"""Source language and translations of a version, for listings."""
|
|
if not manifest or "game.json" not in manifest.get("files", {}):
|
|
return {"source_lang": None, "languages": []}
|
|
game = file_json(manifest, "game.json")
|
|
return {
|
|
"source_lang": game.get("sourceLang"),
|
|
"languages": [{"code": code, "name": info.get("name", code),
|
|
"progress": info.get("progress", 0)}
|
|
for code, info in sorted((game.get("locales") or {}).items())],
|
|
}
|
|
|
|
|
|
# --- reading locales ------------------------------------------------------------------
|
|
|
|
def locale_tables(manifest: dict[str, Any], lang: str) -> tuple[dict, dict, dict]:
|
|
"""(lines by tid, strings by source text, meta) of a language; empty if absent."""
|
|
game = file_json(manifest, "game.json")
|
|
info = (game.get("locales") or {}).get(lang)
|
|
if not info:
|
|
return {}, {}, {}
|
|
lines: dict[str, Any] = {}
|
|
for path in info.get("modules", {}).values():
|
|
lines.update(file_json(manifest, path) or {})
|
|
strings = file_json(manifest, info["strings"]) or {} if info.get("strings") else {}
|
|
meta = file_json(manifest, info["meta"]) or {} if info.get("meta") else {}
|
|
return lines, strings, meta
|
|
|
|
|
|
def line_state(entry: dict[str, Any] | None, info: LineInfo) -> str:
|
|
if not entry or not entry.get("t"):
|
|
return "untranslated"
|
|
if entry.get("src") and entry["src"] != short_hash(info.source):
|
|
return "outdated"
|
|
return entry.get("state", "translated")
|
|
|
|
|
|
def stats(manifest: dict[str, Any], lang: str) -> dict[str, Any]:
|
|
idx = text_index(manifest)
|
|
lines, strings, _ = locale_tables(manifest, lang)
|
|
counts = {s: 0 for s in ("untranslated", "outdated", *STATES)}
|
|
for info in idx.lines:
|
|
counts[line_state(lines.get(info.tid), info)] += 1
|
|
total = len(idx.lines) or 1
|
|
done = len(idx.lines) - counts["untranslated"]
|
|
return {
|
|
"lines": len(idx.lines),
|
|
"translated": done,
|
|
"progress": round(done / total, 3),
|
|
"states": counts,
|
|
"strings": len(idx.strings),
|
|
"stringsTranslated": sum(1 for s in idx.strings if (strings.get(s) or {}).get("t")),
|
|
}
|
|
|
|
|
|
# --- writing -----------------------------------------------------------------------------
|
|
|
|
def _locale_path(lang: str, script_path: str) -> str:
|
|
return f"locales/{lang}/" + str(PurePosixPath(script_path).relative_to("script"))
|
|
|
|
|
|
def renpy_name(lang: str) -> str:
|
|
return next((name for name, code in RENPY_LANGS.items() if code == lang), lang)
|
|
|
|
|
|
def apply_changes(manifest: dict[str, Any], lang: str, *, lines: dict[str, Any] | None = None,
|
|
strings: dict[str, Any] | None = None, meta: dict[str, Any] | None = None,
|
|
name: str | None = None, can_approve: bool = False) -> dict[str, Any]:
|
|
"""Return a new manifest with the given entries changed (``None`` removes an entry)."""
|
|
manifest = {**manifest, "files": dict(manifest["files"])}
|
|
files = manifest["files"]
|
|
game = copy.deepcopy(file_json(manifest, "game.json"))
|
|
if lang == game.get("sourceLang"):
|
|
raise ValueError("Это исходный язык игры")
|
|
idx = text_index({"files": files})
|
|
locales = game.setdefault("locales", {})
|
|
info = locales.setdefault(lang, {"name": name or LANG_NAMES.get(lang, lang),
|
|
"renpy": renpy_name(lang), "modules": {}})
|
|
if name:
|
|
info["name"] = name
|
|
|
|
def entry(value: dict[str, Any], src: str | None) -> dict[str, Any] | None:
|
|
text = (value.get("t") or "").strip("\n")
|
|
if not text.strip():
|
|
return None
|
|
state = value.get("state") or "translated"
|
|
if state not in STATES or (state == "approved" and not can_approve):
|
|
state = "translated"
|
|
e = {"t": text, "state": state}
|
|
if src is not None:
|
|
e["src"] = short_hash(src)
|
|
return e
|
|
|
|
if lines:
|
|
by_module: dict[str, dict[str, Any]] = {}
|
|
for tid, value in lines.items():
|
|
line = idx.by_tid.get(tid)
|
|
if line is None:
|
|
continue
|
|
by_module.setdefault(line.module, {})[tid] = value
|
|
for module, changes in by_module.items():
|
|
path = info["modules"].get(module) or _locale_path(lang, module)
|
|
table = dict(file_json(manifest, path) or {}) if path in files else {}
|
|
for tid, value in changes.items():
|
|
e = entry(value, idx.by_tid[tid].source) if value else None
|
|
if e and game.get("engine") == "naninovel":
|
|
validate_nani_translation(idx.by_tid[tid], e["t"])
|
|
if e:
|
|
table[tid] = e
|
|
else:
|
|
table.pop(tid, None)
|
|
put_json(files, path, table)
|
|
info["modules"][module] = path
|
|
|
|
if strings:
|
|
path = info.get("strings") or f"locales/{lang}/strings.json"
|
|
table = dict(file_json(manifest, path) or {}) if path in files else {}
|
|
for src, value in strings.items():
|
|
e = entry(value, None) if value else None
|
|
if e:
|
|
table[src] = e
|
|
else:
|
|
table.pop(src, None)
|
|
put_json(files, path, table)
|
|
info["strings"] = path
|
|
|
|
if meta is not None:
|
|
path = f"locales/{lang}/meta.json"
|
|
current = dict(file_json(manifest, path) or {}) if path in files else {}
|
|
current.update(meta)
|
|
put_json(files, path, current)
|
|
info["meta"] = path
|
|
info["font"] = current.get("font")
|
|
|
|
info.setdefault("strings", None)
|
|
# stats read the locale index from game.json, so store it before counting
|
|
put_json(files, "game.json", game)
|
|
st = stats(manifest, lang)
|
|
info["lines"] = st["translated"]
|
|
info["progress"] = st["progress"]
|
|
info["strings_count"] = st["stringsTranslated"]
|
|
put_json(files, "game.json", game)
|
|
return manifest
|
|
|
|
|
|
def validate_nani_translation(info: LineInfo, text: str) -> None:
|
|
from collections import Counter
|
|
expressions = r"\{[^{}]+\}|\\[nrt]"
|
|
if Counter(re.findall(expressions, info.source)) != Counter(re.findall(expressions, text)):
|
|
raise ValueError(f"Изменены выражения или управляющие последовательности Naninovel: {info.tid}")
|
|
tags = r"</?[^<>]+>"
|
|
if re.findall(tags, info.source) != re.findall(tags, text):
|
|
raise ValueError(f"Изменены теги Naninovel: {info.tid}")
|
|
for delimiter in (info.resource or {}).get("delimiters", ""):
|
|
if text.count(delimiter) != info.source.count(delimiter):
|
|
raise ValueError(f"Изменён разделитель текстового ресурса Naninovel: {info.tid}")
|
|
|
|
|
|
def merge_languages(target: dict[str, Any], source: dict[str, Any], langs: list[str]
|
|
) -> dict[str, Any]:
|
|
"""Copy whole languages from one manifest into another (publishing a translation)."""
|
|
target = {**target, "files": dict(target["files"])}
|
|
files = target["files"]
|
|
game = copy.deepcopy(file_json(target, "game.json"))
|
|
src_game = file_json(source, "game.json")
|
|
for lang in langs:
|
|
info = (src_game.get("locales") or {}).get(lang)
|
|
if not info:
|
|
continue
|
|
prefix = f"locales/{lang}/"
|
|
for path in [p for p in files if p.startswith(prefix)]:
|
|
del files[path]
|
|
for path, f in source["files"].items():
|
|
if path.startswith(prefix):
|
|
files[path] = f
|
|
game.setdefault("locales", {})[lang] = copy.deepcopy(info)
|
|
put_json(files, "game.json", game)
|
|
for lang in langs:
|
|
if lang in game["locales"]:
|
|
st = stats(target, lang) # target scripts may differ from the fork's
|
|
game["locales"][lang].update(lines=st["translated"], progress=st["progress"])
|
|
put_json(files, "game.json", game)
|
|
return target
|
|
|
|
|
|
def changed_languages(fork: dict[str, Any], base: dict[str, Any] | None) -> list[str]:
|
|
"""Languages whose files differ between a fork and its base."""
|
|
fork_locales = (file_json(fork, "game.json").get("locales") or {})
|
|
if base is None:
|
|
return list(fork_locales)
|
|
out = []
|
|
for lang in fork_locales:
|
|
prefix = f"locales/{lang}/"
|
|
mine = {p: f["hash"] for p, f in fork["files"].items() if p.startswith(prefix)}
|
|
theirs = {p: f["hash"] for p, f in base["files"].items() if p.startswith(prefix)}
|
|
if mine != theirs:
|
|
out.append(lang)
|
|
return out
|
|
|
|
|
|
def merge_translation(base: dict, source: dict, target: dict) -> tuple[dict, list]:
|
|
"""Three-way publication of individual locale entries, preserving source hashes."""
|
|
from app.mods import merge_value
|
|
|
|
def tree(manifest):
|
|
game = file_json(manifest, "game.json")
|
|
infos = {lang: {k: v for k, v in info.items() if k not in ("progress", "lines", "strings_count")}
|
|
for lang, info in game.get("locales", {}).items()}
|
|
files = {p: ({"json": file_json(manifest, p)} if p.endswith(".json") else entry)
|
|
for p, entry in manifest["files"].items() if p.startswith("locales/")}
|
|
return {"locales": infos, "files": files}
|
|
|
|
conflicts = []
|
|
merged = merge_value(tree(base), tree(source), tree(target), "", conflicts, {})
|
|
if conflicts:
|
|
return target, conflicts
|
|
files = {p: f for p, f in target["files"].items() if not p.startswith("locales/")}
|
|
for path, entry in merged["files"].items():
|
|
if set(entry) == {"json"}:
|
|
put_json(files, path, entry["json"])
|
|
else:
|
|
files[path] = entry
|
|
game = copy.deepcopy(file_json(target, "game.json"))
|
|
game["locales"] = merged["locales"]
|
|
put_json(files, "game.json", game)
|
|
manifest = {**target, "files": files}
|
|
for lang, info in game["locales"].items():
|
|
st = stats(manifest, lang)
|
|
info.update(lines=st["translated"], progress=st["progress"], strings_count=st["stringsTranslated"])
|
|
put_json(files, "game.json", game)
|
|
return manifest, []
|
|
|
|
|
|
# --- PO / CSV --------------------------------------------------------------------------
|
|
|
|
def _po_quote(s: str) -> str:
|
|
s = s.replace("\\", "\\\\").replace('"', '\\"').replace("\t", "\\t")
|
|
if "\n" not in s:
|
|
return f'"{s}"'
|
|
parts = s.split("\n")
|
|
return '""\n' + "\n".join(f'"{p}\\n"' if i < len(parts) - 1 else f'"{p}"'
|
|
for i, p in enumerate(parts) if p or i < len(parts) - 1)
|
|
|
|
|
|
def export_po(manifest: dict[str, Any], lang: str, game_title: str) -> str:
|
|
idx = text_index(manifest)
|
|
lines, strings, _ = locale_tables(manifest, lang)
|
|
out = [f'# {game_title}', 'msgid ""', 'msgstr ""',
|
|
'"Content-Type: text/plain; charset=UTF-8\\n"', f'"Language: {lang}\\n"', ""]
|
|
for info in idx.lines:
|
|
e = lines.get(info.tid) or {}
|
|
state = line_state(e, info)
|
|
out.append(f"#. {info.who}" if info.who else "#.")
|
|
out.append(f"#: {info.module} {info.label}")
|
|
if state in ("outdated", "fuzzy", "machine"):
|
|
out.append("#, fuzzy")
|
|
out += [f"msgctxt {_po_quote(info.tid)}", f"msgid {_po_quote(info.source)}",
|
|
f"msgstr {_po_quote(e.get('t', ''))}", ""]
|
|
for src, usage in idx.strings.items():
|
|
e = strings.get(src) or {}
|
|
out += [f"#. {usage}", 'msgctxt "string"', f"msgid {_po_quote(src)}",
|
|
f"msgstr {_po_quote(e.get('t', ''))}", ""]
|
|
return "\n".join(out)
|
|
|
|
|
|
_PO_LINE = re.compile(r'^(msgctxt|msgid|msgstr)\s+(".*")\s*$|^(".*")\s*$')
|
|
|
|
|
|
def _po_unquote(s: str) -> str:
|
|
body = s[1:-1]
|
|
return re.sub(r"\\(.)", lambda m: {"n": "\n", "t": "\t"}.get(m.group(1), m.group(1)), body)
|
|
|
|
|
|
def parse_po(text: str) -> tuple[dict[str, Any], dict[str, Any]]:
|
|
"""Returns (lines by tid, strings by source) from a PO file made by export_po."""
|
|
lines: dict[str, Any] = {}
|
|
strings: dict[str, Any] = {}
|
|
cur: dict[str, str] = {}
|
|
fuzzy = False
|
|
last = None
|
|
|
|
def flush():
|
|
nonlocal cur, fuzzy
|
|
if cur.get("msgid") and "msgstr" in cur:
|
|
value = {"t": cur["msgstr"], "state": "fuzzy" if fuzzy else "translated"}
|
|
if cur.get("msgctxt") == "string":
|
|
strings[cur["msgid"]] = value
|
|
elif cur.get("msgctxt"):
|
|
lines[cur["msgctxt"]] = value
|
|
cur, fuzzy = {}, False
|
|
|
|
for raw in text.splitlines():
|
|
line = raw.strip()
|
|
if not line:
|
|
flush()
|
|
last = None
|
|
continue
|
|
if line.startswith("#,") and "fuzzy" in line:
|
|
fuzzy = True
|
|
continue
|
|
if line.startswith("#"):
|
|
continue
|
|
m = _PO_LINE.match(line)
|
|
if not m:
|
|
continue
|
|
if m.group(1):
|
|
if m.group(1) == "msgctxt" and "msgid" in cur:
|
|
flush()
|
|
last = m.group(1)
|
|
cur[last] = _po_unquote(m.group(2))
|
|
elif last:
|
|
cur[last] += _po_unquote(m.group(3))
|
|
flush()
|
|
return lines, strings
|
|
|
|
|
|
def export_csv(manifest: dict[str, Any], lang: str) -> str:
|
|
idx = text_index(manifest)
|
|
lines, strings, _ = locale_tables(manifest, lang)
|
|
buf = io.StringIO()
|
|
w = csv.writer(buf)
|
|
w.writerow(["kind", "key", "who", "source", "translation", "state"])
|
|
for info in idx.lines:
|
|
e = lines.get(info.tid) or {}
|
|
w.writerow(["line", info.tid, info.who, info.source, e.get("t", ""), line_state(e, info)])
|
|
for src, usage in idx.strings.items():
|
|
e = strings.get(src) or {}
|
|
w.writerow(["string", src, usage, src, e.get("t", ""), e.get("state", "")])
|
|
return buf.getvalue()
|
|
|
|
|
|
def parse_csv(text: str) -> tuple[dict[str, Any], dict[str, Any]]:
|
|
lines: dict[str, Any] = {}
|
|
strings: dict[str, Any] = {}
|
|
for row in csv.DictReader(io.StringIO(text.lstrip("\ufeff"))):
|
|
if not row.get("translation"):
|
|
continue
|
|
state = row.get("state") if row.get("state") in STATES else "translated"
|
|
value = {"t": row["translation"], "state": state}
|
|
if row.get("kind") == "string":
|
|
strings[row["key"]] = value
|
|
elif row.get("kind") == "line":
|
|
lines[row["key"]] = value
|
|
return lines, strings
|