benchmarks.py

"""
benchmarks.py — Collecte ANONYME de métriques d'analyse (Système Schrödinger).

Chaque analyse enregistre UNE ligne JSON (JSONL, append-only) contenant
exclusivement des métriques techniques : type de contenu, score final, scores
par analyseur, durée, résolution. AUCUNE donnée personnelle : pas de nom de
fichier, pas d'IP, pas de surnom, pas de contenu d'image (conforme RGPD —
mention dédiée dans la politique de confidentialité).

Ces données servent à calculer les vrais benchmarks d'Unkor sur des cas réels
et à piloter l'optimisation des poids de fusion (cycle collecte -> analyse ->
amélioration -> réévaluation).

Variables d'environnement :
  BENCHMARKS_PATH  chemin du fichier JSONL (défaut backend/data/benchmarks.jsonl ;
                   repli automatique vers le répertoire temporaire si non inscriptible)
  BENCHMARKS_OFF   "1" pour désactiver toute collecte
"""

import os
import json
import tempfile
import threading
from collections import deque
from datetime import datetime, timezone

MODEL_VERSION = "schrodinger-v2"    # V2 : fusion vidéo par confiance + mode video_frame
SYSTEM_NAME = "Système Schrödinger"

_BASE_DIR = os.path.dirname(os.path.abspath(__file__))
_DEFAULT_PATH = os.path.join(_BASE_DIR, "data", "benchmarks.jsonl")
_LOCK = threading.Lock()
_MAX_ENTRIES = 200_000        # borne mémoire pour l'agrégation
_resolved_path = None


def _off():
    return os.getenv("BENCHMARKS_OFF", "0").lower() in ("1", "true", "yes")


def _path():
    """Chemin inscriptible du JSONL (mémoïsé) — repli /tmp si besoin (HF Spaces)."""
    global _resolved_path
    if _resolved_path:
        return _resolved_path
    cand = os.getenv("BENCHMARKS_PATH", _DEFAULT_PATH)
    try:
        os.makedirs(os.path.dirname(cand) or ".", exist_ok=True)
        with open(cand, "a", encoding="utf-8"):
            pass
        _resolved_path = cand
    except Exception:
        _resolved_path = os.path.join(tempfile.gettempdir(), "unkor_benchmarks.jsonl")
        print(f"[benchmarks] chemin {cand} non inscriptible -> repli {_resolved_path}", flush=True)
    return _resolved_path


def _clean_scores(scores):
    """Ne conserve que des paires clé -> nombre (jamais de texte libre)."""
    out = {}
    for k, v in (scores or {}).items():
        if isinstance(v, (int, float)) and not isinstance(v, bool):
            out[str(k)] = round(float(v), 4)
    return out


def record(kind, final_score, scores=None, duration_ms=None, resolution=None):
    """Enregistre une analyse — anonyme, silencieux en cas d'échec (ne casse
    jamais une analyse)."""
    if _off():
        return
    try:
        entry = {
            "ts": datetime.now(timezone.utc).isoformat(timespec="seconds"),
            "type": str(kind),
            "final_score": round(float(final_score), 4),
            "scores": _clean_scores(scores),
            "duration_ms": int(duration_ms) if duration_ms is not None else None,
            "resolution": str(resolution) if resolution else None,
            "model_version": MODEL_VERSION,
        }
        line = json.dumps(entry, ensure_ascii=False)
        with _LOCK:
            with open(_path(), "a", encoding="utf-8") as f:
                f.write(line + "\n")
    except Exception as e:
        print(f"[benchmarks] enregistrement ignoré : {e}", flush=True)


def _load():
    entries = deque(maxlen=_MAX_ENTRIES)
    try:
        with open(_path(), "r", encoding="utf-8") as f:
            for line in f:
                line = line.strip()
                if not line:
                    continue
                try:
                    entries.append(json.loads(line))
                except Exception:
                    continue
    except FileNotFoundError:
        pass
    except Exception as e:
        print(f"[benchmarks] lecture impossible : {e}", flush=True)
    return list(entries)


def stats():
    """Statistiques agrégées pour GET /benchmarks/stats."""
    entries = _load()
    base = {
        "system": SYSTEM_NAME,
        "model_version": MODEL_VERSION,
        "total_analyses": len(entries),
    }
    if not entries:
        return {**base, "message": "Aucune analyse enregistrée pour le moment."}

    by_type, dist = {}, [0] * 10
    sum_score = 0.0
    durations, scores_acc = {}, {}
    for e in entries:
        kind = e.get("type", "?")
        s = float(e.get("final_score", 0.0))
        sum_score += s
        dist[min(9, max(0, int(s * 10)))] += 1
        t = by_type.setdefault(kind, {"count": 0, "sum_score": 0.0})
        t["count"] += 1
        t["sum_score"] += s
        d = e.get("duration_ms")
        if isinstance(d, (int, float)):
            durations.setdefault(kind, []).append(float(d))
        for k, v in (e.get("scores") or {}).items():
            if isinstance(v, (int, float)):
                acc = scores_acc.setdefault(k, [0.0, 0])
                acc[0] += float(v)
                acc[1] += 1

    all_durations = [d for lst in durations.values() for d in lst]
    return {
        **base,
        "avg_score": round(sum_score / len(entries), 4),
        "avg_duration_ms": round(sum(all_durations) / len(all_durations)) if all_durations else None,
        "by_type": {
            k: {
                "count": t["count"],
                "avg_score": round(t["sum_score"] / t["count"], 4),
                "avg_duration_ms": (round(sum(durations[k]) / len(durations[k]))
                                    if durations.get(k) else None),
            }
            for k, t in sorted(by_type.items())
        },
        "score_distribution": {
            f"{i / 10:.1f}-{(i + 1) / 10:.1f}": dist[i] for i in range(10)
        },
        "per_analyzer_avg": {
            k: round(s / n, 4) for k, (s, n) in sorted(scores_acc.items()) if n > 0
        },
        "period": {"from": entries[0].get("ts"), "to": entries[-1].get("ts")},
    }