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, pve1, pve3


@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)


def _parse_container_sections(path: Path) -> dict:
    """Parse section comments to determine which host each CT_ variable belongs to."""
    section_map = {
        "pve-hetzner": re.compile(r"#.*CONTAINER.*pve-hetzner", re.IGNORECASE),
        "pve1": re.compile(r"#.*CONTAINER.*pve1", re.IGNORECASE),
        "pve3": re.compile(r"#.*CONTAINER.*pve3", re.IGNORECASE),
    }
    ct_var = re.compile(r"^(CT_\d+(?:_\w+)?)\s*=")
    result = {}
    current_host = "pve-hetzner"

    with open(path) as f:
        for line in f:
            stripped = line.strip()
            if stripped.startswith("#"):
                for host, pattern in section_map.items():
                    if pattern.search(stripped):
                        current_host = host
                        break
            else:
                m = ct_var.match(stripped)
                if m:
                    result[m.group(1)] = current_host
    return result


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_(\d+)(?:_(PVE\d+))?$")
    section_hosts = _parse_container_sections(path)

    for k, v in raw.items():
        m = ct_pattern.match(k)
        if m:
            vmid = int(m.group(1))
            explicit_host = m.group(2)
            if explicit_host:
                host = {"PVE1": "pve1", "PVE3": "pve3"}.get(explicit_host, explicit_host.lower())
            else:
                host = section_hosts.get(k, "pve-hetzner")

            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))
    return cfg


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


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)