#!/usr/bin/env python3 """ merge-invites.py -- Merge ~/cloudron-invites.txt into agents.yaml. Parses the loose invite file format (name, url -- extra spaces and blank lines tolerated), extracts email/username/displayName from each invite URL's query params, and updates (or appends) agents in the manifest. Safety: - Detects duplicate invite tokens across entries and SKIPS the later duplicate (stale copy-paste), reporting it loudly. - Never overwrites an existing valid invite with a placeholder. - Names are normalized: vpsecops -> vp-secops, svpknel -> svp-knel, vpinvesting -> vp-investing (coo stays coo). Unknown agents are appended with systems: {} (phase1-only). Input format -- one agent per line, blank lines and # comments skipped: agent-name,https://my.knownelement.com/setupaccount.html?inviteToken=...&email=... The invite URL already contains email/username/displayName as query params; the line only adds the canonical agent name (hyphenated, e.g. vp-secops). Loose names like vpsecops are auto-normalized. Run (reads ro-mounted agents.yaml, writes merged copy to state/): docker compose run --rm --entrypoint python3 \ -v "$HOME/cloudron-invites.txt:/invites.txt:ro" \ provision merge-invites.py cp state/agents-merged.yaml agents.yaml Exit codes: 0 = clean, 1 = skipped duplicates or missing agents (manifest still written with what was valid). """ import re import sys from pathlib import Path from urllib.parse import parse_qs, unquote, urlparse import yaml INVITE_PATH = sys.argv[1] if len(sys.argv) > 1 else "/invites.txt" MANIFEST_IN = sys.argv[2] if len(sys.argv) > 2 else "/app/agents.yaml" MANIFEST_OUT = sys.argv[3] if len(sys.argv) > 3 else "/app/state/agents-merged.yaml" # Loose-file name -> canonical manifest name NAME_MAP = { "vpsecops": "vp-secops", "svpknel": "svp-knel", "svptctc": "svp-tctc", "vptechops": "vp-techops", "vptechcompliance": "vp-techcompliance", } def canonical_name(raw: str) -> str: name = raw.strip().lower().replace("_", "-") if name in NAME_MAP: return NAME_MAP[name] # Generic: vpfoo -> vp-foo, svpfoo -> svp-foo (only when no hyphen yet) m = re.match(r"^(svp|vp)([a-z].*)$", name) if m: return f"{m.group(1)}-{m.group(2)}" return name def parse_invite_file(path: str) -> list[dict]: entries = [] for lineno, line in enumerate(Path(path).read_text().splitlines(), 1): line = line.strip() if not line or line.startswith("#"): continue if "," not in line: print(f" line {lineno}: no comma, skipping: {line[:60]}") continue raw_name, raw_url = line.split(",", 1) raw_url = raw_url.strip() if "inviteToken=" not in raw_url: print(f" line {lineno}: URL has no inviteToken, skipping: {raw_name.strip()}") continue qs = parse_qs(urlparse(raw_url).query) get1 = lambda k: unquote(qs.get(k, [""])[0]) # noqa: E731 token = get1("inviteToken") entries.append({ "line": lineno, "raw_name": raw_name.strip(), "name": canonical_name(raw_name), "url": raw_url, "token": token, "email": get1("email"), "username": get1("username"), "display_name": get1("displayName"), }) return entries def main(): invite_file = Path(INVITE_PATH) if not invite_file.exists(): print(f"ERROR: invite file not found: {INVITE_PATH}") sys.exit(2) print(f"=== Parsing {INVITE_PATH} ===") entries = parse_invite_file(str(invite_file)) if not entries: print("ERROR: no valid entries parsed") sys.exit(2) problems = 0 # Duplicate-token detection (stale copy-paste guard) seen_tokens: dict[str, dict] = {} valid = [] for e in entries: if e["token"] in seen_tokens: first = seen_tokens[e["token"]] print(f" !! DUPLICATE TOKEN line {e['line']} ({e['name']}): same invite as " f"line {first['line']} ({first['name']}) / {first['email']}") print(f" Skipping {e['name']} -- Charles must issue a fresh invite for it.") problems += 1 else: seen_tokens[e["token"]] = e valid.append(e) # Cross-check: entry email/username must not equal ANOTHER entry's # (catches svpknel lines that carry coo's params under a new token) by_email = {} for e in valid: if e["email"] in by_email and e["token"] != by_email[e["email"]]["token"]: print(f" !! {e['name']} (line {e['line']}) reuses email {e['email']} " f"already claimed by {by_email[e['email']]['name']}") problems += 1 else: by_email[e["email"]] = e for e in valid: print(f" {e['name']:20s} email={e['email']:45s} username={e['username']}") # Load manifest (read-only mount) data = yaml.safe_load(Path(MANIFEST_IN).read_text()) agents = data.get("agents", []) by_name = {a["name"]: a for a in agents} print(f"\n=== Merging into {MANIFEST_OUT} ===") for e in valid: agent = by_name.get(e["name"]) if agent is None: print(f" + {e['name']}: new agent, appending (phase1-only)") agent = { "name": e["name"], "display_name": e["display_name"] or e["name"], "cloudron_email": e["email"], "username": e["username"], "priority": "Q4", "cloudron_invite": e["url"], "systems": {}, } agents.append(agent) by_name[e["name"]] = agent else: old = agent.get("cloudron_invite", "") if "REPLACE" in old or not old: agent["cloudron_invite"] = e["url"] print(f" ~ {e['name']}: invite set") elif old.strip() == e["url"]: print(f" = {e['name']}: invite unchanged") else: agent["cloudron_invite"] = e["url"] print(f" ~ {e['name']}: invite REPLACED (was different URL)") # Sync email/username to what Cloudron actually issued if e["email"]: agent["cloudron_email"] = e["email"] if e["username"]: agent["username"] = e["username"] data["agents"] = agents # Report agents still without a valid invite print("\n=== Manifest state ===") for a in agents: inv = a.get("cloudron_invite", "") state = "READY" if inv and "REPLACE" not in inv else "NO VALID INVITE" if state != "READY": problems += 1 print(f" {a['name']:20s} [{state}] {a.get('display_name', '')}") out_path = Path(MANIFEST_OUT) out_path.parent.mkdir(parents=True, exist_ok=True) out_path.write_text(yaml.safe_dump(data, sort_keys=False, allow_unicode=True)) print(f"\nManifest written: {MANIFEST_OUT}") print(f"Next: cp state/agents-merged.yaml agents.yaml") sys.exit(1 if problems else 0) if __name__ == "__main__": main()