feat(clew): extract Ren'Py source files

src/clew/BUILD.bazel

@@ -0,0 +1,11 @@
+load("@rules_python//python:defs.bzl", "py_binary")
+
+py_binary(
+    name = "clew",
+    srcs = [
+        "__init__.py",
+        "__main__.py",
+    ],
+    main = "__main__.py",
+    deps = ["//third_party/python:rpycdec"],
+)

src/clew/README.md

@@ -0,0 +1,17 @@
+# Clew
+
+Clew prepares Ren'Py game artifacts for Ariadne.
+
+Run Clew with Bazel:
+
+```sh
+bazel run //src/clew -- \
+  --game-root /path/to/game \
+  --output-root src/ariadne/output \
+  --force
+```
+
+The first version copies loose `.rpy` files and extracts `.rpy` files from `.rpa` archives.
+It writes output under `source/`, `work/`, and `logs/`.
+
+Use `--force` to replace those Clew-managed directories.

src/clew/__init__.py

@@ -0,0 +1 @@
+"""Ren'Py game preparation tool for Ariadne."""

src/clew/__main__.py

@@ -0,0 +1,165 @@
+from __future__ import annotations
+
+import argparse
+import logging
+import os
+import shutil
+import sys
+from pathlib import Path
+
+from rpycdec.rpa import extract_rpa
+
+LOGGER = logging.getLogger("clew")
+MANAGED_DIRS = ("source", "work", "logs")
+
+
+def parse_args(argv: list[str]) -> argparse.Namespace:
+    parser = argparse.ArgumentParser(
+        prog="clew",
+        description="Prepare Ren'Py game artifacts for Ariadne.",
+    )
+    parser.add_argument(
+        "--game-root",
+        required=True,
+        type=Path,
+        help="Game installation root to scan.",
+    )
+    parser.add_argument(
+        "--output-root",
+        type=Path,
+        default=default_output_root(),
+        help="Preparation output root.",
+    )
+    parser.add_argument(
+        "--force",
+        action="store_true",
+        help="Replace Clew-managed output directories.",
+    )
+    return parser.parse_args(argv)
+
+
+def default_output_root() -> Path:
+    workspace = os.environ.get("BUILD_WORKSPACE_DIRECTORY")
+    if workspace:
+        return Path(workspace) / "src" / "ariadne" / "output"
+    return Path(__file__).resolve().parent / "output"
+
+
+def fail(message: str) -> None:
+    raise SystemExit(f"clew: {message}")
+
+
+def prepare_output(output_root: Path, force: bool) -> tuple[Path, Path, Path]:
+    for name in MANAGED_DIRS:
+        path = output_root / name
+        if path.exists() and not force:
+            fail(f"refusing to overwrite existing preparation path: {path}")
+
+    if force:
+        for name in MANAGED_DIRS:
+            shutil.rmtree(output_root / name, ignore_errors=True)
+
+    source_dir = output_root / "source"
+    work_dir = output_root / "work"
+    logs_dir = output_root / "logs"
+    for path in (source_dir, work_dir, logs_dir):
+        path.mkdir(parents=True, exist_ok=True)
+
+    return source_dir, work_dir, logs_dir
+
+
+def relative_to(root: Path, path: Path) -> Path:
+    return path.resolve().relative_to(root.resolve())
+
+
+def is_renpy_runtime_path(relative: Path) -> bool:
+    parts = relative.parts
+    return bool(parts) and (parts[0] == "renpy" or "renpy" in parts[:-1])
+
+
+def copy_source_file(
+    source_file: Path,
+    input_root: Path,
+    source_dir: Path,
+    label: str,
+    destination_prefix: Path = Path(),
+) -> None:
+    relative = relative_to(input_root, source_file)
+    output_relative = destination_prefix / relative
+    if is_renpy_runtime_path(relative):
+        LOGGER.info("skipped %s Ren'Py runtime source: %s", label, output_relative)
+        return
+
+    destination = source_dir / output_relative
+    destination.parent.mkdir(parents=True, exist_ok=True)
+    shutil.copy2(source_file, destination)
+    LOGGER.info("copied %s source: %s", label, output_relative)
+
+
+def archive_work_path(work_dir: Path, archive_relative: Path) -> Path:
+    return work_dir / "archives" / archive_relative.with_suffix("")
+
+
+def archive_destination_prefix(archive_relative: Path) -> Path:
+    parent = archive_relative.parent
+    if str(parent) == ".":
+        return Path()
+    return parent
+
+
+def extract_archive(archive: Path, game_root: Path, source_dir: Path, work_dir: Path) -> None:
+    archive_relative = relative_to(game_root, archive)
+    output_dir = archive_work_path(work_dir, archive_relative)
+    output_dir.mkdir(parents=True, exist_ok=True)
+
+    LOGGER.info("extracting archive: %s", archive_relative)
+    with archive.open("rb") as handle:
+        extract_rpa(handle, output_dir=str(output_dir), suffixes=[".rpy"])
+    LOGGER.info("extracted archive: %s", archive_relative)
+
+    destination_prefix = archive_destination_prefix(archive_relative)
+    for source_file in sorted(output_dir.rglob("*.rpy")):
+        copy_source_file(source_file, output_dir, source_dir, "archived", destination_prefix)
+
+
+def configure_logging(logs_dir: Path) -> None:
+    log_file = logs_dir / "clew.log"
+    formatter = logging.Formatter("%(message)s")
+
+    file_handler = logging.FileHandler(log_file, encoding="utf-8")
+    file_handler.setFormatter(formatter)
+
+    stream_handler = logging.StreamHandler()
+    stream_handler.setFormatter(formatter)
+
+    logging.basicConfig(level=logging.INFO, handlers=[file_handler, stream_handler])
+
+
+def run(args: argparse.Namespace) -> None:
+    game_root = args.game_root.resolve()
+    output_root = args.output_root.resolve()
+
+    if not game_root.is_dir():
+        fail(f"game root is not a directory: {game_root}")
+
+    source_dir, work_dir, logs_dir = prepare_output(output_root, args.force)
+    configure_logging(logs_dir)
+
+    LOGGER.info("scanning loose source files")
+    for source_file in sorted(game_root.rglob("*.rpy")):
+        copy_source_file(source_file, game_root, source_dir, "loose")
+
+    LOGGER.info("scanning archives")
+    for archive in sorted(game_root.rglob("*.rpa")):
+        extract_archive(archive, game_root, source_dir, work_dir)
+
+    LOGGER.info("done")
+
+
+def main(argv: list[str] | None = None) -> None:
+    args = parse_args(sys.argv[1:] if argv is None else argv)
+    run(args)
+
+
+if __name__ == "__main__":
+    main()