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 by default;
``run-checks`` (or ``release github``) runs the manifest's ``checks`` commands
with ``{python}`` replaced by the running interpreter.
"""

from __future__ import annotations

import argparse
import json
import shutil
import subprocess
import sys
from dataclasses import dataclass
from pathlib import Path


MANIFEST_NAME = "downstream_sync_manifest.json"


@dataclass(frozen=True)
class SyncJob:
    """One source -> destination copy, with optional name patterns to skip."""

    source: Path
    dest: Path
    exclude: tuple[str, ...] = ()


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[SyncJob]:
    dest_root = (root / item["path"]).resolve()
    jobs: list[SyncJob] = []
    for entry in item.get("sync", []):
        src_base = root / entry["from"]
        dest_base = dest_root / entry["to"]
        exclude = tuple(entry.get("exclude") or ())
        includes = entry.get("include")
        if includes:
            for name in includes:
                jobs.append(SyncJob(src_base / name, dest_base / name, exclude))
        else:
            jobs.append(SyncJob(src_base, dest_base, exclude))
    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 job in jobs:
                note = f"  (excluding {', '.join(job.exclude)})" if job.exclude else ""
                print(f"  {job.source} -> {job.dest}{note}")
        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 job in expand_jobs(root, item):
            src_ok = job.source.exists()
            dest_note = "exists" if job.dest.exists() else "missing (ok until apply)"
            status = "ok  " if src_ok else "MISS"
            print(f"{status} source: {job.source}")
            print(f"     dest:   {job.dest} [{dest_note}]")
            if not src_ok:
                missing.append(str(job.source))
        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(job: SyncJob) -> None:
    if not job.source.exists():
        raise FileNotFoundError(job.source)
    if job.source.is_dir():
        # exclude keeps caches out of the public repo (e.g. __pycache__, *.pyc)
        ignore = shutil.ignore_patterns(*job.exclude) if job.exclude else None
        shutil.copytree(job.source, job.dest, dirs_exist_ok=True, ignore=ignore)
        return
    job.dest.parent.mkdir(parents=True, exist_ok=True)
    shutil.copy2(job.source, job.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 job in expand_jobs(root, item):
            copy_job(job)
            copied += 1
            print(f"copied {job.source} -> {job.dest}")
        print()
    print(f"copied {copied} sync job(s).")
    print("checks were not run; execute the listed check commands manually.")
    return 0


def run_checks(root: Path, items: list[dict], *, python: str | None = None) -> int:
    """Run each manifest ``checks`` command inside the downstream repo.

    ``{python}`` in a command is replaced with ``python`` (or the interpreter
    given). Returns the number of failed checks.
    """
    interpreter = python or sys.executable
    failures = 0
    for item in items:
        dest_root = (root / item["path"]).resolve()
        checks = item.get("checks") or []
        if not checks:
            continue
        print(f"== {item['name']} ==")
        for command in checks:
            resolved = command.replace("{python}", interpreter)
            print(f"running: {resolved}")
            completed = subprocess.run(
                resolved, shell=True, cwd=dest_root, check=False
            )
            if completed.returncode != 0:
                print(f"[FAIL] exit {completed.returncode}: {resolved}")
                failures += 1
            else:
                print("[ok  ] check passed")
        print()
    return failures


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")

    run = sub.add_parser("run-checks", help="run the manifest checks in each downstream")
    run.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)
    if args.command == "run-checks":
        return 1 if run_checks(root, items) else 0
    raise SystemExit(f"unknown command: {args.command}")


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