oikos/oikos/gen_topology_lib.py

"""oikos/gen_topology_lib.py — shared Mermaid-view logic.

Split out of oikos/gen-topology.py so it's importable (a hyphenated
filename can't be `import`ed as a module). oikos/gen-topology.py is the
CLI entrypoint that writes knowledge/wiki/infrastructure/topology.md + oikos/cards/;
oikos/console/app.py imports this module directly to render the live
/graph page without shelling out.
"""

from __future__ import annotations

from pathlib import Path

import yaml

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


def load_inventory() -> dict:
    return yaml.safe_load(INVENTORY.read_text())


def node_id(name: str) -> str:
    """Mermaid-safe node id."""
    return name.replace("-", "_").replace(".", "_").replace("/", "_").strip("_")


def guest_label(name: str, entry: dict) -> str:
    pve = entry.get("pve_id")
    role = entry.get("role", "")
    tag = f"LXC {pve}" if entry.get("kind") == "lxc" and pve else \
          f"VM {pve}" if entry.get("kind") == "vm" and pve else entry.get("kind", "")
    ip = entry.get("lan_ip", "")
    parts = [name, tag, role, ip]
    return "<br/>".join(str(p) for p in parts if p)


def compute_view(inv: dict) -> list[str]:
    hosts = inv.get("hosts", {})
    services = inv.get("services", {})
    lines = ["```mermaid", "flowchart LR"]

    hypervisors = {n: e for n, e in hosts.items() if e.get("kind") == "proxmox-host"}
    guests = {n: e for n, e in hosts.items() if e.get("kind") in ("lxc", "vm")}
    others = {n: e for n, e in hosts.items()
              if e.get("kind") in ("workstation", "external")}

    for hv in hypervisors:
        lines.append(f'    subgraph {node_id(hv)}_sub["{hv} (Proxmox)"]')
        for g, e in guests.items():
            if e.get("host") == hv:
                lines.append(f'        {node_id(g)}["{guest_label(g, e)}"]')
        lines.append("    end")

    # guests without a parent hypervisor recorded (e.g. rclone)
    for g, e in guests.items():
        if e.get("host") not in hypervisors:
            lines.append(f'    {node_id(g)}["{guest_label(g, e)}"]')

    for n, e in others.items():
        shape = "([{}])" if e.get("kind") == "workstation" else "[[{}]]"
        lines.append(f'    {node_id(n)}{shape.format(guest_label(n, e))}')

    # ingress: public URL -> backend (routes-to)
    for svc, e in sorted(services.items()):
        if not isinstance(e, dict):
            continue
        backend = e.get("backend")
        url = e.get("url") or (
            f'https://{e["public_host"]}' if e.get("public_host") else None)
        if backend and url and backend in hosts:
            host = url.removeprefix("https://").removeprefix("http://")
            # hypervisors are rendered as subgraphs; point edges at the subgraph id
            target = node_id(backend) + ("_sub" if backend in hypervisors else "")
            lines.append(
                f'    {node_id("url_" + svc)}(["{host}"]) -->|routes-to| {target}')

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


def storage_view(inv: dict) -> list[str]:
    hosts = inv.get("hosts", {})
    lines = ["```mermaid", "flowchart LR"]
    pools: set[str] = set()
    edges: list[str] = []

    for name, e in hosts.items():
        for mount in e.get("mounts", []):
            pools.add(mount)
            edges.append(f'    {node_id(name)}["{name}"] -->|mounts| {node_id(mount)}')

    for pool in sorted(pools):
        lines.append(f'    {node_id(pool)}[("{pool}")]')
    lines.extend(sorted(set(edges)))
    lines.append("```")
    return lines


def archaeology_table(inv: dict) -> list[str]:
    arch = inv.get("archaeology", {})
    if not arch:
        return []
    lines = ["| Node | ID | Destroyed | Reason |", "|---|---|---|---|"]
    entries = sorted(arch.items(), key=lambda kv: str(kv[1].get("destroyed", "")),
                     reverse=True)
    for name, e in entries:
        lines.append(
            f'| {name} | {e.get("pve_id", "")} | {e.get("destroyed", "")} '
            f'| {e.get("reason", "")} |')
    return lines