HitonabiandClaude Opus 5.5
9da666d04e
Der Steward des Homelab-Teils prüft jede Minute, ob das Fenster offen ist, und arbeitet dann einmal je Tag den Plan ab: je Gerät nach seiner Regel — nur melden, Sicherheitsupdates automatisch (Standard, User-Entscheid) oder alles im Fenster —, mit den bekannten Update-Läufen (Rückweg vorher, Prüfung danach, bei Rot zurück). Nach dem Fenster beginnt kein Schritt mehr, Erfolge kommen als eine Meldung am Ende, Fehler sofort; der Proxmox-Host startet nie von selbst neu. Der Ausführer meldet dafür die Namen der Sicherheitspakete und spielt auf Wunsch nur diese ein (Gäste und Host); neuer Baustein „sicherheit“. Dazu „Version überspringen“ je App, die Einstellungen unter Einstellungen → Homelab und „Wartung“ statt „Störung“ im Cockpit, solange das Fenster läuft. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
282 lines
12 KiB
Python
282 lines
12 KiB
Python
"""Wartungsfenster und Update-Regel je Gerät (Ausbauplan Welle 3, Punkt 17, seit 25.09.2026).
|
||
|
||
User-Entscheid 25.09.2026: Fenster sonntags 03:00–04:15; zu Beginn gilt für jedes Gerät „Sicherheitsupdates
|
||
automatisch“, alles andere wird nur gemeldet. Je Gerät einstellbar (Einstellungen → Homelab):
|
||
melden nichts von selbst — das Update steht an und wartet auf den Knopf
|
||
sicherheit im Fenster nur die Pakete, die eine Sicherheitslücke schließen (Debian-Suite „-security“, Alpine über
|
||
die Sicherheitsdatenbank, beim Proxmox-Host ebenso — neu gestartet wird er nie von selbst)
|
||
alles im Fenster alles, was einen Knopf hat: App, Pakete, Docker-Images (nie der Neustart des Hosts)
|
||
Dazu je App „Version überspringen“: Diese Version gilt als aktuell, bis eine neuere kommt.
|
||
|
||
Der Steward (Rolle homelab) ruft jede Minute takt(): Ist das Fenster offen und heute noch nicht gelaufen, arbeitet er
|
||
den Plan Gerät für Gerät ab — mit den Update-Läufen aus updates.py (Rückweg vorher, Prüfung danach, bei Rot zurück).
|
||
Nach dem Ende des Fensters beginnt kein Schritt mehr. Einzelne Erfolge bleiben still (wie bei „Alle aktualisieren“),
|
||
Fehler und Rückwege gehen sofort raus; am Ende kommt eine Zusammenfassung. Solange er läuft, zeigt das Cockpit
|
||
„Wartung“ statt „Störung“.
|
||
"""
|
||
|
||
import asyncio
|
||
import json
|
||
import logging
|
||
import re
|
||
import threading
|
||
import time
|
||
from datetime import datetime, timedelta
|
||
from pathlib import Path
|
||
|
||
from kern.einstellungen import einstellungen
|
||
from kern.zeit import LOCAL_TZ
|
||
|
||
from services import announce
|
||
|
||
log = logging.getLogger(__name__)
|
||
|
||
REGELN = ("melden", "sicherheit", "alles")
|
||
REGEL_TEXT = {"melden": "nur melden", "sicherheit": "Sicherheitsupdates automatisch", "alles": "alles automatisch"}
|
||
TAGE = ("Mo", "Di", "Mi", "Do", "Fr", "Sa", "So")
|
||
STANDARD = {"fenster": {"tag": 6, "von": "03:00", "bis": "04:15"}, "standard": "sicherheit", "regeln": {},
|
||
"uebersprungen": {}}
|
||
IM_FENSTER = {"app", "os", "pakete", "docker"} # nie: neustart, kernel, probe
|
||
BETREFF = "[Homelab-Wartung]"
|
||
TAKT_S = 60
|
||
LAUF_ZEITLIMIT_S = 70 * 60
|
||
_UHR = re.compile(r"([01]\d|2[0-3]):[0-5]\d")
|
||
_ZIEL = re.compile(r"pve|(?:ct|vm)-\d+")
|
||
_lock = threading.Lock()
|
||
|
||
|
||
def _pfad(name: str) -> Path:
|
||
return einstellungen().daten_dir / name
|
||
|
||
|
||
def _json_lesen(name: str, leer: dict) -> dict:
|
||
try:
|
||
daten = json.loads(_pfad(name).read_text(encoding="utf-8"))
|
||
except (OSError, ValueError):
|
||
return json.loads(json.dumps(leer))
|
||
return daten if isinstance(daten, dict) else json.loads(json.dumps(leer))
|
||
|
||
|
||
def _json_schreiben(name: str, daten: dict) -> None:
|
||
pfad = _pfad(name)
|
||
pfad.parent.mkdir(parents=True, exist_ok=True)
|
||
tmp = pfad.with_suffix(".tmp")
|
||
tmp.write_text(json.dumps(daten, ensure_ascii=False, indent=1), encoding="utf-8")
|
||
tmp.replace(pfad)
|
||
|
||
|
||
def lesen() -> dict:
|
||
daten = _json_lesen("homelab-wartung.json", STANDARD)
|
||
return {**json.loads(json.dumps(STANDARD)), **daten}
|
||
|
||
|
||
def stand_lesen() -> dict:
|
||
return _json_lesen("homelab-wartung-stand.json", {})
|
||
|
||
|
||
# --- Regeln, Fenster, Überspringen --------------------------------------------------------------------------------
|
||
|
||
def regel(zid: str) -> str:
|
||
e = lesen()
|
||
return e["regeln"].get(zid) or e["standard"]
|
||
|
||
|
||
def uebersprungen(zid: str, baustein: str) -> str | None:
|
||
return lesen()["uebersprungen"].get(f"{zid}:{baustein}")
|
||
|
||
|
||
def ueberspringen(zid: str, baustein: str, version: str | None) -> dict:
|
||
"""POST/DELETE /api/homelab/ziele/{zid}/ueberspringen: eine Version überspringen oder wieder anbieten."""
|
||
if not _ZIEL.fullmatch(zid) or baustein != "app":
|
||
return {"ok": False, "detail": "Überspringen geht nur für die App eines Geräts."}
|
||
with _lock:
|
||
e = lesen()
|
||
schluessel = f"{zid}:{baustein}"
|
||
if version:
|
||
e["uebersprungen"][schluessel] = str(version)[:80]
|
||
else:
|
||
e["uebersprungen"].pop(schluessel, None)
|
||
_json_schreiben("homelab-wartung.json", e)
|
||
return {"ok": True, "text": (f"Version {version} wird übersprungen, bis eine neuere kommt." if version
|
||
else "Die Version wird wieder angeboten.")}
|
||
|
||
|
||
def _minuten(uhr: str) -> int:
|
||
h, m = uhr.split(":")
|
||
return int(h) * 60 + int(m)
|
||
|
||
|
||
def im_fenster(jetzt: float | None = None, fenster: dict | None = None) -> bool:
|
||
f = fenster or lesen()["fenster"]
|
||
t = datetime.fromtimestamp(time.time() if jetzt is None else jetzt, LOCAL_TZ)
|
||
minute = t.hour * 60 + t.minute
|
||
return t.weekday() == int(f["tag"]) and _minuten(f["von"]) <= minute < _minuten(f["bis"])
|
||
|
||
|
||
def naechstes(jetzt: float | None = None, fenster: dict | None = None) -> float:
|
||
"""Beginn des nächsten Fensters (Unix-Sekunden); läuft eins gerade, dessen Beginn."""
|
||
f = fenster or lesen()["fenster"]
|
||
t = datetime.fromtimestamp(time.time() if jetzt is None else jetzt, LOCAL_TZ)
|
||
von = _minuten(f["von"])
|
||
for tage in range(8):
|
||
tag = (t + timedelta(days=tage)).replace(hour=von // 60, minute=von % 60, second=0, microsecond=0)
|
||
if tag.weekday() == int(f["tag"]) and (tag + timedelta(minutes=_minuten(f["bis"]) - von)) > t:
|
||
return tag.timestamp()
|
||
return t.timestamp()
|
||
|
||
|
||
def fenster_text(fenster: dict | None = None) -> str:
|
||
f = fenster or lesen()["fenster"]
|
||
return f"{TAGE[int(f['tag'])]} {f['von']}–{f['bis']}"
|
||
|
||
|
||
def setzen(neu: dict) -> dict:
|
||
"""POST /api/homelab/wartung: Fenster, Standardregel und Regeln je Gerät (nur was mitkommt, wird geändert)."""
|
||
with _lock:
|
||
e = lesen()
|
||
if "fenster" in neu:
|
||
f = neu["fenster"] if isinstance(neu["fenster"], dict) else {}
|
||
tag, von, bis = f.get("tag"), str(f.get("von") or ""), str(f.get("bis") or "")
|
||
if not (isinstance(tag, int) and 0 <= tag <= 6 and _UHR.fullmatch(von) and _UHR.fullmatch(bis)
|
||
and _minuten(von) < _minuten(bis)):
|
||
return {"ok": False, "detail": "Das Fenster braucht einen Tag und eine Uhrzeit von–bis am selben Tag."}
|
||
e["fenster"] = {"tag": tag, "von": von, "bis": bis}
|
||
if "standard" in neu:
|
||
if neu["standard"] not in REGELN:
|
||
return {"ok": False, "detail": "Unbekannte Regel."}
|
||
e["standard"] = neu["standard"]
|
||
if "regeln" in neu:
|
||
regeln = neu["regeln"] if isinstance(neu["regeln"], dict) else {}
|
||
if not all(_ZIEL.fullmatch(str(k)) and (v in REGELN or v is None) for k, v in regeln.items()):
|
||
return {"ok": False, "detail": "Unbekanntes Gerät oder unbekannte Regel."}
|
||
for k, v in regeln.items():
|
||
if v is None:
|
||
e["regeln"].pop(k, None)
|
||
else:
|
||
e["regeln"][k] = v
|
||
_json_schreiben("homelab-wartung.json", e)
|
||
return {"ok": True, "text": f"Gespeichert. Wartungsfenster {fenster_text(e['fenster'])}."}
|
||
|
||
|
||
# --- Plan und Lauf -------------------------------------------------------------------------------------------------
|
||
|
||
def plan(ziele: list[dict]) -> list[dict]:
|
||
"""Was im Fenster laufen soll: je Gerät nach seiner Regel, in der Reihenfolge der Geräteliste (Host zuerst)."""
|
||
schritte = []
|
||
for z in ziele:
|
||
r = regel(z["id"])
|
||
if r == "melden":
|
||
continue
|
||
for b in z.get("bausteine") or []:
|
||
if b.get("zustand") != "neu" or not b.get("aktion") or b.get("id") not in IM_FENSTER:
|
||
continue
|
||
if r == "alles":
|
||
schritte.append({"ziel": z["id"], "name": z["name"], "baustein": b["id"]})
|
||
elif b["id"] in ("os", "pakete") and b.get("sicherheit"):
|
||
schritte.append({"ziel": z["id"], "name": z["name"], "baustein": "sicherheit",
|
||
"anzahl": b["sicherheit"]})
|
||
return schritte
|
||
|
||
|
||
def _warten_auf(lauf_id: str, updates) -> dict | None:
|
||
ende = time.monotonic() + LAUF_ZEITLIMIT_S
|
||
while time.monotonic() < ende:
|
||
lauf = updates.lauf_lesen(lauf_id)
|
||
if lauf and lauf.get("status") != "laeuft":
|
||
return lauf
|
||
time.sleep(5)
|
||
return None
|
||
|
||
|
||
def _abarbeiten(stand: dict) -> None:
|
||
from services.homelab import updates
|
||
kennung = {"id": f"wartung-{stand['datum']}", "start": stand["start"], "anlass": "Wartungsfenster"}
|
||
for schritt in stand["schritte"]:
|
||
if not im_fenster():
|
||
schritt.update(ergebnis="nicht-mehr", text="Das Fenster war zu Ende.")
|
||
continue
|
||
antwort = updates.starten(schritt["ziel"], schritt["baustein"], sammellauf=kennung)
|
||
if not antwort.get("ok"):
|
||
schritt.update(ergebnis="fehler", text=antwort.get("detail"))
|
||
else:
|
||
lauf = _warten_auf(antwort["lauf"], updates) or {}
|
||
schritt.update(lauf=antwort["lauf"], ergebnis=lauf.get("ergebnis") or "offen", text=lauf.get("text"))
|
||
_json_schreiben("homelab-wartung-stand.json", stand)
|
||
stand.update(status="fertig", ende=time.time())
|
||
_json_schreiben("homelab-wartung-stand.json", stand)
|
||
_melden(zusammenfassung(stand))
|
||
|
||
|
||
def zusammenfassung(stand: dict) -> str:
|
||
"""Die eine Meldung am Ende des Fensters."""
|
||
def namen(ergebnis: str) -> str:
|
||
return ", ".join(f"{s['name']} ({'Sicherheitsupdates' if s['baustein'] == 'sicherheit' else s['baustein']})"
|
||
for s in stand["schritte"] if s.get("ergebnis") == ergebnis)
|
||
teile = []
|
||
if eingespielt := namen("eingespielt"):
|
||
teile.append(f"Eingespielt: {eingespielt}.")
|
||
if zurueck := namen("zurueckgerollt"):
|
||
teile.append(f"Zurückgerollt (läuft wie vorher): {zurueck}.")
|
||
if gescheitert := namen("fehler"):
|
||
teile.append(f"Gescheitert: {gescheitert}.")
|
||
if offen := namen("nicht-mehr"):
|
||
teile.append(f"Nicht mehr im Fenster: {offen} — nächste Woche.")
|
||
kopf = f"Wartung {fenster_text()}: "
|
||
return kopf + (" ".join(teile) if teile else "nichts zu tun.")
|
||
|
||
|
||
def _melden(text: str) -> None:
|
||
try:
|
||
announce.notify_telegram(BETREFF, text)
|
||
except Exception:
|
||
log.warning("wartung: Meldung ging nicht raus", exc_info=True)
|
||
|
||
|
||
def takt(jetzt: float | None = None) -> bool:
|
||
"""Jede Minute (Steward): im offenen Fenster einmal je Tag den Plan starten. Rückgabe: ob ein Lauf begann."""
|
||
from services.homelab import inventar, sammellauf, updates
|
||
jetzt = time.time() if jetzt is None else jetzt
|
||
if not im_fenster(jetzt):
|
||
return False
|
||
heute = datetime.fromtimestamp(jetzt, LOCAL_TZ).date().isoformat()
|
||
with _lock:
|
||
if stand_lesen().get("datum") == heute:
|
||
return False
|
||
# Läuft gerade ein Update von Hand oder „Alle aktualisieren“, wartet das Fenster eine Minute.
|
||
if sammellauf.laeuft() or updates.laufende_ziele():
|
||
return False
|
||
schritte = plan(inventar.ziele()["ziele"])
|
||
stand = {"datum": heute, "status": "laeuft" if schritte else "fertig", "start": jetzt,
|
||
"ende": None if schritte else jetzt, "schritte": schritte}
|
||
_json_schreiben("homelab-wartung-stand.json", stand)
|
||
if not schritte:
|
||
log.info("wartung: Fenster offen, nichts zu tun")
|
||
return False
|
||
log.info("wartung: Fenster offen, %d Schritte", len(schritte))
|
||
threading.Thread(target=_abarbeiten, args=(stand,), name="homelab-wartung", daemon=True).start()
|
||
return True
|
||
|
||
|
||
def laeuft() -> bool:
|
||
s = stand_lesen()
|
||
return s.get("status") == "laeuft" and time.time() - float(s.get("start") or 0) < 6 * 3600
|
||
|
||
|
||
async def wartung_loop() -> None:
|
||
"""Im Steward der Rolle homelab: jede Minute takt(). Ein Fehler kostet eine Minute, nie die Schleife."""
|
||
log.info("wartung: aktiv (Fenster %s)", fenster_text())
|
||
while True:
|
||
try:
|
||
await asyncio.to_thread(takt)
|
||
except Exception:
|
||
log.warning("wartung: Takt gescheitert", exc_info=True)
|
||
await asyncio.sleep(TAKT_S)
|
||
|
||
|
||
def lage(jetzt: float | None = None) -> dict:
|
||
"""Für GET /api/homelab/wartung."""
|
||
e = lesen()
|
||
return {"fenster": e["fenster"], "fenster_text": fenster_text(e["fenster"]), "standard": e["standard"],
|
||
"regeln": e["regeln"], "uebersprungen": e["uebersprungen"], "naechstes": naechstes(jetzt, e["fenster"]),
|
||
"laeuft": laeuft(), "stand": stand_lesen() or None}
|