102 lines
3 KiB
Python
102 lines
3 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Generate downstream connector-hub mirrors from the single source of truth:
|
|
|
|
.aurelio/connectors/registry.yaml
|
|
|
|
Outputs (under .aurelio/connectors/dist/ unless --write-in-place is given,
|
|
which writes the same files — dist/ IS the in-place location here):
|
|
|
|
* mcp.connectors.json — machine-readable catalog the MCP fleet / router
|
|
can load (kind=mcp entries map onto mcp_config.json).
|
|
* theia.connectors.json — slim shape the @aurelio/mcp Connections view reads:
|
|
[{id,label,category,kind,auth,status,host,endpoint}].
|
|
|
|
Usage:
|
|
python3 scripts/generate-connector-mirrors.py --write-in-place
|
|
python3 scripts/generate-connector-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" / "connectors" / "registry.yaml"
|
|
DIST_DIR = REPLICA_ROOT / ".aurelio" / "connectors" / "dist"
|
|
|
|
THEIA_FIELDS = ("id", "label", "category", "kind", "auth", "status", "host", "endpoint", "surfaces")
|
|
|
|
|
|
def load() -> dict:
|
|
with REGISTRY.open("r", encoding="utf-8") as f:
|
|
return yaml.safe_load(f)
|
|
|
|
|
|
def emit_theia(data: dict) -> dict:
|
|
conns = [
|
|
{k: c.get(k) for k in THEIA_FIELDS}
|
|
for c in data["connectors"]
|
|
if "theia" in c.get("surfaces", [])
|
|
]
|
|
return {"version": data.get("version", 1), "connectors": conns}
|
|
|
|
|
|
def emit_mcp(data: dict) -> dict:
|
|
conns = [
|
|
c for c in data["connectors"]
|
|
if "mcp" in c.get("surfaces", [])
|
|
]
|
|
return {"version": data.get("version", 1), "connectors": conns}
|
|
|
|
|
|
def mirrors() -> dict[str, dict]:
|
|
data = load()
|
|
return {
|
|
"theia.connectors.json": emit_theia(data),
|
|
"mcp.connectors.json": emit_mcp(data),
|
|
}
|
|
|
|
|
|
def write_in_place() -> None:
|
|
DIST_DIR.mkdir(parents=True, exist_ok=True)
|
|
for name, obj in mirrors().items():
|
|
(DIST_DIR / name).write_text(json.dumps(obj, indent=2) + "\n", encoding="utf-8")
|
|
print(f"wrote {DIST_DIR / name}")
|
|
|
|
|
|
def check() -> int:
|
|
stale = []
|
|
for name, obj in mirrors().items():
|
|
p = DIST_DIR / name
|
|
want = json.dumps(obj, indent=2) + "\n"
|
|
if not p.exists() or p.read_text(encoding="utf-8") != want:
|
|
stale.append(name)
|
|
if stale:
|
|
print("stale connector mirrors (run --write-in-place):", ", ".join(stale), file=sys.stderr)
|
|
return 1
|
|
print("connector mirrors up to date")
|
|
return 0
|
|
|
|
|
|
def main() -> int:
|
|
ap = argparse.ArgumentParser(description=__doc__)
|
|
ap.add_argument("--write-in-place", action="store_true")
|
|
ap.add_argument("--check", action="store_true")
|
|
args = ap.parse_args()
|
|
if args.check:
|
|
return check()
|
|
write_in_place()
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|