Blame for t2s/evaluate.py

"""Yürütme doğruluğu (execution accuracy, EX) tabanlı değerlendirme.

EX: sistemin SQL'i kendi (çıkarımla oluşturulmuş) veritabanında çalıştırıldığında dönen sonuç
kümesi, altın SQL'in referans veritabanındaki sonucuyla eşleşiyorsa 1.
  - Satır sırası önemsiz (çoklu küme karşılaştırması)
  - Sayılar: |a-b| <= 0,01 + 0,001·|b|  (yuvarlama farklarına tolerans)
  - Metinler: Türkçe katlanmış karşılaştırma (İ/ı, büyük/küçük harf)
  - Sistem fazladan sütun döndürürse, altın sütunlarla eşleşen bir sütun alt kümesi aranır
Cevapsız sorularda (altın SQL yok) sistem cevap vermeyi reddederse doğru sayılır;
cevaplanabilir sorularda reddetmek yanlıştır.
"""
from __future__ import annotations

import itertools
import json
import time
from collections import Counter, defaultdict
from dataclasses import dataclass
from pathlib import Path

from .agent import Text2SQLAgent
from .errors import ErrorLog, diagnose_failure
from .schema import connect
from .trnum import fold

ROOT = Path(__file__).resolve().parent.parent
BENCH = ROOT / "benchmark" / "tr_tablosoru.json"


def load_benchmark(path=BENCH) -> list[dict]:
    return json.loads(Path(path).read_text(encoding="utf-8"))["sorular"]


def _val_eq(a, b) -> bool:
    if a is None or b is None:
        return a is None and b is None
    na = isinstance(a, (int, float)) and not isinstance(a, bool)
    nb = isinstance(b, (int, float)) and not isinstance(b, bool)
    if na and nb:
        return abs(a - b) <= 0.01 + 1e-3 * abs(b)
    if na != nb:
        return False
    return fold(str(a)) == fold(str(b))


def _sort_key(row):
    return tuple((0, round(v, 1)) if isinstance(v, (int, float)) and not isinstance(v, bool)
                 else (1, fold(str(v))) for v in row)


def _rows_eq(p, g) -> bool:
    if len(p) != len(g):
        return False
    ps, gs = sorted(p, key=_sort_key), sorted(g, key=_sort_key)
    if all(all(_val_eq(x, y) for x, y in zip(rp, rg)) for rp, rg in zip(ps, gs)):
        return True
    # sıralama anahtarı yuvarlamadan etkilenmiş olabilir: açgözlü eşleştirme
    used = [False] * len(gs)
    for rp in p:
        for k, rg in enumerate(gs):
            if not used[k] and all(_val_eq(x, y) for x, y in zip(rp, rg)):
                used[k] = True
                break
        else:
            return False
    return True


def results_match(pred: list[tuple], gold: list[tuple]) -> bool:
    if pred is None:
        return False
    if not gold:
        return not pred
    gw = len(gold[0])
    if not pred:
        return False
    pw = len(pred[0])
    if pw == gw and _rows_eq(pred, gold):
        return True
    if pw > gw and pw <= 6:
        for idx in itertools.permutations(range(pw), gw):
            if _rows_eq([tuple(r[i] for i in idx) for r in pred], gold):
                return True
    if pw == gw and gw > 1:
        for idx in itertools.permutations(range(pw)):
            if _rows_eq([tuple(r[i] for i in idx) for r in pred], gold):
                return True
    return False


@dataclass
class SystemSpec:
    name: str
    backend: object            # "kural" | LLMBackend
    db_mode: str = "tr"        # tr | en | text
    self_correct: bool = True
    use_evidence: bool = True
    max_attempts: int = 3
    static_check: bool = True    # statik denetim (ablasyon için kapatılabilir)


def gold_results(items, ref_db) -> dict[str, list]:
    con = connect(ref_db, readonly=True)
    out = {}
    for it in items:
        out[it["id"]] = [tuple(r) for r in con.execute(it["altin_sql"]).fetchall()] if it["altin_sql"] else None
    con.close()
    return out


def evaluate_system(spec: SystemSpec, items, db_paths: dict, ref_db, out_dir: Path,
                    error_log: ErrorLog | None = None, progress=None) -> dict:
    gold = gold_results(items, ref_db)
    con = connect(db_paths[spec.db_mode], readonly=True)
    agent = Text2SQLAgent(con, spec.backend, max_attempts=spec.max_attempts, self_correct=spec.self_correct,
                          static_check=spec.static_check,
                          error_log=error_log, log_source=f"degerlendirme:{spec.name}",
                          fallback_rule=False)  # deneylerde LLM sonucu kural tabanlı cevapla karışmasın
    records = []
    for k, it in enumerate(items):
        t0 = time.time()
        res = agent.run(it["soru"], evidence=it.get("kanit") if spec.use_evidence else None)
        g = gold[it["id"]]
        if not it["cevaplanabilir"]:
            ok = res.abstained
            first_ok = res.first_sql is None   # ilk denemede de reddetti mi
        else:
            ok = (not res.abstained) and results_match(res.rows, g)
            first_ok = res.first_rows is not None and res.first_error is None and results_match(res.first_rows, g)
        err = None if ok else diagnose_failure(it, res.sql, res.rows, g or [], res.abstained, res.exec_error,
                                               spec.db_mode)
        rec = {"id": it["id"], "bolum": it.get("bolum", "gelistirme"), "soru": it["soru"],
               "zorluk": it["zorluk"], "kategoriler": it["kategoriler"], "cevaplanabilir": it["cevaplanabilir"],
               "altin_sql": it["altin_sql"], "tahmin_sql": res.sql, "ilk_sql": res.first_sql,
               "durum": res.status, "cevap": res.answer, "dogru": ok, "ilk_deneme_dogru": first_ok,
               "deneme": res.attempts, "hata_tipi": err,
               "calisma_hatalari": [e.tip for e in res.events], "sure": round(time.time() - t0, 3),
               "tahmin_sonuc": [list(map(_jsonable, r)) for r in (res.rows or [])[:10]],
               "altin_sonuc": [list(map(_jsonable, r)) for r in (g or [])[:10]]}
        records.append(rec)
        if progress:
            progress(k + 1, len(items), rec)
    con.close()
    out_dir.mkdir(parents=True, exist_ok=True)
    (out_dir / f"sonuclar_{spec.name}.jsonl").write_text(
        "\n".join(json.dumps(r, ensure_ascii=False) for r in records), encoding="utf-8")
    return summarize(spec, records)


def _jsonable(v):
    return v if isinstance(v, (int, float, str)) or v is None else str(v)


def _ex(recs):
    return round(100.0 * sum(r["dogru"] for r in recs) / len(recs), 1) if recs else None


def summarize(spec: SystemSpec, records: list[dict]) -> dict:
    ans = [r for r in records if r["cevaplanabilir"]]
    una = [r for r in records if not r["cevaplanabilir"]]
    by = lambda key: {k: _ex(v) for k, v in sorted(_group(records, key).items())}
    cats = defaultdict(list)
    for r in records:
        for c in r["kategoriler"]:
            cats[c].append(r)
    return {
        "sistem": spec.name, "arka_uc": spec.backend if isinstance(spec.backend, str) else spec.backend.name,
        "sayi_kipi": spec.db_mode, "kendini_duzeltme": spec.self_correct, "kanit": spec.use_evidence,
        "n": len(records), "EX": _ex(records), "EX_ilk_deneme": round(100.0 * sum(r["ilk_deneme_dogru"] for r in records) / len(records), 1),
        "EX_cevaplanabilir": _ex(ans),
        "cevapsiz_dogru_red": _ex(una),
        "yanlis_red_orani": round(100.0 * sum(1 for r in ans if r["durum"] != "cevap") / len(ans), 1) if ans else None,
        "bolum": by("bolum"), "zorluk": by("zorluk"),
        "kategori": {k: _ex(v) for k, v in sorted(cats.items())},
        "kategori_n": {k: len(v) for k, v in sorted(cats.items())},
        "hata_tipleri": dict(Counter(r["hata_tipi"] for r in records if r["hata_tipi"]).most_common()),
        "ort_deneme": round(sum(r["deneme"] for r in records) / len(records), 2),
        "ort_sure_sn": round(sum(r["sure"] for r in records) / len(records), 3),
    }


def _group(records, key):
    g = defaultdict(list)
    for r in records:
        g[r[key]].append(r)
    return g


def gold_on_naive_db(items, db_paths: dict, ref_db) -> list[dict]:
    """Altın SQL'i naif sayı kipleriyle kurulmuş veritabanlarında çalıştırır.
    Sorgu doğru olsa bile yalnızca sayı biçimi yüzünden kaç cevabın bozulduğunu ölçer."""
    gold = gold_results(items, ref_db)
    out = []
    for mode, p in db_paths.items():
        con = connect(p, readonly=True)
        ok = err = 0
        broken = []
        n = 0
        for it in items:
            if not it["altin_sql"]:
                continue
            n += 1
            try:
                rows = [tuple(r) for r in con.execute(it["altin_sql"]).fetchall()]
                if results_match(rows, gold[it["id"]]):
                    ok += 1
                else:
                    broken.append(it["id"])
            except Exception:
                err += 1
                broken.append(it["id"])
        con.close()
        out.append({"kip": mode, "n": n, "EX": round(100.0 * ok / n, 1), "yurutme_hatasi": err,
                    "bozulan": broken})
    return out