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

"""
Aufräumen der Modell-Platte (Box-Wart, Umbau 09/2026): was belegt Platz, ohne gebraucht zu werden?

Zwei Arten von Kandidaten:
  • Einträge ohne Rolle   llama-swap-Einträge, die keine der heutigen Rollen tragen (Hirn, Coder, ihre
                          Bild-Zwillinge, Einbettung, Reranker) und in keiner immer-warmen Gruppe stehen.
  • Ordner ohne Eintrag   Unterordner von MODELS_DIR, auf die kein Eintrag mehr zeigt (alte Originale,
                          Reserve-Modelle, Drafts aus früheren Versuchen).

Gelöscht wird nur auf Knopfdruck des Nutzers (mit Rückfrage in der Oberfläche), nie automatisch
(User-Entscheid 23.09.: „nach der Umstellung löschen“ — die Entscheidung je Modell bleibt beim Nutzer).
Ein Eintrag wird aus der Config genommen, danach verschwinden nur Ordner, auf die NICHTS mehr zeigt:
Die Bild-Zwillinge teilen sich ihre Gewichte mit Hirn und Coder, geteilte Dateien bleiben immer liegen.
"""

import os
import shutil
from pathlib import Path

from config import MODELS_DIR

from services import llamaswap

AKTIVE_ROLLEN = {"hermes", "fast", "coder", "heavy", "vision", "coder-bild", "embed", "reranker"}
SYSTEM_ORDNER = {"mc2-backups", "radar", "pruefstand-cache", "lost+found"}
AUF_DER_BOX = os.name == "posix"      # auf dem Windows-PC (Entwicklung) wird nie gelöscht

# Was der Nutzer über bekannte Ordner wissen sollte, bevor er löscht (Stand 24.09.2026).
HINWEISE = {
    "Qwen3.6-35B-A3B-MTP-GGUF": "Original des Hirns vor der Uncensored-Variante (17.09.) — der Rückweg, falls das Hirn "
                                "einmal Probleme macht.",
    "Qwen3.8-27B-GGUF": "Original des Coders vor der Uncensored-Variante (17.09.) — der Rückweg für den Coder.",
    "Nemotron-3.5-Lightning-30B-A3B-GGUF": "Reserve-Hirn aus dem Prüfstand vom 17.09. (gleich gut bewertet, "
                                           "nie in Betrieb).",
    "DFlash-drafts": "Draft-Modelle aus früheren Versuchen, heute ungenutzt.",
    "drafts": "Draft-Modelle aus früheren Versuchen, heute ungenutzt.",
    "gpt-oss-120b": "Früheres heavy-Modell, hat seit dem 04.09. keine Rolle mehr.",
    "Qwen3-VL-30B-A3B-Instruct": "Früheres Bild-Modell, seit dem 24.09. durch die Bild-Zwillinge abgelöst.",
    "Muse-Glimmer-30B": "Debugger — die Rolle entfällt (Entscheid 23.09.: höchstens Hirn und Coder).",
}


def _ordner_von(pfad: str) -> str | None:
    """Oberster Ordner unter MODELS_DIR, in dem eine Datei liegt (oder None, wenn sie woanders liegt)."""
    if not os.path.isabs(pfad):
        return None
    try:
        rel = Path(pfad).resolve().relative_to(Path(MODELS_DIR).resolve())
    except (ValueError, OSError):
        return None
    return rel.parts[0] if len(rel.parts) > 1 else None


def _genutzte_ordner(models: dict) -> dict[str, set[str]]:
    """Ordner → Einträge, deren Befehl eine Datei darin nennt (Gewichte, Projektor, Draft)."""
    genutzt: dict[str, set[str]] = {}
    for name, spec in models.items():
        if not isinstance(spec, dict):
            continue
        for wort in str(spec.get("cmd") or "").split():     # jedes Wort, das unter MODELS_DIR liegt
            if (ordner := _ordner_von(wort.strip("'\""))):
                genutzt.setdefault(ordner, set()).add(name)
    return genutzt


def _groesse(pfad: Path) -> int:
    summe = 0
    for wurzel, _, dateien in os.walk(pfad):
        for datei in dateien:
            try:
                summe += os.path.getsize(os.path.join(wurzel, datei))
            except OSError:
                pass
    return summe


def _immer_warm(cfg: dict) -> set[str]:
    return {m for g in (cfg.get("groups") or {}).values()
            if isinstance(g, dict) and (g.get("persistent") or g.get("persist")) for m in g.get("members") or []}


def kandidaten() -> list[dict]:
    """Alles, was sich löschen ließe — größte zuerst. Löscht nichts."""
    cfg = llamaswap.read_config()
    models = cfg.get("models") or {}
    warm = _immer_warm(cfg)
    genutzt = _genutzte_ordner(models)
    liste: list[dict] = []
    for name, spec in models.items():
        aliase = {str(a).lower() for a in (spec or {}).get("aliases") or []}
        if aliase & AKTIVE_ROLLEN or name in warm:
            continue
        eigene = {o for o, wer in genutzt.items() if wer == {name}}   # nur von diesem Eintrag genutzt
        groesse = sum(_groesse(Path(MODELS_DIR) / o) for o in eigene)
        rolle = f"Rolle „{', '.join(sorted(aliase))}“ gibt es nicht mehr." if aliase else "Hat keine Rolle."
        liste.append({"art": "eintrag", "name": name, "groesse_gb": round(groesse / 1e9, 1),
                      "hinweis": HINWEISE.get(name, rolle)})
    try:
        ordner = sorted(p for p in Path(MODELS_DIR).iterdir() if p.is_dir())
    except OSError:
        ordner = []
    for pfad in ordner:
        if pfad.name in genutzt or pfad.name in SYSTEM_ORDNER or pfad.name.startswith("."):
            continue
        liste.append({"art": "ordner", "name": pfad.name, "groesse_gb": round(_groesse(pfad) / 1e9, 1),
                      "hinweis": HINWEISE.get(pfad.name, "Kein Eintrag nutzt diesen Ordner.")})
    return sorted(liste, key=lambda k: -k["groesse_gb"])


def _ordner_loeschen(name: str) -> int:
    """Einen Ordner direkt unter MODELS_DIR löschen; gibt die freigewordenen Bytes zurück."""
    wurzel = Path(MODELS_DIR).resolve()
    pfad = (wurzel / name).resolve()
    if pfad.parent != wurzel or not pfad.is_dir() or name in SYSTEM_ORDNER or name.startswith("."):
        raise ValueError(f"{name} ist kein löschbarer Modell-Ordner.")
    groesse = _groesse(pfad)
    shutil.rmtree(pfad)
    return groesse


def loeschen(art: str, name: str) -> dict:
    """Knopf „Löschen“: nur, was gerade als Kandidat gilt. Gibt {ok, text} oder {ok: False, detail} zurück."""
    if not any(k["art"] == art and k["name"] == name for k in kandidaten()):
        return {"ok": False, "detail": f"{name} ist nicht (mehr) zum Löschen frei."}
    if not AUF_DER_BOX:
        return {"ok": False, "detail": "Löschen geht nur auf der Box."}
    frei = 0
    try:
        if art == "ordner":
            frei = _ordner_loeschen(name)
        else:
            with llamaswap.config_sperre():
                cfg = llamaswap.read_config()
                models = cfg.get("models") or {}
                vorher = _genutzte_ordner(models)
                del models[name]
                for gruppe in (cfg.get("groups") or {}).values():
                    if isinstance(gruppe, dict) and name in (gruppe.get("members") or []):
                        gruppe["members"] = [m for m in gruppe["members"] if m != name]
                llamaswap.write_config(cfg)
            nachher = _genutzte_ordner(models)
            for ordner in sorted(set(vorher) - set(nachher)):     # nur Ordner, auf die jetzt nichts mehr zeigt
                frei += _ordner_loeschen(ordner)
    except (OSError, ValueError) as e:
        return {"ok": False, "detail": f"Löschen ging nicht: {e}"}
    return {"ok": True, "text": f"{name} gelöscht, {frei / 1e9:.1f} GB frei."}