#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#
# hc — 42i Health-Check.
#
# Duenner, read-only Daemon: iteriert die Drop-in-Scripts in /etc/hc.d, fuehrt
# jedes mit Timeout aus, merged deren JSON-Output (well-known keys) zu einem
# Knoten-Baum und serviert ihn auf Port 4711. Die gesamte Spezifik liegt in den
# Drop-in-Scripts; dieser Daemon kennt nur das Merge- und Status-Schema.
#
# `hc serve` sammelt PROAKTIV: ein Hintergrund-Sweeper laeuft alle
# HC_SWEEP_INTERVAL Sekunden (Default 60) und legt den vollen Baum als Snapshot
# ab. Alle Lese-Routen (/badge, /status, /node, / inkl. Peer-Abruf `/?skip=net`)
# bedienen sich nur aus diesem Snapshot und loesen selbst nie eine Collection
# aus — so haengt kein Abruf je auf einem Live-Sweep. Minimale Aufloesung = das
# Intervall. Nur arbiträre Debug-Queries (`/?tags=…&depth=…`) rechnen live.
#
# Subcommands:
#   hc collect [--tags a,b] [--skip c] [--depth local]   JSON des lokalen Knotens auf stdout
#   hc serve                                              HTTP-Daemon auf :HC_PORT
#
# Konzept und Schema: 42i/intern#236. Schema-Version: "42i-hc/1".
#
# Konfiguration ueber Umgebungsvariablen (Defaults in Klammern):
#   HC_PORT (4711)            HTTP-Port von `hc serve`
#   HC_D_DIR (/etc/hc.d)      Verzeichnis der Drop-in-Scripts
#   HC_SRV_DIR (/srv)         Service-Wurzel (an die Scripts durchgereicht)
#   HC_HOST_YAML (/srv/host.yaml)  Host-Metadaten (name, summary, zones, tags, thresholds)
#   HC_SCRIPT_TIMEOUT (10)    Timeout je Script in Sekunden
#   HC_CACHE_TTL (15)         Default-Cache je Drop-in (Sek.); pro Script via
#                             `# hc-ttl: N` im Kopf ueberschreibbar (Single-Flight)
#   HC_SWEEP_INTERVAL (60)    Takt des proaktiven Sweepers in `hc serve` (Sek.)
#   HC_WARN_PCT (85) / HC_FAIL_PCT (95)  Default-Schwellen, an Scripts durchgereicht

import concurrent.futures
import glob
import json
import os
import socket
import stat
import subprocess
import sys
import threading
import time
from datetime import datetime, timezone
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from urllib.parse import urlparse, parse_qs

try:
    import yaml  # python3-yaml
except Exception:  # pragma: no cover - yaml ist Paket-Dependency, hier nur defensiv
    yaml = None

SCHEMA = "42i-hc/1"
# Inventar-Zustaende (ok/inactive/no-check/no-agent) sind neutral (Ordnung 0)
# und eskalieren den aggregierten Health-Status nicht; nur echte Probleme
# (degraded/warn/unreachable/fail) ziehen ihn hoch.
STATUS_ORDER = {"ok": 0, "inactive": 0, "no-check": 0, "no-agent": 0, "degraded": 2,
                "warn": 3, "unreachable": 4, "timeout": 4, "fail": 5}


def env(name, default):
    """Liest eine Umgebungsvariable mit Default."""
    v = os.environ.get(name)
    return v if v not in (None, "") else default


def cfg():
    """Sammelt die Laufzeit-Konfiguration aus den HC_*-Umgebungsvariablen."""
    return {
        "port": int(env("HC_PORT", "4711")),
        "d_dir": env("HC_D_DIR", "/etc/hc.d"),
        "srv_dir": env("HC_SRV_DIR", "/srv"),
        "host_yaml": env("HC_HOST_YAML", "/srv/host.yaml"),
        "timeout": float(env("HC_SCRIPT_TIMEOUT", "10")),
        "cache_ttl": float(env("HC_CACHE_TTL", "15")),
        "warn_pct": env("HC_WARN_PCT", "85"),
        "fail_pct": env("HC_FAIL_PCT", "95"),
    }


def worst(statuses):
    """Liefert den schlimmsten Status nach STATUS_ORDER; neutrale (Ordnung 0) -> 'ok'."""
    s = [x for x in statuses if x]
    if not s:
        return "ok"
    top = max(s, key=lambda x: STATUS_ORDER.get(x, 0))
    return top if STATUS_ORDER.get(top, 0) > 0 else "ok"


def collect_statuses(node):
    """Sammelt rekursiv alle 'status'-Strings unterhalb eines dict/list-Knotens."""
    out = []
    if isinstance(node, dict):
        for k, v in node.items():
            if k == "status" and isinstance(v, str):
                out.append(v)
            else:
                out.extend(collect_statuses(v))
    elif isinstance(node, list):
        for v in node:
            out.extend(collect_statuses(v))
    return out


def now_iso():
    """Aktueller Zeitstempel in ISO-8601 (UTC, Sekundengenauigkeit)."""
    return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")


def load_host_meta(c):
    """Laedt die Host-Metadaten aus host.yaml und ergaenzt Ableitbares (name, ips)."""
    meta = {}
    path = c["host_yaml"]
    if yaml and os.path.isfile(path):
        try:
            with open(path, "r", encoding="utf-8") as fh:
                meta = yaml.safe_load(fh) or {}
        except Exception as exc:
            meta = {"_meta_error": str(exc)}
    meta.setdefault("name", socket.gethostname().split(".")[0])
    if "ips" not in meta:
        meta["ips"] = host_ips()
    return meta


def host_ips():
    """Ermittelt die IPv4-Adressen des Hosts (best effort, leer wenn nicht ermittelbar)."""
    try:
        out = subprocess.run(["hostname", "-I"], capture_output=True,
                              text=True, timeout=2)
        return [ip for ip in out.stdout.split() if ":" not in ip]
    except Exception:
        return []


def parse_header(path):
    """Liest `# hc-tags:` und `# hc-ttl:` aus dem Script-Kopf (erste 25 Zeilen).

    Liefert (tags:set, ttl:float|None). `ttl` None = Daemon-Default (HC_CACHE_TTL).
    """
    tags, ttl = set(), None
    try:
        with open(path, "r", encoding="utf-8", errors="replace") as fh:
            for _ in range(25):
                line = fh.readline()
                if not line:
                    break
                low = line.lower()
                if "hc-tags:" in low:
                    raw = line.split("hc-tags:", 1)[1]
                    tags = {t.strip().lower() for t in raw.replace(",", " ").split() if t.strip()}
                if "hc-ttl:" in low:
                    field = low.split("hc-ttl:", 1)[1].strip().split()
                    if field:
                        try:
                            ttl = float(field[0])
                        except ValueError:
                            pass
    except Exception:
        pass
    return tags, ttl


def discover_scripts(d_dir):
    """Liefert die ausfuehrbaren Drop-in-Scripts (sortiert) als (name, path, tags, ttl)."""
    found = []
    for path in sorted(glob.glob(os.path.join(d_dir, "*"))):
        if not os.path.isfile(path):
            continue
        try:
            mode = os.stat(path).st_mode
        except OSError:
            continue
        if not (mode & stat.S_IXUSR):
            continue
        tags, ttl = parse_header(path)
        found.append((os.path.basename(path), path, tags, ttl))
    return found


def select_scripts(scripts, want_tags, skip_tags):
    """Teilt Scripts in auszufuehrende und (per Tag-Filter) uebersprungene auf."""
    run, skipped = [], []
    for name, path, tags, ttl in scripts:
        if skip_tags and (tags & skip_tags):
            skipped.append(name)
            continue
        if want_tags and not (tags & want_tags):
            skipped.append(name)
            continue
        run.append((name, path, tags, ttl))
    return run, skipped


def run_script(name, path, c):
    """Fuehrt ein Drop-in-Script aus und gibt dessen geparsten JSON-Output zurueck.

    Liefert bei Timeout/Crash/ungueltigem JSON statt eines Merge-Fragments ein
    `{"checks": [...]}` mit einem Fehler-Eintrag, damit der Status sichtbar bleibt.
    """
    senv = dict(os.environ)
    senv.update({
        "HC_SRV_DIR": c["srv_dir"],
        "HC_WARN_PCT": str(c["warn_pct"]),
        "HC_FAIL_PCT": str(c["fail_pct"]),
    })
    try:
        proc = subprocess.run([path], capture_output=True, text=True,
                              timeout=c["timeout"], env=senv)
    except subprocess.TimeoutExpired:
        return {"checks": [{"name": name, "status": "timeout",
                            "message": "Script-Timeout nach %ss" % c["timeout"]}]}
    except Exception as exc:
        return {"checks": [{"name": name, "status": "fail", "message": str(exc)}]}

    out = (proc.stdout or "").strip()
    try:
        data = json.loads(out) if out else {}
    except json.JSONDecodeError:
        msg = out or (proc.stderr or "").strip() or "kein JSON-Output"
        status = "fail" if proc.returncode != 0 else "no-check"
        return {"checks": [{"name": name, "status": status, "message": msg[:500]}]}
    if not isinstance(data, dict):
        return {"checks": [{"name": name, "status": "fail",
                            "message": "Script-Output ist kein JSON-Objekt"}]}
    if proc.returncode != 0 and not collect_statuses(data):
        data.setdefault("checks", []).append(
            {"name": name, "status": "fail", "message": "Exit-Code %s" % proc.returncode})
    return data


def merge(top, host, fragment):
    """Merged ein Script-Fragment (well-known keys) in host bzw. den Wurzel-Knoten."""
    if not isinstance(fragment, dict):
        return
    metrics = fragment.get("metrics")
    if isinstance(metrics, dict):
        host.setdefault("metrics", {}).update(metrics)
    for key in ("checks",):
        items = fragment.get(key)
        if isinstance(items, list):
            host.setdefault(key, []).extend(items)
    for key in ("services", "children"):
        items = fragment.get(key)
        if isinstance(items, list):
            top.setdefault(key, []).extend(items)


def collect(want_tags=None, skip_tags=None, depth=None):
    """Baut den lokalen Knoten-Baum: Host-Meta + Drop-in-Outputs + Status."""
    c = cfg()
    want_tags = set(want_tags or [])
    skip_tags = set(skip_tags or [])
    if depth == "local":
        skip_tags |= {"recursive"}

    meta = load_host_meta(c)
    host = {k: v for k, v in meta.items() if not k.startswith("_")}
    host.setdefault("metrics", {})
    host.setdefault("checks", [])
    top = {"schema": SCHEMA, "ts": now_iso(), "status": "ok",
           "host": host, "services": [], "children": [], "skipped": []}

    scripts = discover_scripts(c["d_dir"])
    run, skipped = select_scripts(scripts, want_tags, skip_tags)
    top["skipped"] = skipped

    if run:
        with concurrent.futures.ThreadPoolExecutor(max_workers=min(8, len(run))) as ex:
            futures = {ex.submit(cached_run_script, n, p, c, ttl): n for n, p, _, ttl in run}
            for fut in concurrent.futures.as_completed(futures):
                merge(top, host, fut.result())

    # Status rein generisch: host = schlimmster der Host-eigenen Signale,
    # Wurzel = schlimmster aus host + services + children.
    host_signals = collect_statuses(host.get("metrics", {})) + \
        [ch.get("status") for ch in host.get("checks", []) if isinstance(ch, dict)]
    host["status"] = worst(host_signals)
    top["status"] = worst([host["status"]] +
                          collect_statuses(top["services"]) +
                          collect_statuses(top["children"]))
    return top


# --- Badges (pro Tag, Kuma-Ersatz) -----------------------------------------

_BADGE_COLOR = {"ok": "#4c1", "warn": "#dfb317", "degraded": "#dfb317",
                "fail": "#e05d44", "unreachable": "#e05d44", "timeout": "#e05d44"}
_BADGE_TEXT = {"ok": "up", "warn": "warn", "degraded": "degraded",
               "fail": "down", "unreachable": "down", "timeout": "down"}


def walk_nodes(node):
    """Iteriert rekursiv ueber alle Knoten (Wurzel -> children)."""
    yield node
    for ch in node.get("children", []) or []:
        yield from walk_nodes(ch)


def tag_status(tree, tag):
    """Schlimmster Status aller Dinge mit <tag>: Knoten-zones/tags und check.tags.

    None, wenn der Tag nirgends vorkommt (Badge zeigt dann 'n/a').
    """
    statuses, found = [], False
    for n in walk_nodes(tree):
        h = n.get("host", {}) if isinstance(n, dict) else {}
        node_tags = set(h.get("zones") or []) | set(h.get("tags") or [])
        if tag in node_tags:
            statuses.append(h.get("status")); found = True
        for chk in h.get("checks") or []:
            if isinstance(chk, dict) and tag in (chk.get("tags") or []):
                statuses.append(chk.get("status")); found = True
    return worst(statuses) if found else None


def _esc(s):
    """Minimales HTML/XML-Escaping fuer Text in Attributen/Inhalten."""
    return (str(s) if s is not None else "").replace("&", "&amp;").replace(
        "<", "&lt;").replace(">", "&gt;").replace('"', "&quot;")


def _status_color(st):
    """Ampel-Farbe je Status (fuer Badge/Statusseite)."""
    return {"ok": "#2ecc71", "warn": "#f1c40f", "degraded": "#f1c40f",
            "fail": "#e74c3c", "unreachable": "#e74c3c", "timeout": "#e74c3c"}.get(st, "#95a5a6")


def render_badge(label, status):
    """Rendert ein shields-artiges SVG-Status-Badge (links Label, rechts Status)."""
    msg = _BADGE_TEXT.get(status, status or "n/a")
    color = _BADGE_COLOR.get(status, "#9f9f9f")
    lw = 7 * len(label) + 12
    mw = 7 * len(msg) + 12
    w = lw + mw
    return ('<svg xmlns="http://www.w3.org/2000/svg" width="%d" height="20" role="img" '
            'aria-label="%s: %s">'
            '<rect width="%d" height="20" fill="#555"/>'
            '<rect x="%d" width="%d" height="20" fill="%s"/>'
            '<g fill="#fff" text-anchor="middle" font-family="DejaVu Sans,Verdana,Geneva,sans-serif" '
            'font-size="11">'
            '<text x="%d" y="14">%s</text>'
            '<text x="%d" y="14">%s</text>'
            '</g></svg>') % (w, _esc(label), _esc(msg), lw, lw, mw, color,
                             lw // 2, _esc(label), lw + mw // 2, _esc(msg))


_STATUS_PAGE = """<!doctype html><html lang="de"><head><meta charset="utf-8">
<meta name="viewport" content="width=device-width,initial-scale=1">
<meta http-equiv="refresh" content="60"><title>42i-hc Status</title><style>
*{box-sizing:border-box}body{margin:0;background:#1b1f24;color:#e6e6e6;
font-family:system-ui,-apple-system,Segoe UI,Roboto,sans-serif;font-size:14px}
header{padding:14px 20px;border-bottom:1px solid #333;display:flex;align-items:baseline;gap:14px}
h1{font-size:18px;margin:0}.ts{color:#8a929c;font-size:12px}
.grid{display:grid;grid-template-columns:repeat(auto-fill,minmax(280px,1fr));gap:12px;padding:16px 20px}
.card{background:#23282e;border:1px solid #333;border-radius:8px;padding:10px 12px}
.card h2{font-size:14px;margin:0 0 6px;display:flex;align-items:center;gap:8px}
.card h2 small{margin-left:auto;color:#8a929c;font-weight:400}
ul{list-style:none;margin:0;padding:0}li{padding:2px 0;display:flex;align-items:center;gap:7px}
.dot{width:9px;height:9px;border-radius:50%;flex:0 0 auto}.dot.big{width:12px;height:12px}
em{font-style:normal;color:#cfd6dd;font-size:12px}.msg{color:#8a929c;font-size:12px;
overflow:hidden;text-overflow:ellipsis;white-space:nowrap}.ok{color:#6b7480;font-size:12px}
h3{padding:4px 20px 0;margin:8px 0 0;font-size:13px;color:#8a929c;text-transform:uppercase;letter-spacing:.04em}
</style></head><body><header><h1>42i-hc Status</h1>
<span class="badge" style="display:inline-flex;align-items:center;gap:6px">
<span class="dot big" style="background:@@OC@@"></span>@@OV@@</span>
<span class="ts">@@TS@@</span></header>
<h3>Gruppen (Tags)</h3><div class="grid">@@CARDS@@</div>
<h3>Infrastruktur-Probleme</h3><div class="grid"><div class="card"><ul>@@INFRA@@</ul></div></div>
</body></html>"""


def render_status(tree):
    """Rendert eine schlanke HTML-Statusseite: pro Tag der Status + was ihn runterzieht."""
    tagmap = {}
    for n in walk_nodes(tree):
        h = n.get("host", {}) if isinstance(n, dict) else {}
        for chk in h.get("checks") or []:
            if isinstance(chk, dict):
                for tg in chk.get("tags") or []:
                    tagmap.setdefault(tg, []).append(chk)

    def li(status, name, extra=""):
        return ('<li><span class="dot" style="background:%s"></span>%s '
                '<span class="msg">%s</span></li>') % (_status_color(status), _esc(name), _esc(extra))

    cards = []
    for tag in sorted(tagmap, key=lambda t: (-STATUS_ORDER.get(worst([c.get("status") for c in tagmap[t]]), 0), t)):
        checks = tagmap[tag]
        agg = worst([c.get("status") for c in checks])
        bad = [c for c in checks if STATUS_ORDER.get(c.get("status"), 0) > 0]
        bad.sort(key=lambda c: -STATUS_ORDER.get(c.get("status"), 0))
        items = "".join(
            '<li><span class="dot" style="background:%s"></span>%s <em>%s</em> '
            '<span class="msg">%s</span></li>' % (_status_color(c.get("status")), _esc(c.get("name")),
                                                  _esc(c.get("status")), _esc(c.get("message", "")))
            for c in bad)
        okn = len(checks) - len(bad)
        if okn:
            items += '<li class="ok">+ %d ok</li>' % okn
        cards.append('<div class="card"><h2><span class="dot big" style="background:%s"></span>%s '
                     '<small>%d</small></h2><ul>%s</ul></div>'
                     % (_status_color(agg), _esc(tag), len(checks), items or '<li class="ok">—</li>'))

    infra = []
    for n in walk_nodes(tree):
        h = n.get("host", {}) if isinstance(n, dict) else {}
        hn = h.get("name", "?")
        m = h.get("metrics", {}) or {}
        for d in m.get("disk") or []:
            if d.get("status") in ("warn", "fail"):
                infra.append((d.get("status"), hn, "%s %s%%" % (d.get("mount"), d.get("pct"))))
        mem = m.get("mem", {})
        if isinstance(mem, dict) and mem.get("status") in ("warn", "fail"):
            infra.append((mem.get("status"), hn, "RAM %s%%" % mem.get("pct")))
        for svc in n.get("services") or []:
            if isinstance(svc, dict) and svc.get("status") in ("warn", "fail"):
                infra.append((svc.get("status"), "%s/%s" % (hn, svc.get("name", "")), svc.get("message", "")))
        if n.get("status") in ("unreachable", "timeout"):
            infra.append((n.get("status"), hn, "Knoten nicht erreichbar"))
    infra.sort(key=lambda x: -STATUS_ORDER.get(x[0], 0))
    infra_html = "".join(li(s, hn, extra) for s, hn, extra in infra) or '<li class="ok">keine</li>'

    overall = tree.get("status", "ok")
    return (_STATUS_PAGE.replace("@@OC@@", _status_color(overall))
            .replace("@@OV@@", _esc(overall))
            .replace("@@TS@@", _esc(tree.get("ts", "")))
            .replace("@@CARDS@@", "".join(cards))
            .replace("@@INFRA@@", infra_html))


# --- HTTP ------------------------------------------------------------------

_CACHE = {}
_CACHE_LOCK = threading.Lock()
_INFLIGHT = {}                      # key -> Event; markiert einen laufenden collect
_INFLIGHT_LOCK = threading.Lock()


def cached(key, ttl, producer):
    """TTL-Cache + Single-Flight um producer() herum, pro Key.

    Genau ein Leader pro Key ruft producer(); parallele Anfragen auf denselben
    Key warten auf dessen Ergebnis statt selbst zu laufen. So löst ein Dashboard,
    das mehrere Badges gleichzeitig lädt, nicht N teure Script-/collect-Läufe aus
    (Schutz für schwache Collector wie den Raspberry Pi) und alle sehen denselben
    konsistenten Stand statt flackernder Teilergebnisse.
    """
    nowm = time.monotonic()
    with _CACHE_LOCK:
        hit = _CACHE.get(key)
        if hit and (nowm - hit[0]) < ttl:
            return hit[1]
    with _INFLIGHT_LOCK:
        ev = _INFLIGHT.get(key)
        leader = ev is None
        if leader:
            ev = threading.Event()
            _INFLIGHT[key] = ev
    if not leader:
        ev.wait(timeout=ttl + 30)
        with _CACHE_LOCK:
            hit = _CACHE.get(key)
        return hit[1] if hit else producer()    # Leader scheiterte → selbst
    try:
        result = producer()
        with _CACHE_LOCK:
            _CACHE[key] = (time.monotonic(), result)
        return result
    finally:
        with _INFLIGHT_LOCK:
            _INFLIGHT.pop(key, None)
        ev.set()


def cached_run_script(name, path, c, ttl):
    """run_script mit per-Script-TTL (aus `# hc-ttl:`), unabhängig gecacht.

    Jedes Drop-in cached für sich: volatile Werte (RAM/Load) kurz, stabile
    (Plattenplatz, Repo-Erreichbarkeit) lang. Ein collect führt nur die Scripts
    neu aus, deren Cache abgelaufen ist.
    """
    eff = c["cache_ttl"] if ttl is None else ttl
    return cached(("script", path), eff, lambda: run_script(name, path, c))


# --- Proaktiver Sweep-Snapshot ---------------------------------------------
#
# Statt dass jede Anfrage ein collect() ausloest (und so der erste Badge-Abruf
# nach Cache-Ablauf den vollen Probe-Sweep ~7s abwarten muss), sweept ein
# Hintergrund-Thread alle HC_SWEEP_INTERVAL Sekunden (Default 60) von sich aus
# und legt den vollen Baum als Snapshot ab. Alle Lese-Routen (/badge, /status,
# /node, / ohne Debug-Parameter) bedienen sich NUR aus diesem Snapshot und
# loesen selbst nie eine Collection aus. Minimale Aufloesung = das Intervall.
_SWEEP_DEFAULT = 60.0
_SNAPSHOT_LOCK = threading.Lock()
_SNAPSHOT = None                    # (monotonic_ts, tree) oder None vor 1. Sweep


def sweep_interval():
    """Sweep-Intervall in Sekunden (HC_SWEEP_INTERVAL, Default 60, min. 5)."""
    try:
        return max(5.0, float(env("HC_SWEEP_INTERVAL", str(_SWEEP_DEFAULT))))
    except ValueError:
        return _SWEEP_DEFAULT


def run_sweep():
    """Fuehrt einen vollen collect()-Lauf aus und legt ihn als Snapshot ab."""
    tree = collect()
    with _SNAPSHOT_LOCK:
        global _SNAPSHOT
        _SNAPSHOT = (time.monotonic(), tree)
    return tree


def current_tree():
    """(tree, stale) aus dem Snapshot.

    stale=True, wenn der Snapshot aelter als 3x Intervall ist (Sweeper haengt/
    tot) oder noch keiner existiert -> dann tree=None. So bleibt ein toter
    Sweeper sichtbar (Badge 'stale', /healthz degraded), statt still alte Daten
    auszuliefern.
    """
    with _SNAPSHOT_LOCK:
        snap = _SNAPSHOT
    if snap is None:
        return None, True
    age = time.monotonic() - snap[0]
    return snap[1], age > 3 * sweep_interval()


def snapshot_age():
    """Alter des Snapshots in Sekunden, oder None wenn noch keiner da ist."""
    with _SNAPSHOT_LOCK:
        snap = _SNAPSHOT
    return None if snap is None else time.monotonic() - snap[0]


def without_children(tree):
    """Snapshot-Root ohne children (Peer-Sicht `/?skip=net`).

    Der Parent zieht jedes Kind ueber `/?skip=net`, damit dessen eigener
    Peer-Fanout nicht rekursiv weiterlaeuft. Aus dem vollen Snapshot wird diese
    Sicht durch Weglassen der `children` abgeleitet (kein Recompute) und der
    Wurzel-Status wie in collect() neu als worst(host + services) gebildet.
    """
    node = dict(tree)
    node["children"] = []
    host_status = node.get("host", {}).get("status", "ok")
    node["status"] = worst([host_status] + collect_statuses(node.get("services", [])))
    return node


def sweeper_loop():
    """Hintergrund-Loop: alle sweep_interval() Sekunden ein Sweep, ohne Overlap."""
    interval = sweep_interval()
    while True:
        t0 = time.monotonic()
        try:
            run_sweep()
        except Exception as exc:  # pragma: no cover - Sweeper darf nie sterben
            sys.stderr.write("42i-hc sweep-Fehler: %s\n" % exc)
        time.sleep(max(1.0, interval - (time.monotonic() - t0)))


class Handler(BaseHTTPRequestHandler):
    """HTTP-Routen: /healthz (Liveness), /node (nur Knoten), / (voller Subtree)."""

    server_version = "42i-hc"

    def log_message(self, *args):  # leiser Default
        pass

    def _send(self, code, payload):
        body = json.dumps(payload, ensure_ascii=False, indent=2).encode("utf-8")
        self.send_response(code)
        self.send_header("Content-Type", "application/json; charset=utf-8")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def _send_svg(self, svg):
        body = svg.encode("utf-8")
        self.send_response(200)
        self.send_header("Content-Type", "image/svg+xml; charset=utf-8")
        # Passt zur Sweep-Kadenz: ein Refresh innerhalb des Intervalls holt das
        # Badge nicht mal vom Server; stale-while-revalidate haelt ihn fluessig.
        self.send_header("Cache-Control", "public, max-age=60, stale-while-revalidate=60")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def _send_html(self, html):
        body = html.encode("utf-8")
        self.send_response(200)
        self.send_header("Content-Type", "text/html; charset=utf-8")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def do_GET(self):
        parsed = urlparse(self.path)
        path = parsed.path.rstrip("/") or "/"
        q = parse_qs(parsed.query)
        want = {t for v in q.get("tags", []) for t in v.split(",") if t}
        skip = {t for v in q.get("skip", []) for t in v.split(",") if t}
        depth = (q.get("depth", [None])[0])

        if path == "/healthz":
            # Liveness inkl. Sweeper-Gesundheit: fehlt der Snapshot oder ist er
            # zu alt (Sweeper haengt/tot), meldet healthz 'degraded'/503.
            age = snapshot_age()
            stale = age is None or age > 3 * sweep_interval()
            self._send(503 if stale else 200, {
                "schema": SCHEMA, "service": "42i-hc",
                "status": "degraded" if stale else "ok",
                "sweep_age_s": None if age is None else round(age, 1),
                "ts": now_iso()})
            return

        if path == "/node":
            # Nur der Knoten selbst, aus dem Snapshot abgeleitet: host ist der
            # lokale Knoten (Checks/Metriken) ohne services/children.
            tree, _stale = current_tree()
            if tree is None:
                self._send(503, {"schema": SCHEMA, "status": "warming", "ts": now_iso()})
                return
            host = tree.get("host", {})
            out = {"schema": SCHEMA, "ts": tree.get("ts"), "status": host.get("status", "ok"),
                   "host": host, "skipped": tree.get("skipped", [])}
            self._send(503 if out["status"] == "fail" else 200, out)
            return

        if path.startswith("/badge/"):
            # SVG-Status-Badge fuer einen Tag, rein aus dem Snapshot (kein collect).
            tag = path[len("/badge/"):].strip("/")
            label = (q.get("label", [""])[0] or tag)
            tree, stale = current_tree()
            status = "stale" if (tree is None or stale) else tag_status(tree, tag)
            self._send_svg(render_badge(label, status))
            return

        if path == "/status":
            # Schlanke HTML-Statusseite (Kuma-Dashboard-Ersatz), aus dem Snapshot.
            tree, _stale = current_tree()
            if tree is None:
                self._send_html("<!doctype html><meta charset=utf-8>"
                                "<meta http-equiv=refresh content=5>"
                                "<p>42i-hc waermt auf …</p>")
                return
            self._send_html(render_status(tree))
            return

        if path == "/":
            # Konsumenten- und Peer-Abruf (`/`, `/?skip=net`) aus dem Snapshot;
            # nur arbiträre Debug-Queries (tags/depth/andere skips) rechnen live.
            snapshot_servable = not want and not depth and (not skip or skip == {"net"})
            if snapshot_servable:
                tree, _stale = current_tree()
                if tree is None:
                    self._send(503, {"schema": SCHEMA, "status": "warming", "ts": now_iso()})
                    return
                node = without_children(tree) if skip == {"net"} else tree
            else:
                node = collect(want_tags=want, skip_tags=skip, depth=depth)
            self._send(503 if node["status"] == "fail" else 200, node)
            return

        self._send(404, {"status": "fail", "message": "unbekannte Route: %s" % path})


def serve():
    """Startet den HTTP-Daemon auf 0.0.0.0:HC_PORT (siehe keepalived-VIP-Hinweis).

    Vor dem Binden laeuft ein synchroner Kaltstart-Sweep, damit schon die erste
    Anfrage Daten sieht; danach haelt der Hintergrund-Sweeper den Snapshot warm.
    """
    c = cfg()
    sys.stderr.write("42i-hc serve auf :%d (hc.d=%s, sweep=%ss)\n"
                     % (c["port"], c["d_dir"], sweep_interval()))
    run_sweep()  # Kaltstart: erster Snapshot, bevor wir Anfragen annehmen
    threading.Thread(target=sweeper_loop, name="hc-sweeper", daemon=True).start()
    httpd = ThreadingHTTPServer(("0.0.0.0", c["port"]), Handler)
    try:
        httpd.serve_forever()
    except KeyboardInterrupt:
        pass


def main(argv):
    """CLI-Einstieg: collect (JSON auf stdout) oder serve (HTTP-Daemon)."""
    args = argv[1:]
    cmd = args[0] if args else "collect"
    if cmd == "serve":
        serve()
        return 0
    if cmd == "collect":
        want, skip, depth = set(), set(), None
        i = 1
        while i < len(args):
            if args[i] == "--tags" and i + 1 < len(args):
                want |= {t for t in args[i + 1].split(",") if t}
                i += 2
            elif args[i] == "--skip" and i + 1 < len(args):
                skip |= {t for t in args[i + 1].split(",") if t}
                i += 2
            elif args[i] == "--depth" and i + 1 < len(args):
                depth = args[i + 1]
                i += 2
            else:
                i += 1
        json.dump(collect(want_tags=want, skip_tags=skip, depth=depth),
                  sys.stdout, ensure_ascii=False, indent=2)
        sys.stdout.write("\n")
        return 0
    sys.stderr.write("Usage: hc [collect|serve]\n")
    return 2


if __name__ == "__main__":
    sys.exit(main(sys.argv))
