From a96d3cdd16ce6baac3b89dbe1e7fd56403a85797 Mon Sep 17 00:00:00 2001 From: alexion Date: Mon, 7 Sep 2026 15:31:33 -0400 Subject: [PATCH] feat(clew): copy game directory tree --- src/clew/clew.py | 129 ++++++++++------------------------------------- 1 file changed, 27 insertions(+), 102 deletions(-) diff --git a/src/clew/clew.py b/src/clew/clew.py index 6db58e4..4fe07a0 100644 --- a/src/clew/clew.py +++ b/src/clew/clew.py @@ -5,14 +5,10 @@ import logging import os import shutil import sys -from datetime import UTC, datetime +from collections import deque from pathlib import Path -from rpycdec.rpa import extract_rpa - -LOGGER = logging.getLogger("clew") -MANAGED_DIRS = ("extracted",) - +LOG = logging.getLogger(__name__) def parse_args(argv: list[str]) -> argparse.Namespace: parser = argparse.ArgumentParser( @@ -55,91 +51,25 @@ def resolve_output_root(output_root: Path) -> Path: return output_root.resolve() -def prepare_output(output_root: Path, overwrite: bool) -> Path: - for name in MANAGED_DIRS: - path = output_root / name - if path.exists() and not overwrite: - LOGGER.critical("refusing to overwrite existing output path: %s", path) - raise SystemExit(2) - - if overwrite: - for name in MANAGED_DIRS: - shutil.rmtree(output_root / name, ignore_errors=True) - - return output_root / "extracted" +def get_output_directory(output_root: Path, folder: str) -> Path: + directory = output_root / folder + directory.mkdir(parents=True, exist_ok=True) + return directory -def get_folder(path: Path) -> Path: - path.mkdir(parents=True, exist_ok=True) - return path +def extract_game_files(game_root: Path, output_path: Path) -> None: + queue = deque(game_root.iterdir()) + while queue: + item = queue.popleft() + destination = output_path / item.relative_to(game_root) -class LazyFileHandler(logging.FileHandler): - def __init__(self, path: Path, encoding: str) -> None: - super().__init__(path, encoding=encoding, delay=True) - - def _open(self): - get_folder(Path(self.baseFilename).parent) - return super()._open() - - -def relative_to(root: Path, path: Path) -> Path: - return path.resolve().relative_to(root.resolve()) - - -def normalize_suffixes(suffixes: list[str] | None) -> list[str] | None: - if suffixes is None: - return None - - normalized = [] - for suffix in suffixes: - if not suffix: - LOGGER.critical("suffix must not be empty") - raise SystemExit(2) - if not suffix.startswith("."): - suffix = f".{suffix}" - normalized.append(suffix) - return normalized - - -def copy_loose_file(file: Path, game_root: Path, extracted_dir: Path) -> None: - relative = relative_to(game_root, file) - destination = extracted_dir / relative - get_folder(destination.parent) - shutil.copy2(file, destination) - LOGGER.info("copied loose file: %s", relative) - - -def extract_archive( - archive: Path, - game_root: Path, - extracted_dir: Path, - suffixes: list[str] | None, -) -> None: - archive_relative = relative_to(game_root, archive) - output_dir = get_folder(extracted_dir / archive_relative.with_suffix("")) - - 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) - - -def configure_logging(output_root: Path) -> None: - log_level = logging.INFO - timestamp = datetime.now(UTC).strftime("%Y%m%dT%H%M%SZ") - logs_dir = output_root / "logs" - log_file = logs_dir / f"{logging.getLevelName(log_level)}-clew-{timestamp}.log" - - formatter = logging.Formatter("%(message)s") - - file_handler = LazyFileHandler(log_file, encoding="utf-8") - file_handler.setFormatter(formatter) - - stream_handler = logging.StreamHandler() - stream_handler.setFormatter(formatter) - - logging.basicConfig(level=log_level, handlers=[file_handler, stream_handler]) + if item.is_dir(): + destination.mkdir(parents=True, exist_ok=True) + queue.extend(item.iterdir()) + elif item.is_file(): + destination.parent.mkdir(parents=True, exist_ok=True) + shutil.copy2(item, destination) def run(args: argparse.Namespace) -> None: @@ -147,25 +77,20 @@ def run(args: argparse.Namespace) -> None: output_root = resolve_output_root(args.output_root) if not game_root.is_dir(): - LOGGER.critical("game root is not a directory: %s", game_root) + LOG.critical(f'game root is not a directory: {game_root}') raise SystemExit(2) - suffixes = normalize_suffixes(args.suffix) - extracted_dir = prepare_output(output_root, args.overwrite) - - LOGGER.info("copying loose files") - for file in sorted(game_root.rglob("*")): - if file.is_file() and file.suffix.lower() != ".rpa": - copy_loose_file(file, game_root, extracted_dir) - - LOGGER.info("extracting archives") - for archive in sorted(game_root.rglob("*.rpa")): - extract_archive(archive, game_root, extracted_dir, suffixes) - - LOGGER.info("done") + extracted = get_output_directory(output_root, "extracted") + extract_game_files(game_root, extracted) if __name__ == "__main__": args = parse_args(sys.argv[1:]) - configure_logging(resolve_output_root(args.output_root)) + + logging.basicConfig( + level=logging.INFO, + format="[%(asctime)s] %(levelname)s %(filename)s:%(lineno)d | %(message)s", + datefmt="%Y-%m-%d][%H:%M:%S", + ) + run(args)