monitoring/metrics.py

"""Suivi des performances en production : parse `logs/api.log` (une ligne
JSON par appel à `/analyze`, écrite par `app/main.py`) et calcule des
métriques agrégées (taux de succès, confiance moyenne, causes de blocage).

Ne dépend d'aucun état en mémoire : peut tourner en script séparé pendant
que l'API tourne (lecture seule du fichier de log), ou être appelé depuis
`GET /metrics` pour un aperçu rapide en démo.
"""
import json
import os
from pathlib import Path
from typing import Any, Dict, List, Optional
from collections import Counter

MODULE_ROOT = Path(__file__).resolve().parent.parent

# YELY_DATA_DIR pointe vers un répertoire persistant (ex. le stockage
# persistant d'un Hugging Face Space, monté sur /data) quand il est défini —
# sinon on retombe sur le dossier du module (éphémère en conteneur Docker
# sans stockage persistant). Doit rester cohérent avec DATA_DIR dans
# `app/main.py` et `monitoring/feedback.py`, sinon /metrics et /failures
# liraient un fichier différent de celui où l'API écrit réellement.
DATA_DIR = Path(os.environ.get("YELY_DATA_DIR", str(MODULE_ROOT)))
DEFAULT_LOG_PATH = DATA_DIR / "logs" / "api.log"


def parse_log_entries(log_path: Optional[Path] = None) -> List[Dict[str, Any]]:
    """Extrait les entrées `{"request": ..., "response": ...}` de `api.log`.

    Le format de ligne est `"<timestamp> | <niveau> | <message>"` (voir le
    `Formatter` dans `app/main.py`) ; seules les lignes INFO dont le message
    est le JSON structuré loggé après chaque `/analyze` sont retenues — les
    autres lignes (ex. "Modèle CRNN chargé sur cpu.") sont ignorées.
    """
    log_path = log_path or DEFAULT_LOG_PATH
    if not log_path.exists():
        return []

    entries = []
    with open(log_path, "r", encoding="utf-8") as f:
        for line in f:
            parts = line.rstrip("\n").split(" | ", 2)
            if len(parts) != 3:
                continue
            _, _, message = parts
            try:
                data = json.loads(message)
            except json.JSONDecodeError:
                continue
            if isinstance(data, dict) and "response" in data:
                entries.append(data)
    return entries


def compute_metrics(log_path: Optional[Path] = None) -> Dict[str, Any]:
    """Calcule les métriques agrégées sur l'ensemble des entrées loggées.

    Retourne `total_requests=0` et des métriques à `None` si aucun appel n'a
    encore été journalisé (démo pas encore lancée), plutôt qu'une erreur.
    """
    entries = parse_log_entries(log_path)
    total = len(entries)
    if total == 0:
        return {
            "total_requests": 0,
            "success_rate": None,
            "avg_confidence_score": None,
            "blocking_causes": {},
            "image_quality_distribution": {},
        }

    responses = [e["response"] for e in entries]
    success_count = sum(1 for r in responses if r.get("success"))

    confidences = [r["confidence_score"] for r in responses if r.get("confidence_score") is not None]
    avg_confidence = sum(confidences) / len(confidences) if confidences else None

    blocking_causes = Counter(r["message"] for r in responses if not r.get("success"))
    quality_distribution = Counter(r.get("image_quality") for r in responses)

    return {
        "total_requests": total,
        "success_rate": round(success_count / total, 3),
        "avg_confidence_score": round(avg_confidence, 3) if avg_confidence is not None else None,
        "blocking_causes": dict(blocking_causes.most_common()),
        "image_quality_distribution": dict(quality_distribution.most_common()),
    }


def list_failed_requests(log_path: Optional[Path] = None, limit: int = 20) -> List[Dict[str, Any]]:
    """Retourne les `limit` dernières réponses bloquées (`success=false`),
    les plus récentes en premier — matière première de la revue manuelle
    (voir `web/stats.html` -> `web/feedback.html` et docs/MONITORING.md).

    Ne renvoie que les champs utiles à l'affichage/la correction, pas
    l'entrée de log complète (pas de `request.filename`, `model_version`...).
    """
    entries = parse_log_entries(log_path)
    failed = [e["response"] for e in entries if not e["response"].get("success")]
    failed.reverse()
    out = []
    for r in failed[:limit]:
        out.append({
            "photo_reference": r.get("photo_reference"),
            "message": r.get("message"),
            "image_quality": r.get("image_quality"),
            "detected_liters": r.get("detected_liters"),
            "detected_amount": r.get("detected_amount"),
            "fuel_price": r.get("fuel_price"),
            "confidence_score": r.get("confidence_score"),
            "transaction_datetime": r.get("transaction_datetime"),
        })
    return out


def _print_report(metrics: Dict[str, Any]) -> None:
    print("=== Suivi des performances YELY (logs/api.log) ===")
    print(f"Requêtes totales      : {metrics['total_requests']}")
    if metrics["total_requests"] == 0:
        print("Aucun appel journalisé pour l'instant.")
        return
    print(f"Taux de succès        : {metrics['success_rate'] * 100:.1f}%")
    conf = metrics["avg_confidence_score"]
    print(f"Confiance moyenne      : {conf if conf is None else round(conf * 100, 1)}%")
    print("Répartition qualité image :")
    for label, count in metrics["image_quality_distribution"].items():
        print(f"  - {label}: {count}")
    print("Causes de blocage (hors succès) :")
    for message, count in metrics["blocking_causes"].items():
        print(f"  - {message}: {count}")


if __name__ == "__main__":
    _print_report(compute_metrics())