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

"""
Automatische Modell-Entdeckung ("aktuell beste Modelle"): fragt vertrauenswürdige
HF-Orgs live ab, kategorisiert per Stichwort, rankt nach Hardware-Fit + Beliebtheit
und cached. Portiert aus Mission Control v1 (cookbook.py-Discover).

Wichtig (Greenfield-Fix gegen v1): EIN gemeinsamer Ranking-Helfer `rank_runnable`
ist die Quelle der Wahrheit — sowohl die „beste Empfehlung" je Kategorie als auch
spätere Auto-Setups nutzen ihn, damit sie nie auseinanderlaufen.
"""

import json
import logging
import math
import os
import time
from datetime import datetime

import httpx
from config import DISCOVER_CACHE_PATH, DISCOVER_TTL

from services import catalog
from services.caps import capabilities
from services.fit import evaluate_fit, extract_params_b, max_ctx_for
from services.sources import CATEGORIES, SKIP_TOKENS, TRUSTED_AUTHORS

log = logging.getLogger(__name__)

_FIT_ORDER = {"perfect": 0, "marginal": 1, "too_tight": 2}


def _categorize(repo_id: str) -> str:
    low = repo_id.lower()
    for cat in CATEGORIES:
        if any(k in low for k in cat["kw"]):
            return cat["role"]
    return "scout"


def _fetch_author_models(author: str) -> list:
    url = (f"https://huggingface.co/api/models?author={author}"
           f"&filter=gguf&sort=downloads&direction=-1&limit=40")
    try:
        with httpx.Client(timeout=12.0) as c:
            data = c.get(url).json()
        return data if isinstance(data, list) else []
    except Exception:
        log.debug("discover: Abfrage für Autor %s fehlgeschlagen", author, exc_info=True)
        return []


def _age_days(last_modified, now_ts: float) -> float:
    """Alter eines HF-Modells in Tagen (lastModified ISO). Unbekannt → ~1.5 Jahre."""
    if not last_modified:
        return 540.0
    try:
        dt = datetime.fromisoformat(str(last_modified).replace("Z", "+00:00"))
        return max((now_ts - dt.timestamp()) / 86400.0, 0.0)
    except Exception:
        return 540.0


def _score(m: dict, now_ts: float) -> float:
    """Zukunftssicherer Rang-Score für DIESE Hardware. Kombiniert:
    - Fit: perfect dominiert (Bonus 3.0 > Summe der übrigen Terme → passt-komfortabel zuerst),
    - Recency: neuere Generationen bevorzugt (Halbwertszeit ~9 Monate über lastModified),
    - Capability: mehr Parameter (log-skaliert),
    - Popularity: Downloads (log-skaliert).
    So gewinnt bei vergleichbarer Größe die NEUERE Generation (z.B. Qwen3-Coder vor
    Qwen2.5-Coder), ohne dass kleine Populär-Modelle große verdrängen."""
    fit_bonus = 3.0 if m["fit"]["level"] == "perfect" else 0.0
    recency = 0.5 ** (_age_days(m.get("lastModified"), now_ts) / 270.0)
    cap = math.log2(max(float(m.get("params_b") or 1.0), 1.0) + 1.0) / 8.0
    pop = math.log10(float(m.get("downloads") or 0) + 1.0) / 7.0
    return fit_bonus + 1.2 * recency + 1.2 * cap + 0.5 * pop


def rank_runnable(models: list[dict]) -> list[dict]:
    """EINE Quelle der Wahrheit fürs Ranking lauffähiger Modelle für DIESE Hardware.
    Nur was passt (too_tight fliegt raus), dann nach `_score` (Fit + Recency + Capability
    + Popularity). Bevorzugt neuere, fähige Modelle → zukunftssicher; „Modelle finden"
    schlägt nie ein Downgrade vor (Downgrade-Sperre zusätzlich in maintenance)."""
    now_ts = time.time()
    return sorted(
        [m for m in models if m["fit"]["level"] != "too_tight"],
        key=lambda m: -_score(m, now_ts),
    )


def refresh_discover(ram_gb: float) -> dict:
    """Quellen live abfragen, kategorisieren, ranken, cachen. Wirft nur, wenn KEINE
    Quelle erreichbar war."""
    raw, seen, ok = [], set(), 0
    for author in TRUSTED_AUTHORS:
        models = _fetch_author_models(author)
        if models:
            ok += 1
        for m in models:
            rid = m.get("id")
            if not rid or rid in seen:
                continue
            seen.add(rid)
            raw.append(m)
    if ok == 0 and not catalog.entries():
        raise RuntimeError("Keine Quelle erreichbar.")

    by_cat: dict[str, list] = {c["role"]: [] for c in CATEGORIES}
    for m in raw:
        rid = m["id"]
        low = rid.lower()
        if any(tok in low for tok in SKIP_TOKENS):
            continue
        role = _categorize(rid)
        params_b = extract_params_b(rid)
        quant = "Q4_K_M"  # Referenz-Quant für die Fit-Einschätzung
        fit = evaluate_fit(params_b, quant, 8192, ram_gb, name=rid)
        tags = [str(t) for t in (m.get("tags") or [])]
        by_cat[role].append({
            "name": rid.split("/")[-1], "author": rid.split("/")[0], "repo": rid,
            "role": role, "params_b": params_b, "quant": quant, "tags": tags,
            "downloads": int(m.get("downloads") or 0), "likes": int(m.get("likes") or 0),
            "lastModified": m.get("lastModified"),
            "fit": fit, "optimal_ctx": max_ctx_for(params_b, quant, ram_gb),
            "caps": capabilities(name=rid, hf={"tags": tags}),
        })

    cats = []
    for c in CATEGORIES:
        role = c["role"]
        # 1) KATALOG zuerst (kuratierte, korrekte Metadaten, MoE-bewusst gerankt) —
        #    macht die Empfehlung präzise statt Namens-Raterei.
        cat_entries = sorted(catalog.entries_for_role(role),
                             key=lambda e: -catalog.stack_score(e, ram_gb))
        cat_models = [m for m in (catalog.to_model_dict(e, ram_gb) for e in cat_entries)
                      if m["fit"]["level"] != "too_tight"]
        # 2) HF-Dynamik als Ergänzung (nicht-kuratierte Funde), dedupliziert.
        hf_ranked = rank_runnable(by_cat[role])
        seen = {catalog._norm(m["repo"]) for m in cat_models}
        extra = [h for h in hf_ranked if catalog._norm(h["repo"]) not in seen]
        combined = cat_models + extra
        if combined:
            cats.append({
                "role": role, "title": c["title"], "icon": c["icon"],
                "models": combined[:6],
                # Empfehlung = bester KURATIERTER Eintrag, sonst beste HF-Fundstelle.
                "recommended": (cat_models[0]["repo"] if cat_models
                                else (hf_ranked[0]["repo"] if hf_ranked else None)),
            })

    data = {"updated": time.time(), "categories": cats}
    try:
        DISCOVER_CACHE_PATH.parent.mkdir(parents=True, exist_ok=True)
        tmp = DISCOVER_CACHE_PATH.with_name(DISCOVER_CACHE_PATH.name + ".tmp")
        tmp.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")
        os.replace(tmp, DISCOVER_CACHE_PATH)
    except Exception:
        log.debug("discover: Cache-Schreiben fehlgeschlagen (nur Beschleunigung)", exc_info=True)
    return data


def load_discover() -> dict | None:
    try:
        if DISCOVER_CACHE_PATH.exists():
            return json.loads(DISCOVER_CACHE_PATH.read_text(encoding="utf-8"))
    except Exception:
        log.debug("discover: Cache-Lesen fehlgeschlagen", exc_info=True)
    return None


def safe_discover(ram_gb: float) -> dict | None:
    """Aus Cache (wenn frisch) oder live; wirft nie — None wenn nichts da."""
    cached = load_discover()
    if cached and (time.time() - cached.get("updated", 0) < DISCOVER_TTL):
        return cached
    try:
        return refresh_discover(ram_gb)
    except Exception:
        log.warning("discover: Live-Refresh fehlgeschlagen, nutze Cache", exc_info=True)
        return cached