Converter: cover real Ren'Py games better, add folder import and dry run

Checked against six real games (Ren'Py 7.5 and 8.x):
- nested labels and named menus become jump anchors inside label bodies
- call with arguments and labels with parameters (dynamically scoped)
- frame-by-frame ATL images, Movie frames, layeredimage, camera, Shake,
  custom audio channels via register_channel, show text with expressions
- generator expressions in Python, no SyntaxWarning noise from decompiled code

Also:
- app.import_dir converts an unpacked game folder without changing it
- app.converter.analyze dry-runs the script part of the converter
- imports use their own "importing" status so a worker restart can't hijack them

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
mrleo1nidandClaude Opus 5.5 committed 2026-10-25 03:23:20 +03:00
1 parent 35033fb86c
commit a97b46079e
18 files changed
+3456 -2695

No files matched your search

+144
View File
@@ -0,0 +1,144 @@
"""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)
+57 -5
View File
@@ -36,6 +36,7 @@ class Result:
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)
@@ -44,9 +45,14 @@ def convert_archive(archive: Path, filename: str, work: Path, progress: Progress
out_dir.mkdir(parents=True)
data = ScriptData()
progress("Распаковка архива")
extract(archive, filename, src_root)
game_dir = find_game_dir(src_root)
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):
@@ -115,7 +121,8 @@ def convert_archive(archive: Path, filename: str, work: Path, progress: Progress
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}, ensure_ascii=False,
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}
@@ -138,6 +145,7 @@ def convert_archive(archive: Path, filename: str, work: Path, progress: Progress
"transforms": data.transforms,
"transitions": data.transitions,
"audio": audio,
"channels": data.channels,
"files": aliases,
"defaults": data.defaults,
"init": data.init_code,
@@ -169,12 +177,18 @@ def convert_archive(archive: Path, filename: str, work: Path, progress: Progress
{"message": msg, "count": n, "example": data.warning_examples.get(msg)}
for msg, n in data.warnings.most_common()
],
"sourceSize": archive.stat().st_size,
"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")
@@ -203,9 +217,47 @@ def _resolve_image(img: dict[str, Any], aliases: dict[str, str], missing: set[st
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():
+189 -185
View File
@@ -1,185 +1,189 @@
"""Translate the subset of Python used by visual novels into PGVN IR.
Expressions become JSON trees ``{"k": kind, ...}``; statements become
``{"op": ..., ...}`` nodes executed by the player's JS interpreter.
Anything outside the subset raises :class:`Unsupported` so the caller can
report it and degrade gracefully.
"""
import ast
from typing import Any
Expr = dict[str, Any]
Stmt = dict[str, Any]
BIN_OPS = {
ast.Add: "+", ast.Sub: "-", ast.Mult: "*", ast.Div: "/", ast.FloorDiv: "//",
ast.Mod: "%", ast.Pow: "**", ast.BitAnd: "&", ast.BitOr: "|", ast.BitXor: "^",
ast.LShift: "<<", ast.RShift: ">>",
}
CMP_OPS = {
ast.Eq: "==", ast.NotEq: "!=", ast.Lt: "<", ast.LtE: "<=", ast.Gt: ">", ast.GtE: ">=",
ast.In: "in", ast.NotIn: "not in", ast.Is: "is", ast.IsNot: "is not",
}
UNARY_OPS = {ast.Not: "not", ast.USub: "-", ast.UAdd: "+", ast.Invert: "~"}
# Modules whose import is harmless: the player provides the functions it supports.
KNOWN_MODULES = {"random", "math", "renpy", "store", "time", "datetime"}
class Unsupported(Exception):
pass
def const(v: Any) -> Expr:
return {"k": "const", "v": v}
def parse_expr(source: str) -> Expr:
try:
tree = ast.parse(source.strip(), mode="eval")
except SyntaxError as e:
raise Unsupported(f"синтаксическая ошибка в выражении: {source!r}") from e
return expr(tree.body)
def parse_stmts(source: str) -> list[Stmt]:
try:
tree = ast.parse(_dedent(source))
except SyntaxError as e:
raise Unsupported(f"синтаксическая ошибка в python: {e.msg} ({source[:60]!r})") from e
return stmts(tree.body)
def _dedent(source: str) -> str:
import textwrap
return textwrap.dedent(source)
def literal(source: str) -> Any:
"""Evaluate a constant expression at conversion time, or raise Unsupported."""
try:
return ast.literal_eval(source.strip())
except (ValueError, SyntaxError, TypeError, MemoryError, RecursionError) as e:
raise Unsupported(f"не константа: {source!r}") from e
def expr(node: ast.AST) -> Expr:
match node:
case ast.Constant(value=v):
if isinstance(v, bytes):
v = v.decode("utf-8", "replace")
if isinstance(v, complex) or v is Ellipsis:
raise Unsupported("неподдерживаемая константа")
return const(v)
case ast.Name(id=name):
return {"k": "name", "id": name}
case ast.Attribute(value=ast.Name(id="store"), attr=attr):
return {"k": "name", "id": attr}
case ast.Attribute(value=value, attr=attr):
return {"k": "attr", "o": expr(value), "a": attr}
case ast.Subscript(value=value, slice=ast.Slice(lower=lo, upper=hi, step=st)):
return {"k": "slice", "o": expr(value), "lo": _opt(lo), "hi": _opt(hi), "st": _opt(st)}
case ast.Subscript(value=value, slice=index):
return {"k": "sub", "o": expr(value), "i": expr(index)}
case ast.BinOp(left=l, op=op, right=r):
if type(op) not in BIN_OPS:
raise Unsupported(f"оператор {type(op).__name__}")
return {"k": "bin", "op": BIN_OPS[type(op)], "l": expr(l), "r": expr(r)}
case ast.BoolOp(op=op, values=values):
return {"k": "bool", "op": "and" if isinstance(op, ast.And) else "or",
"v": [expr(v) for v in values]}
case ast.UnaryOp(op=op, operand=operand):
return {"k": "unary", "op": UNARY_OPS[type(op)], "v": expr(operand)}
case ast.Compare(left=l, ops=ops, comparators=rs):
return {"k": "cmp", "l": expr(l), "ops": [CMP_OPS[type(o)] for o in ops],
"r": [expr(r) for r in rs]}
case ast.IfExp(test=t, body=b, orelse=e):
return {"k": "if", "c": expr(t), "t": expr(b), "e": expr(e)}
case ast.List(elts=elts) | ast.Set(elts=elts):
return {"k": "list", "v": [expr(e) for e in elts]}
case ast.Tuple(elts=elts):
return {"k": "tuple", "v": [expr(e) for e in elts]}
case ast.Dict(keys=keys, values=values):
if any(k is None for k in keys):
raise Unsupported("распаковка ** в словаре")
return {"k": "dict", "v": [[expr(k), expr(v)] for k, v in zip(keys, values)]}
case ast.Call(func=func, args=args, keywords=kws):
if any(isinstance(a, ast.Starred) for a in args) or any(k.arg is None for k in kws):
raise Unsupported("распаковка аргументов *args/**kwargs")
return {"k": "call", "f": expr(func), "a": [expr(a) for a in args],
"kw": {k.arg: expr(k.value) for k in kws}}
case ast.JoinedStr(values=values):
parts: list[Any] = []
for v in values:
if isinstance(v, ast.Constant):
parts.append(v.value)
elif isinstance(v, ast.FormattedValue):
parts.append(expr(v.value))
return {"k": "fmt", "v": parts}
case ast.ListComp(elt=elt, generators=[ast.comprehension(target=t, iter=it, ifs=ifs,
is_async=0)]):
return {"k": "comp", "e": expr(elt), "t": target(t), "it": expr(it),
"if": [expr(i) for i in ifs]}
case ast.Lambda():
raise Unsupported("lambda")
raise Unsupported(f"выражение {type(node).__name__}")
def _opt(node: ast.AST | None) -> Expr | None:
return None if node is None else expr(node)
def target(node: ast.AST) -> Expr:
match node:
case ast.Name() | ast.Attribute() | ast.Subscript():
return expr(node)
case ast.Tuple(elts=elts) | ast.List(elts=elts):
return {"k": "tuple", "v": [target(e) for e in elts]}
raise Unsupported(f"присваивание в {type(node).__name__}")
def stmts(nodes: list[ast.stmt]) -> list[Stmt]:
out: list[Stmt] = []
for n in nodes:
out.extend(stmt(n))
return out
def stmt(node: ast.stmt) -> list[Stmt]:
match node:
case ast.Assign(targets=targets, value=value):
v = expr(value)
return [{"op": "set", "t": target(t), "v": v} for t in targets]
case ast.AnnAssign(target=t, value=value) if value is not None:
return [{"op": "set", "t": target(t), "v": expr(value)}]
case ast.AugAssign(target=t, op=op, value=value):
if type(op) not in BIN_OPS:
raise Unsupported(f"оператор {type(op).__name__}")
return [{"op": "aug", "t": target(t), "o": BIN_OPS[type(op)], "v": expr(value)}]
case ast.Expr(value=ast.Constant()):
return [] # docstring / bare literal
case ast.Expr(value=value):
return [{"op": "eval", "e": expr(value)}]
case ast.If(test=test, body=body, orelse=orelse):
return [{"op": "if", "c": expr(test), "then": stmts(body), "else": stmts(orelse)}]
case ast.While(test=test, body=body, orelse=[]):
return [{"op": "while", "c": expr(test), "body": stmts(body)}]
case ast.For(target=t, iter=it, body=body, orelse=[]):
return [{"op": "for", "t": target(t), "it": expr(it), "body": stmts(body)}]
case ast.Delete(targets=targets):
return [{"op": "del", "t": target(t)} for t in targets]
case ast.Pass() | ast.Global() | ast.Nonlocal():
return []
case ast.Break():
return [{"op": "break"}]
case ast.Continue():
return [{"op": "continue"}]
case ast.Import(names=names) | ast.ImportFrom(names=names):
mod = getattr(node, "module", None) or names[0].name
if mod.split(".")[0] in KNOWN_MODULES:
return []
raise Unsupported(f"импорт модуля {mod}")
case ast.FunctionDef(name=name) | ast.ClassDef(name=name):
raise Unsupported(f"определение {name}()")
raise Unsupported(f"оператор {type(node).__name__}")
"""Translate the subset of Python used by visual novels into PGVN IR.
Expressions become JSON trees ``{"k": kind, ...}``; statements become
``{"op": ..., ...}`` nodes executed by the player's JS interpreter.
Anything outside the subset raises :class:`Unsupported` so the caller can
report it and degrade gracefully.
"""
import ast
from typing import Any
Expr = dict[str, Any]
Stmt = dict[str, Any]
BIN_OPS = {
ast.Add: "+", ast.Sub: "-", ast.Mult: "*", ast.Div: "/", ast.FloorDiv: "//",
ast.Mod: "%", ast.Pow: "**", ast.BitAnd: "&", ast.BitOr: "|", ast.BitXor: "^",
ast.LShift: "<<", ast.RShift: ">>",
}
CMP_OPS = {
ast.Eq: "==", ast.NotEq: "!=", ast.Lt: "<", ast.LtE: "<=", ast.Gt: ">", ast.GtE: ">=",
ast.In: "in", ast.NotIn: "not in", ast.Is: "is", ast.IsNot: "is not",
}
UNARY_OPS = {ast.Not: "not", ast.USub: "-", ast.UAdd: "+", ast.Invert: "~"}
# Modules whose import is harmless: the player provides the functions it supports.
KNOWN_MODULES = {"random", "math", "renpy", "store", "time", "datetime"}
class Unsupported(Exception):
pass
def const(v: Any) -> Expr:
return {"k": "const", "v": v}
def parse_expr(source: str) -> Expr:
try:
tree = ast.parse(source.strip(), mode="eval")
except SyntaxError as e:
raise Unsupported(f"синтаксическая ошибка в выражении: {source!r}") from e
return expr(tree.body)
def parse_stmts(source: str) -> list[Stmt]:
try:
tree = ast.parse(_dedent(source))
except SyntaxError as e:
raise Unsupported(f"синтаксическая ошибка в python: {e.msg} ({source[:60]!r})") from e
return stmts(tree.body)
def _dedent(source: str) -> str:
import textwrap
return textwrap.dedent(source)
def literal(source: str) -> Any:
"""Evaluate a constant expression at conversion time, or raise Unsupported."""
try:
return ast.literal_eval(source.strip())
except (ValueError, SyntaxError, TypeError, MemoryError, RecursionError) as e:
raise Unsupported(f"не константа: {source!r}") from e
def expr(node: ast.AST) -> Expr:
match node:
case ast.Constant(value=v):
if isinstance(v, bytes):
v = v.decode("utf-8", "replace")
if isinstance(v, complex) or v is Ellipsis:
raise Unsupported("неподдерживаемая константа")
return const(v)
case ast.Name(id=name):
return {"k": "name", "id": name}
case ast.Attribute(value=ast.Name(id="store"), attr=attr):
return {"k": "name", "id": attr}
case ast.Attribute(value=value, attr=attr):
return {"k": "attr", "o": expr(value), "a": attr}
case ast.Subscript(value=value, slice=ast.Slice(lower=lo, upper=hi, step=st)):
return {"k": "slice", "o": expr(value), "lo": _opt(lo), "hi": _opt(hi), "st": _opt(st)}
case ast.Subscript(value=value, slice=index):
return {"k": "sub", "o": expr(value), "i": expr(index)}
case ast.BinOp(left=l, op=op, right=r):
if type(op) not in BIN_OPS:
raise Unsupported(f"оператор {type(op).__name__}")
return {"k": "bin", "op": BIN_OPS[type(op)], "l": expr(l), "r": expr(r)}
case ast.BoolOp(op=op, values=values):
return {"k": "bool", "op": "and" if isinstance(op, ast.And) else "or",
"v": [expr(v) for v in values]}
case ast.UnaryOp(op=op, operand=operand):
return {"k": "unary", "op": UNARY_OPS[type(op)], "v": expr(operand)}
case ast.Compare(left=l, ops=ops, comparators=rs):
return {"k": "cmp", "l": expr(l), "ops": [CMP_OPS[type(o)] for o in ops],
"r": [expr(r) for r in rs]}
case ast.IfExp(test=t, body=b, orelse=e):
return {"k": "if", "c": expr(t), "t": expr(b), "e": expr(e)}
case ast.List(elts=elts) | ast.Set(elts=elts):
return {"k": "list", "v": [expr(e) for e in elts]}
case ast.Tuple(elts=elts):
return {"k": "tuple", "v": [expr(e) for e in elts]}
case ast.Dict(keys=keys, values=values):
if any(k is None for k in keys):
raise Unsupported("распаковка ** в словаре")
return {"k": "dict", "v": [[expr(k), expr(v)] for k, v in zip(keys, values)]}
case ast.Call(func=func, args=args, keywords=kws):
if any(isinstance(a, ast.Starred) for a in args) or any(k.arg is None for k in kws):
raise Unsupported("распаковка аргументов *args/**kwargs")
return {"k": "call", "f": expr(func), "a": [expr(a) for a in args],
"kw": {k.arg: expr(k.value) for k in kws}}
case ast.JoinedStr(values=values):
parts: list[Any] = []
for v in values:
if isinstance(v, ast.Constant):
parts.append(v.value)
elif isinstance(v, ast.FormattedValue):
parts.append(expr(v.value))
return {"k": "fmt", "v": parts}
case (ast.ListComp(elt=elt, generators=[ast.comprehension(target=t, iter=it, ifs=ifs,
is_async=0)])
| ast.GeneratorExp(elt=elt, generators=[ast.comprehension(target=t, iter=it,
ifs=ifs, is_async=0)])
| ast.SetComp(elt=elt, generators=[ast.comprehension(target=t, iter=it, ifs=ifs,
is_async=0)])):
return {"k": "comp", "e": expr(elt), "t": target(t), "it": expr(it),
"if": [expr(i) for i in ifs]}
case ast.Lambda():
raise Unsupported("lambda")
raise Unsupported(f"выражение {type(node).__name__}")
def _opt(node: ast.AST | None) -> Expr | None:
return None if node is None else expr(node)
def target(node: ast.AST) -> Expr:
match node:
case ast.Name() | ast.Attribute() | ast.Subscript():
return expr(node)
case ast.Tuple(elts=elts) | ast.List(elts=elts):
return {"k": "tuple", "v": [target(e) for e in elts]}
raise Unsupported(f"присваивание в {type(node).__name__}")
def stmts(nodes: list[ast.stmt]) -> list[Stmt]:
out: list[Stmt] = []
for n in nodes:
out.extend(stmt(n))
return out
def stmt(node: ast.stmt) -> list[Stmt]:
match node:
case ast.Assign(targets=targets, value=value):
v = expr(value)
return [{"op": "set", "t": target(t), "v": v} for t in targets]
case ast.AnnAssign(target=t, value=value) if value is not None:
return [{"op": "set", "t": target(t), "v": expr(value)}]
case ast.AugAssign(target=t, op=op, value=value):
if type(op) not in BIN_OPS:
raise Unsupported(f"оператор {type(op).__name__}")
return [{"op": "aug", "t": target(t), "o": BIN_OPS[type(op)], "v": expr(value)}]
case ast.Expr(value=ast.Constant()):
return [] # docstring / bare literal
case ast.Expr(value=value):
return [{"op": "eval", "e": expr(value)}]
case ast.If(test=test, body=body, orelse=orelse):
return [{"op": "if", "c": expr(test), "then": stmts(body), "else": stmts(orelse)}]
case ast.While(test=test, body=body, orelse=[]):
return [{"op": "while", "c": expr(test), "body": stmts(body)}]
case ast.For(target=t, iter=it, body=body, orelse=[]):
return [{"op": "for", "t": target(t), "it": expr(it), "body": stmts(body)}]
case ast.Delete(targets=targets):
return [{"op": "del", "t": target(t)} for t in targets]
case ast.Pass() | ast.Global() | ast.Nonlocal():
return []
case ast.Break():
return [{"op": "break"}]
case ast.Continue():
return [{"op": "continue"}]
case ast.Import(names=names) | ast.ImportFrom(names=names):
mod = getattr(node, "module", None) or names[0].name
if mod.split(".")[0] in KNOWN_MODULES:
return []
raise Unsupported(f"импорт модуля {mod}")
case ast.FunctionDef(name=name) | ast.ClassDef(name=name):
raise Unsupported(f"определение {name}()")
raise Unsupported(f"оператор {type(node).__name__}")
+5 -2
View File
@@ -7,6 +7,7 @@ unpickler that only allows plain containers — archives are untrusted input.
import io
import pickle
import zlib
from collections.abc import Callable
from pathlib import Path
@@ -65,13 +66,15 @@ def read_index(path: Path) -> dict[str, list[tuple[int, int, bytes]]]:
return index
def extract_rpa(path: Path, dest: Path) -> int:
"""Extract all files of an archive into dest. Returns the number of files."""
def extract_rpa(path: Path, dest: Path, only: Callable[[str], bool] | None = None) -> int:
"""Extract files of an archive into dest (all, or those `only` accepts)."""
index = read_index(path)
root = dest.resolve()
count = 0
with path.open("rb") as f:
for name, entries in index.items():
if only is not None and not only(name):
continue
target = (dest / name).resolve()
if not target.is_relative_to(root):
continue
+283 -31
View File
@@ -1,6 +1,8 @@
"""Ren'Py script (.rpy) → PGVN IR."""
import ast
import re
import warnings
from collections import Counter
from dataclasses import dataclass, field
from typing import Any
@@ -43,7 +45,7 @@ POSITION_PROPS = {"xpos", "ypos", "xanchor", "yanchor", "xalign", "yalign", "xce
"ycenter", "zoom", "xzoom", "yzoom", "alpha", "rotate", "xoffset", "yoffset",
"align", "pos", "anchor", "xysize", "size"}
CHANNELS = {"music", "sound", "voice", "audio", "movie"}
SKIP_BLOCKS = {"screen", "style", "translate", "testcase", "layeredimage", "camera"}
SKIP_BLOCKS = {"screen", "style", "translate", "testcase"}
BUILTIN_CHARACTERS = {"narrator", "centered", "vcentered", "extend", "nvl", "adv", "name_only"}
@@ -57,6 +59,7 @@ class ScriptData:
defaults: list[list[Any]] = field(default_factory=list)
init_code: list[Stmt] = field(default_factory=list) # defines + init python, in order
gui: dict[str, Any] = field(default_factory=dict)
channels: dict[str, dict[str, Any]] = field(default_factory=dict)
config: dict[str, Any] = field(default_factory=dict)
modules: dict[str, list[Node]] = field(default_factory=dict) # file -> labels
warnings: Counter = field(default_factory=Counter)
@@ -73,6 +76,11 @@ class RpyParser:
def __init__(self, data: ScriptData, filename: str):
self.d = data
self.filename = filename
self.global_label = ""
def label_ref(self, name: str) -> str:
"""Resolve `.local` label names against the enclosing global label."""
return self.global_label + name if name.startswith(".") else name
# --- helpers -------------------------------------------------------------
@@ -92,7 +100,7 @@ class RpyParser:
def python(self, source: str, line: Line) -> list[Stmt]:
"""Translate python code statement by statement, skipping unsupported ones."""
try:
tree = ast.parse(_dedent(source))
tree = _parse_py(source)
except SyntaxError as e:
self.d.warn(f"Python: синтаксическая ошибка ({e.msg})", self.where(line))
return []
@@ -114,13 +122,12 @@ class RpyParser:
c = self.cursor(line)
first = c.peek_word()
if first == "label":
c.keyword("label")
name = c.label_name()
if not name:
raise c.error("ожидалось имя метки")
if c.char("("):
self.d.warn("Параметры меток не поддерживаются", self.where(line))
current = {"name": name, "body": self.block(line.block)}
name, params = self.label_header(line)
if not name.startswith("."):
self.global_label = name
current = {"name": self.label_ref(name), "body": self.block(line.block)}
if params:
current["params"] = params
labels.append(current)
continue
nodes = self.statement(line, top_level=True)
@@ -143,7 +150,15 @@ class RpyParser:
out.append({"op": "if", "branches": branches})
continue
if word == "label":
self.d.warn("Вложенные метки не поддерживаются", self.where(line))
# a label inside a block is an entry point in the middle of the flow:
# keep a marker the player can jump to and inline the label's block
name, params = self.label_header(line)
if not name.startswith("."):
self.global_label = name
marker: Node = {"op": "label", "name": self.label_ref(name)}
if params:
marker["params"] = params
out.append(marker)
out.extend(self.block(line.block))
i += 1
continue
@@ -151,6 +166,37 @@ class RpyParser:
i += 1
return out
def label_header(self, line: Line) -> tuple[str, list[list[Any]] | None]:
c = self.cursor(line)
c.keyword("label")
name = c.label_name()
if not name:
raise c.error("ожидалось имя метки")
params = None
if c.peek("("):
sig = c.until(set(), ":")
try:
fn = _parse_py(f"def _f{sig}: pass").body[0]
args = fn.args
defaults = [None] * (len(args.args) - len(args.defaults)) + list(args.defaults)
params = [[a.arg, pyexpr.expr(d) if d is not None else None]
for a, d in zip(args.args, defaults)]
except (SyntaxError, Unsupported):
self.d.warn("Не разобраны параметры метки", self.where(line))
return name, params
def call_args(self, c: Cursor, line: Line) -> dict[str, Any] | None:
if not c.peek("("):
return None
src = c.until({"from"})
try:
call = _parse_py(f"_f{src}", mode="eval").body
return {"a": [pyexpr.expr(a) for a in call.args],
"kw": {k.arg: pyexpr.expr(k.value) for k in call.keywords}}
except (SyntaxError, Unsupported, AttributeError):
self.d.warn("Не разобраны аргументы call", self.where(line))
return None
def if_chain(self, lines: list[Line], i: int) -> tuple[list[Node], int]:
branches: list[Node] = []
while i < len(lines):
@@ -282,11 +328,15 @@ class RpyParser:
def init_python(self, source: str, line: Line) -> None:
try:
tree = ast.parse(_dedent(source))
tree = _parse_py(source)
except SyntaxError as e:
self.d.warn(f"Python: синтаксическая ошибка ({e.msg})", self.where(line))
return
for node in tree.body:
if (isinstance(node, ast.Expr) and isinstance(node.value, ast.Call)
and ast.unparse(node.value.func) == "renpy.music.register_channel"):
self.register_channel(node.value)
continue
if isinstance(node, ast.Assign) and len(node.targets) == 1:
tgt = node.targets[0]
name = _assign_name(tgt)
@@ -306,6 +356,21 @@ class RpyParser:
except Unsupported as e:
self.d.warn(f"init python: {e}", f"{self.filename}:{line.lineno + node.lineno}")
def register_channel(self, call: ast.Call) -> None:
args = list(call.args) + [None] * 3
kw = {k.arg: k.value for k in call.keywords}
def lit(node, default=None):
try:
return ast.literal_eval(node) if node is not None else default
except ValueError:
return default
name = lit(args[0])
if isinstance(name, str):
self.d.channels[name] = {"mixer": lit(kw.get("mixer", args[1]), "sfx"),
"loop": bool(lit(kw.get("loop", args[2]), False))}
# define / default / image / transform -------------------------------------------
def st_define(self, line: Line) -> list[Node]:
@@ -499,25 +564,182 @@ class RpyParser:
def atl_image(self, line: Line) -> dict[str, Any] | None:
frames: list[dict[str, Any]] = []
props: dict[str, Any] = {}
for child in line.block:
repeat = self.atl_frames(line.block, frames, props)
if not frames:
self.d.warn("ATL-изображение без кадров", self.where(line))
return None
if len(frames) == 1:
img = dict(frames[0]["img"])
else:
img = {"frames": frames, "repeat": repeat}
if props:
img["props"] = {**img.get("props", {}), **props}
return img
def atl_frames(self, lines: list[Line], frames: list[dict[str, Any]],
props: dict[str, Any]) -> bool:
"""Collect displayable frames of an ATL block. Returns whether it repeats."""
repeat = False
for child in lines:
c = self.cursor(child)
word = c.peek_word()
if word in ("contains", "block", "parallel", "choice"):
if not frames or word == "block":
repeat = self.atl_frames(child.block, frames, props) or repeat
continue
if word == "repeat":
repeat = True
continue
if word in ("pause",) or _NUMBER_LINE.match(child.text):
c.keyword("pause")
d = _num(c.number())
if frames and d is not None:
frames[-1]["d"] = frames[-1].get("d", 0) + d
continue
if word in ("function", "on", "event", "time", "transform"):
continue
s = c.string()
if s is not None:
img = self.displayable_expr(const(s), child)
if img:
frames.append(img)
frames.append({"img": img})
continue
m = _CALL_LINE.match(child.text)
if m and m.group(1) in _DISPLAYABLE_CALLS:
e = self.expr(child.text.split(" with ")[0], child)
img = self.displayable_expr(e, child) if e else None
if img:
frames.append({"img": img})
continue
if m and not frames and not _atl_props(child.text):
# e.g. get_face_path("base"): an image computed by game code
e = self.expr(child.text, child)
if e is not None:
self.d.warn("Изображение задано выражением", self.where(child))
frames.append({"img": {"expr": e}})
continue
if not frames or not frames[-1].get("d"):
props.update(_atl_props(child.text))
return repeat
def st_layeredimage(self, line: Line) -> list[Node]:
c = self.cursor(line)
c.keyword("layeredimage")
parts: list[str] = []
while (p := c.image_part()) is not None:
parts.append(p)
name = " ".join(parts)
layers: list[dict[str, Any]] = []
at: list[dict[str, Any]] = []
for child in line.block:
cc = self.cursor(child)
kw = cc.word()
if kw == "always":
img = self.layer_image(cc, child, None)
if img:
layers.append({"always": img})
elif kw == "attribute":
layers.append(self.layer_attribute(cc, child, name, None))
elif kw == "group":
group = cc.word() or ""
auto = False
while not cc.eol():
w = cc.word()
if w == "auto":
auto = True
elif w in ("multiple", "prefix", "variant"):
cc.word() if w in ("prefix", "variant") else None
elif w is None:
break
for sub in child.block:
sc = self.cursor(sub)
if sc.keyword("attribute"):
layers.append(self.layer_attribute(sc, sub, name, group))
if auto:
layers.append({"autoGroup": group})
elif kw == "if":
cond = self.expr(cc.rest().removesuffix(":"), child, const(False))
img = self.layer_image(Cursor(""), child, None)
if img:
layers.append({"cond": [[cond, img]]})
elif kw in ("elif", "else"):
img = self.layer_image(Cursor(""), child, None)
if img and layers and "cond" in layers[-1]:
cond = None
if kw == "elif":
cond = self.expr(cc.rest().removesuffix(":"), child, const(False))
layers[-1]["cond"].append([cond, img])
elif kw == "at":
for item in _split_top(cc.rest()):
at.append(self.transform_ref(item, child))
self.d.images[name] = {"layered": layers, "at": at}
self.d.stats["layeredimage"] += 1
return []
def layer_attribute(self, c: Cursor, line: Line, image: str, group: str | None
) -> dict[str, Any]:
attr = c.image_part() or ""
layer: dict[str, Any] = {"attr": attr, "group": group}
img = None
while not c.eol():
if c.peek('"') or c.peek("'"):
img = self.displayable_expr(const(c.string()), line)
continue
w = c.word()
if w == "default":
layer["default"] = True
elif w == "null":
layer["null"] = True
elif w in ("if_any", "if_all", "if_not"):
src = c.until({"if_any", "if_all", "if_not", "default", "null", "variant", "at"},
":")
try:
val = pyexpr.literal(src)
layer[w] = [val] if isinstance(val, str) else list(val)
except Unsupported:
pass
elif w in ("variant", "at"):
c.until({"if_any", "if_all", "if_not", "default", "null"}, ":")
elif w is None:
break
if img is None and line.block:
img = self.layer_image(Cursor(""), line, None)
if img is None and not layer.get("null"):
# Ren'Py's default naming: <image>_<group>_<attribute>
img = {"ref": "_".join(x for x in (image.replace(" ", "_"), group, attr) if x)}
if img:
layer["img"] = img
return layer
def layer_image(self, c: Cursor, line: Line, default) -> dict[str, Any] | None:
s = c.string() if c.text else None
if s is not None:
return self.displayable_expr(const(s), line)
for child in line.block:
cc = self.cursor(child)
s = cc.string()
if s is not None:
return self.displayable_expr(const(s), child)
m = _CALL_LINE.match(child.text)
if m and m.group(1) in _DISPLAYABLE_CALLS:
e = self.expr(child.text, child)
return self.displayable_expr(e, child) if e else None
return default
def st_camera(self, line: Line) -> list[Node]:
c = self.cursor(line)
c.keyword("camera")
c.word() if not c.peek("at") and not c.peek(":") else None
node: Node = {"op": "camera", "at": []}
if c.keyword("at"):
for item in _split_top(c.until(set(), ":")):
node["at"].append(self.transform_ref(item, line))
props: dict[str, Any] = {}
for child in line.block:
props.update(_atl_props(child.text))
if not frames:
self.d.warn("ATL-изображение без кадров", self.where(line))
return None
if len(frames) > 1:
self.d.warn("Анимированные ATL-изображения упрощены до первого кадра",
self.where(line))
img = dict(frames[0])
if props:
img["props"] = {**img.get("props", {}), **props}
return img
node["at"].append({"props": props})
return [node]
def st_transform(self, line: Line) -> list[Node]:
c = self.cursor(line)
@@ -588,6 +810,10 @@ class RpyParser:
fname = _dotted_name(e["f"])
if fname in ("Position", "Transform"):
return {"props": _static_props(e["kw"])}
if fname in ("Shake", "SnoopyShake"):
args = [_const(a) for a in e["a"]]
return {"effect": "shake", "d": args[1] if len(args) > 1 else 0.5,
"dist": _const(e["kw"].get("dist")) or 20}
if fname in self.d.transforms:
return {"props": self.d.transforms[fname], "name": fname}
if e and e["k"] == "name" and e["id"] not in self.d.transforms:
@@ -615,8 +841,12 @@ class RpyParser:
self.d.warn("Не поддерживается: screen", self.where(line))
return []
if c.keyword("text"):
text = c.string() or ""
node: Node = {"op": "show", "img": {"text": text}}
text = c.string()
if text is not None:
node: Node = {"op": "show", "img": {"text": text}}
else:
node = {"op": "show", "img": {"textExpr": self.expr(c.until(_SHOW_KEYWORDS), line,
const(""))}}
self.show_clauses(c, line, node)
return [node]
node = {"op": "show", "img": self.image_spec(c, line)}
@@ -668,6 +898,8 @@ class RpyParser:
fin = nums[2] if len(nums) > 2 else (nums[1] if len(nums) > 1 else out)
return {"t": "fade", "out": out, "hold": hold, "in": fin,
"color": kw.get("color") or "#000000"}
if fname in ("Shake", "SnoopyShake"):
return {"t": "punch", "dir": "h", "d": nums[0] if nums else 0.5}
if fname in ("MoveTransition", "Move"):
return {"t": "move", "d": nums[0] if nums else 0.5}
if fname in TRANSITIONS:
@@ -690,8 +922,6 @@ class RpyParser:
c = self.cursor(line)
c.keyword(op)
ch = c.word() or "music"
if ch not in CHANNELS:
self.d.warn(f"Нестандартный аудиоканал {ch}", self.where(line))
node: Node = {"op": op, "ch": ch, "files": self.audio_files(c, line)}
while not c.eol():
if c.keyword("fadein"):
@@ -737,7 +967,7 @@ class RpyParser:
c.keyword("jump")
if c.keyword("expression"):
return [{"op": "jump", "expr": self.expr(c.rest(), line, const(""))}]
return [{"op": "jump", "target": c.label_name()}]
return [{"op": "jump", "target": self.label_ref(c.label_name() or "")}]
def st_call(self, line: Line) -> list[Node]:
c = self.cursor(line)
@@ -748,10 +978,12 @@ class RpyParser:
if c.keyword("expression"):
node: Node = {"op": "call", "expr": self.expr(c.until({"pass", "from"}), line,
const(""))}
c.keyword("pass")
else:
node = {"op": "call", "target": c.label_name()}
if c.peek("("):
self.d.warn("Аргументы call не поддерживаются", self.where(line))
node = {"op": "call", "target": self.label_ref(c.label_name() or "")}
args = self.call_args(c, line)
if args:
node["args"] = args
return [node]
def st_return(self, line: Line) -> list[Node]:
@@ -791,11 +1023,19 @@ class RpyParser:
def st_menu(self, line: Line) -> list[Node]:
node: Node = {"op": "menu", "items": [], "say": None}
head = self.cursor(line)
head.keyword("menu")
name = head.label_name()
out: list[Node] = []
if name:
out.append({"op": "label", "name": self.label_ref(name)})
for child in line.block:
c = self.cursor(child)
if c.keyword("set"):
node["set"] = self.expr(c.rest(), child)
continue
if c.keyword("with"):
continue
text = c.string()
if text is not None and child.text.rstrip().endswith(":"):
cond = None
@@ -813,7 +1053,7 @@ class RpyParser:
else:
self.d.warn("Не разобран пункт меню", f"{self.where(child)}: {child.text[:60]}")
self.d.stats["menu"] += 1
return [node]
return out + [node]
# say ------------------------------------------------------------------------------
@@ -871,6 +1111,11 @@ class RpyParser:
_SHOW_KEYWORDS = {"at", "as", "behind", "onlayer", "with", "zorder"}
_NUMBER_LINE = re.compile(r"^\d+(\.\d*)?$|^\.\d+$")
_CALL_LINE = re.compile(r"^([A-Za-z_][\w.]*)\s*\(")
_DISPLAYABLE_CALLS = {"Movie", "Solid", "Transform", "Composite", "LiveComposite", "Fixed",
"ConditionSwitch", "Image", "im.Scale", "im.FactorScale", "im.Flip",
"im.Composite", "im.Crop", "im.Grayscale", "im.Sepia", "Frame", "Flatten"}
_MEDIA_EXTS = {"png", "jpg", "jpeg", "webp", "gif", "bmp", "avif", "ogg", "mp3", "wav", "opus",
"flac", "webm", "ogv", "mp4", "mkv", "avi", "m4a"}
@@ -880,6 +1125,13 @@ def _dedent(source: str) -> str:
return textwrap.dedent(source)
def _parse_py(source: str, mode: str = "exec") -> ast.AST:
# decompiled scripts are full of "\{" in plain strings: don't spam SyntaxWarning
with warnings.catch_warnings():
warnings.simplefilter("ignore")
return ast.parse(_dedent(source), mode=mode)
def _dotted_name(e: Expr) -> str:
if e["k"] == "name":
return e["id"]