Welle 2 · Zeitleiste je Gerät: Protokoll nach Gerät, Updates von → nach, Hinweise mit ihrem Gerät

/homelab/protokoll#ct-104 zeigt die Zeitleiste eines Geräts (Auswahl oben, Link „Zeitleiste“ je Gerät in der
Geräte-Tabelle); die Punkte eines Tages hängen an einer Linie. Das Protokoll filtert dafür im Backend (?ziel=…),
Update-Läufe heißen „Update eingespielt · Gitea 1.27.2 → 1.27.3“, und Hinweise finden ihr Gerät über ihre Quelle
(der Wächter schreibt sie seit heute in seinen Verlauf) — auch Zertifikate, Docker und Sicherungen.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Hitonabi
2026-09-25 18:06:41 +02:00
co-authored by Claude Opus 5.5
parent 1c66daad0b
commit 83bb227969
7 changed files with 84 additions and 29 deletions
+3 -2
View File
@@ -162,9 +162,10 @@ def sammellauf_quittieren(sl_id: str) -> dict:
@router.get("/protokoll")
def protokoll_lesen(grenze: int = 200, berichte: bool = False) -> dict:
def protokoll_lesen(grenze: int = 200, berichte: bool = False, ziel: str | None = None) -> dict:
"""Was geschah, neueste zuerst; mit ziel nur ein Gerät (Zeitleiste je Gerät, seit 25.09.2026)."""
from services.homelab import protokoll
return {"eintraege": protokoll.eintraege(grenze, berichte)}
return {"eintraege": protokoll.eintraege(grenze, berichte, ziel or None)}
@router.get("/speicher")
+19 -9
View File
@@ -347,7 +347,10 @@ def _lauf(x: dict, schw: Callable) -> dict | None:
elif ergebnis == "eingespielt":
stufe = "ok"
if baustein == "app":
titel = f"Update eingespielt · {name}" + (f" {x['version_neu']}" if x.get("version_neu") else "")
# Seit 25.09.2026 „von → nach“, wo der Lauf beide kennt (Zeitleiste je Gerät).
von_nach = (f" {x['version_alt']} → {x['version_neu']}" if x.get("version_alt") and x.get("version_neu")
else f" {x['version_neu']}" if x.get("version_neu") else "")
titel = f"Update eingespielt · {name}{von_nach}"
elif baustein == "docker":
titel = f"Docker-Images aktualisiert · {name}"
elif baustein == "pakete":
@@ -407,7 +410,13 @@ def _waechter_pfad() -> Path:
return Path(os.environ.get("MC_WAECHTER_STORE", str(einstellungen().daten_dir / "mc2-waechter.json")))
def _ziel_aus(hinweis_id: str) -> str | None:
_GERAET = re.compile(r"pve|(?:ct|vm)-\d+")
def _ziel_aus(hinweis_id: str, quelle: object = None) -> str | None:
"""Das Gerät eines Hinweises: seit 25.09.2026 aus seiner Quelle, wenn sie ein Gerät ist, sonst aus der ID."""
if isinstance(quelle, str) and _GERAET.fullmatch(quelle):
return quelle
if m := re.search(r"\b(ct|vm)-(\d+)\b", hinweis_id):
return f"{m.group(1)}-{m.group(2)}"
if m := re.fullmatch(r"gast-platte:(\d+)", hinweis_id):
@@ -428,15 +437,15 @@ def _hinweise(schw: Callable) -> list[dict]:
continue
art, hid = v.get("art"), str(v.get("id") or "")
titel, h = str(v.get("text") or hid), aktuell.get(hid) if isinstance(aktuell.get(hid), dict) else {}
ziel = _ziel_aus(hid, v.get("quelle") or h.get("quelle"))
if art == "neu":
erschienen.add(hid)
e = _eintrag(v.get("ts"), "hinweis", f"Hinweis: {titel}", h.get("text"),
STUFE_HINWEIS.get(str(v.get("stufe") or h.get("stufe")), "warn"), _ziel_aus(hid), None, schw)
STUFE_HINWEIS.get(str(v.get("stufe") or h.get("stufe")), "warn"), ziel, None, schw)
elif art == "erledigt":
e = _eintrag(v.get("ts"), "hinweis", f"Erledigt: {titel}", None, "ok", _ziel_aus(hid), None, schw)
e = _eintrag(v.get("ts"), "hinweis", f"Erledigt: {titel}", None, "ok", ziel, None, schw)
elif art == "auto":
e = _eintrag(v.get("ts"), "hinweis", titel, "Selbstreparatur des Wächters.", "info", _ziel_aus(hid),
None, schw)
e = _eintrag(v.get("ts"), "hinweis", titel, "Selbstreparatur des Wächters.", "info", ziel, None, schw)
else:
continue
eintraege.append(e)
@@ -552,8 +561,9 @@ def _pflege(geraete: dict, schw: Callable) -> list[dict]:
# --- Zusammen -------------------------------------------------------------------------------------------
def eintraege(grenze: int = 200, berichte: bool = False) -> list[dict]:
"""Alle Einträge, neueste zuerst, höchstens `grenze` (1 bis GRENZE_MAX)."""
def eintraege(grenze: int = 200, berichte: bool = False, ziel: str | None = None) -> list[dict]:
"""Alle Einträge, neueste zuerst, höchstens `grenze` (1 bis GRENZE_MAX). Mit ziel (seit 25.09.2026) nur die eines
Geräts — die Zeitleiste je Gerät."""
grenze = max(1, min(int(grenze), GRENZE_MAX))
schw = _schwaerzer()
geraete = _geraete()
@@ -568,6 +578,6 @@ def eintraege(grenze: int = 200, berichte: bool = False) -> list[dict]:
alle += _hinweise(schw)
alle += _meldungen(_betreffe(laeufe, sammellaeufe), schw)
alle += _pflege(geraete, schw)
liste = [e for e in alle if e is not None]
liste = [e for e in alle if e is not None and (ziel is None or e.get("ziel") == ziel)]
liste.sort(key=lambda e: e["zeit"], reverse=True)
return liste[:grenze]
+5 -3
View File
@@ -664,10 +664,12 @@ def _speichere() -> None:
log.warning("waechter: Stand %s nicht schreibbar", STORE_PATH, exc_info=True)
def _verlauf(art: str, hinweis_id: str, text: str, stufe: str | None = None) -> None:
def _verlauf(art: str, hinweis_id: str, text: str, stufe: str | None = None, quelle: str | None = None) -> None:
eintrag = {"ts": time.time(), "art": art, "id": hinweis_id, "text": text}
if stufe: # seit 24.09.2026: das Protokoll des Homelab-Teils zeigt, wie ernst ein Hinweis war
eintrag["stufe"] = stufe
if quelle: # seit 25.09.2026: welches Gerät — für die Zeitleiste je Gerät
eintrag["quelle"] = quelle
_stand["verlauf"].append(eintrag)
del _stand["verlauf"][:-VERLAUF_MAX]
@@ -744,7 +746,7 @@ def takt() -> None:
h.pop("fehlt", None)
hinweise[b.id] = h
if neu:
_verlauf("neu", b.id, b.titel, b.stufe)
_verlauf("neu", b.id, b.titel, b.stufe, b.quelle)
if b.stufe == "rot" and (neu or jetzt - h.get("gemeldet", 0.0) >= REMIND_S) \
and not ist_stumm(zustaende.get(b.id), b.stufe, jetzt):
vorsatz = "" if neu else "Immer noch: "
@@ -763,7 +765,7 @@ def takt() -> None:
continue
h = hinweise.pop(hid)
h.pop("fehlt", None)
_verlauf("erledigt", hid, h.get("titel", hid), h.get("stufe"))
_verlauf("erledigt", hid, h.get("titel", hid), h.get("stufe"), h.get("quelle"))
if h.get("stufe") == "rot" and h.get("gemeldet"):
_telegram(BETREFF_OK, f"Erledigt: {h.get('titel', hid)}.")
for kid in list(kandidaten):
+18 -2
View File
@@ -168,8 +168,8 @@ def test_alle_quellen_mit_sprechenden_titeln(geschichte):
assert lang.startswith("…\nZeile ") and len(lang) <= protokoll.DETAILS_MAX and lang.endswith("Zeile 199: " + "x" * 90)
# Update-Läufe mit ihren Schritten
assert e["Update eingespielt · Gitea 1.27.3"]["stufe"] == "ok"
assert e["Update eingespielt · Gitea 1.27.3"]["details"] == "snapshot: ok\nupdate: ok"
assert e["Update eingespielt · Gitea 1.27.2 → 1.27.3"]["stufe"] == "ok"
assert e["Update eingespielt · Gitea 1.27.2 → 1.27.3"]["details"] == "snapshot: ok\nupdate: ok"
assert e["Paket-Update zurückgerollt · Proxmox Backup Server"]["stufe"] == "warn"
assert e["Update nicht begonnen · NetBird"]["stufe"] == "warn"
host_lauf = e["5 Host-Pakete eingespielt · Proxmox-Host"]
@@ -333,3 +333,19 @@ def test_pflege_merkt_sich_anstoss_fehlschlag_und_erholung(daten, monkeypatch):
assert titel == Counter({"Wöchentliche Suche angestoßen · AdGuard Home": 2,
"Wöchentliche Suche gescheitert · AdGuard Home": 1,
"Wöchentliche Suche klappt wieder · AdGuard Home": 1})
def test_zeitleiste_je_geraet(geschichte):
"""Seit 25.09.2026: nur die Einträge eines Geräts — Updates von → nach, Aufträge, Hinweise."""
gitea = protokoll.eintraege(ziel="ct-104")
assert gitea and all(e["ziel"] == "ct-104" for e in gitea)
assert "Update eingespielt · Gitea 1.27.2 → 1.27.3" in {e["titel"] for e in gitea}
assert protokoll.eintraege(ziel="vm-999") == []
def test_hinweise_finden_ihr_geraet_ueber_die_quelle():
assert protokoll._ziel_aus("zertifikat:npm", "ct-101") == "ct-101"
assert protokoll._ziel_aus("docker:rippy-api-1", "vm-106") == "vm-106"
assert protokoll._ziel_aus("gast-platte:106", None) == "ct-106" # alte Einträge ohne Quelle
assert protokoll._ziel_aus("sicherung:probe", "pve") == "pve"
assert protokoll._ziel_aus("platte", "platte") is None