oikos/archive/mcp/build_host_files.py

#!/usr/bin/env python3
"""
Generate hosts/<name>.yaml from inventory.yaml.

Run from the repo root:
    python3 mcp/build_host_files.py            # writes files, exits non-zero on diff
    python3 mcp/build_host_files.py --check    # exits non-zero if any output differs

Designed to be wired into a pre-commit hook or Gitea Action so generated
hosts/*.yaml never drift from inventory.yaml.
"""

from __future__ import annotations

import argparse
import difflib
import os
import sys
from pathlib import Path

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


REPO = Path(__file__).resolve().parent.parent
INVENTORY = REPO / "inventory.yaml"
HOSTS_DIR = REPO / "hosts"

GENERATED_BANNER = (
    "# Generated by mcp/build_host_files.py from inventory.yaml.\n"
    "# Do NOT edit by hand — your changes will be overwritten.\n"
    "# Source of truth: ../inventory.yaml\n"
)


def narrative_page(name: str, kind: str, pve_id: int | None) -> str | None:
    """Best-guess path to the human-authored narrative page for this host."""
    if kind == "proxmox-host":
        candidate = REPO / "hosts" / f"{name}.md"
    elif kind == "lxc":
        candidate = REPO / "containers" / f"{pve_id}-{name}.md"
    elif kind == "vm":
        candidate = REPO / "vms" / f"{pve_id}-{name}.md"
    else:
        return None
    if candidate.exists():
        return str(candidate.relative_to(REPO))
    return None


def build_one(name: str, entry: dict, inventory: dict) -> dict:
    """Project the entry for a single host into a per-host yaml record."""
    services = inventory.get("services", {})
    mesh = inventory.get("mesh", {})
    pve_id = entry.get("pve_id")

    # Services this host runs: scan inventory.services for matching backend.
    runs_services = sorted(
        svc for svc, sentry in services.items()
        if isinstance(sentry, dict) and sentry.get("backend") == name
    )

    record = {
        "name": name,
        "kind": entry.get("kind"),
        "os": entry.get("os"),
        "role": entry.get("role"),
        # Oikos lifecycle (oikos/ontology.yaml); absent in inventory = active
        "state": entry.get("state", "active"),
        "host": entry.get("host"),
        "pve_id": pve_id,
        "storage": entry.get("storage"),
        "depends_on": entry.get("depends_on", []),
        "lan_ip": entry.get("lan_ip"),
        "mesh": entry.get("mesh", {}),
        "mesh_globals": {
            "primary": mesh.get("primary"),
            "accepted": mesh.get("accepted"),
        },
        "peers": entry.get("peers", []),
        "mounts": entry.get("mounts", []),
        "public_host": entry.get("public_host"),
        "public_hosts": entry.get("public_hosts", []),
        "ssh": entry.get("ssh", {}),
        "runs": entry.get("runs", []) + runs_services,
        "services_hosted": [
            {"name": svc, **services[svc]} for svc in runs_services
        ],
        "notes": entry.get("notes", []),
        "age_pubkey": entry.get("age_pubkey", ""),
        "see_also": [
            page for page in [narrative_page(name, entry.get("kind", ""), pve_id)]
            if page
        ],
        "mcp_endpoint": services.get("homelab_mcp", {}).get("endpoint"),
        "secrets_issuance_endpoint": (
            services.get("secrets_issuance", {}).get("endpoint")
        ),
    }
    # Strip None and empty containers so the file stays readable.
    return {k: v for k, v in record.items() if v not in (None, {}, [], "")}


def serialize(record: dict) -> str:
    return GENERATED_BANNER + yaml.safe_dump(
        record, sort_keys=False, default_flow_style=False, width=100
    )


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--check", action="store_true",
                        help="exit 1 if any output would change (don't write)")
    args = parser.parse_args()

    inventory = yaml.safe_load(INVENTORY.read_text())
    hosts = inventory.get("hosts", {})

    HOSTS_DIR.mkdir(exist_ok=True)
    desired: dict[Path, str] = {}
    for name, entry in hosts.items():
        desired[HOSTS_DIR / f"{name}.yaml"] = serialize(build_one(name, entry, inventory))

    diff_count = 0
    for path, content in desired.items():
        existing = path.read_text() if path.exists() else ""
        if existing != content:
            diff_count += 1
            if args.check:
                diff = difflib.unified_diff(
                    existing.splitlines(keepends=True),
                    content.splitlines(keepends=True),
                    fromfile=str(path),
                    tofile=str(path) + " (generated)",
                )
                sys.stdout.writelines(diff)
            else:
                path.write_text(content)
                print(f"wrote {path.relative_to(REPO)}")

    # Clean up orphans (file exists but host removed from inventory).
    for existing_path in HOSTS_DIR.glob("*.yaml"):
        if existing_path not in desired:
            diff_count += 1
            if args.check:
                print(f"orphan: {existing_path.relative_to(REPO)} (would delete)")
            else:
                existing_path.unlink()
                print(f"deleted orphan {existing_path.relative_to(REPO)}")

    if args.check and diff_count > 0:
        print(f"\n{diff_count} file(s) would change. Run without --check to write.",
              file=sys.stderr)
        return 1
    return 0


if __name__ == "__main__":
    sys.exit(main())