homelab-brain/homelab-ai-bot/core/config.py

"""Parses homelab.conf — the single source of truth for infrastructure facts."""

import os
import re
from pathlib import Path
from dataclasses import dataclass, field

HOMELAB_CONF_PATHS = [
    Path("/root/homelab-brain/homelab.conf"),
    Path("/opt/homelab-brain/homelab.conf"),
]


@dataclass
class Container:
    vmid: int
    name: str
    tailscale_ip: str
    services: str
    host: str  # pve-hetzner, pve-ka-1, pve-mu-3, etc.


@dataclass
class Tunnel:
    ct_id: int
    domain: str
    target: str
    status: str


@dataclass
class HomelabConfig:
    raw: dict = field(default_factory=dict)
    domains: dict = field(default_factory=dict)
    servers: dict = field(default_factory=dict)
    passwords: dict = field(default_factory=dict)
    containers: list = field(default_factory=list)
    telegram: dict = field(default_factory=dict)
    api_keys: dict = field(default_factory=dict)
    tunnels: list = field(default_factory=list)


HOST_CODE_MAP = {
    "HZ": "pve-hetzner",
    "KA1": "pve-ka-1",
    "KA2": "pve-ka-2",
    "KA3": "pve-ka-3",
    "MU1": "pve-mu-1",
    "MU2": "pve-mu-2",
    "MU3": "pve-mu-3",
    "HE": "pve-he",
}


def find_config() -> Path:
    for p in HOMELAB_CONF_PATHS:
        if p.exists():
            return p
    raise FileNotFoundError(f"homelab.conf not found in {HOMELAB_CONF_PATHS}")


def parse_config(path: Path = None) -> HomelabConfig:
    if path is None:
        path = find_config()

    raw = {}
    with open(path) as f:
        for line in f:
            line = line.strip()
            if not line or line.startswith("#"):
                continue
            m = re.match(r'^([A-Za-z_][A-Za-z0-9_]*)="?(.*?)"?\s*$', line)
            if m:
                raw[m.group(1)] = m.group(2)

    cfg = HomelabConfig(raw=raw)

    for k, v in raw.items():
        if k.startswith("DOMAIN_"):
            cfg.domains[k.replace("DOMAIN_", "").lower()] = v
        elif k.startswith("SRV_"):
            cfg.servers[k.replace("SRV_", "").lower()] = v
        elif k.startswith("PW_"):
            cfg.passwords[k.replace("PW_", "").lower()] = v
        elif k.startswith("TG_"):
            cfg.telegram[k.lower()] = v
        elif k.startswith("FORGEJO_") or k.startswith("GITHUB_") or k.startswith("OPENROUTER_"):
            cfg.api_keys[k.lower()] = v

    ct_pattern = re.compile(r"^(?:CT|VM)_(\d+)_([A-Z][A-Z0-9]+)$")

    for k, v in raw.items():
        m = ct_pattern.match(k)
        if m:
            vmid = int(m.group(1))
            host_code = m.group(2)
            host = HOST_CODE_MAP.get(host_code, host_code.lower())

            parts = v.split("|")
            if len(parts) >= 3:
                cfg.containers.append(Container(
                    vmid=vmid,
                    name=parts[0],
                    tailscale_ip=parts[1] if parts[1] != "—" else "",
                    services=parts[2],
                    host=host,
                ))

    for k, v in raw.items():
        m = re.match(r"^TUNNEL_(\d+)(?:_\w+)?$", k)
        if m:
            ct_id = int(m.group(1))
            parts = v.split("|")
            if len(parts) >= 3:
                cfg.tunnels.append(Tunnel(
                    ct_id=ct_id,
                    domain=parts[0],
                    target=parts[1],
                    status=parts[2],
                ))

    cfg.containers.sort(key=lambda c: (c.host, c.vmid))

    from core import proxmox_client, loki_client
    srv_to_host = {
        "hetzner": "pve-hetzner",
        "ka1": "pve-ka-1", "ka2": "pve-ka-2", "ka3": "pve-ka-3",
        "mu1": "pve-mu-1", "mu2": "pve-mu-2", "mu3": "pve-mu-3",
        "he": "pve-he",
    }
    for key, ip in cfg.servers.items():
        host_name = srv_to_host.get(key, key)
        if not key.endswith("_local") and not key.endswith("_hostname") and "pbs" not in key:
            proxmox_client.PROXMOX_HOSTS[host_name] = ip

    if raw.get("LOKI_URL"):
        loki_client.LOKI_URL = raw["LOKI_URL"]

    return cfg


def get_container(cfg: HomelabConfig, vmid: int = None, name: str = None, host: str = None) -> Container | None:
    for c in cfg.containers:
        if host and c.host != host:
            continue
        if vmid and c.vmid == vmid:
            return c
        if name and name.lower() in c.name.lower():
            return c
    return None


def get_containers_by_host(cfg: HomelabConfig, host: str) -> list[Container]:
    return [c for c in cfg.containers if c.host == host]


def format_overview(cfg: HomelabConfig) -> str:
    lines = ["# Homelab Infrastructure (from homelab.conf)\n"]

    lines.append("## Domains")
    for k, v in cfg.domains.items():
        lines.append(f"- {k}: {v}")

    lines.append("\n## Servers (Tailscale)")
    for k, v in cfg.servers.items():
        lines.append(f"- {k}: {v}")

    current_host = None
    for c in cfg.containers:
        if c.host != current_host:
            current_host = c.host
            lines.append(f"\n## Containers on {current_host}")
            lines.append("| CT | Name | Tailscale | Services |")
            lines.append("|---|---|---|---|")
        ts = c.tailscale_ip or "—"
        lines.append(f"| {c.vmid} | {c.name} | {ts} | {c.services} |")

    if cfg.tunnels:
        lines.append("\n## Cloudflare Tunnels")
        for t in cfg.tunnels:
            lines.append(f"- CT {t.ct_id}: {t.domain} → {t.target} ({t.status})")

    return "\n".join(lines)