nanobot-voice-interface/scripts/sync_card_templates.py

#!/usr/bin/env python3
from __future__ import annotations

import argparse
import filecmp
import os
import shutil
from pathlib import Path


LIVE_TEMPLATES_DIR = Path.home() / ".nanobot" / "cards" / "templates"
EXAMPLE_TEMPLATES_DIR = Path(__file__).resolve().parent.parent / "examples" / "cards" / "templates"


def iter_template_dirs(root: Path) -> dict[str, Path]:
    templates: dict[str, Path] = {}
    if not root.exists():
        return templates
    for child in sorted(root.iterdir()):
        if not child.is_dir():
            continue
        if not (child / "manifest.json").exists():
            continue
        templates[child.name] = child
    return templates


def sync_directory(source: Path, dest: Path) -> None:
    dest.mkdir(parents=True, exist_ok=True)

    source_entries = {path.name: path for path in source.iterdir()}
    dest_entries = {path.name: path for path in dest.iterdir()} if dest.exists() else {}

    for name, source_path in source_entries.items():
        dest_path = dest / name
        if source_path.is_dir():
            sync_directory(source_path, dest_path)
            continue
        dest_path.parent.mkdir(parents=True, exist_ok=True)
        if not dest_path.exists() or not filecmp.cmp(source_path, dest_path, shallow=False):
            shutil.copy2(source_path, dest_path)

    for name, dest_path in dest_entries.items():
        if name in source_entries:
            continue
        if dest_path.is_dir():
            shutil.rmtree(dest_path)
        else:
            dest_path.unlink(missing_ok=True)


def compare_directory(source: Path, dest: Path, drifts: list[str], *, prefix: str) -> None:
    source_entries = {path.name: path for path in source.iterdir()} if source.exists() else {}
    dest_entries = {path.name: path for path in dest.iterdir()} if dest.exists() else {}

    for name, source_path in source_entries.items():
        rel = f"{prefix}/{name}" if prefix else name
        dest_path = dest / name
        if name not in dest_entries:
            drifts.append(f"missing mirror: {rel}")
            continue
        if source_path.is_dir():
            if not dest_path.is_dir():
                drifts.append(f"type mismatch: {rel}")
                continue
            compare_directory(source_path, dest_path, drifts, prefix=rel)
            continue
        if dest_path.is_dir():
            drifts.append(f"type mismatch: {rel}")
            continue
        if not filecmp.cmp(source_path, dest_path, shallow=False):
            drifts.append(f"content drift: {rel}")

    for name in dest_entries:
        if name not in source_entries:
            rel = f"{prefix}/{name}" if prefix else name
            drifts.append(f"stale mirror: {rel}")


def run_check(source_root: Path, dest_root: Path) -> int:
    drifts: list[str] = []
    source_templates = iter_template_dirs(source_root)
    dest_templates = iter_template_dirs(dest_root)

    for name, source_dir in source_templates.items():
        dest_dir = dest_root / name
        if name not in dest_templates:
            drifts.append(f"missing mirror: {name}")
            continue
        compare_directory(source_dir, dest_dir, drifts, prefix=name)

    for name in dest_templates:
        if name not in source_templates:
            drifts.append(f"stale mirror: {name}")

    if drifts:
        for drift in drifts:
            print(drift)
        return 1
    print("template sync ok")
    return 0


def run_sync(source_root: Path, dest_root: Path) -> int:
    source_templates = iter_template_dirs(source_root)
    dest_root.mkdir(parents=True, exist_ok=True)

    for name, source_dir in source_templates.items():
        sync_directory(source_dir, dest_root / name)

    for child in list(dest_root.iterdir()):
        if child.name in source_templates:
            continue
        if child.is_dir():
            shutil.rmtree(child)
        else:
            child.unlink(missing_ok=True)

    print(f"synced {len(source_templates)} template directories")
    return 0


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--check", action="store_true")
    parser.add_argument("--source", type=Path, default=LIVE_TEMPLATES_DIR)
    parser.add_argument("--dest", type=Path, default=EXAMPLE_TEMPLATES_DIR)
    args = parser.parse_args()

    source_root = args.source.expanduser()
    dest_root = args.dest.expanduser()

    if args.check:
        return run_check(source_root, dest_root)
    return run_sync(source_root, dest_root)


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