From bd8417259c61a926b8aa6609a31be35567ef7fa0 Mon Sep 17 00:00:00 2001 From: alexion Date: Mon, 7 Sep 2026 17:35:07 -0400 Subject: [PATCH] feat(clew): dump Ren'Py compiled script objects --- AGENTS.md | 2 + src/clew/BUILD.bazel | 1 + src/clew/clew.py | 101 +++++++++++-- src/clew/renpy_object_dump.py | 258 ++++++++++++++++++++++++++++++++++ 4 files changed, 354 insertions(+), 8 deletions(-) create mode 100644 src/clew/renpy_object_dump.py diff --git a/AGENTS.md b/AGENTS.md index bd8749d..169ea08 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -98,3 +98,5 @@ Do not use `nix fmt` as a validation command until a formatter is added. - Ariadne extraction must preserve the game-root season prefix for archived source. Different game installs can contain compiled scripts with the same internal path. +- Clew owns Ren'Py extraction, preparation, and compiled-script AST export. Ariadne owns analysis over Clew-produced data. +- Bazel Python targets require `PYTHON_TOOLCHAIN_ROOT` to point at a Python installation root. diff --git a/src/clew/BUILD.bazel b/src/clew/BUILD.bazel index 06a8412..c464f89 100644 --- a/src/clew/BUILD.bazel +++ b/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"], ) diff --git a/src/clew/clew.py b/src/clew/clew.py index 76a3723..dffee11 100644 --- a/src/clew/clew.py +++ b/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__": diff --git a/src/clew/renpy_object_dump.py b/src/clew/renpy_object_dump.py new file mode 100644 index 0000000..d206798 --- /dev/null +++ b/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:])