docker/scripts/docs/generate_host_topology.py

#!/usr/bin/env python3
"""Generate host topology Markdown/Mermaid from Terraform inventory JSON."""

from __future__ import annotations

import argparse
import json
from datetime import datetime, timezone
from pathlib import Path
from typing import Any

PARENT_KEYS = [
    "node",
    "node_name",
    "host",
    "physical_host",
    "hypervisor_host",
    "proxmox_node",
]


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Generate Markdown topology from terraform output JSON"
    )
    parser.add_argument("--input", required=True, help="Input JSON path")
    parser.add_argument("--output", required=True, help="Output Markdown path")
    return parser.parse_args()


def load_inventory(raw: dict[str, Any]) -> dict[str, Any]:
    payload = raw.get("value", raw)
    if not isinstance(payload, dict):
        raise ValueError("Inventory payload must be an object")
    return payload


def infer_name(key: str, item: dict[str, Any]) -> str:
    for candidate in ("name", "hostname", "vm_name", "id"):
        value = item.get(candidate)
        if value:
            return str(value)
    return key


def to_records(host_map: Any) -> list[dict[str, Any]]:
    if isinstance(host_map, dict):
        records: list[dict[str, Any]] = []
        for key, val in host_map.items():
            if isinstance(val, dict):
                rec = dict(val)
                rec.setdefault("_key", str(key))
                rec.setdefault("name", infer_name(str(key), rec))
                records.append(rec)
        return sorted(records, key=lambda x: str(x.get("name", "")).lower())
    return []


def escape_cell(value: Any) -> str:
    if value is None:
        return ""
    text = str(value).replace("|", "\\|").replace("\n", "<br>")
    return text


def get_first(item: dict[str, Any], *keys: str) -> str:
    for key in keys:
        value = item.get(key)
        if value is not None and value != "":
            return str(value)
    return ""


def detect_virtual_hosts(inv: dict[str, Any]) -> list[dict[str, Any]]:
    virtual: list[dict[str, Any]] = []
    for key in ("virtual_hosts", "vms"):
        virtual.extend(to_records(inv.get(key, {})))
    return sorted(virtual, key=lambda x: str(x.get("name", "")).lower())


def build_mermaid(physical: list[dict[str, Any]], virtual: list[dict[str, Any]]) -> list[str]:
    lines = ["```mermaid", "flowchart TD"]
    for p in physical:
        pname = get_first(p, "name", "hostname", "_key")
        pid = f"phys_{pname.lower().replace('-', '_').replace(' ', '_')}"
        lines.append(f'  {pid}["{pname}\\nphysical"]')

    for v in virtual:
        vname = get_first(v, "name", "hostname", "_key")
        vid = f"virt_{vname.lower().replace('-', '_').replace(' ', '_')}"
        parent = ""
        for k in PARENT_KEYS:
            parent = get_first(v, k)
            if parent:
                break

        lines.append(f'  {vid}["{vname}\\nvirtual"]')
        if parent:
            pid = f"phys_{parent.lower().replace('-', '_').replace(' ', '_')}"
            lines.append(f"  {pid} --> {vid}")

    lines.append("```")
    return lines


def build_table(headers: list[str], rows: list[list[str]]) -> list[str]:
    out = ["| " + " | ".join(headers) + " |", "| " + " | ".join(["---"] * len(headers)) + " |"]
    for row in rows:
        out.append("| " + " | ".join(escape_cell(cell) for cell in row) + " |")
    return out


def main() -> int:
    args = parse_args()
    raw = json.loads(Path(args.input).read_text(encoding="utf-8"))
    inv = load_inventory(raw)

    physical = to_records(inv.get("physical_hosts", {}))
    virtual = detect_virtual_hosts(inv)

    now = datetime.now(timezone.utc).replace(microsecond=0).isoformat()
    lines: list[str] = [
        "# Host Topology",
        "",
        f"> Generated by `scripts/docs/generate_host_topology.py` on {now}.",
        "",
        "## Topology Diagram",
        "",
    ]
    lines.extend(build_mermaid(physical, virtual))
    lines.extend(["", "## Physical Hosts", ""])

    physical_rows = [
        [
            get_first(p, "name", "hostname", "_key"),
            get_first(p, "type"),
            get_first(p, "role"),
            get_first(p, "management", "management_ip"),
            get_first(p, "os", "os_family"),
            get_first(p, "hypervisor"),
            get_first(p, "location"),
            get_first(p, "notes"),
        ]
        for p in physical
    ]
    lines.extend(
        build_table(
            ["Name", "Type", "Role", "Management", "OS", "Hypervisor", "Location", "Notes"],
            physical_rows,
        )
    )

    lines.extend(["", "## Virtual Hosts", ""])
    if virtual:
        virtual_rows = []
        for v in virtual:
            parent = ""
            for k in PARENT_KEYS:
                parent = get_first(v, k)
                if parent:
                    break
            virtual_rows.append(
                [
                    get_first(v, "name", "hostname", "_key"),
                    get_first(v, "type"),
                    get_first(v, "role"),
                    parent,
                    get_first(v, "management", "management_ip", "ip"),
                    get_first(v, "os", "os_family"),
                    get_first(v, "notes"),
                ]
            )
        lines.extend(
            build_table(
                ["Name", "Type", "Role", "Parent/Node", "Management", "OS", "Notes"],
                virtual_rows,
            )
        )
    else:
        lines.append("No VM/virtual host data found in the current Terraform inventory output.")

    out_path = Path(args.output)
    out_path.parent.mkdir(parents=True, exist_ok=True)
    out_path.write_text("\n".join(lines) + "\n", encoding="utf-8")
    return 0


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