"""Protokoll des Homelab-Teils (24.09.2026): was wann geschah, aus allen Quellen, neueste zuerst. Quellen, alle nur gelesen: auftrag Aufträge an den Ausführer (kanal.py): Aktion, Gerät, Status, Code; die Ausgabe als Einzelheiten. Berichte nur auf Wunsch (berichte=1): die Aufträge „bericht“ der Update-Läufe und die Berichte, die der Ausführer von sich aus schickt (alle zehn Minuten; kanal.py vermerkt ihren Eingang). lauf „Jetzt updaten“-Läufe mit ihren Schritten (updates.py) sammellauf „Alle aktualisieren“ (sammellauf.py) hinweis Hinweise des Wächters: erschienen, erledigt, Selbstreparatur (der Verlauf in mc2-waechter.json; ein Hinweis, der dort schon herausgefallen ist, erscheint mit seinem Beginn) meldung was notify.sh an Telegram gab oder für die Morgenmeldung zurücklegte (Melde-Log des Containers, MC_NOTIFY_LOG). Den Betreff schreibt notify.sh nicht mit; er folgt aus dem Absender: Lauf, Sammellauf, Wächter, Telegram-Test oder Morgenmeldung. pflege das wöchentliche Suchen (pflege.py): angestoßen, gescheitert, wieder ok Kein Eintrag enthält Geheimnisse: schwaerzen() nimmt die bekannten Schlüssel dieser Instanz (Arcane, Ausführer) und alles, was wie ein Zugangs-Token aussieht, aus Titeln, Texten und Ausgaben; Steuerzeichen der Konsole fallen weg. """ import json import os import re from collections.abc import Callable from datetime import datetime from pathlib import Path from kern.einstellungen import einstellungen from kern.zeit import LOCAL_TZ from services.einstellungen import melde_log from services.homelab import apps, arcane, kanal, pflege, sammellauf, updates GRENZE_MAX = 1000 DETAILS_MAX = 4000 TEXT_MAX = 300 MELDELOG_MAX = 512 * 1024 # so viel vom Ende des Melde-Logs STUFE_HINWEIS = {"rot": "fehler", "gelb": "warn"} # Betreffzeilen der Absender im Homelab-Teil (notify.sh schreibt sie nicht ins Melde-Log). BETREFF_WAECHTER = ("[Homelab-Problem]", "[Homelab wieder ok]") # waechter.BETREFF_* in der Rolle homelab BETREFF_TEST = "[Test]" # services/einstellungen.telegram_test BETREFF_MORGEN = "[Morgenmeldung Homelab]" # deploy/homelab/mc2-homelab-morgenmeldung.service LUCY_ZUSATZ = " (Diese Meldung kam auch an Lucy.)" # hängt waechter._telegram an # Aktion → (erledigt, gescheitert, offen, kurz). Titel sind „Was · Gerät“. _AKTION = { "bericht": ("Bericht geholt", "Bericht gescheitert", "Bericht angefordert", "Bericht"), "snapshot": ("Snapshot angelegt", "Snapshot gescheitert", "Snapshot wird angelegt", "Snapshot"), "update": ("Update-Skript gelaufen", "Update-Skript gescheitert", "Update-Skript läuft", "Update-Skript"), "os_update": ("Pakete eingespielt", "Pakete einspielen gescheitert", "Pakete werden eingespielt", "Pakete einspielen"), "suchen": ("Paketlisten erneuert", "Paketlisten erneuern gescheitert", "Paketlisten werden erneuert", "Paketlisten erneuern"), "zurueck": ("Auf den Snapshot zurückgesetzt", "Zurücksetzen gescheitert", "Wird auf den Snapshot zurückgesetzt", "Zurücksetzen"), "snapshot_loeschen": ("Alter Snapshot gelöscht", "Snapshot löschen gescheitert", "Alter Snapshot wird gelöscht", "Snapshot löschen"), "sichern": ("Sicherung angelegt", "Sicherung gescheitert", "Sicherung läuft", "Sicherung"), "sicherung_zurueck": ("Sicherung zurückgespielt", "Zurückspielen gescheitert", "Sicherung wird zurückgespielt", "Zurückspielen"), "sicherung_loeschen": ("Alte Sicherung gelöscht", "Sicherung löschen gescheitert", "Alte Sicherung wird gelöscht", "Sicherung löschen"), "host_update": ("Host-Pakete eingespielt", "Host-Pakete gescheitert", "Host-Pakete werden eingespielt", "Host-Pakete"), "host_neustart": ("Neustart ausgelöst", "Neustart gescheitert", "Neustart angefordert", "Neustart"), "kernel_aufraeumen": ("Alte Kernel entfernt", "Kernel aufräumen gescheitert", "Alte Kernel werden entfernt", "Kernel aufräumen"), } _HOST_AKTIONEN = {"bericht", "host_update", "host_neustart", "kernel_aufraeumen"} _LAUF_WAS = {"app": "Update", "os": "Paket-Update", "pakete": "Paket-Update", "docker": "Docker-Update", "neustart": "Neustart", "kernel": "Kernel-Aufräumen"} _SCHRITT_WORT = {"wartet": "wartet", "laeuft": "läuft", "eingespielt": "eingespielt", "zurueckgerollt": "zurückgerollt", "fehler": "gescheitert", "uebersprungen": "übersprungen"} _MELDUNG = re.compile(r"^(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}) " r"(OK telegram direkt|OK telegram|QUEUED für Morgen-Digest|FALLBACK \((.*?)\)): ?(.*)$") _WEG = {"OK telegram direkt": ("Meldung raus", "info", "Direkt an die Telegram-Bot-API gesendet."), "OK telegram": ("Meldung raus", "info", "Über Hermes an Telegram gesendet."), "QUEUED für Morgen-Digest": ("Meldung zurückgelegt", "info", "Nachtruhe: Sie kommt mit der Morgenmeldung um 07:00.")} _FEHLERZEILE = ("E:", "ERROR", "Error", "error:", "FATAL", "Fehler", "Abgelehnt", "Zeitlimit", "Traceback") # --- Geheimnisse und Konsolen-Zeichen ------------------------------------------------------------------- _ANSI = re.compile(r"\x1b(?:\[[0-?]*[ -/]*[@-~]|\][^\x07\x1b]*(?:\x07|\x1b\\)|[@-Z\\-_])") _STEUER = re.compile(r"[\x00-\x08\x0b\x0c\x0e-\x1f\x7f]") _GEHEIM = ( # Telegram-Bot-Token, frei oder in Adressen (…/bot/sendMessage: dort ohne Wortgrenze davor) (re.compile(r"(? str: """Das Geheimnis des Kanals, falls es schon eines gibt — ohne es zu erzeugen (kanal.token() täte das).""" if wert := os.environ.get("MC_AUSFUEHRER_TOKEN", "").strip(): return wert try: return (einstellungen().daten_dir / "ausfuehrer.token").read_text(encoding="utf-8").strip() except (OSError, UnicodeDecodeError): return "" def _schwaerzer() -> Callable[[str | None], str | None]: """Eine Schwärz-Funktion mit den bekannten Schlüsseln dieser Instanz (einmal je Protokoll gelesen).""" bekannt = sorted({w for w in (arcane.schluessel(), _ausfuehrer_token()) if len(w) >= 8}, key=len, reverse=True) def schwaerzen(text: str | None) -> str | None: if text is None: return None text = _ANSI.sub("", str(text)) # Fortschrittsbalken überschreiben ihre Zeile mit \r: Stehen bleibt, was die Konsole zuletzt zeigte. zeilen = [] for zeile in text.split("\n"): teile = [t for t in zeile.split("\r") if t.strip()] zeilen.append(teile[-1] if teile else "") text = _STEUER.sub("", "\n".join(zeilen)) for wert in bekannt: text = text.replace(wert, "***") for muster, ersatz in _GEHEIM: text = muster.sub(ersatz, text) return text return schwaerzen def schwaerzen(text: str | None) -> str | None: """Geheimnisse und Steuerzeichen aus einem Text (für Tests und Einzelfälle; eintraege() liest einmal).""" return _schwaerzer()(text) def _kuerzen(text: str | None, laenge: int = DETAILS_MAX) -> str | None: """Das Ende langer Ausgaben (dort stehen Ergebnis und Fehler), wenn möglich ab einem Zeilenanfang.""" if not text or not text.strip(): return None text = text.strip("\n") if len(text) <= laenge: return text rest = text[-(laenge - 2):] if "\n" in rest[:300]: rest = rest.split("\n", 1)[1] return "…\n" + rest def _zeile(text: str | None) -> str | None: """Ein Text für das Feld „text“: eine Zeile, höchstens TEXT_MAX Zeichen.""" if not text or not str(text).strip(): return None ganz = str(text).strip() erste = ganz.splitlines()[0].strip() if len(erste) > TEXT_MAX: return erste[:TEXT_MAX - 1].rstrip() + "…" return erste + ("…" if "\n" in ganz else "") # --- Geräte --------------------------------------------------------------------------------------------- def _geraete() -> dict[int, tuple[str, str]]: """VMID → (Ziel-ID, Name) aus dem letzten Bericht des Ausführers.""" geraete: dict[int, tuple[str, str]] = {} for g in ((kanal.bericht() or {}).get("bericht") or {}).get("gaeste") or []: if not isinstance(g, dict) or not isinstance(g.get("vmid"), int): continue app = apps.app_fuer((g.get("app") or {}).get("kennung"), g.get("name")) zid = f"{'ct' if g.get('art') == 'lxc' else 'vm'}-{g['vmid']}" geraete[g["vmid"]] = (zid, app.name if app else str(g.get("name") or zid)) return geraete def _eintrag(zeit: object, art: str, titel: str, text: str | None, stufe: str, ziel: str | None, details: str | None, schw: Callable) -> dict | None: if not isinstance(zeit, (int, float)): return None return {"zeit": float(zeit), "art": art, "titel": schw(titel), "text": schw(text), "stufe": stufe, "ziel": ziel, "details": schw(details)} # --- Aufträge an den Ausführer -------------------------------------------------------------------------- def _dauer(bis: object, von: object) -> str | None: if not (isinstance(bis, (int, float)) and isinstance(von, (int, float))) or bis < von: return None s = round(bis - von) if s < 60: return f"{s} s" if s < 3600: return f"{s // 60} min" return f"{s // 3600} h {s % 3600 // 60} min" def _gegenstand(aktion: str, p: dict, ausgabe: str) -> str | None: """Worum es ging: der Snapshot bzw. die Sicherung.""" if aktion in ("zurueck", "snapshot_loeschen"): return str(p.get("snapshot") or "") or None if aktion in ("sicherung_zurueck", "sicherung_loeschen"): return str(p.get("sicherung") or "").rsplit("/", 1)[-1] or None if aktion == "snapshot" and (m := re.search(r"snapshot=(\S+)", ausgabe)): return m.group(1) if aktion == "sichern" and (m := re.search(r"^sicherung=(\S+)$", ausgabe, re.MULTILINE)): return m.group(1).rsplit("/", 1)[-1] return None def _fehlerzeile(ausgabe: str) -> str | None: zeilen = [z.strip() for z in ausgabe.splitlines() if z.strip()] wichtig = [z for z in zeilen if z.startswith(_FEHLERZEILE)] zeile = (wichtig or zeilen or [None])[-1] return zeile[:200] if zeile else None def _bericht_satz(version: object, gaeste: object, laufend: object, host_updates: object) -> str: teile = [f"Proxmox VE {version}" if version else None, f"{gaeste} Gäste, davon {laufend} laufend", f"{host_updates} Paket-Updates auf dem Host" if isinstance(host_updates, int) else None] return ", ".join(t for t in teile if t) + "." def _bericht_kurz(ausgabe: str) -> str | None: """Ein Bericht ist JSON; statt seines Endes eine Zusammenfassung.""" try: b = json.loads(ausgabe) except ValueError: return None if not isinstance(b, dict): return None host = b.get("host") if isinstance(b.get("host"), dict) else {} gaeste = [g for g in b.get("gaeste") or [] if isinstance(g, dict)] return _bericht_satz(host.get("version"), len(gaeste), sum(1 for g in gaeste if g.get("status") == "running"), len(host["updates"]) if isinstance(host.get("updates"), list) else None) def _bericht_eingang(e: dict, schw: Callable) -> dict | None: """Ein Bericht, den der Ausführer von sich aus geschickt hat (kein Auftrag).""" return _eintrag(e.get("empfangen"), "auftrag", "Bericht empfangen · Proxmox-Host", _bericht_satz(e.get("host_version"), e.get("gaeste"), e.get("laufend"), e.get("host_updates")), "info", "pve", None, schw) def _auftrag(a: dict, geraete: dict, schw: Callable) -> dict | None: aktion = str(a.get("aktion") or "?") erledigt, gescheitert, offen, kurz = _AKTION.get(aktion, (aktion, f"{aktion} gescheitert", f"{aktion} läuft", aktion)) p = a.get("parameter") if isinstance(a.get("parameter"), dict) else {} ziel, name = None, None if aktion in _HOST_AKTIONEN: ziel, name = "pve", "Proxmox-Host" elif isinstance(p.get("vmid"), int): ziel, name = geraete.get(p["vmid"], (None, f"Gast {p['vmid']}")) status, code = a.get("status"), a.get("code") ausgabe = str(a.get("text") or "") gegenstand = _gegenstand(aktion, p, ausgabe) dauer = _dauer(a.get("fertig"), a.get("abgeholt")) kopf = ", ".join(t for t in (gegenstand, f"nach {dauer}" if dauer else None, f"Code {code}" if code is not None else None) if t) if status == "fertig": titel, stufe, text = erledigt, "ok", kopf or None elif status == "fehler" and code == 2 and ausgabe.startswith("Abgelehnt"): titel, stufe, text = f"{kurz} abgelehnt", "warn", _fehlerzeile(ausgabe) elif status == "fehler": grund = _fehlerzeile(ausgabe) titel, stufe, text = gescheitert, "fehler", f"{kopf}: {grund}" if grund else kopf or None elif status == "verloren": titel, stufe, text = f"{kurz}: keine Antwort", "fehler", "Der Ausführer hat nie geantwortet." elif status == "laeuft": titel, stufe, text = offen, "info", "Der Ausführer ist dran." else: titel, stufe, text = offen, "info", "Wartet auf den Ausführer." details = _bericht_kurz(ausgabe) if aktion == "bericht" else None return _eintrag(a.get("fertig") or a.get("abgeholt") or a.get("erstellt"), "auftrag", f"{titel} · {name}" if name else titel, text, stufe, ziel, details or _kuerzen(schw(ausgabe)), schw) # --- Läufe und Sammelläufe ------------------------------------------------------------------------------ def _anzahl_pakete(n: object, wort: str) -> str: return f"{n} {wort}{'' if n == 1 else 'e'} eingespielt" if isinstance(n, int) and n > 0 else f"{wort}e eingespielt" def _lauf(x: dict, schw: Callable) -> dict | None: name = str(x.get("name") or x.get("ziel") or "?") baustein, ergebnis, status, text = x.get("baustein"), x.get("ergebnis"), x.get("status"), x.get("text") was = _LAUF_WAS.get(str(baustein), "Update") if status == "laeuft": titel, stufe = f"{was} läuft · {name}", "info" elif status == "unterbrochen": titel, stufe = f"{was} unterbrochen · {name}", "fehler" text = text or "Der Homelab-Teil wurde während des Laufs neu gestartet; ob das Update durchlief, zeigt der " \ "nächste Bericht." elif ergebnis == "eingespielt": stufe = "ok" if baustein == "app": titel = f"Update eingespielt · {name}" + (f" {x['version_neu']}" if x.get("version_neu") else "") elif baustein == "docker": titel = f"Docker-Images aktualisiert · {name}" elif baustein == "pakete": titel = f"{_anzahl_pakete(x.get('pakete'), 'Host-Paket')} · {name}" else: titel = f"{_anzahl_pakete(x.get('pakete'), 'Paket')} · {name}" elif ergebnis == "zurueckgerollt": titel, stufe = f"{was} zurückgerollt · {name}", "warn" elif ergebnis == "fehler" and "Update nicht begonnen" in str(text or ""): titel, stufe = f"{was} nicht begonnen · {name}", "warn" elif ergebnis == "fehler": titel, stufe = f"{was} gescheitert · {name}", "fehler" elif ergebnis == "neustart": titel, stufe = f"Neustart ausgelöst · {name}", "info" elif ergebnis == "aufgeraeumt": titel, stufe = f"Alte Kernel entfernt · {name}", "ok" elif ergebnis == "offen": titel, stufe = f"Docker-Probelauf · {name}", "info" else: titel, stufe = f"{was} beendet · {name}", "info" zeilen = [str(s) for s in x.get("schritte") or []] if x.get("sammellauf"): zeilen.append("Teil von „Alle aktualisieren“.") if x.get("gemeldet") is False: zeilen.append("Keine eigene Meldung: Die Sammelmeldung am Ende sagt es.") return _eintrag(x.get("ende") or x.get("start"), "lauf", titel, text, stufe, x.get("ziel"), "\n".join(zeilen) or None, schw) def _sammellauf(sl: dict, schw: Callable) -> dict | None: schritte = [s for s in sl.get("schritte") or [] if isinstance(s, dict)] anzahl = len(schritte) eingespielt = sum(1 for s in schritte if s.get("status") == "eingespielt") status = sl.get("status") if status == "laeuft": i = sl.get("aktuell") titel = "Alle aktualisieren läuft" + (f" · Schritt {i + 1} von {anzahl}" if isinstance(i, int) else "") stufe = "info" elif status == "abgebrochen": bei = next((s.get("name") for s in schritte if s.get("status") in ("fehler", "zurueckgerollt")), None) titel, stufe = "Alle aktualisieren abgebrochen" + (f" · {bei}" if bei else ""), "fehler" else: titel = f"Alle aktualisieren fertig · {eingespielt} von {anzahl} eingespielt" stufe = "ok" if eingespielt == anzahl else "info" details = "\n".join(f"{s.get('name')} ({s.get('was')}): {_SCHRITT_WORT.get(str(s.get('status')), s.get('status'))}" + (f" – {s['text']}" if s.get("text") else "") for s in schritte) return _eintrag(sl.get("ende") or sl.get("start"), "sammellauf", titel, sl.get("text"), stufe, None, details or None, schw) # --- Hinweise des Wächters ------------------------------------------------------------------------------ def _waechter_pfad() -> Path: """Wie waechter.STORE_PATH — hier gelesen, ohne den Wächter zu laden (der zieht Module der KI-Box nach).""" return Path(os.environ.get("MC_WAECHTER_STORE", str(einstellungen().daten_dir / "mc2-waechter.json"))) def _ziel_aus(hinweis_id: str) -> str | None: 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): return f"ct-{m.group(1)}" return "pve" if hinweis_id == "ausfuehrer" else None def _hinweise(schw: Callable) -> list[dict]: try: daten = json.loads(_waechter_pfad().read_text(encoding="utf-8")) except (OSError, ValueError): return [] daten = daten if isinstance(daten, dict) else {} aktuell = daten.get("hinweise") if isinstance(daten.get("hinweise"), dict) else {} eintraege, erschienen = [], set() for v in daten.get("verlauf") or []: if not isinstance(v, 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 {} 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) elif art == "erledigt": e = _eintrag(v.get("ts"), "hinweis", f"Erledigt: {titel}", None, "ok", _ziel_aus(hid), None, schw) elif art == "auto": e = _eintrag(v.get("ts"), "hinweis", titel, "Selbstreparatur des Wächters.", "info", _ziel_aus(hid), None, schw) else: continue eintraege.append(e) for hid, h in aktuell.items(): if hid not in erschienen and isinstance(h, dict): eintraege.append(_eintrag(h.get("seit"), "hinweis", f"Hinweis: {h.get('titel') or hid}", h.get("text"), STUFE_HINWEIS.get(str(h.get("stufe")), "warn"), _ziel_aus(hid), None, schw)) return eintraege # --- Meldungen ------------------------------------------------------------------------------------------ def _meldelog() -> list[list]: """[Zeit, Weg, Grund, Text] je Eintrag des Melde-Logs; Zeilen ohne Zeitstempel gehören zur Meldung davor.""" try: with melde_log().open("rb") as f: f.seek(0, os.SEEK_END) groesse = f.tell() f.seek(max(0, groesse - MELDELOG_MAX)) roh = f.read().decode("utf-8", "replace") except OSError: return [] if groesse > MELDELOG_MAX: roh = roh.split("\n", 1)[-1] # angeschnittene erste Zeile eintraege: list[list] = [] for zeile in roh.splitlines(): if m := _MELDUNG.match(zeile): try: zeit = datetime.strptime(m.group(1), "%Y-%m-%d %H:%M:%S").replace(tzinfo=LOCAL_TZ).timestamp() except ValueError: continue weg = "FALLBACK" if m.group(2).startswith("FALLBACK") else m.group(2) eintraege.append([zeit, weg, m.group(3), m.group(4)]) elif eintraege: eintraege[-1][3] += "\n" + zeile return eintraege def _betreffe(laeufe: list[dict], sammellaeufe: list[dict]) -> dict[str, str]: """Meldungstext → Betreff für das, was Läufe und Sammelläufe gemeldet haben.""" bekannt = {} for x in laeufe: if x.get("text") and x.get("gemeldet") is not False: dringend = x.get("dringend") if dringend is None: # Läufe von vor dem 24.09.2026 abends: wie updates._ende es entschied dringend = x.get("ergebnis") in ("zurueckgerollt", "fehler") and "nicht begonnen" not in x["text"] bekannt[x["text"]] = updates.BETREFF_ALARM if dringend else updates.BETREFF for sl in sammellaeufe: if sl.get("meldung"): bekannt[sl["meldung"]] = updates.BETREFF_ALARM if sl.get("dringend") else updates.BETREFF return bekannt def _betreff(text: str, bekannt: dict[str, str]) -> str | None: if text in bekannt: return bekannt[text] if text.endswith(LUCY_ZUSATZ): return BETREFF_WAECHTER[1] if text.startswith("Erledigt:") else BETREFF_WAECHTER[0] if text.startswith("Testmeldung aus den Einstellungen"): return BETREFF_TEST if text.startswith("Guten Morgen, Commander."): return BETREFF_MORGEN return None def _meldungen(bekannt: dict[str, str], schw: Callable) -> list[dict]: eintraege = [] for zeit, weg, grund, text in _meldelog(): betreff = _betreff(text, bekannt) if weg == "FALLBACK": titel, stufe, wie = "Meldung ging nicht raus", "fehler", f"Telegram ging nicht: {grund or 'ohne Angabe'}" else: titel, stufe, wie = _WEG[weg] kurz = _zeile(text) details = wie if kurz == text.strip() else f"{wie}\n\n{text.strip()}" eintraege.append(_eintrag(zeit, "meldung", f"{titel} · {betreff}" if betreff else titel, kurz, stufe, None, details, schw)) return eintraege # --- Pflege --------------------------------------------------------------------------------------------- _PFLEGE = {"angestossen": ("Wöchentliche Suche angestoßen", "info"), "gescheitert": ("Wöchentliche Suche gescheitert", "warn"), "wieder_ok": ("Wöchentliche Suche klappt wieder", "ok")} def _pflege(geraete: dict, schw: Callable) -> list[dict]: daten = pflege.zustand() ereignisse = [e for e in daten.get("ereignisse") or [] if isinstance(e, dict)] eintraege = [] for e in ereignisse: if e.get("art") not in _PFLEGE or not isinstance(e.get("vmid"), int): continue titel, stufe = _PFLEGE[e["art"]] name = e.get("name") or geraete.get(e["vmid"], (None, f"Gast {e['vmid']}"))[1] eintraege.append(_eintrag(e.get("ts"), "pflege", f"{titel} · {name}", e.get("text"), stufe, f"ct-{e['vmid']}", None, schw)) # Stände von vor dem 24.09.2026 abends haben keine Ereignisliste: dann wenigstens der letzte Anstoß je Gast. schon = {(e.get("vmid"), e.get("ts")) for e in ereignisse if e.get("art") == "angestossen"} for vmid, g in (daten.get("gaeste") or {}).items(): if not (isinstance(g, dict) and isinstance(g.get("angestossen"), (int, float)) and str(vmid).isdigit()): continue if (int(vmid), g["angestossen"]) not in schon: name = g.get("name") or geraete.get(int(vmid), (None, f"Gast {vmid}"))[1] eintraege.append(_eintrag(g["angestossen"], "pflege", f"{_PFLEGE['angestossen'][0]} · {name}", "Die Paketlisten waren älter als 7 Tage.", "info", f"ct-{vmid}", None, schw)) return eintraege # --- Zusammen ------------------------------------------------------------------------------------------- def eintraege(grenze: int = 200, berichte: bool = False) -> list[dict]: """Alle Einträge, neueste zuerst, höchstens `grenze` (1 bis GRENZE_MAX).""" grenze = max(1, min(int(grenze), GRENZE_MAX)) schw = _schwaerzer() geraete = _geraete() laeufe = updates.laeufe(100) sammellaeufe = sammellauf.liste() alle: list[dict | None] = [] alle += [_auftrag(a, geraete, schw) for a in kanal.liste() if berichte or a.get("aktion") != "bericht"] if berichte: alle += [_bericht_eingang(e, schw) for e in kanal.berichte()] alle += [_lauf(x, schw) for x in laeufe if isinstance(x, dict)] alle += [_sammellauf(sl, schw) for sl in sammellaeufe] 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.sort(key=lambda e: e["zeit"], reverse=True) return liste[:grenze]