Blame for evaluator.py

"""
evaluator.py
────────────
Métricas de calidad para las correcciones del sistema RAG.

Las tres comparaciones que realiza el evaluador:

  1. GT vs HTR       → error de PARTIDA (qué tan malo era el HTR)
  2. GT vs Corregido → error FINAL      (qué tan bueno es el RAG)
  3. HTR vs Corregido → MODERNISMOS     (qué cambió el LLM, no debería modernizar)

  Un cer_improvement positivo significa que el RAG mejoró el texto.
  Un cer_improvement negativo significa que el LLM empeoró el texto.

Métricas:
  - CER  (Character Error Rate) : distancia Levenshtein a nivel carácter
  - WER  (Word Error Rate)      : distancia Levenshtein a nivel palabra
  - Modernism score             : penalización por grafías s.XVI modernizadas
  - Regression score            : detecta si el LLM empeoró respecto al HTR

Uso:
    from evaluator import Evaluator
    ev = Evaluator()

    # Un solo par
    metrics = ev.evaluate_pair(htr="...", corrected="...", gt="...")
    print(ev.format_pair_report(metrics))

    # Batch
    report = ev.batch_evaluate(corrector, pairs[:50])
"""

import re
from typing import List, Dict
from knowledge_base import GRAFIA_PATTERNS


class Evaluator:

    # ── Métricas de edición ──────────────────────────────────────────────────

    @staticmethod
    def cer(reference: str, hypothesis: str) -> float:
        """
        Character Error Rate: fracción de caracteres incorrectos.
        0.0 = perfecto, 1.0 = todo mal.
        Compara: reference (GT) vs hypothesis (HTR o Corregido).
        """
        r, h = list(reference), list(hypothesis)
        return Evaluator._levenshtein(r, h) / max(len(r), 1)

    @staticmethod
    def wer(reference: str, hypothesis: str) -> float:
        """
        Word Error Rate: fracción de palabras incorrectas.
        0.0 = perfecto, 1.0 = todo mal.
        Compara: reference (GT) vs hypothesis (HTR o Corregido).
        """
        r = reference.split()
        h = hypothesis.split()
        return Evaluator._levenshtein(r, h) / max(len(r), 1)

    @staticmethod
    def _levenshtein(seq1: list, seq2: list) -> int:
        """Distancia de edición mínima entre dos secuencias."""
        m, n = len(seq1), len(seq2)
        dp = list(range(n + 1))
        for i in range(1, m + 1):
            prev = dp[:]
            dp[0] = i
            for j in range(1, n + 1):
                if seq1[i - 1] == seq2[j - 1]:
                    dp[j] = prev[j - 1]
                else:
                    dp[j] = 1 + min(prev[j], dp[j - 1], prev[j - 1])
        return dp[n]

    # ── Detector de modernismos ──────────────────────────────────────────────

    @staticmethod
    def modernism_penalty(htr: str, corrected: str) -> Dict:
        """
        Comparación 3: HTR vs Corregido.

        Detecta formas modernas que el LLM introdujo y que NO estaban
        en el HTR original. Estas son correcciones incorrectas porque
        el sistema NO debe modernizar grafías del s.XVI.

        Ejemplo de error detectado:
          HTR:       "fizo merçed"
          Corregido: "hizo merced"   ← modernizó f→h y ç→c (INCORRECTO)

        score: 1.0 = sin modernismos, 0.0 = muchos modernismos
        """
        issues = []
        htr_lower  = htr.lower()
        corr_lower = corrected.lower()

        for p in GRAFIA_PATTERNS:
            modern        = p["modern"].lower()
            ancient_forms = [f.strip().lower() for f in p["ancient"].split("/")]

            # El corregido tiene la forma moderna Y el HTR no la tenía
            if modern in corr_lower and modern not in htr_lower:
                # Además el HTR tampoco tenía la forma antigua
                # (si la tenía, es una expansión de abreviatura válida)
                if not any(af in htr_lower for af in ancient_forms):
                    issues.append({
                        "modern":   p["modern"],
                        "ancient":  p["ancient"],
                        "rule":     p["rule"],
                        "category": p.get("category", ""),
                    })

        score = max(0.0, 1.0 - len(issues) * 0.1)
        return {
            "count":  len(issues),
            "issues": issues,
            "score":  round(score, 4),
        }

    # ── Detector de regresiones ──────────────────────────────────────────────

    @staticmethod
    def regression_check(htr: str, corrected: str, gt: str) -> Dict:
        """
        Detecta si el LLM empeoró el texto respecto al HTR original.

        Un resultado positivo en cer_improvement no garantiza que todo
        esté bien — el LLM podría haber corregido unos errores e
        introducido otros. Esta función identifica palabras que estaban
        bien en el HTR y el LLM cambió incorrectamente.
        """
        htr_words  = htr.split()
        corr_words = corrected.split()
        gt_words   = gt.split()

        regressions = []
        # Comparar palabra a palabra hasta el mínimo de las tres listas
        for i, gt_w in enumerate(gt_words):
            htr_w  = htr_words[i]  if i < len(htr_words)  else ""
            corr_w = corr_words[i] if i < len(corr_words) else ""

            # El HTR estaba bien, el corregido está mal
            if htr_w == gt_w and corr_w != gt_w:
                regressions.append({
                    "position": i,
                    "gt":       gt_w,
                    "htr":      htr_w,   # correcto en HTR
                    "corrected":corr_w,  # empeorado por el LLM
                })

        return {
            "count":       len(regressions),
            "regressions": regressions,
        }

    # ── Evaluación de un par ─────────────────────────────────────────────────

    def evaluate_pair(self, htr: str, corrected: str, gt: str) -> Dict:
        """
        Evalúa una corrección con las tres comparaciones:

          Comparación 1 — GT vs HTR:
            Mide el error de partida. Cuánto se alejaba el HTR del GT.

          Comparación 2 — GT vs Corregido:
            Mide el error final. Cuánto se aleja la corrección del GT.
            cer_improvement > 0 → el RAG mejoró el texto
            cer_improvement < 0 → el RAG empeoró el texto

          Comparación 3 — HTR vs Corregido:
            Detecta modernismos introducidos por el LLM.
            El LLM NO debe cambiar grafías válidas del s.XVI.
        """
        # ── Comparación 1: GT vs HTR (error de partida) ────────────────────
        cer_htr = self.cer(gt, htr)
        wer_htr = self.wer(gt, htr)

        # ── Comparación 2: GT vs Corregido (error final) ───────────────────
        cer_corr = self.cer(gt, corrected)
        wer_corr = self.wer(gt, corrected)

        cer_improvement = cer_htr - cer_corr   # positivo = mejoró
        wer_improvement = wer_htr - wer_corr

        # Veredicto de la corrección
        if cer_improvement > 0.02:
            verdict = "✓ MEJORADO"
        elif cer_improvement < -0.02:
            verdict = "✗ EMPEORADO"
        else:
            verdict = "~ SIN CAMBIO SIGNIFICATIVO"

        # ── Comparación 3: HTR vs Corregido (modernismos) ─────────────────
        modernism  = self.modernism_penalty(htr, corrected)
        regression = self.regression_check(htr, corrected, gt)

        return {
            # ── Error de partida (GT vs HTR) ──────────────────────────────
            "cer_before":         round(cer_htr, 4),
            "wer_before":         round(wer_htr, 4),

            # ── Error final (GT vs Corregido) ─────────────────────────────
            "cer_after":          round(cer_corr, 4),
            "wer_after":          round(wer_corr, 4),

            # ── Mejora neta ───────────────────────────────────────────────
            "cer_improvement":    round(cer_improvement, 4),
            "wer_improvement":    round(wer_improvement, 4),
            "verdict":            verdict,

            # ── Modernismos (HTR vs Corregido) ────────────────────────────
            "modernism_score":    modernism["score"],
            "modernism":          modernism,

            # ── Regresiones ───────────────────────────────────────────────
            "regression":         regression,
        }

    # ── Reporte legible de un par ────────────────────────────────────────────

    @staticmethod
    def format_pair_report(metrics: Dict) -> str:
        """
        Formatea las métricas de un par en texto legible para la UI.
        """
        lines = []
        lines.append("─" * 50)
        lines.append(" EVALUACIÓN DE LA CORRECCIÓN")
        lines.append("─" * 50)

        # Comparación 1 — error de partida
        lines.append("\n① Error de partida  (GT vs HTR original)")
        lines.append(f"   CER: {metrics['cer_before']:.2%}   WER: {metrics['wer_before']:.2%}")

        # Comparación 2 — error final
        lines.append("\n② Error final       (GT vs Texto corregido)")
        lines.append(f"   CER: {metrics['cer_after']:.2%}   WER: {metrics['wer_after']:.2%}")

        # Mejora neta
        cer_imp = metrics["cer_improvement"]
        wer_imp = metrics["wer_improvement"]
        sign_c  = "+" if cer_imp >= 0 else ""
        sign_w  = "+" if wer_imp >= 0 else ""
        lines.append(f"\n   Mejora CER: {sign_c}{cer_imp:.2%}   Mejora WER: {sign_w}{wer_imp:.2%}")
        lines.append(f"   {metrics['verdict']}")

        # Comparación 3 — modernismos
        mod = metrics["modernism"]
        lines.append(f"\n③ Modernismos introducidos  (HTR vs Corregido)")
        if mod["count"] == 0:
            lines.append("   ✓ Ninguno — el LLM respetó las grafías del s.XVI")
        else:
            lines.append(f"   ✗ {mod['count']} modernismo(s) detectado(s):")
            for issue in mod["issues"]:
                lines.append(
                    f"     • '{issue['modern']}' introducido "
                    f"(debería ser '{issue['ancient']}'): {issue['rule']}"
                )

        # Regresiones
        reg = metrics["regression"]
        if reg["count"] > 0:
            lines.append(f"\n⚠ Regresiones: {reg['count']} palabra(s) correctas en HTR empeoradas por el LLM:")
            for r in reg["regressions"][:5]:
                lines.append(
                    f"     • posición {r['position']}: "
                    f"HTR='{r['htr']}' → Corregido='{r['corrected']}' "
                    f"(GT='{r['gt']}')"
                )

        lines.append("─" * 50)
        return "\n".join(lines)

    # ── Evaluación en batch ──────────────────────────────────────────────────

    def batch_evaluate(
        self, corrector, pairs: List[Dict], verbose: bool = True
    ) -> Dict:
        """
        Evalúa el sistema sobre una lista de pares con groundtruth.
        Retorna métricas agregadas + detalle por par.
        """
        results = []
        for i, pair in enumerate(pairs):
            if verbose:
                print(f"  Evaluando {i+1}/{len(pairs)}: {pair['id']}")
            try:
                out = corrector.correct(pair["htr"])
                metrics = self.evaluate_pair(
                    htr=pair["htr"],
                    corrected=out["corrected"],
                    gt=pair["gt"],
                )
                metrics["id"]        = pair["id"]
                metrics["htr"]       = pair["htr"]
                metrics["corrected"] = out["corrected"]
                metrics["gt"]        = pair["gt"]
                results.append(metrics)
            except Exception as e:
                print(f"   Error en {pair['id']}: {e}")

        if not results:
            return {"error": "Sin resultados"}

        def avg(key):
            return round(sum(r[key] for r in results) / len(results), 4)

        n = len(results)
        mejoras    = sum(1 for r in results if r["cer_improvement"] >  0.02)
        empeorados = sum(1 for r in results if r["cer_improvement"] < -0.02)
        sin_cambio = n - mejoras - empeorados

        summary = {
            "n_evaluated":           n,

            # ── Comparación 1: GT vs HTR ──────────────────────────────────
            "avg_cer_before":        avg("cer_before"),
            "avg_wer_before":        avg("wer_before"),

            # ── Comparación 2: GT vs Corregido ────────────────────────────
            "avg_cer_after":         avg("cer_after"),
            "avg_wer_after":         avg("wer_after"),

            # ── Mejora neta ───────────────────────────────────────────────
            "avg_cer_improvement":   avg("cer_improvement"),
            "avg_wer_improvement":   avg("wer_improvement"),
            "n_mejorados":           mejoras,
            "n_empeorados":          empeorados,
            "n_sin_cambio":          sin_cambio,

            # ── Comparación 3: Modernismos ────────────────────────────────
            "avg_modernism_score":   avg("modernism_score"),

            "detail": results,
        }

        if verbose:
            print(f"\n{'─'*55}")
            print(f" RESUMEN EVALUACIÓN ({n} pares)")
            print(f"{'─'*55}")
            print(f"\n① Error de partida  (GT vs HTR):")
            print(f"   CER: {summary['avg_cer_before']:.2%}   WER: {summary['avg_wer_before']:.2%}")
            print(f"\n② Error final       (GT vs Corregido):")
            print(f"   CER: {summary['avg_cer_after']:.2%}   WER: {summary['avg_wer_after']:.2%}")
            print(f"\n   Mejora CER: {summary['avg_cer_improvement']:+.2%}   "
                  f"Mejora WER: {summary['avg_wer_improvement']:+.2%}")
            print(f"\n   Mejorados    : {mejoras}/{n}  ({mejoras/n:.0%})")
            print(f"   Empeorados   : {empeorados}/{n}  ({empeorados/n:.0%})")
            print(f"   Sin cambio   : {sin_cambio}/{n}  ({sin_cambio/n:.0%})")
            print(f"\n③ Modernismos  (HTR vs Corregido):")
            print(f"   Score promedio: {summary['avg_modernism_score']:.2%} "
                  f"(1.0 = sin modernismos)")
            print(f"{'─'*55}")

        return summary