feat(clew): copy game directory tree
This commit is contained in:
129
src/clew/clew.py
129
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)
|
||||
|
||||
Reference in New Issue
Block a user