Blame for scripts/eval/evaluer_recherche.py

"""
Mesure le classement de la PAGE DE RECHERCHE, document par document.

La page /search envoie une recherche en mode « text », filtree sur
status=published, 20 resultats par page. Ce script rejoue exactement cette
requete, sans cache et sans journaliser d'evenement de recherche (lecture
seule), et mesure le rang de la loi attendue :

- les 102 questions relues de tests/fixtures/eval/retrieval_eval_v1.json
  (loi attendue : expected_law_id) : hit@1, hit@3, hit@10, MRR@20 ;
- des requetes TEMOINS, chacune avec ce qu'on attend d'elle : la requete qui
  a revele le defaut (« peine de mort au cameroun »), et celles qui fondent la
  passe floue sur les titres (fautes de frappe : « nominaton », « code miner »).

Usage:
    python -m scripts.eval.evaluer_recherche
    python -m scripts.eval.evaluer_recherche --modes text hybrid --sortie data/eval_runs/recherche.json
"""

import argparse
import asyncio
import json
import sys
import unicodedata
from pathlib import Path
from typing import Any, Callable, Dict, List, Optional, Sequence

RACINE = Path(__file__).resolve().parents[2]
sys.path.insert(0, str(RACINE))

from app.core.database import AsyncSessionLocal
from app.schemas.search import SearchFilters, SearchRequest
from app.services.search_service import SearchService

JEU = RACINE / "tests" / "fixtures" / "eval" / "retrieval_eval_v1.json"
TAILLE_PAGE = 20  # TAILLE_PAGE_RECHERCHE du front (src/lib/pagination.ts)

CODE_PENAL, JUSTICE_MILITAIRE = 2236, 2085
CODES_MINIERS = {2210, 2228}


def _sans_accents(texte: str) -> str:
    return "".join(c for c in unicodedata.normalize("NFD", texte.lower())
                   if unicodedata.category(c) != "Mn")


def _titres_contiennent(mot: str, n: int) -> Callable[[Dict[str, Any]], bool]:
    """Les n premiers titres contiennent tous `mot` (sans accents)."""
    def verifier(reponse: Dict[str, Any]) -> bool:
        titres = [_sans_accents(r["title"]) for r in reponse["results"][:n]]
        return len(titres) == n and all(mot in t for t in titres)
    return verifier


def _rang_parmi(ids: set, n: int) -> Callable[[Dict[str, Any]], bool]:
    """Une des lois `ids` figure dans les n premiers resultats."""
    def verifier(reponse: Dict[str, Any]) -> bool:
        return any(r["law_id"] in ids for r in reponse["results"][:n])
    return verifier


def _navigation_directe(ids: set) -> Callable[[Dict[str, Any]], bool]:
    def verifier(reponse: Dict[str, Any]) -> bool:
        return bool(reponse["direct_navigation"]) and reponse["target_law_id"] in ids
    return verifier


# (requete, attente lisible, verification)
TEMOINS: List[tuple] = [
    ("peine de mort au cameroun", "Code penal ou justice militaire dans le top 3",
     _rang_parmi({CODE_PENAL, JUSTICE_MILITAIRE}, 3)),
    ("peine de mort", "Code penal ou justice militaire dans le top 3",
     _rang_parmi({CODE_PENAL, JUSTICE_MILITAIRE}, 3)),
    ("justice militaire", "Code de justice militaire premier",
     _rang_parmi({JUSTICE_MILITAIRE}, 1)),
    ("code pénal", "Code penal (loi 2236 ou decret 1769) dans le top 2",
     _rang_parmi({CODE_PENAL, 1769}, 2)),
    ("code minier", "un code minier dans le top 2", _rang_parmi(CODES_MINIERS, 2)),
    ("code miner", "faute de frappe : un code minier dans le top 3", _rang_parmi(CODES_MINIERS, 3)),
    ("cod minier", "faute de frappe : un code minier dans le top 3", _rang_parmi(CODES_MINIERS, 3)),
    ("nominaton", "faute de frappe : les 5 premiers titres parlent de nomination",
     _titres_contiennent("nomination", 5)),
    ("avancment", "faute de frappe : les 3 premiers titres parlent d'avancement",
     _titres_contiennent("avancement", 3)),
    ("article 35 du code minier", "navigation directe vers un code minier",
     _navigation_directe(CODES_MINIERS)),
]


async def rejouer(service: SearchService, requete: str, mode: str) -> Dict[str, Any]:
    """La requete de la page, sans cache ni journal d'evenement."""
    demande = SearchRequest(
        query=requete, mode=mode, limit=TAILLE_PAGE, offset=0,
        filters=SearchFilters(status="published"),
    )
    chunks = await service._execute_search_by_mode(demande)
    reponse = await service._build_search_response(demande, chunks, 0)
    return {
        "results": [{"law_id": r.law_id, "title": r.title, "score": round(r.relevance_score, 3),
                     "portee": r.match_scope} for r in reponse.results],
        "total": reponse.total,
        "direct_navigation": reponse.direct_navigation,
        "target_law_id": reponse.target_law_id,
    }


def _rang(reponse: Dict[str, Any], law_id: int) -> Optional[int]:
    for position, r in enumerate(reponse["results"], start=1):
        if r["law_id"] == law_id:
            return position
    return None


def synthese(rangs: Sequence[Optional[int]]) -> Dict[str, Any]:
    n = len(rangs) or 1
    return {
        "n": len(rangs),
        "hit@1": round(sum(1 for r in rangs if r and r <= 1) / n, 3),
        "hit@3": round(sum(1 for r in rangs if r and r <= 3) / n, 3),
        "hit@10": round(sum(1 for r in rangs if r and r <= 10) / n, 3),
        "mrr@20": round(sum(1 / r for r in rangs if r) / n, 3),
        "aucun_resultat": None,
    }


async def evaluer(modes: Sequence[str]) -> Dict[str, Any]:
    items = json.loads(JEU.read_text(encoding="utf-8"))["items"]
    sortie: Dict[str, Any] = {"modes": {}}
    async with AsyncSessionLocal() as session:
        service = SearchService(session, use_cache=False)
        for mode in modes:
            par_item, rangs, vides = [], {"keyword": [], "paraphrase": [], "tous": []}, 0
            for item in items:
                reponse = await rejouer(service, item["question"], mode)
                rang = _rang(reponse, item["expected_law_id"])
                vides += not reponse["results"]
                rangs["tous"].append(rang)
                rangs.setdefault(item.get("difficulty", "?"), []).append(rang)
                par_item.append({"question": item["question"], "attendu": item["expected_law_id"],
                                 "rang": rang, "total": reponse["total"]})
            temoins = []
            for requete, attente, verifier in TEMOINS:
                reponse = await rejouer(service, requete, mode)
                temoins.append({"requete": requete, "attente": attente, "ok": verifier(reponse),
                                "top5": [f"{r['law_id']} {r['portee']} {r['score']} {r['title'][:70]}"
                                         for r in reponse["results"][:5]]})
            resume = {cle: synthese(valeurs) for cle, valeurs in rangs.items() if valeurs}
            resume["tous"]["aucun_resultat"] = vides
            sortie["modes"][mode] = {"synthese": resume, "temoins": temoins, "items": par_item}
    return sortie


def afficher(sortie: Dict[str, Any]) -> None:
    for mode, donnees in sortie["modes"].items():
        print(f"\n=== mode {mode}")
        for cle, s in donnees["synthese"].items():
            print(f"  {cle:10} n={s['n']:3}  hit@1 {s['hit@1']:.3f}  hit@3 {s['hit@3']:.3f}  "
                  f"hit@10 {s['hit@10']:.3f}  mrr@20 {s['mrr@20']:.3f}"
                  + (f"  sans resultat {s['aucun_resultat']}" if s["aucun_resultat"] is not None else ""))
        ok = sum(t["ok"] for t in donnees["temoins"])
        print(f"  temoins : {ok}/{len(donnees['temoins'])}")
        for t in donnees["temoins"]:
            print(f"    [{'ok' if t['ok'] else 'KO'}] {t['requete']!r} : {t['attente']}")
            if not t["ok"]:
                for ligne in t["top5"]:
                    print(f"         {ligne}")


def main(argv: Optional[Sequence[str]] = None) -> int:
    parseur = argparse.ArgumentParser(description=__doc__.split("\n\n")[0])
    parseur.add_argument("--modes", nargs="+", default=["text"], choices=["text", "semantic", "hybrid"])
    parseur.add_argument("--sortie", type=Path, help="Ecrit le detail en JSON")
    args = parseur.parse_args(argv)
    sortie = asyncio.run(evaluer(args.modes))
    afficher(sortie)
    if args.sortie:
        args.sortie.parent.mkdir(parents=True, exist_ok=True)
        args.sortie.write_text(json.dumps(sortie, ensure_ascii=False, indent=1), encoding="utf-8")
        print(f"\nDetail : {args.sortie}")
    return 0


if __name__ == "__main__":
    sys.exit(main())