- Converter: def/return/global/nonlocal, class (inheritance, property, staticmethod, classmethod), lambda, *args/**kwargs and call-site unpacking. Unsupported dunder methods, generators and unknown decorators are still refused - Player: closures, super(), instances that survive save/load, isinstance, hasattr/getattr/setattr, map/filter/sorted(key=), recursion limit - audit_ir accepts calls to names, classes and methods the game defines itself; unknown renpy.* calls still block the import. Volume/pan/pause/replay calls are accepted as presentation-only no-ops; build.* settings are skipped - Tests: IR fixture compiled from tests/fixtures/classes.py (tests/make_ir.py) and run in the interpreter; converter tests updated Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
190 lines
8.9 KiB
Python
190 lines
8.9 KiB
Python
"""Engine detection and fail-closed import diagnostics. Never runs game executables."""
|
|
import os
|
|
from dataclasses import asdict, dataclass
|
|
from pathlib import Path, PurePosixPath
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Issue:
|
|
code: str
|
|
message: str
|
|
location: str = ""
|
|
count: int = 1
|
|
|
|
|
|
class CompatibilityError(RuntimeError):
|
|
def __init__(self, engine: str, issues: list[Issue]):
|
|
self.engine, self.issues = engine, issues
|
|
super().__init__("Импорт остановлен: " + "; ".join(
|
|
i.message + (f" ({i.location})" if i.location else "") for i in issues[:5]))
|
|
|
|
def report(self) -> dict:
|
|
return {"error": str(self), "engine": self.engine, "compatible": False,
|
|
"issues": [asdict(i) for i in self.issues]}
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Detection:
|
|
engine: str
|
|
root: Path
|
|
|
|
|
|
# Development copies and installed patches must not be mistaken for additional games.
|
|
IGNORED = {".git", ".godot", ".local-translator", "migration-staging", "migration-tools",
|
|
"node_modules", "__macosx"}
|
|
|
|
|
|
def detect(root: Path) -> Detection:
|
|
candidates: dict[tuple[str, Path], Detection] = {}
|
|
for directory, dirs, files in os.walk(root, followlinks=False):
|
|
dirs[:] = sorted(d for d in dirs if d.lower() not in IGNORED and
|
|
not Path(directory, d).is_symlink())
|
|
path = Path(directory)
|
|
names = {f.lower() for f in files}
|
|
lower = {d.lower() for d in dirs}
|
|
engine = None
|
|
if path.name.lower().endswith("_data") and ("managed" in lower or "il2cpp_data" in lower):
|
|
managed = path / next((d for d in dirs if d.lower() == "managed"), "Managed")
|
|
dlls = {p.name.lower() for p in managed.glob("*.dll")}
|
|
engine = "naninovel" if any("naninovel" in n for n in dlls) else "unity"
|
|
elif "data.win" in names:
|
|
engine = "gamemaker"
|
|
elif "project.godot" in names or any(n.endswith(".pck") for n in names):
|
|
engine = "godot"
|
|
elif "game.rgss3a" in names or "game.rgss2a" in names or "game.rgssad" in names:
|
|
engine = "rpgmaker"
|
|
elif "rpg_core.js" in names or "rmmz_core.js" in names:
|
|
engine = "rpgmaker"
|
|
elif path.name.lower() == "game" and any(
|
|
PurePosixPath(n).suffix in {".rpy", ".rpyc", ".rpa"} for n in names):
|
|
engine = "renpy"
|
|
if engine:
|
|
game_root = path.parent if engine in {"naninovel", "unity", "renpy"} else path
|
|
candidates[(engine, game_root)] = Detection(engine, game_root)
|
|
# Engine internals are not independent games.
|
|
dirs.clear()
|
|
if not candidates:
|
|
# Unpacked Ren'Py script-only archives do not always have a game/ wrapper.
|
|
from app.converter.archive import ArchiveError, find_game_dir
|
|
try:
|
|
game = find_game_dir(root)
|
|
return Detection("renpy", game)
|
|
except ArchiveError:
|
|
raise CompatibilityError("unknown", [Issue("unknown_engine",
|
|
"Движок не распознан; нужен полный дистрибутив игры, а не отдельный EXE или патч")]) from None
|
|
if len(candidates) != 1:
|
|
raise CompatibilityError("ambiguous", [Issue("multiple_games",
|
|
"В одном импорте обнаружено несколько игр; загрузите каждую отдельно",
|
|
", ".join(f"{c.engine}: {c.root.relative_to(root)}" for c in candidates.values()))])
|
|
result = next(iter(candidates.values()))
|
|
from app.games.registry import all_profiles
|
|
if result.engine not in {profile.engine for profile in all_profiles()}:
|
|
raise CompatibilityError(result.engine, [Issue("unsupported_engine",
|
|
f"Движок {result.engine} распознан, но его исполнение в PGames не поддерживается")])
|
|
return result
|
|
|
|
|
|
def require_supported_scripts(data, engine: str = "renpy") -> None:
|
|
issues = [Issue("unsupported_script", message, data.blocker_examples.get(message, ""), count)
|
|
for message, count in data.blockers.items()]
|
|
if issues:
|
|
raise CompatibilityError(engine, issues)
|
|
|
|
|
|
CALLS = set("len str int float bool abs round min max sum any all range list tuple set dict sorted reversed enumerate zip isinstance _ __".split())
|
|
CALLS |= {prefix + method for prefix in ("random.", "renpy.random.")
|
|
for method in ("randint", "random", "choice", "shuffle", "uniform")}
|
|
CALLS |= {"math." + method for method in ("floor", "ceil", "sqrt", "pow")}
|
|
CALLS |= {"renpy." + method for method in ("play", "notify", "jump", "call", "input", "has_label",
|
|
"seen_label", "showing", "get_attributes", "full_restart", "quit", "version")}
|
|
CALLS |= {f"renpy.{channel}.{method}" for channel in ("music", "sound")
|
|
for method in ("play", "queue", "stop", "get_playing", "is_playing")}
|
|
# Presentation-only calls the player accepts and ignores (volume/pan/pause tweaks, replay and
|
|
# rollback bookkeeping). They never change the story state.
|
|
CALLS |= {f"renpy.{channel}.{method}" for channel in ("music", "sound")
|
|
for method in ("set_volume", "set_pause", "set_pan")}
|
|
CALLS |= {"renpy.end_replay", "renpy.block_rollback"}
|
|
CALLS |= {"map", "filter", "hasattr", "getattr", "setattr", "callable", "super"}
|
|
METHODS = set("append extend insert pop remove index count reverse sort clear copy add discard get keys values items setdefault update lower upper capitalize title strip lstrip rstrip startswith endswith replace split join format find isdigit isalpha".split())
|
|
|
|
|
|
def _bound_names(root) -> tuple[set[str], set[str]]:
|
|
"""Names the game binds itself and the method names of its classes.
|
|
|
|
Calling any of them is the game's own code, which the player interprets.
|
|
"""
|
|
names: set[str] = set()
|
|
methods: set[str] = set()
|
|
|
|
def target(value):
|
|
if isinstance(value, dict):
|
|
if value.get("k") == "name":
|
|
names.add(value["id"])
|
|
elif value.get("k") == "tuple":
|
|
for item in value["v"]:
|
|
target(item)
|
|
|
|
def visit(value, in_class=False):
|
|
if isinstance(value, list):
|
|
for item in value:
|
|
visit(item, in_class)
|
|
elif isinstance(value, dict):
|
|
op, kind = value.get("op"), value.get("k")
|
|
if op == "def":
|
|
(methods if in_class else names).add(value["name"])
|
|
names.update(p["n"] for p in value["params"])
|
|
elif op == "class":
|
|
names.add(value["name"])
|
|
elif op in ("set", "aug", "for"):
|
|
target(value.get("t"))
|
|
elif kind == "lambda":
|
|
names.update(p["n"] for p in value["params"])
|
|
elif kind == "comp":
|
|
target(value.get("t"))
|
|
for key, child in value.items():
|
|
visit(child, op == "class" and key == "body")
|
|
|
|
visit(root)
|
|
return names, methods
|
|
|
|
|
|
def audit_ir(data) -> None:
|
|
"""Reject calls which the browser would otherwise silently replace by None."""
|
|
user_names, user_methods = _bound_names([data.modules, data.init_code, data.defaults, data.images])
|
|
user_names |= {name for name, _ in data.defaults}
|
|
|
|
def dotted(value):
|
|
if value.get("k") == "name":
|
|
return value["id"]
|
|
if value.get("k") == "attr":
|
|
return dotted(value["o"]) + "." + value["a"]
|
|
return ""
|
|
|
|
def visit(value, location):
|
|
if isinstance(value, list):
|
|
for item in value:
|
|
visit(item, location)
|
|
elif isinstance(value, dict):
|
|
if value.get("k") == "call":
|
|
function = value.get("f", {})
|
|
name = dotted(function)
|
|
is_method = (function.get("k") == "attr" and not name.startswith(("renpy.", "math.", "random."))
|
|
and (function.get("a") in METHODS or function.get("a") in user_methods
|
|
or function.get("a") in user_names))
|
|
if name not in CALLS and name not in user_names and not is_method:
|
|
data.warn(f"Вызов не поддерживается плеером: {name or 'динамическая функция'}", location)
|
|
if value.get("op") in {"jump", "call"} and value.get("target"):
|
|
if value["target"] not in labels:
|
|
data.warn(f"Не найдена цель перехода: {value['target']}", location)
|
|
for child in value.values():
|
|
visit(child, location)
|
|
|
|
from app.converter.pipeline import anchors
|
|
labels = {label["name"] for module in data.modules.values() for label in module}
|
|
labels |= {name for module in data.modules.values() for name in anchors(module)}
|
|
for path, module in data.modules.items():
|
|
visit(module, path)
|
|
visit(data.defaults, "default")
|
|
visit(data.init_code, "init")
|
|
visit(data.images, "image")
|