236 lines
6.9 KiB
Python
236 lines
6.9 KiB
Python
from __future__ import annotations
|
|
|
|
import argparse
|
|
import logging
|
|
import os
|
|
import shutil
|
|
import sys
|
|
import textwrap
|
|
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="Extract Ren'Py game files.",
|
|
formatter_class=argparse.RawDescriptionHelpFormatter,
|
|
epilog=textwrap.dedent(
|
|
"""\
|
|
examples:
|
|
clew --game-root /path/to/game --force
|
|
clew --game-root /path/to/game --output-root /tmp/clew --force
|
|
clew --game-root /path/to/game --suffix rpy --suffix rpyc --force
|
|
"""
|
|
),
|
|
)
|
|
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. Defaults to repository output/.",
|
|
)
|
|
parser.add_argument(
|
|
"--force",
|
|
action="store_true",
|
|
help="Replace Clew-managed output directories.",
|
|
)
|
|
parser.add_argument(
|
|
"--suffix",
|
|
action="append",
|
|
help="Extract only archive paths with this suffix. Repeatable.",
|
|
)
|
|
return parser.parse_args(argv)
|
|
|
|
|
|
def default_output_root() -> Path:
|
|
workspace = os.environ.get("BUILD_WORKSPACE_DIRECTORY")
|
|
if workspace:
|
|
return Path(workspace) / "output"
|
|
return Path(__file__).resolve().parent / "output"
|
|
|
|
|
|
def fail(message: str) -> None:
|
|
raise SystemExit(f"clew: {message}")
|
|
|
|
|
|
def resolve_output_root(output_root: Path) -> Path:
|
|
if output_root.is_absolute():
|
|
return output_root.resolve()
|
|
|
|
workspace = os.environ.get("BUILD_WORKSPACE_DIRECTORY")
|
|
if workspace:
|
|
return (Path(workspace) / output_root).resolve()
|
|
|
|
return output_root.resolve()
|
|
|
|
|
|
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 copy_compiled_file(compiled_file: Path, input_root: Path, work_dir: Path) -> None:
|
|
relative = relative_to(input_root, compiled_file)
|
|
if is_renpy_runtime_path(relative):
|
|
LOGGER.info("skipped Ren'Py runtime compiled script: %s", relative)
|
|
return
|
|
|
|
destination = work_dir / "compiled" / relative
|
|
destination.parent.mkdir(parents=True, exist_ok=True)
|
|
shutil.copy2(compiled_file, destination)
|
|
LOGGER.info("copied compiled script: %s", 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 normalize_suffixes(suffixes: list[str] | None) -> list[str] | None:
|
|
if suffixes is None:
|
|
return None
|
|
|
|
normalized = []
|
|
for suffix in suffixes:
|
|
if not suffix:
|
|
fail("suffix must not be empty")
|
|
if not suffix.startswith("."):
|
|
suffix = f".{suffix}"
|
|
normalized.append(suffix)
|
|
return normalized
|
|
|
|
|
|
def extract_archive(
|
|
archive: Path,
|
|
game_root: Path,
|
|
source_dir: Path,
|
|
work_dir: Path,
|
|
suffixes: list[str] | None,
|
|
) -> 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=suffixes)
|
|
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 = resolve_output_root(args.output_root)
|
|
|
|
if not game_root.is_dir():
|
|
fail(f"game root is not a directory: {game_root}")
|
|
|
|
suffixes = normalize_suffixes(args.suffix)
|
|
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 loose compiled scripts")
|
|
for suffix in ("*.rpyc", "*.rpymc"):
|
|
for compiled_file in sorted(game_root.rglob(suffix)):
|
|
copy_compiled_file(compiled_file, game_root, work_dir)
|
|
|
|
LOGGER.info("scanning archives")
|
|
for archive in sorted(game_root.rglob("*.rpa")):
|
|
extract_archive(archive, game_root, source_dir, work_dir, suffixes)
|
|
|
|
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()
|