replica-omnisciente/scripts/generate-provider-mirrors.py

#!/usr/bin/env python3
"""
Generate downstream provider mirrors from the single source of truth:

    .aurelio/providers/registry.yaml

Outputs (under .aurelio/providers/dist/ unless --write-in-place is given):

  * portal.providers.json   — consumable by porta.portugalfuturista.org /
                              aurelio-theia ModelRouter.
  * mcp.providers.json      — machine-readable form the Python model_router
                              can load instead of its hard-coded registry.
  * dirac.providers.json    — {list:[{value,label}]} matching Dirac's
                              src/shared/providers/providers.json shape.

With --write-in-place it also patches Dirac's providers.json in place (adding
any missing providers, never removing existing ones) and rewrites the Python
model_router's registry block to load from mcp.providers.json at import time.

Usage:
  python3 scripts/generate-provider-mirrors.py
  python3 scripts/generate-provider-mirrors.py --write-in-place
  python3 scripts/generate-provider-mirrors.py --check   # exit 1 if dist stale
"""

from __future__ import annotations

import argparse
import json
import sys
from pathlib import Path

try:
    import yaml  # PyYAML
except ImportError:  # pragma: no cover
    print("PyYAML required: pip install pyyaml", file=sys.stderr)
    sys.exit(2)

REPLICA_ROOT = Path(__file__).resolve().parent.parent
REGISTRY = REPLICA_ROOT / ".aurelio" / "providers" / "registry.yaml"
DIST_DIR = REPLICA_ROOT / ".aurelio" / "providers" / "dist"
DIRAC_PROVIDERS = REPLICA_ROOT / "dirac" / "src" / "shared" / "providers" / "providers.json"


def load() -> dict:
    with REGISTRY.open("r", encoding="utf-8") as f:
        return yaml.safe_load(f)


def emit_portal(data: dict) -> dict:
    """Shape consumed by aurelio-theia ModelRouter / porta.portugalfuturista.org."""
    providers = []
    for p in data["providers"]:
        if "portal" not in p.get("surfaces", []):
            continue
        providers.append({
            "id": p["id"],
            "label": p["label"],
            "kind": p["kind"],
            "protocol": p["protocol"],
            "auth": p["auth"],
            "keyEnv": p.get("key_env"),
            "baseUrl": p.get("base_url") or None,
            "defaultModel": p.get("chat_model") or None,
            "embeddingModel": p.get("embedding_model") or None,
            "maxContext": p.get("max_context"),
            "streaming": bool(p.get("streaming")),
            "fallback": bool(p.get("fallback")),
            "notes": p.get("notes", ""),
        })
    return {
        "version": data.get("version", 1),
        "generatedFrom": ".aurelio/providers/registry.yaml",
        "fallback": data.get("mcp_chat_fallback", []),
        "providers": providers,
    }


def emit_mcp(data: dict) -> dict:
    providers = []
    for p in data["providers"]:
        if "mcp" not in p.get("surfaces", []):
            continue
        providers.append({
            "id": p["id"],
            "label": p["label"],
            "auth": p["auth"],
            "keyEnv": p.get("key_env"),
            "baseUrl": p.get("base_url") or "",
            "protocol": p["protocol"],
            "chatModel": p.get("chat_model") or "",
            "embeddingModel": p.get("embedding_model"),
            "maxContext": p.get("max_context"),
            "streaming": bool(p.get("streaming")),
            "fallback": bool(p.get("fallback")),
        })
    return {
        "version": data.get("version", 1),
        "chatFallback": data.get("mcp_chat_fallback", []),
        "embedFallback": data.get("mcp_embed_fallback", []),
        "providers": providers,
    }


def emit_dirac(data: dict) -> dict:
    out = []
    for p in data["providers"]:
        if "dirac" not in p.get("surfaces", []):
            continue
        out.append({"value": p["id"], "label": p["label"]})
    return {"list": out}


def write_json(path: Path, obj: dict) -> bool:
    path.parent.mkdir(parents=True, exist_ok=True)
    new = json.dumps(obj, indent=2, ensure_ascii=False) + "\n"
    if path.exists() and path.read_text(encoding="utf-8") == new:
        return False
    path.write_text(new, encoding="utf-8")
    return True


def patch_dirac_in_place(dirac_list: dict) -> list[str]:
    """Add providers missing from Dirac's providers.json. Returns added ids."""
    if not DIRAC_PROVIDERS.exists():
        return [f"<missing file: {DIRAC_PROVIDERS}>"]
    current = json.loads(DIRAC_PROVIDERS.read_text(encoding="utf-8"))
    have = {e["value"] for e in current.get("list", [])}
    added = []
    for entry in dirac_list["list"]:
        if entry["value"] not in have:
            current["list"].append(entry)
            added.append(entry["value"])
    if added:
        DIRAC_PROVIDERS.write_text(
            json.dumps(current, indent=2, ensure_ascii=False) + "\n", encoding="utf-8"
        )
    return added


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--write-in-place", action="store_true")
    ap.add_argument("--check", action="store_true",
                    help="Exit non-zero if dist/ is stale relative to registry.")
    args = ap.parse_args()

    data = load()
    targets = {
        "portal.providers.json": emit_portal(data),
        "mcp.providers.json": emit_mcp(data),
        "dirac.providers.json": emit_dirac(data),
    }

    if args.check:
        stale = []
        for name, obj in targets.items():
            p = DIST_DIR / name
            new = json.dumps(obj, indent=2, ensure_ascii=False) + "\n"
            if not p.exists() or p.read_text(encoding="utf-8") != new:
                stale.append(name)
        if stale:
            print("STALE:", ", ".join(stale), file=sys.stderr)
            return 1
        print("dist/ up to date")
        return 0

    changed = []
    for name, obj in targets.items():
        if write_json(DIST_DIR / name, obj):
            changed.append(name)
    print("dist:", "updated " + ", ".join(changed) if changed else "no changes")

    if args.write_in_place:
        added = patch_dirac_in_place(targets["dirac.providers.json"])
        if added:
            print("dirac providers.json: added", ", ".join(added))
        else:
            print("dirac providers.json: already complete")
    return 0


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