"""Vom Bericht des Ausführers zu Zielen im gemeinsamen Modell (kern/ziele.py) — Phase 3, 24.09.2026. Je Gerät ein Ziel: der Proxmox-Host (Pakete, Neustart) und jeder freigegebene Gast (Etikett community-script oder watcher, nicht watcher-aus) mit seinen Bausteinen „App“ (Community-Script, Vergleich mit der neuesten Veröffentlichung auf GitHub) und „Pakete“ (Betriebssystem im Gast). Dazu prüft dieser Teil selbst, ob die Weboberfläche jedes Gasts antwortet. Wie ehrlich der Stand ist, steht dabei: Sind die Paketlisten im Gast älter als zwei Wochen, heißt es „unbekannt“ (mit Knopf zum Suchen) statt „aktuell“; ist der Bericht älter als eine halbe Stunde, sagt die Übersicht, dass der Ausführer schweigt. Ist das Update-Skript einer App frisch geändert, bleibt sie „neu“, aber ohne Knopf (karenz.py). Rückweg und Rückfrage sagen, was vor dem Update geschieht: Snapshot, wo keiner geht eine Sicherung, sonst keiner. """ import re import threading import time from datetime import datetime import httpx from kern.github import neueste_version from kern.zeit import LOCAL_TZ from kern.ziele import Aktion, BausteinStand, ZielStand from services.homelab import apps, arcane, kanal, karenz BERICHT_ALT_S = 30 * 60 LISTEN_ALT_S = 14 * 24 * 3600 EIGENES_ETIKETT = "mc2" ERREICHBAR_CACHE_S = 60 _erreichbar_cache: dict[str, tuple[float, bool]] = {} _erreichbar_lock = threading.Lock() def erreichbar(url: str) -> bool: """Antwortet die Weboberfläche? Alles unter 500 zählt (eine Anmeldeseite ist eine Antwort).""" jetzt = time.time() with _erreichbar_lock: if (eintrag := _erreichbar_cache.get(url)) and jetzt - eintrag[0] < ERREICHBAR_CACHE_S: return eintrag[1] try: # Selbstsignierte Zertifikate sind im Heimnetz üblich (PBS, NPMplus); geprüft wird nur, ob jemand antwortet. ok = httpx.get(url, timeout=5.0, verify=False, follow_redirects=False).status_code < 500 except httpx.HTTPError: ok = False with _erreichbar_lock: _erreichbar_cache[url] = (jetzt, ok) return ok def erreichbar_frisch(url: str) -> bool: """Wie erreichbar(), aber ohne den Zwischenspeicher — für die Prüfung direkt nach einem Update.""" with _erreichbar_lock: _erreichbar_cache.pop(url, None) return erreichbar(url) def _teile(version: str) -> tuple[int, ...]: return tuple(int(x) for x in re.findall(r"\d+", version.split("-")[0])[:4]) def neuer(verfuegbar: str, installiert: str) -> bool | None: """Ist verfuegbar neuer als installiert? None, wenn sich das nicht sagen lässt.""" a, b = _teile(verfuegbar), _teile(installiert) if not a or not b: return None return a > b def _datum(ts: float | None) -> str: return datetime.fromtimestamp(ts, LOCAL_TZ).strftime("%d.%m.%Y") if ts else "?" def _rueckweg(gast: dict) -> str: """Die Zeile „Rückweg“ der Karte: was vor jedem Update geschieht.""" if gast.get("snapshot_moeglich"): return "Snapshot vor jedem Update" grund = gast.get("snapshot_grund") or "kein Snapshot möglich" if gast.get("sicherung_moeglich"): return f"Sicherung auf „{gast.get('sicherung_speicher') or 'dem Host'}“ vor jedem Update ({grund})" if gast.get("sicherung_grund"): return f"Kein automatischer Rückweg: {grund}. Auch keine Sicherung: {gast['sicherung_grund']}" return grund def _rueckweg_frage(gast: dict) -> str: """Der Satz zum Rückweg in der Rückfrage vor „Jetzt updaten“.""" if gast.get("snapshot_moeglich"): return "Vorher wird ein Snapshot angelegt; ist die Prüfung danach rot, geht es von selbst zurück." if gast.get("sicherung_moeglich"): return "Vorher wird eine Sicherung angelegt; ist die Prüfung danach rot, wird sie zurückgespielt." satz = "Ein Snapshot ist hier nicht möglich — scheitert das Update, gibt es keinen automatischen Rückweg." if gast.get("sicherung_grund"): satz += f" Eine Sicherung geht auch nicht: {gast['sicherung_grund']}" return satz def _app_baustein(gast: dict, app: apps.App, zid: str) -> BausteinStand: installiert = (gast.get("app") or {}).get("version") stand = BausteinStand(id="app", name=app.name, zustand="aktuell", installiert=installiert) if not app.quelle: return stand try: neu = neueste_version(app.quelle) except Exception: neu = None if not neu: stand.zustand, stand.grund = "unbekannt", "Die neueste Version ist gerade nicht abrufbar (GitHub)." return stand stand.verfuegbar = neu["version"] if installiert and installiert.startswith("untagged-"): # Manche Projekte veröffentlichen ohne Versions-Tag; die App kennt dann nur dieses Tag (PVE Scripts Local). if installiert == neu.get("tag_roh"): stand.installiert = neu["version"] return stand stand.zustand, stand.grund = "unbekannt", "Die installierte Fassung lässt sich keiner Version zuordnen." return stand if app.vergleich == "datum": # Docker-Image ohne Versionsnummer: neuer, wenn die Veröffentlichung jünger ist als das Image. bild = re.search(r"\d{4}-\d{2}-\d{2}", installiert or "") ist_neuer = (neu["datum"] or "") > bild.group(0) if bild else None else: ist_neuer = neuer(neu["version"], installiert) if installiert else None if ist_neuer is None: stand.zustand, stand.grund = "unbekannt", "Die installierte Version ließ sich nicht lesen." elif ist_neuer and app.weg != "skript": stand.zustand, stand.kurz = "neu", f"{installiert} → {neu['version']}" stand.grund = f"{app.name} aktualisiert sich über die eigene Oberfläche; das Community-Script tut es nicht." elif ist_neuer: stand.zustand, stand.kurz = "neu", f"{installiert} → {neu['version']}" # Frisch geänderte Update-Skripte erst nach der Wartezeit (karenz.py): „neu“, aber ohne Knopf. wartezeit = karenz.pruefen(app.kennung) if wartezeit["gesperrt"]: stand.grund = wartezeit["gesperrt"] return stand frage = f"{app.name} jetzt aktualisieren? {_rueckweg_frage(gast)}" if not wartezeit["geprueft"]: frage += f" {karenz.NICHT_GEPRUEFT}" stand.aktion = Aktion("POST", f"/api/homelab/ziele/{zid}/update", frage) return stand def _os_baustein(gast: dict, zid: str, jetzt: float) -> BausteinStand: os_stand = gast.get("os_updates") or {} stand = BausteinStand(id="os", name="Pakete", zustand="aktuell") anzahl, listen = os_stand.get("anzahl"), os_stand.get("listen_stand") suchen = Aktion("POST", f"/api/homelab/ziele/{zid}/suchen", "Die Paketlisten im Gast jetzt erneuern?", label="Nach Updates suchen") if anzahl is None: stand.zustand, stand.grund = "unbekannt", "Die Pakete ließen sich nicht lesen." elif listen and jetzt - listen > LISTEN_ALT_S: stand.zustand, stand.aktion = "unbekannt", suchen stand.grund = f"Die Paketlisten sind vom {_datum(listen)}; was fehlt, weiß erst eine neue Suche." elif anzahl > 0: stand.zustand, stand.kurz = "neu", f"{anzahl} Pakete" stand.aktion = Aktion("POST", f"/api/homelab/ziele/{zid}/os-update", f"{anzahl} Pakete im Gast jetzt einspielen? {_rueckweg_frage(gast)}") return stand def _arcane_bausteine(basis: str | None, zid: str) -> list[BausteinStand]: """Arcane selbst (öffentliche Version) und die Docker-Images (mit API-Schlüssel).""" app = arcane.app_version(basis) if basis else None selbst = BausteinStand(id="arcane", name="Arcane", zustand="unbekannt", installiert=app["installiert"] if app else None, verfuegbar=app["neu"] if app else None) if not app: selbst.grund = "Arcane antwortet nicht." elif app["update"]: selbst.zustand, selbst.kurz = "neu", f"{app['installiert']} → {app['neu']}" selbst.grund = "Arcane spielt sein eigenes Update über die eigene Oberfläche ein (Einstellungen)." else: selbst.zustand = "aktuell" docker = BausteinStand(id="docker", name="Docker-Images", zustand="unbekannt") if not arcane.schluessel(): docker.grund = ("Es fehlt ein Arcane-API-Schlüssel (MC_ARCANE_KEY): ohne ihn sieht der Orchestrator " "die Images nicht.") elif basis: try: images = arcane.images_mit_update(basis) except Exception: docker.grund = "Arcane verweigert die Auskunft oder antwortet nicht (Schlüssel prüfen)." else: if images: docker.zustand, docker.kurz = "neu", f"{len(images)} Images" docker.grund = ", ".join(i["name"] for i in images[:6]) + (" …" if len(images) > 6 else "") probe = not arcane.echt() docker.aktion = Aktion( "POST", f"/api/homelab/ziele/{zid}/docker", ("Arcane prüft im Probelauf, welche Container es aktualisieren würde — geändert wird noch nichts." if probe else "Arcane aktualisiert jetzt die Container mit neuen Images; Daten-Volumes bleiben."), label="Probelauf" if probe else "Jetzt updaten") else: docker.zustand = "aktuell" return [selbst, docker] def _gast_ziel(gast: dict, jetzt: float) -> ZielStand: art = "container" if gast["art"] == "lxc" else "vm" zid = f"{'ct' if art == 'container' else 'vm'}-{gast['vmid']}" app = apps.app_fuer((gast.get("app") or {}).get("kennung"), gast.get("name")) url = apps.adresse(app, gast.get("ip")) laeuft = gast.get("status") == "running" ziel = ZielStand(id=zid, name=app.name if app else str(gast.get("name") or zid), art=art, instanz="homelab", erreichbar=(erreichbar(url) if laeuft else False) if url else (laeuft or None), rueckweg=_rueckweg(gast)) if not laeuft: ziel.bausteine.append(BausteinStand(id="status", name="Status", zustand="unbekannt", grund=f"Der Gast ist {gast.get('status') or 'unbekannt'}.")) return ziel if app and app.kennung == "arcane": # Arcane läuft über seine eigene Schnittstelle, nicht über den Ausführer: kein Etikett nötig. ziel.bausteine += _arcane_bausteine(url.rstrip("/") if url else None, zid) return ziel if not gast.get("erlaubt"): ziel.bausteine.append(BausteinStand( id="freigabe", name="Freigabe", zustand="unbekannt", grund="Nicht freigegeben: Das Etikett „watcher“ fehlt (oder „watcher-aus“ ist gesetzt).")) return ziel if app and app.kennung in apps.KATALOG: ziel.bausteine.append(_app_baustein(gast, app, zid)) if gast["art"] == "lxc": ziel.bausteine.append(_os_baustein(gast, zid, jetzt)) return ziel def _host_ziel(host: dict) -> ZielStand: ziel = ZielStand(id="pve", name="Proxmox-Host", art="proxmox-host", instanz="homelab", erreichbar=True, rueckweg="Kein Snapshot möglich; der Host bleibt beim Update in Betrieb, neu gestartet wird getrennt.") updates = host.get("updates") pakete = BausteinStand(id="pakete", name="Proxmox VE und Debian", zustand="aktuell", installiert=host.get("version")) if updates is None: pakete.zustand, pakete.grund = "unbekannt", host.get("fehler") or "Die Update-Liste fehlt im Bericht." elif updates: pakete.zustand, pakete.kurz = "neu", f"{len(updates)} Pakete" pakete.aktion = Aktion("POST", "/api/homelab/ziele/pve/update", f"{len(updates)} Pakete auf dem Proxmox-Host einspielen? Alle Gäste laufen weiter. " "Einen Neustart löst das nicht aus — der ist ein eigener Knopf.") ziel.bausteine.append(pakete) if host.get("neustart_noetig"): ziel.bausteine.append(BausteinStand( id="neustart", name="Neustart", zustand="neu", kurz="steht an", aktion=Aktion("POST", "/api/homelab/ziele/pve/neustart", "Den Proxmox-Host jetzt neu starten? ALLE Container und die Arcane-VM sind dann ein paar " "Minuten weg. Die KI-Box prüft danach von außen, ob alles wiederkommt.", label="Jetzt neu starten"))) return ziel def ziele() -> dict: """Alle Ziele des Homelabs aus dem letzten Bericht, dazu wie frisch er ist.""" eingang = kanal.bericht() jetzt = time.time() zuletzt = kanal.zuletzt() ausfuehrer = {"verbunden": bool(zuletzt and jetzt - zuletzt < BERICHT_ALT_S), "zuletzt": zuletzt} if not eingang: return {"ziele": [], "bericht": None, "ausfuehrer": ausfuehrer} b = eingang["bericht"] liste = [_host_ziel(b.get("host") or {})] # Der Container des Orchestrators selbst (Etikett „mc2“) ist kein Gerät, das er pflegt. liste += [_gast_ziel(g, jetzt) for g in b.get("gaeste") or [] if EIGENES_ETIKETT not in (g.get("etiketten") or [])] for z in liste: z.geprueft = eingang["empfangen"] return {"ziele": [z.als_dict() for z in liste], "bericht": {"empfangen": eingang["empfangen"], "veraltet": jetzt - eingang["empfangen"] > BERICHT_ALT_S}, "ausfuehrer": ausfuehrer} def erreichbarkeit() -> list[dict]: """Für den Wächter: jede Weboberfläche laufender, freigegebener Gäste — ohne GitHub-Abfragen. Gäste mitten in einem Update-Lauf bleiben außen vor: Sicherung, Update und Zurückspielen legen sie kurz still, und was dabei herauskommt, meldet der Lauf selbst (sonst käme ein zweiter Alarm samt „wieder ok“).""" from services.homelab import updates # updates importiert dieses Modul eingang = kanal.bericht() im_update = updates.laufende_ziele() ergebnis = [] for g in (eingang or {}).get("bericht", {}).get("gaeste") or []: app = apps.app_fuer((g.get("app") or {}).get("kennung"), g.get("name")) url = apps.adresse(app, g.get("ip")) zid = f"{'ct' if g['art'] == 'lxc' else 'vm'}-{g['vmid']}" if url and g.get("erlaubt") and zid not in im_update: ergebnis.append({"id": zid, "name": app.name if app else g.get("name"), "url": url, "laeuft": g.get("status") == "running", "ok": g.get("status") == "running" and erreichbar(url)}) return ergebnis def gast(zid: str) -> dict | None: """Der Gast zu einer Ziel-ID aus dem letzten Bericht (ct-104 → vmid 104).""" m = re.fullmatch(r"(ct|vm)-(\d+)", zid) eingang = kanal.bericht() if not m or not eingang: return None return next((g for g in eingang["bericht"].get("gaeste") or [] if g.get("vmid") == int(m.group(2))), None)