mission-control-v2/backend/services/update_verlauf.py

"""
Update-Verlauf und festgehaltene Bausteine (Box-Wart, Umbau 09/2026).

Was ein Update-Lauf wirklich gebracht hat, steht nicht im Hermes-Cron-Status — der sagt nur
„Skript lief". Die Wahrheit ist das, was deploy/autoupdate.sh am Ende meldet: eine Zeile je
Baustein („255 → 256 eingespielt", „zurückgerollt + GEPINNT" …). notify.sh schreibt jede
Meldung mit Zeitstempel in ~/mc2-notify.log; der Verlauf wird von dort gelesen, nur lesend.

Das Pin-Register (/srv/models/mc2-pins.json) führt autoupdate.sh: Besteht ein Update den Check
nicht, wird zurückgerollt und der Baustein festgehalten — künftige Sonntage überspringen ihn.
Vom 06. bis 17.09.2026 hat das niemand gesehen, die Box bekam elf Tage keine Updates. Darum
zeigt der Wächter jeden festgehaltenen Baustein als Hinweis, und hier lässt er sich freigeben.
"""

import json
import os
import re
import threading
from datetime import datetime
from pathlib import Path
from zoneinfo import ZoneInfo

LOCAL_TZ = ZoneInfo(os.environ.get("MC_LOCAL_TZ", "Europe/Berlin"))
NOTIFY_LOG = Path(os.environ.get("MC_NOTIFY_LOG", str(Path.home() / "mc2-notify.log")))
PINS_FILE = Path(os.environ.get("MC_PINS_FILE", "/srv/models/mc2-pins.json"))
LESE_MAX = 3_000_000          # Bytes vom Ende des Protokolls — reicht für Monate
LAUF_LUECKE_S = 45 * 60       # Meldungen mit größerem Abstand gehören zu verschiedenen Läufen

NAMEN = {"swap": "llama-swap", "engine": "Motor (llama.cpp)", "hermes": "Hermes"}
FREIGEGEBEN_TEXT = "Freigegeben. Der nächste Sonntags-Lauf versucht das Update erneut."

# So heißen die Bausteine in den Meldungen von autoupdate.sh → kurzer Name fürs Cockpit.
_BAUSTEIN_IM_TEXT = (
    (re.compile(r"Router \(llama-swap\)"), "llama-swap"),
    (re.compile(r"Engine \(llama\.cpp\)"), "Motor"),
    (re.compile(r"Hermes(-Agent)?\b"), "Hermes"),
)

# Kopfzeile eines Protokolleintrags (notify.sh): Zeitstempel, Zustellweg, Nachricht.
_EINTRAG = re.compile(
    r"^(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}) "
    r"(?:OK telegram|QUEUED für Morgen-Digest|FALLBACK \(.*?\)): ?(.*)$")

_ZUSAMMENFASSUNG = re.compile(r"Wochenpflege der Box ist durch")
_UPDATE_MELDUNG = re.compile(
    r"^(KRITISCH: )?(Router \(llama-swap\)|Engine \(llama\.cpp\)|Hermes-Agent|Hermes)[ -](aktualisiert|Update)"
    r"|^Auto-Update abgebrochen|^Sonntags-Update")
_NEUSTART = re.compile(r"^Die Box startet jetzt neu|^Neustart (NICHT|verschoben|FEHLGESCHLAGEN)")
# Die Morgenmeldung (07:00) wiederholt die Nachtmeldungen gesammelt — sie ist kein eigener Lauf.
_MORGENMELDUNG = re.compile(r"^Guten Morgen, Commander\. Heute Nacht")

# Einordnung einer Ergebniszeile. Reihenfolge zählt: „zurückgerollt + GEPINNT“ ist ein Rückbau,
# „Check fehlgeschlagen“ ist offen und kein Stack-Schaden.
_EINORDNUNG = (
    ("fehler", re.compile(r"kritisch", re.IGNORECASE)),
    ("zurueckgerollt", re.compile(r"zurückgerollt", re.IGNORECASE)),
    ("festgehalten", re.compile(r"gepinnt", re.IGNORECASE)),
    ("offen", re.compile(r"wartet|unklar|check fehlgeschlagen|job-start|nicht ausgefuehrt|verschoben",
                         re.IGNORECASE)),
    ("fehler", re.compile(r"fehlgeschlagen|abgebrochen|ausgefallen|mit fehler", re.IGNORECASE)),
    ("neustart", re.compile(r"startet jetzt neu", re.IGNORECASE)),
    ("eingespielt", re.compile(r"eingespielt|aktualisiert|repariert", re.IGNORECASE)),
    ("aktuell", re.compile(r"aktuell", re.IGNORECASE)),
)
STUFE_DER_ZEILE = {"fehler": "rot", "zurueckgerollt": "gelb", "festgehalten": "gelb", "offen": "gelb",
                   "neustart": "info", "eingespielt": "gruen", "aktuell": "grau"}
_RANG = {"rot": 4, "gelb": 3, "gruen": 2, "info": 1, "grau": 0}

_cache_lock = threading.Lock()
_cache: dict = {"schluessel": None, "laeufe": []}


# --- Protokoll lesen ----------------------------------------------------------------

def eintraege(text: str) -> list[tuple[datetime, str]]:
    """Protokoll → (Zeitpunkt, Nachricht). Zeilen ohne Zeitstempel gehören zur Nachricht davor
    (mehrzeilige Meldungen wie die Wochen-Zusammenfassung oder die Nachrichten-Übersicht)."""
    ergebnis: list[tuple[datetime, str]] = []
    for zeile in text.splitlines():
        if m := _EINTRAG.match(zeile):
            try:
                zeit = datetime.strptime(m.group(1), "%Y-%m-%d %H:%M:%S").replace(tzinfo=LOCAL_TZ)
            except ValueError:
                continue
            ergebnis.append((zeit, m.group(2)))
        elif ergebnis:
            zeit, bisher = ergebnis[-1]
            ergebnis[-1] = (zeit, f"{bisher}\n{zeile}")
    return ergebnis


def _einordnen(text: str) -> str:
    return next((art for art, muster in _EINORDNUNG if muster.search(text)), "offen")


def _baustein_aus(text: str) -> str:
    return next((name for muster, name in _BAUSTEIN_IM_TEXT if muster.search(text)), "Update-Lauf")


def _lesbar(text: str) -> str:
    """Meldungstext fürs Cockpit: Fachwörter der Skripte in Alltagssprache."""
    text = re.sub(r"\s*\+\s*GEPINNT", " und festgehalten", text)
    text = re.sub(r"gepinnt, übersprungen", "festgehalten, übersprungen", text, flags=re.IGNORECASE)
    text = text.replace("GEPINNT", "festgehalten").replace("FEHLGESCHLAGEN", "fehlgeschlagen")
    text = re.sub(r"\b1 Commits\b", "1 Änderung", text)
    text = re.sub(r"\b(\d+) Commits\b", r"\1 Änderungen", text)
    # Nur der erste Satz: der Rest der Meldungen ist Begleittext für Telegram.
    return re.split(r"(?<=[.!])\s", text.strip(), maxsplit=1)[0][:240]


def _zeile(baustein: str, text: str) -> dict:
    ergebnis = _einordnen(text)
    return {"baustein": baustein, "ergebnis": ergebnis, "stufe": STUFE_DER_ZEILE[ergebnis], "text": _lesbar(text)}


def _lauf(gruppe: list[tuple[datetime, str]]) -> dict:
    zusammenfassung = next((t for _, t in reversed(gruppe) if _ZUSAMMENFASSUNG.search(t)), None)
    zeilen: list[dict] = []
    if zusammenfassung:
        for z in zusammenfassung.splitlines()[1:]:
            if m := re.match(r"^\s*•\s*(.+?):\s*(.+)$", z):
                zeilen.append(_zeile(_baustein_aus(m.group(1)), m.group(2)))
    for _, text in gruppe:
        erste = text.splitlines()[0] if text else ""
        if _NEUSTART.search(erste):
            zeilen.append(_zeile("Neustart", erste))
        elif erste.startswith(("Sonntags-Update", "Auto-Update abgebrochen")):
            zeilen.append(_zeile("Update-Lauf", erste))
        elif not zusammenfassung and _UPDATE_MELDUNG.search(erste):
            # Ohne Abschlussmeldung (Lauf abgebrochen) sind die Einzelmeldungen alles, was es gibt.
            zeilen.append(_zeile(_baustein_aus(erste), re.sub(r"^KRITISCH:\s*", "Kritisch: ", erste)))

    stufe = max((z["stufe"] for z in zeilen), key=lambda s: _RANG[s], default="grau")
    arten = {z["ergebnis"] for z in zeilen}
    if stufe == "rot":
        titel = "Fehler"
    elif stufe == "gelb":
        titel = ("Zurückgerollt" if "zurueckgerollt" in arten
                 else "Festgehalten" if "festgehalten" in arten else "Unvollständig")
    elif stufe == "gruen":
        titel = "Eingespielt"
    elif stufe == "info":
        titel = "Neu gestartet"
    else:
        titel = "Alles aktuell"
    start, ende = gruppe[0][0], gruppe[-1][0]
    sonntag = start.weekday() == 6 and 4 <= start.hour < 6
    return {"start": start.isoformat(), "ende": ende.isoformat(),
            "anlass": "Updates am Sonntag" if sonntag else "Update von Hand",
            "stufe": stufe, "titel": titel, "zeilen": zeilen}


def laeufe_aus(text: str) -> list[dict]:
    """Alle Update-Läufe im Protokolltext, neueste zuerst."""
    relevant = [(z, t) for z, t in eintraege(text)
                if not _MORGENMELDUNG.match(t)
                and (_ZUSAMMENFASSUNG.search(t) or _UPDATE_MELDUNG.search(t.splitlines()[0] if t else "")
                     or _NEUSTART.search(t.splitlines()[0] if t else ""))]
    gruppen: list[list[tuple[datetime, str]]] = []
    for eintrag in relevant:
        if gruppen and (eintrag[0] - gruppen[-1][-1][0]).total_seconds() <= LAUF_LUECKE_S:
            gruppen[-1].append(eintrag)
        else:
            gruppen.append([eintrag])
    # Ein Neustart allein ist kein Update-Lauf (z. B. von Hand ausgelöst).
    laeufe = [_lauf(g) for g in gruppen if any(not _NEUSTART.search(t.splitlines()[0]) for _, t in g)]
    return list(reversed(laeufe))


def _protokoll_ende() -> str:
    try:
        with NOTIFY_LOG.open("rb") as f:
            f.seek(0, os.SEEK_END)
            groesse = f.tell()
            f.seek(max(0, groesse - LESE_MAX))
            roh = f.read()
    except OSError:
        return ""
    text = roh.decode("utf-8", errors="replace")
    if groesse > LESE_MAX:
        text = text.split("\n", 1)[-1]   # angeschnittene erste Zeile verwerfen
    return text


def laeufe(anzahl: int = 8) -> list[dict]:
    """Die letzten Update-Läufe (neueste zuerst). Neu gelesen nur, wenn das Protokoll wuchs."""
    try:
        st = NOTIFY_LOG.stat()
        schluessel = (st.st_mtime_ns, st.st_size)
    except OSError:
        return []
    with _cache_lock:
        if _cache["schluessel"] != schluessel:
            _cache["laeufe"] = laeufe_aus(_protokoll_ende())
            _cache["schluessel"] = schluessel
        return _cache["laeufe"][:anzahl]


# --- Festgehaltene Bausteine (Pin-Register von autoupdate.sh) -----------------------

def _pins_lesen() -> dict:
    try:
        daten = json.loads(PINS_FILE.read_text(encoding="utf-8"))
    except (OSError, ValueError):
        return {}
    return daten if isinstance(daten, dict) else {}


def festgehalten() -> list[dict]:
    """Bausteine, die der Sonntags-Lauf nach einem gescheiterten Update überspringt."""
    ergebnis = []
    for baustein, eintrag in _pins_lesen().items():
        if not (isinstance(eintrag, dict) and eintrag.get("pinned")):
            continue
        datum = str(eintrag.get("datum") or "")
        m = re.fullmatch(r"\d{4}-(\d{2})-(\d{2})", datum)
        seit = f"{m.group(2)}.{m.group(1)}." if m else (datum or "?")
        ergebnis.append({"baustein": baustein, "name": NAMEN.get(baustein, baustein),
                         "version": str(eintrag.get("version") or "?"), "seit": seit,
                         "grund": _lesbar(str(eintrag.get("grund") or ""))})
    return ergebnis


def freigeben(baustein: str) -> bool:
    """Pin lösen (atomar schreiben). False, wenn der Baustein gar nicht festgehalten war."""
    daten = _pins_lesen()
    if baustein not in daten:
        return False
    daten.pop(baustein)
    tmp = PINS_FILE.with_suffix(".tmp")
    try:
        tmp.write_text(json.dumps(daten, ensure_ascii=False, indent=2), encoding="utf-8")
        os.replace(tmp, PINS_FILE)
    except OSError:
        return False
    return True