docker/scripts/docs/generate-diagrams.py

#!/usr/bin/env python3
import argparse
import re
import subprocess
import shutil
from pathlib import Path
import yaml

INTERNAL_DOMAIN_RE = re.compile(r"\b[a-zA-Z0-9.-]+\.lan\.ddnsgeek\.com\b")
HOST_MATCH_RE = re.compile(r"Host\(([^)]*)\)")
TICKED_HOST_RE = re.compile(r"`([^`]+)`")


def require_dot() -> None:
    if not shutil.which("dot"):
        raise SystemExit(
            "Graphviz 'dot' not found in environment. Install graphviz before running docs generation."
        )


def load_compose(path: Path) -> dict:
    with path.open() as handle:
        return yaml.safe_load(handle) or {}


def sanitize_domain(value: str, known: dict[str, str]) -> str:
    if value in known:
        return known[value]
    if INTERNAL_DOMAIN_RE.search(value):
        mapped = f"service-{len(known)+1}.<internal-domain>"
        known[value] = mapped
        return mapped
    return value


def parse_labels(service: dict) -> dict[str, str]:
    labels = service.get("labels") or {}
    if isinstance(labels, list):
        mapped = {}
        for item in labels:
            if "=" in str(item):
                k, v = str(item).split("=", 1)
                mapped[k] = v
        return mapped
    if isinstance(labels, dict):
        return {str(k): str(v) for k, v in labels.items()}
    return {}


def extract_hosts(rule: str) -> list[str]:
    hosts: list[str] = []
    for m in HOST_MATCH_RE.findall(rule or ""):
        for host in TICKED_HOST_RE.findall(m):
            hosts.append(host.strip())
    return hosts


def render_svg(dot_path: Path, svg_path: Path) -> None:
    subprocess.run(["dot", "-Tsvg", str(dot_path), "-o", str(svg_path)], check=True)


def write_dot(path: Path, lines: list[str]) -> None:
    path.write_text("\n".join(lines) + "\n")


def infer_host(service_name: str, service: dict) -> str:
    labels = parse_labels(service)
    for key in ("com.docker.compose.project", "infra.host", "host"):
        value = labels.get(key)
        if value:
            return value.lower()
    s = service_name.lower()
    if "raspi" in s or "pi" in s:
        return "raspberrypi"
    if "proxmox" in s:
        return "proxmox"
    return "docker"


def generate_physical_topology(compose: dict, out_dot: Path, out_svg: Path) -> None:
    services = compose.get("services") or {}
    hosts: dict[str, list[str]] = {}
    for name, svc in services.items():
        hosts.setdefault(infer_host(name, svc), []).append(name)

    lines = [
        "digraph PhysicalTopology {",
        "  graph [rankdir=LR, compound=true, splines=ortho, nodesep=0.45, ranksep=0.75, fontname=\"Helvetica\"];",
        "  node [fontname=\"Helvetica\", fontsize=10, style=\"rounded,filled\", fillcolor=\"#ffffff\"];",
        "  edge [fontname=\"Helvetica\", fontsize=9];",
    ]

    for host, host_services in sorted(hosts.items()):
        lines.extend(
            [
                f'  subgraph "cluster_{host}" {{',
                f'    label="{host}";',
                '    style="rounded,filled";',
                '    color="#c7d6f5";',
                '    fillcolor="#eef3ff";',
                f'    "host:{host}" [label="{host}", shape=box3d, fillcolor="#d4e3ff"];',
            ]
        )
        for service in sorted(host_services):
            lines.append(
                f'    "svc:{service}" [label="{service}", shape=box, fillcolor="#dff2e1"];'
            )
            lines.append(f'    "host:{host}" -> "svc:{service}" [style=dashed, color="#6b7280"];')
        lines.append("  }")

    lines.append("}")
    write_dot(out_dot, lines)
    render_svg(out_dot, out_svg)


def generate_docker_traefik_dynu(compose: dict, out_dot: Path, out_svg: Path) -> None:
    services = compose.get("services") or {}
    networks = compose.get("networks") or {}
    known_domains: dict[str, str] = {}

    lines = [
        "digraph DockerTraefikDynu {",
        "  graph [rankdir=LR, compound=true, splines=true, nodesep=0.5, ranksep=1.0, fontname=\"Helvetica\"];",
        "  node [fontname=\"Helvetica\", fontsize=10, style=\"rounded,filled\"];",
        "  edge [fontname=\"Helvetica\", fontsize=9];",
        '  "ext:dynu" [label="Dynu / Public DNS", shape=oval, fillcolor="#fde68a"];',
        '  "svc:traefik" [label="traefik", shape=box, fillcolor="#bfdbfe"];',
    ]

    for net in sorted(networks.keys()):
        lines.append(f'  "net:{net}" [label="{net}", shape=ellipse, fillcolor="#f3f4f6"];')

    for svc_name, svc in sorted(services.items()):
        lines.append(f'  "svc:{svc_name}" [label="{svc_name}", shape=box, fillcolor="#dcfce7"];')
        svc_nets = svc.get("networks") or []
        if isinstance(svc_nets, dict):
            svc_nets = svc_nets.keys()
        for net in svc_nets:
            lines.append(f'  "svc:{svc_name}" -> "net:{net}" [color="#6b7280"];')

        labels = parse_labels(svc)
        router_prefix = "traefik.http.routers."
        service_prefix = "traefik.http.services."

        routers = sorted({k[len(router_prefix):].split(".", 1)[0] for k in labels if k.startswith(router_prefix)})
        for router in routers:
            rule = labels.get(f"{router_prefix}{router}.rule", "")
            router_service = labels.get(f"{router_prefix}{router}.service", svc_name)
            middlewares = labels.get(f"{router_prefix}{router}.middlewares", "")
            entrypoints = labels.get(f"{router_prefix}{router}.entrypoints", "")
            tls = labels.get(f"{router_prefix}{router}.tls", "false")
            lines.append(f'  "router:{router}" [label="router:{router}\\nentry:{entrypoints} tls:{tls}", shape=diamond, fillcolor="#fbcfe8"];')
            lines.append(f'  "svc:traefik" -> "router:{router}";')
            target_node = f"traefik-service:{router_service}"
            lines.append(f'  "{target_node}" [label="service:{router_service}", shape=component, fillcolor="#fecaca"];')
            lines.append(f'  "router:{router}" -> "{target_node}";')

            for host in extract_hosts(rule):
                clean = sanitize_domain(host, known_domains)
                lines.append(f'  "dns:{clean}" [label="{clean}", shape=note, fillcolor="#fde68a"];')
                lines.append(f'  "ext:dynu" -> "dns:{clean}";')
                lines.append(f'  "dns:{clean}" -> "router:{router}";')

            for middleware in [m.strip() for m in middlewares.split(",") if m.strip()]:
                lines.append(f'  "mw:{middleware}" [label="{middleware}", shape=hexagon, fillcolor="#ddd6fe"];')
                lines.append(f'  "router:{router}" -> "mw:{middleware}" [style=dashed];')

        lb_services = sorted({k[len(service_prefix):].split(".", 1)[0] for k in labels if k.startswith(service_prefix)})
        for lb in lb_services:
            port = labels.get(f"{service_prefix}{lb}.loadbalancer.server.port", "")
            lines.append(f'  "traefik-service:{lb}" [label="service:{lb}\\nport:{port}", shape=component, fillcolor="#fecaca"];')
            lines.append(f'  "traefik-service:{lb}" -> "svc:{svc_name}";')

    lines.append("}")
    write_dot(out_dot, lines)
    render_svg(out_dot, out_svg)


def generate_compose_topology(compose: dict, out_dot: Path, out_svg: Path) -> None:
    services = compose.get("services") or {}
    networks = compose.get("networks") or {}
    lines = [
        "digraph Compose {",
        "  rankdir=LR;",
        '  node [fontname="Helvetica"];',
    ]
    for service in sorted(services):
        lines.append(f'  "svc:{service}" [label="{service}", shape=box, style=filled, fillcolor="#dfefff"];')
    for net in sorted(networks):
        lines.append(f'  "net:{net}" [label="{net}", shape=ellipse, style=filled, fillcolor="#f4f4f4"];')

    for service, svc in sorted(services.items()):
        svc_nets = svc.get("networks") or []
        if isinstance(svc_nets, dict):
            svc_nets = svc_nets.keys()
        for net in svc_nets:
            lines.append(f'  "svc:{service}" -> "net:{net}";')

    lines.append("}")
    write_dot(out_dot, lines)
    render_svg(out_dot, out_svg)


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("legacy", nargs="*")
    parser.add_argument("--compose")
    parser.add_argument("--out-dir")
    args = parser.parse_args()

    require_dot()

    if args.compose and args.out_dir:
        compose_path = Path(args.compose)
        out_dir = Path(args.out_dir)
    elif len(args.legacy) == 3:
        compose_path = Path(args.legacy[0])
        out_dir = Path(args.legacy[1]).parent
    else:
        raise SystemExit("Usage: generate-diagrams.py --compose <compose.yml> --out-dir <dir>")

    out_dir.mkdir(parents=True, exist_ok=True)
    compose = load_compose(compose_path)

    generate_docker_traefik_dynu(compose, out_dir / "docker-traefik-dynu.dot", out_dir / "docker-traefik-dynu.svg")
    generate_physical_topology(compose, out_dir / "physical-topology.dot", out_dir / "physical-topology.svg")
    generate_compose_topology(compose, out_dir / "docker-compose.dot", out_dir / "docker-compose.svg")


if __name__ == "__main__":
    main()