feat(clew): dump Ren'Py compiled script objects

src/clew/BUILD.bazel

@@ -3,6 +3,7 @@ load("@rules_python//python:defs.bzl", "py_binary")
 py_binary(
     name = "clew",
     srcs = ["clew.py"],
+    data = ["renpy_object_dump.py"],
     main = "clew.py",
     deps = ["//third_party/python:rpycdec"],
 )

src/clew/clew.py

@@ -4,6 +4,7 @@ import argparse
 import logging
 import os
 import shutil
+import subprocess
 import sys
 from collections import deque
 from pathlib import Path
@@ -19,7 +20,6 @@ def parse_args(argv: list[str]) -> argparse.Namespace:
     )
     parser.add_argument(
         "--game-root",
-        required=True,
         type=Path,
         help="Game installation root to scan.",
     )
@@ -34,6 +34,11 @@ def parse_args(argv: list[str]) -> argparse.Namespace:
         action="store_true",
         help="Replace Clew-managed output directories.",
     )
+    parser.add_argument(
+        "--dump-renpy-objects",
+        action="store_true",
+        help="Dump compiled Ren'Py scripts from output/extracted/ to output/ast-raw/.",
+    )
     return parser.parse_args(argv)
 
 
@@ -77,16 +82,96 @@ def extract_game_files(game_root: Path, output_path: Path) -> None:
             shutil.copy2(item, destination)
 
 
-def run(args: argparse.Namespace) -> None:
-    game_root = args.game_root.resolve()
-    output_root = resolve_output_root(args.output_root)
+def is_game_root(path: Path) -> bool:
+    return (
+        (path / "game").is_dir()
+        and (path / "renpy").is_dir()
+        and (path / "lib" / "linux-x86_64" / "python").is_file()
+    )
 
-    if not game_root.is_dir():
-        LOG.critical(f'game root is not a directory: {game_root}')
+
+def find_game_roots(extracted_root: Path) -> list[Path]:
+    game_roots = []
+    for root, directories, _files in os.walk(extracted_root):
+        path = Path(root)
+        if is_game_root(path):
+            game_roots.append(path)
+            directories[:] = []
+    return sorted(game_roots)
+
+
+def find_compiled_scripts(game_root: Path) -> list[Path]:
+    scripts = []
+    game_directory = game_root / "game"
+    for suffix in ("*.rpyc", "*.rpymc"):
+        scripts.extend(game_directory.rglob(suffix))
+    return sorted(scripts)
+
+
+def object_dump_output_path(output_root: Path, extracted_root: Path, game_root: Path, script: Path) -> Path:
+    namespace = game_root.relative_to(extracted_root)
+    relative_script = script.relative_to(game_root)
+    return output_root / "ast-raw" / namespace / relative_script.with_suffix(relative_script.suffix + ".json")
+
+
+def run_renpy_object_dump(game_root: Path, script: Path, output: Path) -> None:
+    exporter = Path(__file__).with_name("renpy_object_dump.py").resolve()
+    bundled_python = game_root / "lib" / "linux-x86_64" / "python"
+    output.parent.mkdir(parents=True, exist_ok=True)
+    command = [
+        "steam-run",
+        str(bundled_python),
+        str(exporter),
+        "--game-root",
+        str(game_root),
+        "--input",
+        str(script),
+        "--output",
+        str(output),
+    ]
+    LOG.info("dumping Ren'Py object graph: %s", script)
+    subprocess.run(command, check=True)
+
+
+def dump_renpy_objects(output_root: Path, overwrite: bool) -> None:
+    extracted_root = output_root / "extracted"
+    if not extracted_root.is_dir():
+        LOG.critical("extracted output directory does not exist: %s", extracted_root)
         raise SystemExit(2)
 
-    extracted = get_output_directory(output_root, "extracted")
-    extract_game_files(game_root, extracted)
+    game_roots = find_game_roots(extracted_root)
+    if not game_roots:
+        LOG.warning("no Ren'Py game roots found under %s", extracted_root)
+        return
+
+    for game_root in game_roots:
+        scripts = find_compiled_scripts(game_root)
+        LOG.info("found %d compiled Ren'Py scripts under %s", len(scripts), game_root)
+        for script in scripts:
+            output = object_dump_output_path(output_root, extracted_root, game_root, script)
+            if output.exists() and not overwrite:
+                LOG.info("skipping existing Ren'Py object dump: %s", output)
+                continue
+            run_renpy_object_dump(game_root, script, output)
+
+
+def run(args: argparse.Namespace) -> None:
+    output_root = resolve_output_root(args.output_root)
+
+    if args.game_root is not None:
+        game_root = args.game_root.resolve()
+        if not game_root.is_dir():
+            LOG.critical(f"game root is not a directory: {game_root}")
+            raise SystemExit(2)
+
+        extracted = get_output_directory(output_root, "extracted")
+        extract_game_files(game_root, extracted)
+    elif not args.dump_renpy_objects:
+        LOG.critical("missing --game-root unless --dump-renpy-objects is set")
+        raise SystemExit(2)
+
+    if args.dump_renpy_objects:
+        dump_renpy_objects(output_root, args.overwrite)
 
 
 if __name__ == "__main__":

src/clew/renpy_object_dump.py

@@ -0,0 +1,258 @@
+from __future__ import print_function, unicode_literals
+
+import argparse
+import json
+import os
+import struct
+import sys
+import zlib
+
+try:
+    basestring
+except NameError:
+    basestring = str
+
+try:
+    unicode
+except NameError:
+    unicode = str
+
+try:
+    long
+except NameError:
+    long = int
+
+
+RPYC2_HEADER = b"RENPY RPC2"
+SCHEMA = "clew-renpy-object-dump-v1"
+
+
+def parse_args(argv):
+    parser = argparse.ArgumentParser(
+        prog="renpy-object-dump",
+        description="Dump Ren'Py compiled script objects as generic JSON.",
+    )
+    parser.add_argument("--game-root", required=True)
+    parser.add_argument("--input", required=True)
+    parser.add_argument("--output", required=True)
+    parser.add_argument("--max-depth", default=24, type=int)
+    return parser.parse_args(argv)
+
+
+def read_rpyc_data(path, slot):
+    with open(path, "rb") as handle:
+        header_data = handle.read(1024)
+
+        if header_data[:len(RPYC2_HEADER)] != RPYC2_HEADER:
+            if slot != 1:
+                return None
+            handle.seek(0)
+            return zlib.decompress(handle.read())
+
+        position = len(RPYC2_HEADER)
+        while True:
+            header_slot, start, length = struct.unpack("III", header_data[position:position + 12])
+            if slot == header_slot:
+                break
+            if header_slot == 0:
+                return None
+            position += 12
+
+        handle.seek(start)
+        return zlib.decompress(handle.read(length))
+
+
+def load_compiled_script(path):
+    import renpy.script
+
+    for slot in (2, 1):
+        data = read_rpyc_data(path, slot)
+        if data:
+            metadata, roots = renpy.script.loads(data)
+            return slot, metadata, roots
+
+    raise RuntimeError("no loadable slot found in {0}".format(path))
+
+
+def class_name(value):
+    return value.__class__.__module__ + "." + value.__class__.__name__
+
+
+def collect_slots(value):
+    slots = []
+    try:
+        classes = value.__class__.mro()
+    except AttributeError:
+        classes = [value.__class__]
+
+    for cls in reversed(classes):
+        cls_slots = getattr(cls, "__slots__", [])
+        if isinstance(cls_slots, basestring):
+            cls_slots = [cls_slots]
+        for slot in cls_slots:
+            if slot not in slots:
+                slots.append(slot)
+    return slots
+
+
+def safe_text(value):
+    if isinstance(value, unicode):
+        return value
+    if isinstance(value, str):
+        try:
+            return value.decode("utf-8")
+        except UnicodeDecodeError:
+            return {"type": "bytes", "repr": repr(value)}
+    return value
+
+
+def is_primitive(value):
+    return value is None or isinstance(value, (bool, int, long, float, basestring))
+
+
+class Dumper(object):
+    def __init__(self, max_depth):
+        self.max_depth = max_depth
+        self.next_id = 1
+        self.seen = {}
+        self.active_containers = set()
+
+    def dump(self, value, depth=0):
+        if is_primitive(value):
+            return safe_text(value)
+
+        if depth >= self.max_depth:
+            return {
+                "type": "truncated",
+                "class": class_name(value),
+                "repr": repr(value),
+            }
+
+        if isinstance(value, (list, tuple)):
+            object_id = id(value)
+            if object_id in self.active_containers:
+                return {"type": "recursion", "class": class_name(value)}
+            self.active_containers.add(object_id)
+            items = [self.dump(item, depth + 1) for item in value]
+            self.active_containers.remove(object_id)
+            return {"type": "tuple" if isinstance(value, tuple) else "list", "items": items}
+
+        if isinstance(value, dict):
+            object_id = id(value)
+            if object_id in self.active_containers:
+                return {"type": "recursion", "class": class_name(value)}
+            self.active_containers.add(object_id)
+            items = []
+            for key, item in sorted(value.items(), key=lambda pair: repr(pair[0])):
+                items.append({"key": self.dump(key, depth + 1), "value": self.dump(item, depth + 1)})
+            self.active_containers.remove(object_id)
+            return {"type": "dict", "items": items}
+
+        if isinstance(value, type):
+            return {
+                "type": "repr",
+                "class": class_name(value),
+                "repr": repr(value),
+            }
+
+        identity = id(value)
+        if identity in self.seen:
+            return {"type": "ref", "id": self.seen[identity]}
+
+        dump_id = self.next_id
+        self.next_id += 1
+        self.seen[identity] = dump_id
+
+        attributes = {}
+        for slot in collect_slots(value):
+            if slot.startswith("__"):
+                continue
+            try:
+                slot_value = getattr(value, slot)
+            except Exception as error:
+                attributes[slot] = {
+                    "type": "attribute-error",
+                    "class": class_name(error),
+                    "repr": repr(error),
+                }
+                continue
+            attributes[slot] = self.dump(slot_value, depth + 1)
+
+        value_dict = getattr(value, "__dict__", None)
+        if value_dict:
+            for key, item in sorted(value_dict.items(), key=lambda pair: repr(pair[0])):
+                if isinstance(key, basestring):
+                    key_text = safe_text(key)
+                else:
+                    key_text = repr(key)
+                attributes[key_text] = self.dump(item, depth + 1)
+
+        if not attributes:
+            del self.seen[identity]
+            return {
+                "type": "repr",
+                "class": class_name(value),
+                "repr": repr(value),
+            }
+
+        return {
+            "type": "object",
+            "id": dump_id,
+            "class": class_name(value),
+            "attributes": attributes,
+        }
+
+
+def write_json(path, document):
+    parent = os.path.dirname(path)
+    if parent and not os.path.isdir(parent):
+        os.makedirs(parent)
+
+    data = json.dumps(document, indent=2, sort_keys=True)
+    if not isinstance(data, bytes):
+        data = data.encode("utf-8")
+
+    with open(path, "wb") as handle:
+        handle.write(data)
+        handle.write(b"\n")
+
+
+def import_renpy(game_root):
+    sys.path.insert(0, game_root)
+
+    import renpy
+    renpy.import_all()
+
+    class Dummy(object):
+        pass
+
+    renpy.game.script = Dummy()
+    renpy.game.script.all_pyexpr = None
+    renpy.game.script.record_pycode = False
+    renpy.game.log = Dummy()
+    renpy.game.log.mutated = {}
+
+    return renpy
+
+
+def main(argv):
+    args = parse_args(argv)
+    renpy = import_renpy(args.game_root)
+    slot, metadata, roots = load_compiled_script(args.input)
+
+    dumper = Dumper(args.max_depth)
+    document = {
+        "schema": SCHEMA,
+        "source": os.path.abspath(args.input),
+        "game_root": os.path.abspath(args.game_root),
+        "slot": slot,
+        "python_version": sys.version,
+        "renpy_version": getattr(renpy, "version", None),
+        "metadata": dumper.dump(metadata),
+        "roots": dumper.dump(roots),
+    }
+    write_json(args.output, document)
+
+
+if __name__ == "__main__":
+    main(sys.argv[1:])