CASAN/scripts/copy-runtime.py

#!/usr/bin/env python3
"""Copy one CASAN production component through the canonical runtime allowlist."""

from __future__ import annotations

import argparse
import json
from pathlib import Path
import shutil
import sys


IGNORED_NAMES = {
    "__pycache__", ".DS_Store", "node_modules", "dist", "build", "coverage",
}
IGNORED_SUFFIXES = {".pyc", ".pyo", ".log", ".tmp"}


class LayoutError(RuntimeError):
    pass


def load_layout(source_root: Path) -> dict:
    path = source_root / "packaging" / "runtime-layout.json"
    try:
        layout = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, ValueError) as error:
        raise LayoutError(f"cannot load {path}: {error}") from error
    if layout.get("schema_version") != 1:
        raise LayoutError("runtime layout schema_version must be 1")
    return layout


def is_excluded(relative: Path, exclusions: set[str]) -> bool:
    normalized = relative.as_posix()
    if any(part in IGNORED_NAMES for part in relative.parts):
        return True
    if relative.suffix in IGNORED_SUFFIXES:
        return True
    return any(
        normalized == excluded or normalized.startswith(excluded.rstrip("/") + "/")
        for excluded in exclusions
    )


def transformed_content(path: Path, transform: str) -> bytes:
    if transform != "empty_project_registry":
        raise LayoutError(f"unknown runtime transform: {transform}")
    try:
        payload = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, ValueError) as error:
        raise LayoutError(f"cannot transform project registry: {error}") from error
    payload["projects"] = []
    return (json.dumps(payload, ensure_ascii=False, indent=2) + "\n").encode("utf-8")


def copy_component(
        source_root: Path,
        destination_root: Path,
        name: str,
        preserved: set[str] | None = None,
        clean: bool = False) -> dict:
    layout = load_layout(source_root)
    component = layout.get("components", {}).get(name)
    if not isinstance(component, dict):
        raise LayoutError(f"unknown runtime component: {name}")

    relative_source = Path(str(component.get("source", "")))
    source = (source_root / relative_source).resolve()
    destination = destination_root / relative_source
    if not source.is_dir():
        raise LayoutError(f"component source is missing: {source}")
    if source_root not in source.parents:
        raise LayoutError(f"component source escapes source root: {source}")

    exclusions = {
        str(path).replace("\\", "/").strip("/")
        for path in component.get("exclude", [])
        if isinstance(path, str) and path.strip("/")
    }
    unsafe = [
        path for path in exclusions
        if Path(path).is_absolute() or ".." in Path(path).parts
    ]
    if unsafe:
        raise LayoutError(
            f"runtime component {name} has unsafe exclusions: {', '.join(unsafe)}")
    transforms = component.get("transforms", {})
    if not isinstance(transforms, dict):
        raise LayoutError(f"runtime component {name} transforms must be an object")
    removed = 0
    if clean and destination.exists():
        for excluded in sorted(exclusions, reverse=True):
            candidate = destination / excluded
            if candidate.is_dir() and not candidate.is_symlink():
                shutil.rmtree(candidate)
                removed += 1
            elif candidate.exists() or candidate.is_symlink():
                candidate.unlink()
                removed += 1
        for candidate in sorted(destination.rglob("*"), reverse=True):
            if (candidate.name in IGNORED_NAMES or
                    (candidate.is_file() and candidate.suffix in IGNORED_SUFFIXES)):
                if candidate.is_dir() and not candidate.is_symlink():
                    shutil.rmtree(candidate)
                else:
                    candidate.unlink()
                removed += 1
    copied = 0
    skipped = 0
    preserved = preserved or set()
    for path in sorted(source.rglob("*")):
        relative = path.relative_to(source)
        if is_excluded(relative, exclusions):
            skipped += path.is_file()
            continue
        target = destination / relative
        if path.is_dir():
            continue
        if not path.is_file():
            continue
        target_relative = (relative_source / relative).as_posix()
        if target_relative in preserved and target.exists():
            skipped += 1
            continue
        target.parent.mkdir(parents=True, exist_ok=True)
        transform = transforms.get(relative.as_posix())
        if transform:
            target.write_bytes(transformed_content(path, str(transform)))
            shutil.copymode(path, target)
        else:
            shutil.copy2(path, target)
        copied += 1

    if clean and destination.exists():
        for candidate in sorted(destination.rglob("*"), reverse=True):
            if candidate.is_dir() and not candidate.is_symlink():
                try:
                    candidate.rmdir()
                except OSError:
                    pass

    missing = [
        required for required in component.get("required", [])
        if not (destination / required).is_file()
    ]
    if missing:
        raise LayoutError(
            f"runtime component {name} is incomplete; missing: {', '.join(missing)}")
    return {
        "component": name,
        "source": str(source),
        "destination": str(destination),
        "copied_files": copied,
        "excluded_files": skipped,
        "removed_paths": removed,
    }


def parser() -> argparse.ArgumentParser:
    value = argparse.ArgumentParser(description=__doc__)
    value.add_argument("--source-root", required=True)
    value.add_argument("--destination-root", required=True)
    value.add_argument("--component", action="append", required=True)
    value.add_argument(
        "--preserve", action="append", default=[],
        help="destination-root-relative file to retain when it already exists")
    value.add_argument(
        "--clean", action="store_true",
        help="remove excluded source-only paths left by an older installation")
    value.add_argument("--json", action="store_true")
    return value


def main(argv: list[str] | None = None) -> int:
    args = parser().parse_args(argv)
    source_root = Path(args.source_root).expanduser().resolve()
    destination_root = Path(args.destination_root).expanduser().resolve()
    destination_root.mkdir(parents=True, exist_ok=True)
    preserved = {
        str(path).replace("\\", "/").strip("/")
        for path in args.preserve
        if str(path).strip("/")
    }
    try:
        results = [
            copy_component(
                source_root, destination_root, component, preserved, args.clean)
            for component in args.component
        ]
    except LayoutError as error:
        sys.stderr.write(f"copy-runtime: {error}\n")
        return 1
    if args.json:
        print(json.dumps({"status": "ok", "components": results}, indent=2))
    else:
        for result in results:
            print(
                "copy-runtime: {component} copied={copied_files} "
                "excluded={excluded_files} removed={removed_paths}".format(**result))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())