Blame for t2s/qa_eval.py

"""Cevap düzeyinde soru-cevap değerlendirmesi (altın SQL gerektirmez).

Pilot, stres ve ağır LLM kümeleri bu modülle çalışır; arayüzdeki "Model karşılaştırma" bölümü ve
pilot/pilot_calistir.py aynı kodu kullanır.

Değerlendirme kuralları
  sayi     : sonucun ilk satırındaki sayılardan biri |a−b| ≤ 0,01 + 10⁻³|b| ile eşleşirse doğru
             (isaret_serbest: mutlak değerler de denenir)
  metin    : ilk satırdaki bir değer Türkçe katlama sonrası beklenene eşitse doğru; tarih/dönem ISO'ya çevrilir.
             'kabul' listesi varsa (ör. kurum kısa adı) değerin içinde geçmesi yeterlidir
  liste    : ilk sütundaki değer kümesi beklenen kümeye eşitse doğru
  cevapsiz : sistem cevap üretmezse doğru
"""
from __future__ import annotations

import hashlib
import json
import re
import time
from dataclasses import dataclass
from pathlib import Path

from .agent import Text2SQLAgent, format_answer
from .ai import AIHelper
from .llm import USAGE
from .pipeline import build_database
from .schema import connect
from .trnum import fold, parse_date

ROOT = Path(__file__).resolve().parent.parent
TOL_A, TOL_R = 0.01, 1e-3


# ====================================================================== test kümeleri
@dataclass
class QASet:
    key: str
    belgeler: Path
    sorular: Path
    ham: Path | None
    belge_basina: bool          # True: her belge ayrı veritabanı
    filtre: str | None = None   # ör. "hizli": yalnız q["hizli"] olan sorular

    def questions(self) -> list[dict]:
        spec = json.loads(self.sorular.read_text(encoding="utf-8"))
        qs = spec["sorular"] if isinstance(spec, dict) else spec
        if self.filtre:
            qs = [q for q in qs if q.get(self.filtre)]
        return qs

    def available(self) -> bool:
        return self.belgeler.exists() and self.sorular.exists()

    def docs(self) -> list[Path]:
        return sorted(p for p in self.belgeler.iterdir() if p.suffix.lower() in (".pdf", ".html", ".htm"))


SETS = {
    "agir": QASet("agir", ROOT / "agir_kume" / "belgeler", ROOT / "agir_kume" / "sorular.json",
                  ROOT / "agir_kume" / "ham_degerler.json", False),
    "agir_hizli": QASet("agir_hizli", ROOT / "agir_kume" / "belgeler", ROOT / "agir_kume" / "sorular.json",
                        ROOT / "agir_kume" / "ham_degerler.json", False, "hizli"),
    "stres": QASet("stres", ROOT / "stres_kumesi" / "belgeler", ROOT / "stres_kumesi" / "sorular.json",
                   ROOT / "stres_kumesi" / "ham_degerler.json", True),
    "pilot": QASet("pilot", ROOT / "pilot" / "belgeler", ROOT / "pilot" / "sorular.json",
                   ROOT / "pilot" / "ham_degerler.json", False),
}


def code_hash() -> str:
    return hashlib.sha256(b"".join(p.read_bytes() for p in sorted((ROOT / "t2s").glob("*.py")))).hexdigest()[:16]


def extraction_hash() -> str:
    """Çıkarımı etkileyen modüllerin özeti: değişmedikçe kurulu veritabanı yeniden kullanılır."""
    names = ("extract.py", "pipeline.py", "schema.py", "trnum.py")
    return hashlib.sha256(b"".join((ROOT / "t2s" / n).read_bytes() for n in names)).hexdigest()[:16]


# ====================================================================== hakem
def _num(v):
    if isinstance(v, bool):
        return None
    if isinstance(v, (int, float)):
        return float(v)
    return None


def _txt(v):
    if isinstance(v, float) and v.is_integer():
        return str(int(v))
    return str(v)


def _key(v) -> str:
    """Metin karşılaştırma anahtarı: Türkçe katlama; tarih/dönem ise ISO ('Ekim 2025' == '2025-10' == '2025/10')."""
    s = str(v).strip()
    m = re.fullmatch(r"(\d{4})[/.](\d{1,2})", s)
    if m:
        return f"{m.group(1)}-{int(m.group(2)):02d}"
    return parse_date(s) or fold(s)


def judge(q: dict, res) -> tuple[bool, str]:
    tip = q["tip"]
    if tip == "cevapsiz":
        return res.abstained, "red" if res.abstained else "cevap üretti"
    if res.abstained or not res.rows:
        return False, f"red/boş ({res.status})"
    first = res.rows[0]
    if tip == "sayi":
        g = float(q["cevap"])
        vals = [x for x in (_num(v) for v in first) if x is not None]
        if q.get("isaret_serbest"):
            vals += [abs(x) for x in vals]
        ok = any(abs(x - g) <= TOL_A + TOL_R * abs(g) for x in vals)
        return ok, f"bulunan {vals[:3]}"
    if tip == "metin" and q.get("kabul"):
        keys = [fold(str(k)) for k in q["kabul"]] + [fold(str(q["cevap"]))]
        ok = any(v is not None and any(k in fold(_txt(v)) for k in keys) for v in first)
        return ok, f"bulunan {[v for v in first if v is not None][:3]}"
    if tip == "metin":
        g = _key(q["cevap"])
        ok = any(v is not None and _key(_txt(v)) == g for v in first)
        return ok, f"bulunan {[v for v in first if v is not None][:3]}"
    if tip == "liste":
        got = {_key(_txt(r[0])) for r in res.rows if r and r[0] is not None}
        exp = {_key(x) for x in q["cevap"]}
        return got == exp, f"{len(got & exp)}/{len(exp)} ortak, {len(got - exp)} fazla"
    return False, "bilinmeyen tip"


def value_recall(con, doc: str, raw_values: list[float]) -> dict:
    """Belgenin ham sayısal değerlerinden kaçı, o belgeden çıkarılan tablolarda bir sayı olarak bulunuyor?"""
    tabs = [t for (t,) in con.execute("SELECT tablo FROM _tablolar WHERE kaynak=?", (doc,))]
    found = set()
    for t in tabs:
        for (col, typ) in con.execute(f'SELECT name, type FROM pragma_table_info("{t}")'):
            if typ.upper() not in ("REAL", "INTEGER"):
                continue
            for (v,) in con.execute(f'SELECT "{col}" FROM "{t}" WHERE "{col}" IS NOT NULL'):
                found.add(round(float(v), 4))
    need = [round(float(x), 4) for x in raw_values]
    hit = sum(1 for x in need if x in found or -x in found)
    return {"tablo": len(tabs), "ham_deger": len(need), "bulunan": hit,
            "oran": round(100 * hit / len(need), 1) if need else None, "tablolar": tabs}


# ====================================================================== veritabanları
def prepare_dbs(qs: QASet, out_dir: Path, docs: list[Path] | None = None, reuse: bool = True) -> list[tuple[Path, list[Path]]]:
    """[(veritabanı, belgeler)] döndürür. Çıkarım kodu ve belgeler değişmedikçe kurulu veritabanı yeniden kullanılır."""
    docs = docs or qs.docs()
    out_dir.mkdir(parents=True, exist_ok=True)
    groups = [(out_dir / f"{d.stem}.db", [d]) for d in docs] if qs.belge_basina else [(out_dir / f"{out_dir.name}.db", docs)]
    sig = extraction_hash() + "|" + hashlib.sha256(b"".join(d.read_bytes()[:4096] + str(d.stat().st_size).encode()
                                                             for d in docs)).hexdigest()[:12]
    stamp = out_dir / ".imza"
    fresh = reuse and stamp.exists() and stamp.read_text() == sig and all(p.exists() for p, _ in groups)
    if not fresh:
        for db, gdocs in groups:
            db.unlink(missing_ok=True)
            build_database(db, docs=gdocs, mode="tr")
        stamp.write_text(sig)
    return groups


# ====================================================================== çalıştırma
def run_set(qs: QASet, backend, out_dir: Path, *, questions: list[dict] | None = None, max_attempts: int = 3,
            self_correct: bool = True, use_ai_tables: bool = True, progress=None, limit: int | None = None,
            reuse_db: bool = True) -> dict:
    """Bir test kümesini bir arka uçla çalıştırır; {'ozet':…, 'kayitlar':…} döndürür ve out_dir'e yazar."""
    t0 = time.time()
    u0 = (USAGE.calls, USAGE.input_tokens, USAGE.output_tokens)
    questions = questions if questions is not None else qs.questions()
    if limit:
        questions = questions[:limit]
    dbkey = qs.key.replace("_hizli", "")
    if qs.key == "ozel":
        dbkey = "ozel_" + hashlib.sha256(str(qs.belgeler.resolve()).encode()).hexdigest()[:8]
    groups = prepare_dbs(qs, ROOT / "outputs" / "qa" / "db" / dbkey, reuse=reuse_db)
    ham = json.loads(qs.ham.read_text(encoding="utf-8")) if qs.ham and qs.ham.exists() else {}
    llm = backend != "kural"
    ai = AIHelper(backend, cache=False) if (llm and use_ai_tables) else None
    recs, cikarim, done, n = [], {}, 0, len(questions)
    for db, gdocs in groups:
        names = {d.name for d in gdocs}
        mine = [q for q in questions if (not qs.belge_basina) or q["belge"] in names]
        con = connect(db, readonly=True)
        for d in gdocs:
            info = {"yazilan_tablo": [t for (t,) in con.execute("SELECT tablo FROM _tablolar WHERE kaynak=?", (d.name,))]}
            if d.name in ham:
                info["deger_geri_cagirma"] = value_recall(con, d.name, ham[d.name])
            cikarim[d.name] = info
        if not mine:
            con.close()
            continue
        agent = Text2SQLAgent(con, backend, max_attempts=max_attempts, self_correct=self_correct,
                              fallback_rule=False, log_source="qa_eval", ai=ai)
        for q in mine:
            try:
                res = agent.run(q["soru"])
                ok, why = judge(q, res)
                rec = {**q, "dogru": ok, "aciklama": why, "durum": res.status, "sql": res.sql,
                       "sonuc": [list(r) for r in res.rows[:5]], "cevap_metni": format_answer(res, con),
                       "deneme": res.attempts, "sure": round(res.duration, 2)}
            except Exception as e:  # noqa: BLE001 - tek soru hatası tüm çalıştırmayı durdurmasın
                rec = {**q, "dogru": False, "aciklama": f"istisna: {str(e)[:200]}", "durum": "istisna", "sql": None,
                       "sonuc": [], "cevap_metni": "", "deneme": 0, "sure": 0}
            recs.append(rec)
            done += 1
            if progress:
                progress(done, n, rec)
        con.close()
    summary = summarize(recs)
    summary.update({"arka_uc": "kural" if not llm else backend.name, "kume": qs.key, "cikarim": cikarim,
                    "sure_sn": round(time.time() - t0, 1), "kod_ozeti": code_hash(),
                    "llm_cagri": USAGE.calls - u0[0], "girdi_belirteci": USAGE.input_tokens - u0[1],
                    "cikti_belirteci": USAGE.output_tokens - u0[2],
                    "zaman": time.strftime("%Y-%m-%dT%H:%M:%S")})
    out_dir.mkdir(parents=True, exist_ok=True)
    (out_dir / "sonuclar.json").write_text(json.dumps({"ozet": summary, "kayitlar": recs}, ensure_ascii=False,
                                                      indent=1, default=str), encoding="utf-8")
    return {"ozet": summary, "kayitlar": recs}


def _rate(rs):
    return round(100 * sum(r["dogru"] for r in rs) / len(rs), 1) if rs else None


def summarize(recs: list[dict]) -> dict:
    def by(field):
        ks = sorted({str(r.get(field)) for r in recs if r.get(field) is not None})
        return {k: {"n": sum(1 for r in recs if str(r.get(field)) == k),
                    "basari": _rate([r for r in recs if str(r.get(field)) == k])} for k in ks}

    ans = [r for r in recs if r["tip"] != "cevapsiz"]
    una = [r for r in recs if r["tip"] == "cevapsiz"]
    return {
        "soru": len(recs), "dogru": sum(r["dogru"] for r in recs), "basari": _rate(recs),
        "cevaplanabilir_basari": _rate(ans), "cevapsiz_dogru_red": _rate(una),
        "yanlis_red": round(100 * sum(r["durum"] != "cevap" for r in ans) / len(ans), 1) if ans else None,
        "sessiz_yanlis": round(100 * sum((not r["dogru"]) and r["durum"] == "cevap" for r in recs) / len(recs), 1)
        if recs else None,
        "seviyeye_gore": by("seviye"), "dile_gore": by("dil"), "aileye_gore": by("aile"), "tipe_gore": by("tip"),
        "cizime_gore": by("cizim"), "belge_turune_gore": by("belge_turu"), "zorluga_gore": by("zorluk"),
    }