Localization stage 1: stable IDs, import of tl/ translations, language picker

- every script node gets a permanent id, say nodes a Ren'Py-style tid
- all languages from the game's tl/ are imported at conversion: dialogue blocks
  matched by their source-location comment or by label + original text,
  `translate strings` become a per-language strings table
- source language from config.language or detected from the dialogue
- manifest stores locales/<lang>/<module>.json and strings.json; game.json
  lists sourceLang and available locales with progress
- player: language picker in main menu and settings, fallback to source,
  __(), [var!t] and _preferences.language
- parser: `default` with init priority, and one bad statement no longer drops
  the whole file
- docs/LOCALIZATION.md: design for translations, publishing and mods

Line endings of several converter/player files go back to LF: an earlier
commit accidentally stored them as CRLF.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
mrleo1nidandClaude Opus 5.5 committed 2026-10-25 03:39:40 +03:00
1 parent 5d9ddb6376
commit e1d7b29f77
16 files changed
+4280 -3530

No files matched your search

+155 -144
View File
@@ -1,144 +1,155 @@
"""Dry run of the script part of the converter on unpacked games (no media work).
python -m app.converter.analyze /games/Some-Game [/games/Other ...]
Prints, per game, how much of the script was understood and the top warnings.
"""
import json
import re
import shutil
import sys
import tempfile
import time
from collections import Counter
from pathlib import Path
from app.converter import assets
from app.converter.archive import ArchiveError, find_game_dir
from app.converter.pipeline import _auto_images, _is_engine_file, _resolve_image, anchors
from app.converter.rpa import RpaError, extract_rpa, read_index
from app.converter.rpy_lexer import ParseError
from app.converter.rpy_parser import RpyParser, ScriptData
from app.converter.rpyc import decompile_missing
SCRIPT_EXTS = (".rpy", ".rpyc", ".rpym", ".rpymc")
def renpy_version(root: Path) -> str:
for vc in root.rglob("renpy/vc_version.py"):
m = re.search(r"version\s*=\s*u?['\"]([^'\"]+)", vc.read_text(errors="replace"))
if m:
return m.group(1)
for init in root.rglob("renpy/__init__.py"):
m = re.search(r"version_tuple\s*=\s*\(([^)]*)\)", init.read_text(errors="replace"))
if m:
return ".".join(x.strip() for x in m.group(1).split(",")[:3])
return "?"
def analyze(src: Path, dump: Path | None = None) -> dict:
started = time.monotonic()
game_dir = find_game_dir(src)
data = ScriptData()
work = Path(tempfile.mkdtemp(prefix="analyze-"))
try:
files: dict[str, str] = {} # lowercase path -> path, for asset resolution
for p in game_dir.rglob("*"):
if p.is_file():
rel = p.relative_to(game_dir).as_posix()
if p.suffix.lower() in SCRIPT_EXTS:
dest = work / rel
dest.parent.mkdir(parents=True, exist_ok=True)
shutil.copyfile(p, dest)
elif p.suffix.lower() != ".rpa":
files[rel.lower()] = rel
rpa_errors = []
for rpa in game_dir.rglob("*.rpa"):
try:
for name in read_index(rpa):
if not name.lower().endswith(SCRIPT_EXTS):
files[name.lower()] = name
extract_rpa(rpa, work, only=lambda n: n.lower().endswith(SCRIPT_EXTS))
except RpaError as e:
rpa_errors.append(str(e))
decompiled = decompile_missing(work, data.warn)
if dump is not None:
shutil.copytree(work, dump / src.name, dirs_exist_ok=True,
ignore=shutil.ignore_patterns("*.rpyc", "*.rpymc"))
rpy = sorted((p for p in work.rglob("*.rpy") if not _is_engine_file(work, p)),
key=lambda p: p.relative_to(work).as_posix().lower())
lines = 0
for p in rpy:
rel = p.relative_to(work).as_posix()
text = p.read_text(encoding="utf-8", errors="replace")
lines += text.count("\n")
try:
RpyParser(data, rel).parse(text)
except ParseError as e:
data.warn(f"Ошибка разбора файла: {e.msg}", f"{e.filename}:{e.lineno}")
except RecursionError:
data.warn("Слишком глубокая вложенность", rel)
# media aliases as the real pipeline would produce them
aliases = {k: str(Path(v).with_suffix(assets.target_ext(Path(v))).as_posix())
for k, v in files.items() if assets.kind_of(Path(v))}
_auto_images(data, aliases)
missing: set[str] = set()
for img in data.images.values():
_resolve_image(img, aliases, missing)
labels = {lbl["name"] for mod in data.modules.values() for lbl in mod}
labels |= {name for mod in data.modules.values() for name in anchors(mod)}
targets = Counter()
_collect_targets(data.modules, targets)
media = Counter(assets.kind_of(Path(v)) for v in files.values() if assets.kind_of(Path(v)))
return {
"game": src.name,
"renpy": renpy_version(src),
"seconds": round(time.monotonic() - started, 1),
"rpyFiles": len(rpy),
"decompiled": decompiled,
"lines": lines,
"labels": len(labels),
"hasStart": "start" in labels,
"stats": dict(data.stats),
"media": dict(media),
"images": len(data.images),
"characters": len(data.characters),
"missingJumpTargets": sorted(t for t in targets if t not in labels)[:15],
"missingFiles": sorted(missing)[:15],
"rpaErrors": rpa_errors,
"warningsTotal": sum(data.warnings.values()),
"warnings": [{"n": n, "msg": m, "ex": data.warning_examples.get(m)}
for m, n in data.warnings.most_common(25)],
}
finally:
shutil.rmtree(work, ignore_errors=True)
def _collect_targets(modules, out: Counter) -> None:
def walk(nodes):
for n in nodes:
if n["op"] in ("jump", "call") and n.get("target"):
out[n["target"]] += 1
for b in n.get("branches", []):
walk(b["body"])
for it in n.get("items", []):
walk(it["body"])
walk(n.get("body", []))
for mod in modules.values():
for lbl in mod:
walk(lbl["body"])
if __name__ == "__main__":
args = sys.argv[1:]
dump = None
if args[:1] == ["--dump"]:
dump, args = Path(args[1]), args[2:]
for arg in args:
try:
result = analyze(Path(arg), dump)
except (ArchiveError, OSError) as e:
result = {"game": Path(arg).name, "error": str(e)}
print(json.dumps(result, ensure_ascii=False), flush=True)
"""Dry run of the script part of the converter on unpacked games (no media work).
python -m app.converter.analyze /games/Some-Game [/games/Other ...]
Prints, per game, how much of the script was understood and the top warnings.
"""
import json
import re
import shutil
import sys
import tempfile
import time
from collections import Counter
from pathlib import Path
from app.converter import assets, locales
from app.converter.archive import ArchiveError, find_game_dir
from app.converter.pipeline import _auto_images, _is_engine_file, _resolve_image, anchors
from app.converter.rpa import RpaError, extract_rpa, read_index
from app.converter.rpy_lexer import ParseError
from app.converter.rpy_parser import RpyParser, ScriptData
from app.converter.rpyc import decompile_missing
SCRIPT_EXTS = (".rpy", ".rpyc", ".rpym", ".rpymc")
def renpy_version(root: Path) -> str:
for vc in root.rglob("renpy/vc_version.py"):
m = re.search(r"version\s*=\s*u?['\"]([^'\"]+)", vc.read_text(errors="replace"))
if m:
return m.group(1)
for init in root.rglob("renpy/__init__.py"):
m = re.search(r"version_tuple\s*=\s*\(([^)]*)\)", init.read_text(errors="replace"))
if m:
return ".".join(x.strip() for x in m.group(1).split(",")[:3])
return "?"
def analyze(src: Path, dump: Path | None = None) -> dict:
started = time.monotonic()
game_dir = find_game_dir(src)
data = ScriptData()
work = Path(tempfile.mkdtemp(prefix="analyze-"))
try:
files: dict[str, str] = {} # lowercase path -> path, for asset resolution
for p in game_dir.rglob("*"):
if p.is_file():
rel = p.relative_to(game_dir).as_posix()
if p.suffix.lower() in SCRIPT_EXTS:
dest = work / rel
dest.parent.mkdir(parents=True, exist_ok=True)
shutil.copyfile(p, dest)
elif p.suffix.lower() != ".rpa":
files[rel.lower()] = rel
rpa_errors = []
for rpa in game_dir.rglob("*.rpa"):
try:
for name in read_index(rpa):
if not name.lower().endswith(SCRIPT_EXTS):
files[name.lower()] = name
extract_rpa(rpa, work, only=lambda n: n.lower().endswith(SCRIPT_EXTS))
except RpaError as e:
rpa_errors.append(str(e))
decompiled = decompile_missing(work, data.warn)
if dump is not None:
shutil.copytree(work, dump / src.name, dirs_exist_ok=True,
ignore=shutil.ignore_patterns("*.rpyc", "*.rpymc"))
rpy = sorted((p for p in work.rglob("*.rpy") if not _is_engine_file(work, p)),
key=lambda p: p.relative_to(work).as_posix().lower())
lines = 0
for p in rpy:
rel = p.relative_to(work).as_posix()
text = p.read_text(encoding="utf-8", errors="replace")
lines += text.count("\n")
try:
RpyParser(data, rel).parse(text)
except ParseError as e:
data.warn(f"Ошибка разбора файла: {e.msg}", f"{e.filename}:{e.lineno}")
except RecursionError:
data.warn("Слишком глубокая вложенность", rel)
by_line = locales.assign_ids(data.modules)
translations = locales.import_tl(work, data.modules, by_line, data.warn)
says = len(by_line) or 1
language = data.config.get("language")
source_lang = (locales.lang_code(language) if isinstance(language, str)
else locales.detect_language(data.modules))
# media aliases as the real pipeline would produce them
aliases = {k: str(Path(v).with_suffix(assets.target_ext(Path(v))).as_posix())
for k, v in files.items() if assets.kind_of(Path(v))}
_auto_images(data, aliases)
missing: set[str] = set()
for img in data.images.values():
_resolve_image(img, aliases, missing)
labels = {lbl["name"] for mod in data.modules.values() for lbl in mod}
labels |= {name for mod in data.modules.values() for name in anchors(mod)}
targets = Counter()
_collect_targets(data.modules, targets)
media = Counter(assets.kind_of(Path(v)) for v in files.values() if assets.kind_of(Path(v)))
return {
"game": src.name,
"renpy": renpy_version(src),
"seconds": round(time.monotonic() - started, 1),
"rpyFiles": len(rpy),
"decompiled": decompiled,
"lines": lines,
"labels": len(labels),
"hasStart": "start" in labels,
"stats": dict(data.stats),
"media": dict(media),
"images": len(data.images),
"sourceLang": source_lang,
"locales": {code: {"lines": len(t.lines), "strings": len(t.strings),
"progress": round(len(t.lines) / says, 3), "unmatched": t.unmatched}
for code, t in translations.items()},
"characters": len(data.characters),
"missingJumpTargets": sorted(t for t in targets if t not in labels)[:15],
"missingFiles": sorted(missing)[:15],
"rpaErrors": rpa_errors,
"warningsTotal": sum(data.warnings.values()),
"warnings": [{"n": n, "msg": m, "ex": data.warning_examples.get(m)}
for m, n in data.warnings.most_common(25)],
}
finally:
shutil.rmtree(work, ignore_errors=True)
def _collect_targets(modules, out: Counter) -> None:
def walk(nodes):
for n in nodes:
if n["op"] in ("jump", "call") and n.get("target"):
out[n["target"]] += 1
for b in n.get("branches", []):
walk(b["body"])
for it in n.get("items", []):
walk(it["body"])
walk(n.get("body", []))
for mod in modules.values():
for lbl in mod:
walk(lbl["body"])
if __name__ == "__main__":
args = sys.argv[1:]
dump = None
if args[:1] == ["--dump"]:
dump, args = Path(args[1]), args[2:]
for arg in args:
try:
result = analyze(Path(arg), dump)
except (ArchiveError, OSError) as e:
result = {"game": Path(arg).name, "error": str(e)}
print(json.dumps(result, ensure_ascii=False), flush=True)
+280
View File
@@ -0,0 +1,280 @@
"""Stable IDs for script nodes and strings, and import of Ren'Py translations (tl/).
* every statement node gets a permanent ``id``; say nodes also get a ``tid``
(translation id) — Ren'Py's own id when the game's tl/ files know it;
* ``tl/<lang>/*.rpy`` is turned into per-language tables: dialogue lines keyed by
tid and a strings table (names, menu choices, ``_()``) keyed by source text.
"""
import hashlib
import re
from collections import Counter, defaultdict
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any
from app.converter.rpy_lexer import Cursor, Line, ParseError, parse_tree
# Ren'Py language names -> BCP-47
RENPY_LANGS = {
"english": "en", "russian": "ru", "ukrainian": "uk", "belarusian": "be", "german": "de",
"french": "fr", "spanish": "es", "spanish_latam": "es-419", "latam": "es-419",
"portuguese": "pt", "brazilian": "pt-BR", "portuguese_br": "pt-BR", "italian": "it",
"polish": "pl", "czech": "cs", "slovak": "sk", "dutch": "nl", "norwegian": "no",
"swedish": "sv", "danish": "da", "finnish": "fi", "turkish": "tr", "greek": "el",
"hungarian": "hu", "romanian": "ro", "bulgarian": "bg", "serbian": "sr", "croatian": "hr",
"bosnian": "bs", "japanese": "ja", "korean": "ko", "chinese": "zh-Hans",
"schinese": "zh-Hans", "tchinese": "zh-Hant", "chinese_traditional": "zh-Hant",
"thai": "th", "vietnamese": "vi", "indonesian": "id", "malay": "ms", "filipino": "fil",
"tagalog": "fil", "arabic": "ar", "hebrew": "he", "persian": "fa", "hindi": "hi",
"hinglish": "hi-Latn",
}
LANG_NAMES = {
"en": "English", "ru": "Русский", "uk": "Українська", "de": "Deutsch", "fr": "Français",
"es": "Español", "es-419": "Español (Latinoamérica)", "pt": "Português",
"pt-BR": "Português (Brasil)", "it": "Italiano", "pl": "Polski", "cs": "Čeština",
"nl": "Nederlands", "no": "Norsk", "tr": "Türkçe", "ja": "日本語", "ko": "한국어",
"zh-Hans": "简体中文", "zh-Hant": "繁體中文", "th": "ไทย", "vi": "Tiếng Việt",
"id": "Bahasa Indonesia", "bs": "Bosanski", "hi-Latn": "Hinglish",
}
_STOPWORDS = {
"en": {"the", "you", "and", "is", "to", "it", "what", "i'm", "that", "of"},
"fr": {"le", "la", "les", "et", "est", "je", "tu", "vous", "pas", "une", "des", "c'est"},
"de": {"der", "die", "und", "ist", "ich", "du", "nicht", "das", "ein", "zu"},
"es": {"el", "la", "que", "y", "es", "no", "un", "los", "por", "qué"},
"pt": {"o", "a", "que", "não", "é", "um", "uma", "você", "eu", "de"},
"it": {"il", "che", "non", "è", "un", "di", "sono", "io", "per", "la"},
}
_TAGS = re.compile(r"\{[^{}]*\}|\[[^\[\]]*\]")
_WORDS = re.compile(r"[^\W\d_]+(?:'[^\W\d_]+)?", re.UNICODE)
_SRC_COMMENT = re.compile(r"#\s*(?:game/)?(.+?\.rpym?):(\d+)\s*$")
_TRANSLATE = re.compile(r"^translate\s+(\w+)\s+([\w.]+)\s*:\s*$")
def lang_code(renpy_name: str) -> str:
return RENPY_LANGS.get(renpy_name.lower(), renpy_name.lower())
def short_hash(text: str) -> str:
return hashlib.md5(text.encode("utf-8")).hexdigest()[:8]
@dataclass
class LocaleData:
renpy: str
lines: dict[str, str] = field(default_factory=dict) # tid -> text
strings: dict[str, str] = field(default_factory=dict) # source text -> text
unmatched: int = 0
def walk(nodes: list[dict[str, Any]]):
"""All statement nodes in a body, depth first, including menu captions."""
for n in nodes:
yield n
if n.get("op") == "menu" and n.get("say"):
yield n["say"]
for b in n.get("branches", []):
yield from walk(b["body"])
for it in n.get("items", []):
yield from walk(it["body"])
if n.get("op") == "while":
yield from walk(n["body"])
def assign_ids(modules: dict[str, list[dict[str, Any]]]) -> dict[tuple[str, int], dict]:
"""Give every node an ``id`` and say nodes a ``tid``.
Returns an index (file, line) -> say node used to attach translations.
"""
by_line: dict[tuple[str, int], dict] = {}
used: Counter = Counter()
for file, labels in modules.items():
n = 0
for label in labels:
group = label["name"]
for node in walk(label["body"]):
n += 1
node["id"] = short_hash(f"{file}:{n}")
if node.get("op") == "label":
group = node["name"]
if node.get("op") != "say":
continue
# Ren'Py style: <label>_<hash of the line>, de-duplicated
base = f"{group}_{short_hash((node.get('who') or '') + chr(0) + node['what'])}"
used[base] += 1
node["tid"] = base if used[base] == 1 else f"{base}_{used[base] - 1}"
node["_group"] = group
if "_ln" in node:
by_line[(file, node["_ln"])] = node
return by_line
def strip_temp(modules: dict[str, list[dict[str, Any]]]) -> None:
for labels in modules.values():
for label in labels:
for node in walk(label["body"]):
node.pop("_ln", None)
node.pop("_group", None)
def import_tl(game_dir: Path, modules: dict[str, list[dict[str, Any]]],
by_line: dict[tuple[str, int], dict], warn) -> dict[str, LocaleData]:
"""Read tl/<lang>/ of a game into per-language tables (lines are re-keyed to our tids)."""
tl_root = next((p for p in game_dir.iterdir() if p.is_dir() and p.name.lower() == "tl"), None)
if tl_root is None:
return {}
# fallback matching: Ren'Py id prefix is the label; compare the line text
by_group: dict[str, list[dict]] = defaultdict(list)
for labels in modules.values():
for label in labels:
for node in walk(label["body"]):
if node.get("op") == "say":
by_group[node.get("_group", label["name"])].append(node)
result: dict[str, LocaleData] = {}
for lang_dir in sorted(p for p in tl_root.iterdir() if p.is_dir()):
if lang_dir.name == "None":
continue # strings for the source language itself
loc = LocaleData(renpy=lang_dir.name)
for f in sorted(lang_dir.rglob("*.rpy")):
try:
_read_tl_file(f.read_text(encoding="utf-8", errors="replace"), loc, by_line,
by_group)
except ParseError as e:
warn(f"Перевод {lang_dir.name}: ошибка разбора", f"{f.name}:{e.lineno}")
if loc.lines or loc.strings:
result[lang_code(lang_dir.name)] = loc
if loc.unmatched:
warn(f"Перевод {lang_dir.name}: реплики без пары в скрипте ({loc.unmatched})")
return result
def _read_tl_file(text: str, loc: LocaleData, by_line, by_group) -> None:
tree = parse_tree(text)
raw_lines = text.replace("\r\n", "\n").split("\n")
for line in tree:
m = _TRANSLATE.match(line.text)
if not m:
continue
ident = m.group(2)
if ident == "strings":
_read_strings(line.block, loc)
continue
if ident == "python" or ident == "style":
continue
texts = [s for s in (_say_text(child) for child in line.block) if s is not None]
if not texts:
continue
translated = "\n".join(texts) if len(texts) > 1 else texts[0]
target = None
src = _source_location(raw_lines, line.lineno)
if src is not None:
target = by_line.get(src)
if target is None:
original = _source_comment(line.block, raw_lines)
group = ident.rsplit("_", 1)[0]
for node in by_group.get(group, []):
if original is not None and node["what"] == original:
target = node
break
if target is None:
loc.unmatched += 1
continue
loc.lines[target["tid"]] = translated
def _source_location(raw_lines: list[str], lineno: int) -> tuple[str, int] | None:
"""The `# game/file.rpy:N` comment Ren'Py writes above each translate block."""
for i in range(lineno - 2, max(lineno - 5, -1), -1):
m = _SRC_COMMENT.match(raw_lines[i].strip())
if m:
return m.group(1), int(m.group(2))
return None
def _source_comment(block: list[Line], raw_lines: list[str]) -> str | None:
# comments are stripped from logical lines; look at the raw text before the first line
if not block:
return None
for raw in raw_lines[max(0, block[0].lineno - 4):block[0].lineno - 1]:
s = raw.strip()
if s.startswith("#"):
c = Cursor(s[1:].strip())
c.dotted() if not (c.peek('"') or c.peek("'")) else None
while c.image_part() is not None and not (c.peek('"') or c.peek("'")):
pass
text = c.string()
if text is not None:
return text
return None
def _say_text(line: Line) -> str | None:
c = Cursor(line.text)
first = c.string()
if first is not None:
second = c.string() if not c.eol() else None
return second if second is not None else first
if c.dotted() is None:
return None
while True:
save = c.pos
if c.string_raw() is not None:
c.pos = save
break
if c.image_part() is None and not c.char("@"):
c.pos = save
break
return c.string()
def _read_strings(block: list[Line], loc: LocaleData) -> None:
old = None
for line in block:
c = Cursor(line.text)
kw = c.word()
if kw == "old":
old = c.string()
elif kw == "new" and old is not None:
new = c.string()
if new is not None and new != old:
loc.strings[old] = new
old = None
def detect_language(modules: dict[str, list[dict[str, Any]]]) -> str | None:
"""Guess the source language of the dialogue from a sample of lines."""
scores: Counter = Counter()
letters = Counter()
seen = 0
for labels in modules.values():
for label in labels:
for node in walk(label["body"]):
if node.get("op") != "say" or seen > 4000:
continue
seen += 1
text = _TAGS.sub(" ", node["what"]).lower()
for ch in text:
if "Ѐ" <= ch <= "ӿ":
letters["cyr"] += 1
elif "぀" <= ch <= "ヿ":
letters["ja"] += 1
elif "一" <= ch <= "鿿":
letters["zh"] += 1
elif ch.isalpha():
letters["lat"] += 1
for w in _WORDS.findall(text):
for lang, words in _STOPWORDS.items():
if w in words:
scores[lang] += 1
if not seen:
return None
total = sum(letters.values()) or 1
if letters["cyr"] / total > 0.4:
return "ru"
if letters["ja"] / total > 0.2:
return "ja"
if letters["zh"] / total > 0.3:
return "zh-Hans"
return scores.most_common(1)[0][0] if scores else None
+349 -289
View File
@@ -1,289 +1,349 @@
"""Full conversion: uploaded archive → blobs + version manifest."""
import json
import logging
import os
import shutil
import time
from collections.abc import Callable
from concurrent.futures import ThreadPoolExecutor, as_completed
from dataclasses import dataclass
from pathlib import Path, PurePosixPath
from typing import Any
from app.blobs import put_bytes, put_file
from app.config import get_settings
from app.converter import assets
from app.converter.archive import extract, find_game_dir
from app.converter.rpa import RpaError, extract_rpa
from app.converter.rpy_lexer import ParseError
from app.converter.rpy_parser import RpyParser, ScriptData
from app.converter.rpyc import decompile_missing
log = logging.getLogger(__name__)
FORMAT = "pgvn/1"
Progress = Callable[[str], None]
@dataclass
class Result:
manifest: dict[str, Any]
report: dict[str, Any]
title: str | None
cover_source: str | None # blob id of an image suitable as a cover
size: int
def convert_archive(archive: Path, filename: str, work: Path, progress: Progress) -> Result:
"""Convert an uploaded archive, or an already unpacked game directory (left untouched)."""
started = time.monotonic()
if work.exists():
shutil.rmtree(work)
src_root = work / "src"
out_dir = work / "out"
out_dir.mkdir(parents=True)
data = ScriptData()
if archive.is_dir():
progress("Копирование файлов игры")
game_dir = src_root / "game"
shutil.copytree(find_game_dir(archive), game_dir)
else:
progress("Распаковка архива")
extract(archive, filename, src_root)
game_dir = find_game_dir(src_root)
rpas = sorted(game_dir.rglob("*.rpa"))
for i, rpa in enumerate(rpas, 1):
progress(f"Распаковка {rpa.name} ({i}/{len(rpas)})")
try:
extract_rpa(rpa, game_dir)
except RpaError as e:
data.warn(str(e))
rpa.unlink()
progress("Декомпиляция .rpyc")
decompiled = decompile_missing(game_dir, data.warn)
progress("Разбор скриптов")
rpy_files = sorted((p for p in game_dir.rglob("*.rpy") if not _is_engine_file(game_dir, p)),
key=lambda p: p.relative_to(game_dir).as_posix().lower())
for p in rpy_files:
rel = p.relative_to(game_dir).as_posix()
try:
RpyParser(data, rel).parse(p.read_text(encoding="utf-8", errors="replace"))
except ParseError as e:
data.warn(f"Ошибка разбора файла: {e.msg}", f"{e.filename}:{e.lineno}")
except RecursionError:
data.warn("Слишком глубокая вложенность", rel)
if not any(lbl["name"] == "start" for labels in data.modules.values() for lbl in labels):
raise RuntimeError("В скриптах не найдена метка start — нечего запускать")
files: dict[str, dict[str, Any]] = {}
aliases: dict[str, str] = {}
media = [p for p in game_dir.rglob("*")
if p.is_file() and assets.kind_of(p) and not _is_engine_file(game_dir, p)]
counts = {"image": 0, "audio": 0, "video": 0, "font": 0}
failed: list[str] = []
done = 0
def work_one(p: Path) -> tuple[str, str, str, int]:
out, ext = assets.convert(p, out_dir)
blob_id, size = put_file(out, ext, move=True)
rel = p.relative_to(game_dir).as_posix()
logical = str(PurePosixPath(rel).with_suffix(ext))
return rel, logical, blob_id, size
with ThreadPoolExecutor(max_workers=get_settings().worker_threads) as pool:
futures = {pool.submit(work_one, p): p for p in media}
for fut in as_completed(futures):
p = futures[fut]
done += 1
if done % 20 == 0 or done == len(media):
progress(f"Конвертация медиа {done}/{len(media)}")
try:
rel, logical, blob_id, size = fut.result()
except Exception as e: # noqa: BLE001 — one broken file must not fail the game
failed.append(p.name)
data.warn("Не удалось сконвертировать файл", f"{p.name}: {e}")
continue
files[logical] = {"hash": blob_id, "size": size}
aliases[rel.lower()] = logical
counts[assets.kind_of(p)] += 1
progress("Сборка формата PGVN")
_auto_images(data, aliases)
missing: set[str] = set()
images = {name: _resolve_image(img, aliases, missing) for name, img in data.images.items()}
audio = {name: _resolve(path, aliases) or path for name, path in data.audio.items()}
scripts: list[str] = []
for rel, labels in data.modules.items():
logical = "script/" + str(PurePosixPath(rel).with_suffix(".json"))
payload = json.dumps({"file": rel, "labels": labels, "anchors": anchors(labels)},
ensure_ascii=False,
separators=(",", ":")).encode()
blob_id, size = put_bytes(payload, ".json")
files[logical] = {"hash": blob_id, "size": size}
scripts.append(logical)
gui = data.gui
resolution = gui.get("resolution") or [
data.config.get("screen_width") or 1280, data.config.get("screen_height") or 720]
title = data.config.get("name") if isinstance(data.config.get("name"), str) else None
game = {
"format": FORMAT,
"engine": "renpy",
"title": title,
"version": data.config.get("version"),
"resolution": resolution,
"entry": "start",
"scripts": scripts,
"characters": data.characters,
"images": images,
"transforms": data.transforms,
"transitions": data.transitions,
"audio": audio,
"channels": data.channels,
"files": aliases,
"defaults": data.defaults,
"init": data.init_code,
"ui": _ui(gui, aliases),
}
payload = json.dumps(game, ensure_ascii=False, separators=(",", ":")).encode()
blob_id, size = put_bytes(payload, ".json")
files["game.json"] = {"hash": blob_id, "size": size}
cover = None
for candidate in ("gui/main_menu.png", "gui/main_menu.jpg", "gui/main_menu.webp",
"gui/window_icon.png"):
if (logical := aliases.get(candidate)) and logical in files:
cover = files[logical]["hash"]
break
total = sum(f["size"] for f in files.values())
report = {
"format": FORMAT,
"durationSec": round(time.monotonic() - started, 1),
"scripts": len(rpy_files),
"decompiled": decompiled,
"labels": sum(len(v) for v in data.modules.values()),
"stats": dict(data.stats),
"media": counts,
"failedMedia": failed[:50],
"missingFiles": sorted(missing)[:100],
"warnings": [
{"message": msg, "count": n, "example": data.warning_examples.get(msg)}
for msg, n in data.warnings.most_common()
],
"sourceSize": _size(archive),
"outputSize": total,
}
return Result({"format": FORMAT, "files": files}, report, title, cover, total)
def _size(path: Path) -> int:
if path.is_file():
return path.stat().st_size
return sum(p.stat().st_size for p in path.rglob("*") if p.is_file())
def _is_engine_file(game_dir: Path, p: Path) -> bool:
parts = [x.lower() for x in p.relative_to(game_dir).parts]
return len(parts) > 1 and parts[0] in ("cache", "saves", "tl", "renpy")
def _resolve(path: str, aliases: dict[str, str]) -> str | None:
key = path.replace("\\", "/").lower().lstrip("/")
for prefix in ("", "images/", "audio/", "music/", "sound/", "sounds/"):
if (hit := aliases.get(prefix + key)) is not None:
return hit
return None
def _resolve_image(img: dict[str, Any], aliases: dict[str, str], missing: set[str]
) -> dict[str, Any]:
img = dict(img)
for key in ("file", "movie"):
if isinstance(img.get(key), str):
hit = _resolve(img[key], aliases)
if hit is None:
missing.add(img[key])
else:
img[key] = hit
if "switch" in img:
img["switch"] = [[c, _resolve_image(sub, aliases, missing)] for c, sub in img["switch"]]
if "composite" in img:
img["composite"] = [{**layer, "img": _resolve_image(layer["img"], aliases, missing)}
for layer in img["composite"]]
if "frames" in img:
img["frames"] = [{**f, "img": _resolve_image(f["img"], aliases, missing)}
for f in img["frames"]]
if "layered" in img:
layers = []
for layer in img["layered"]:
layer = dict(layer)
for key in ("always", "img"):
if key in layer:
layer[key] = _resolve_image(layer[key], aliases, missing)
if "cond" in layer:
layer["cond"] = [[c, _resolve_image(sub, aliases, missing)]
for c, sub in layer["cond"]]
layers.append(layer)
img["layered"] = layers
return img
def anchors(labels: list[dict[str, Any]]) -> dict[str, dict[str, Any]]:
"""Positions of `label` markers (nested labels, named menus) inside label bodies."""
found: dict[str, dict[str, Any]] = {}
def walk(nodes: list[dict[str, Any]], top: str, path: list[int]) -> None:
for i, n in enumerate(nodes):
here = path + [i]
if n["op"] == "label":
found[n["name"]] = {"label": top, "path": here, "params": n.get("params")}
if n["op"] == "if":
for b, branch in enumerate(n["branches"]):
walk(branch["body"], top, here + [b])
elif n["op"] == "menu":
for b, item in enumerate(n["items"]):
walk(item["body"], top, here + [b])
elif n["op"] == "while":
walk(n["body"], top, here + [0])
for label in labels:
walk(label["body"], label["name"], [])
return found
def _auto_images(data: ScriptData, aliases: dict[str, str]) -> None:
"""Ren'Py defines an image for every file under images/ (name = lowercase basename)."""
for rel, logical in aliases.items():
if not rel.startswith("images/") or assets.kind_of(Path(rel)) != "image":
continue
name = os.path.splitext(os.path.basename(rel))[0].lower().strip()
name = " ".join(name.split())
if name and name not in data.images:
data.images[name] = {"file": rel}
def _ui(gui: dict[str, Any], aliases: dict[str, str]) -> dict[str, Any]:
ui: dict[str, Any] = {}
for key, rel in (("textbox", "gui/textbox.png"), ("namebox", "gui/namebox.png"),
("mainMenu", "gui/main_menu.png"), ("nvl", "gui/nvl.png"),
("choice", "gui/button/choice_idle_background.png"),
("choiceHover", "gui/button/choice_hover_background.png")):
if hit := aliases.get(rel):
ui[key] = hit
for key in ("text_font", "name_text_font"):
if isinstance(gui.get(key), str) and (hit := _resolve(gui[key], aliases)):
ui[key] = hit
for key in ("text_size", "name_text_size", "text_color", "accent_color", "textbox_height",
"textbox_yalign", "dialogue_xpos", "dialogue_ypos", "dialogue_width",
"name_xpos", "name_ypos", "interface_text_color", "idle_color", "hover_color",
"choice_button_text_idle_color", "choice_button_text_hover_color"):
if isinstance(gui.get(key), (int, float, str)):
ui[key] = gui[key]
return ui
"""Full conversion: uploaded archive → blobs + version manifest."""
import json
import logging
import os
import shutil
import time
from collections.abc import Callable
from concurrent.futures import ThreadPoolExecutor, as_completed
from dataclasses import dataclass
from pathlib import Path, PurePosixPath
from typing import Any
from app.blobs import put_bytes, put_file
from app.config import get_settings
from app.converter import assets, locales
from app.converter.archive import extract, find_game_dir
from app.converter.rpa import RpaError, extract_rpa
from app.converter.rpy_lexer import ParseError
from app.converter.rpy_parser import RpyParser, ScriptData
from app.converter.rpyc import decompile_missing
log = logging.getLogger(__name__)
FORMAT = "pgvn/1"
Progress = Callable[[str], None]
@dataclass
class Result:
manifest: dict[str, Any]
report: dict[str, Any]
title: str | None
cover_source: str | None # blob id of an image suitable as a cover
size: int
def convert_archive(archive: Path, filename: str, work: Path, progress: Progress) -> Result:
"""Convert an uploaded archive, or an already unpacked game directory (left untouched)."""
started = time.monotonic()
if work.exists():
shutil.rmtree(work)
src_root = work / "src"
out_dir = work / "out"
out_dir.mkdir(parents=True)
data = ScriptData()
if archive.is_dir():
progress("Копирование файлов игры")
game_dir = src_root / "game"
shutil.copytree(find_game_dir(archive), game_dir)
else:
progress("Распаковка архива")
extract(archive, filename, src_root)
game_dir = find_game_dir(src_root)
rpas = sorted(game_dir.rglob("*.rpa"))
for i, rpa in enumerate(rpas, 1):
progress(f"Распаковка {rpa.name} ({i}/{len(rpas)})")
try:
extract_rpa(rpa, game_dir)
except RpaError as e:
data.warn(str(e))
rpa.unlink()
progress("Декомпиляция .rpyc")
decompiled = decompile_missing(game_dir, data.warn)
progress("Разбор скриптов")
rpy_files = sorted((p for p in game_dir.rglob("*.rpy") if not _is_engine_file(game_dir, p)),
key=lambda p: p.relative_to(game_dir).as_posix().lower())
for p in rpy_files:
rel = p.relative_to(game_dir).as_posix()
try:
RpyParser(data, rel).parse(p.read_text(encoding="utf-8", errors="replace"))
except ParseError as e:
data.warn(f"Ошибка разбора файла: {e.msg}", f"{e.filename}:{e.lineno}")
except RecursionError:
data.warn("Слишком глубокая вложенность", rel)
if not any(lbl["name"] == "start" for labels in data.modules.values() for lbl in labels):
raise RuntimeError("В скриптах не найдена метка start — нечего запускать")
files: dict[str, dict[str, Any]] = {}
aliases: dict[str, str] = {}
media = [p for p in game_dir.rglob("*")
if p.is_file() and assets.kind_of(p) and not _is_engine_file(game_dir, p)]
counts = {"image": 0, "audio": 0, "video": 0, "font": 0}
failed: list[str] = []
done = 0
def work_one(p: Path) -> tuple[str, str, str, int]:
out, ext = assets.convert(p, out_dir)
blob_id, size = put_file(out, ext, move=True)
rel = p.relative_to(game_dir).as_posix()
logical = str(PurePosixPath(rel).with_suffix(ext))
return rel, logical, blob_id, size
with ThreadPoolExecutor(max_workers=get_settings().worker_threads) as pool:
futures = {pool.submit(work_one, p): p for p in media}
for fut in as_completed(futures):
p = futures[fut]
done += 1
if done % 20 == 0 or done == len(media):
progress(f"Конвертация медиа {done}/{len(media)}")
try:
rel, logical, blob_id, size = fut.result()
except Exception as e: # noqa: BLE001 — one broken file must not fail the game
failed.append(p.name)
data.warn("Не удалось сконвертировать файл", f"{p.name}: {e}")
continue
files[logical] = {"hash": blob_id, "size": size}
aliases[rel.lower()] = logical
counts[assets.kind_of(p)] += 1
progress("Сборка формата PGVN")
_auto_images(data, aliases)
missing: set[str] = set()
images = {name: _resolve_image(img, aliases, missing) for name, img in data.images.items()}
audio = {name: _resolve(path, aliases) or path for name, path in data.audio.items()}
progress("Извлечение текстов и переводов")
by_line = locales.assign_ids(data.modules)
translations = locales.import_tl(game_dir, data.modules, by_line, data.warn)
locales.strip_temp(data.modules)
language = data.config.get("language")
source_lang = (locales.lang_code(language) if isinstance(language, str)
else locales.detect_language(data.modules)) or "und"
scripts: list[str] = []
for rel, labels in data.modules.items():
logical = "script/" + str(PurePosixPath(rel).with_suffix(".json"))
payload = json.dumps({"file": rel, "labels": labels, "anchors": anchors(labels)},
ensure_ascii=False,
separators=(",", ":")).encode()
blob_id, size = put_bytes(payload, ".json")
files[logical] = {"hash": blob_id, "size": size}
scripts.append(logical)
locale_index = _write_locales(translations, data.modules, files)
gui = data.gui
resolution = gui.get("resolution") or [
data.config.get("screen_width") or 1280, data.config.get("screen_height") or 720]
title = data.config.get("name") if isinstance(data.config.get("name"), str) else None
game = {
"format": FORMAT,
"engine": "renpy",
"title": title,
"version": data.config.get("version"),
"resolution": resolution,
"entry": "start",
"scripts": scripts,
"characters": data.characters,
"images": images,
"transforms": data.transforms,
"transitions": data.transitions,
"audio": audio,
"channels": data.channels,
"files": aliases,
"defaults": data.defaults,
"init": data.init_code,
"ui": _ui(gui, aliases),
"sourceLang": source_lang,
"locales": locale_index,
}
payload = json.dumps(game, ensure_ascii=False, separators=(",", ":")).encode()
blob_id, size = put_bytes(payload, ".json")
files["game.json"] = {"hash": blob_id, "size": size}
cover = None
for candidate in ("gui/main_menu.png", "gui/main_menu.jpg", "gui/main_menu.webp",
"gui/window_icon.png"):
if (logical := aliases.get(candidate)) and logical in files:
cover = files[logical]["hash"]
break
total = sum(f["size"] for f in files.values())
report = {
"format": FORMAT,
"durationSec": round(time.monotonic() - started, 1),
"scripts": len(rpy_files),
"decompiled": decompiled,
"labels": sum(len(v) for v in data.modules.values()),
"stats": dict(data.stats),
"media": counts,
"failedMedia": failed[:50],
"missingFiles": sorted(missing)[:100],
"sourceLang": source_lang,
"locales": {code: {"lines": meta["lines"], "strings": meta["strings_count"],
"progress": meta["progress"]}
for code, meta in locale_index.items()},
"warnings": [
{"message": msg, "count": n, "example": data.warning_examples.get(msg)}
for msg, n in data.warnings.most_common()
],
"sourceSize": _size(archive),
"outputSize": total,
}
return Result({"format": FORMAT, "files": files}, report, title, cover, total)
def _write_locales(translations: dict[str, "locales.LocaleData"],
modules: dict[str, list[dict[str, Any]]],
files: dict[str, dict[str, Any]]) -> dict[str, dict[str, Any]]:
"""Store each language as per-module line tables plus one strings table."""
tid_module: dict[str, str] = {}
tid_source: dict[str, str] = {}
for rel, labels in modules.items():
for label in labels:
for node in locales.walk(label["body"]):
if node.get("op") == "say":
tid_module[node["tid"]] = rel
tid_source[node["tid"]] = node["what"]
index: dict[str, dict[str, Any]] = {}
def put_json(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}
for code, loc in sorted(translations.items()):
grouped: dict[str, dict[str, Any]] = {}
for tid, text in loc.lines.items():
grouped.setdefault(tid_module[tid], {})[tid] = {
"t": text, "src": locales.short_hash(tid_source[tid]), "state": "translated"}
module_files = {}
for rel, entries in grouped.items():
path = f"locales/{code}/" + str(PurePosixPath(rel).with_suffix(".json"))
put_json(path, entries)
module_files["script/" + str(PurePosixPath(rel).with_suffix(".json"))] = path
strings_path = f"locales/{code}/strings.json"
put_json(strings_path, {old: {"t": new, "state": "translated"}
for old, new in loc.strings.items()})
index[code] = {
"name": locales.LANG_NAMES.get(code, loc.renpy),
"renpy": loc.renpy,
"modules": module_files,
"strings": strings_path,
"lines": len(loc.lines),
"strings_count": len(loc.strings),
"progress": round(len(loc.lines) / len(tid_module), 3) if tid_module else 0,
}
return index
def _size(path: Path) -> int:
if path.is_file():
return path.stat().st_size
return sum(p.stat().st_size for p in path.rglob("*") if p.is_file())
def _is_engine_file(game_dir: Path, p: Path) -> bool:
parts = [x.lower() for x in p.relative_to(game_dir).parts]
return len(parts) > 1 and parts[0] in ("cache", "saves", "tl", "renpy")
def _resolve(path: str, aliases: dict[str, str]) -> str | None:
key = path.replace("\\", "/").lower().lstrip("/")
for prefix in ("", "images/", "audio/", "music/", "sound/", "sounds/"):
if (hit := aliases.get(prefix + key)) is not None:
return hit
return None
def _resolve_image(img: dict[str, Any], aliases: dict[str, str], missing: set[str]
) -> dict[str, Any]:
img = dict(img)
for key in ("file", "movie"):
if isinstance(img.get(key), str):
hit = _resolve(img[key], aliases)
if hit is None:
missing.add(img[key])
else:
img[key] = hit
if "switch" in img:
img["switch"] = [[c, _resolve_image(sub, aliases, missing)] for c, sub in img["switch"]]
if "composite" in img:
img["composite"] = [{**layer, "img": _resolve_image(layer["img"], aliases, missing)}
for layer in img["composite"]]
if "frames" in img:
img["frames"] = [{**f, "img": _resolve_image(f["img"], aliases, missing)}
for f in img["frames"]]
if "layered" in img:
layers = []
for layer in img["layered"]:
layer = dict(layer)
for key in ("always", "img"):
if key in layer:
layer[key] = _resolve_image(layer[key], aliases, missing)
if "cond" in layer:
layer["cond"] = [[c, _resolve_image(sub, aliases, missing)]
for c, sub in layer["cond"]]
layers.append(layer)
img["layered"] = layers
return img
def anchors(labels: list[dict[str, Any]]) -> dict[str, dict[str, Any]]:
"""Positions of `label` markers (nested labels, named menus) inside label bodies."""
found: dict[str, dict[str, Any]] = {}
def walk(nodes: list[dict[str, Any]], top: str, path: list[int]) -> None:
for i, n in enumerate(nodes):
here = path + [i]
if n["op"] == "label":
found[n["name"]] = {"label": top, "path": here, "params": n.get("params")}
if n["op"] == "if":
for b, branch in enumerate(n["branches"]):
walk(branch["body"], top, here + [b])
elif n["op"] == "menu":
for b, item in enumerate(n["items"]):
walk(item["body"], top, here + [b])
elif n["op"] == "while":
walk(n["body"], top, here + [0])
for label in labels:
walk(label["body"], label["name"], [])
return found
def _auto_images(data: ScriptData, aliases: dict[str, str]) -> None:
"""Ren'Py defines an image for every file under images/ (name = lowercase basename)."""
for rel, logical in aliases.items():
if not rel.startswith("images/") or assets.kind_of(Path(rel)) != "image":
continue
name = os.path.splitext(os.path.basename(rel))[0].lower().strip()
name = " ".join(name.split())
if name and name not in data.images:
data.images[name] = {"file": rel}
def _ui(gui: dict[str, Any], aliases: dict[str, str]) -> dict[str, Any]:
ui: dict[str, Any] = {}
for key, rel in (("textbox", "gui/textbox.png"), ("namebox", "gui/namebox.png"),
("mainMenu", "gui/main_menu.png"), ("nvl", "gui/nvl.png"),
("choice", "gui/button/choice_idle_background.png"),
("choiceHover", "gui/button/choice_hover_background.png")):
if hit := aliases.get(rel):
ui[key] = hit
for key in ("text_font", "name_text_font"):
if isinstance(gui.get(key), str) and (hit := _resolve(gui[key], aliases)):
ui[key] = hit
for key in ("text_size", "name_text_size", "text_color", "accent_color", "textbox_height",
"textbox_yalign", "dialogue_xpos", "dialogue_ypos", "dialogue_width",
"name_xpos", "name_ypos", "interface_text_color", "idle_color", "hover_color",
"choice_button_text_idle_color", "choice_button_text_hover_color"):
if isinstance(gui.get(key), (int, float, str)):
ui[key] = gui[key]
return ui
File diff suppressed because it is too large. Load diff