orisys-sdk/tools/sync_downstreams.py

#!/usr/bin/env python3
"""Plan and apply downstream file syncs from tools/downstream_sync_manifest.json.

The ``apply`` command copies files/directories listed in the manifest. It does
not delete extra downstream files and does not run downstream tests.
"""

from __future__ import annotations

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


MANIFEST_NAME = "downstream_sync_manifest.json"


def sdk_root() -> Path:
    return Path(__file__).resolve().parent.parent


def load_manifest(root: Path) -> dict:
    path = Path(__file__).resolve().parent / MANIFEST_NAME
    with path.open(encoding="utf-8") as handle:
        data = json.load(handle)
    if data.get("version") != 1:
        raise SystemExit(f"unsupported manifest version: {data.get('version')}")
    if not isinstance(data.get("downstreams"), list):
        raise SystemExit("manifest is missing a downstreams list")
    return data


def selected_downstreams(manifest: dict, target: str | None) -> list[dict]:
    items = manifest["downstreams"]
    if target is None:
        return items
    matches = [item for item in items if item.get("name") == target]
    if not matches:
        names = ", ".join(item.get("name", "<unnamed>") for item in items)
        raise SystemExit(f"unknown target {target!r}; known: {names}")
    return matches


def expand_jobs(root: Path, item: dict) -> list[tuple[Path, Path]]:
    dest_root = (root / item["path"]).resolve()
    jobs: list[tuple[Path, Path]] = []
    for entry in item.get("sync", []):
        src_base = root / entry["from"]
        dest_base = dest_root / entry["to"]
        includes = entry.get("include")
        if includes:
            for name in includes:
                jobs.append((src_base / name, dest_base / name))
        else:
            jobs.append((src_base, dest_base))
    return jobs


def print_plan(root: Path, items: list[dict]) -> None:
    print(f"SDK root: {root}")
    print()
    for item in items:
        dest_root = (root / item["path"]).resolve()
        print(f"== {item['name']} ==")
        print(f"path: {dest_root}")
        jobs = expand_jobs(root, item)
        if jobs:
            print("sync:")
            for src, dest in jobs:
                print(f"  {src} -> {dest}")
        else:
            print("sync: (none)")
        checks = item.get("checks") or []
        if checks:
            print("checks:")
            for command in checks:
                print(f"  {command}")
        else:
            print("checks: (none)")
        print()


def check_paths(root: Path, items: list[dict]) -> int:
    missing: list[str] = []
    print(f"SDK root: {root}")
    print()
    for item in items:
        dest_root = (root / item["path"]).resolve()
        print(f"== {item['name']} ==")
        if dest_root.is_dir():
            print(f"ok   downstream root: {dest_root}")
        else:
            print(f"MISS downstream root: {dest_root}")
            missing.append(str(dest_root))
        for src, dest in expand_jobs(root, item):
            src_ok = src.exists()
            dest_note = "exists" if dest.exists() else "missing (ok until apply)"
            status = "ok  " if src_ok else "MISS"
            print(f"{status} source: {src}")
            print(f"     dest:   {dest} [{dest_note}]")
            if not src_ok:
                missing.append(str(src))
        print()
    if missing:
        print(f"{len(missing)} missing path(s).")
        return 1
    print("all required source and downstream roots exist.")
    return 0


def copy_job(src: Path, dest: Path) -> None:
    if not src.exists():
        raise FileNotFoundError(src)
    if src.is_dir():
        shutil.copytree(src, dest, dirs_exist_ok=True)
        return
    dest.parent.mkdir(parents=True, exist_ok=True)
    shutil.copy2(src, dest)


def apply_sync(root: Path, items: list[dict]) -> int:
    if check_paths(root, items) != 0:
        print("apply aborted because required paths are missing.")
        return 1

    copied = 0
    for item in items:
        print(f"== {item['name']} ==")
        for src, dest in expand_jobs(root, item):
            copy_job(src, dest)
            copied += 1
            print(f"copied {src} -> {dest}")
        print()
    print(f"copied {copied} sync job(s).")
    print("checks were not run; execute the listed check commands manually.")
    return 0


def parse_args(argv: list[str]) -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__)
    sub = parser.add_subparsers(dest="command", required=True)

    plan = sub.add_parser("plan", help="print intended sync jobs and checks")
    plan.add_argument("--target", help="limit to one downstream name")

    check = sub.add_parser("check-paths", help="verify source and dest roots exist")
    check.add_argument("--target", help="limit to one downstream name")

    apply_cmd = sub.add_parser("apply", help="copy manifest sources to downstreams")
    apply_cmd.add_argument("--target", help="limit to one downstream name")

    return parser.parse_args(argv)


def main(argv: list[str] | None = None) -> int:
    args = parse_args(argv if argv is not None else sys.argv[1:])
    root = sdk_root()
    items = selected_downstreams(load_manifest(root), getattr(args, "target", None))
    if args.command == "plan":
        print_plan(root, items)
        return 0
    if args.command == "check-paths":
        return check_paths(root, items)
    if args.command == "apply":
        return apply_sync(root, items)
    raise SystemExit(f"unknown command: {args.command}")


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