Blame for t2s/errors.py

"""Hata kataloğu: tip tanımları, çalışma zamanı sınıflandırıcısı ve değerlendirme hatası analizi."""
from __future__ import annotations

import json
import re
import sqlite3
from dataclasses import asdict, dataclass, field
from datetime import datetime
from pathlib import Path

from .trnum import AYLAR, fold

# --------------------------------------------------------------------------- katalog
CATALOG = {
    "SEMA_ESLEME": {
        "ad": "Şema eşleme hatası",
        "tanim": "Soru, yanlış veya var olmayan tablo/sütuna bağlandı (ör. 'gerçekleşme' yerine 'bütçe' sütunu).",
        "ornek": "no such column: gerceklesme2024",
        "cozum": "Benzer sütun adlarını öner; orijinal başlıkları ve birimleri şemada göster.",
    },
    "DEGER_ESLEME": {
        "ad": "Değer eşleme hatası",
        "tanim": "Metin değeri tablodakiyle birebir eşleşmedi: İ/ı-I/i büyük-küçük harf, ek almış adlar, kısaltmalar.",
        "ornek": "WHERE il = 'Istanbul'  (tabloda 'İstanbul')",
        "cozum": "TR_NORM() ile katlanmış karşılaştırma; tablodaki en yakın değerleri geri besle.",
    },
    "BIRIM": {
        "ad": "Birim hatası",
        "tanim": "Ölçek/birim dönüşümü unutuldu veya yanlış yapıldı (bin TL ↔ TL, bin MWh ↔ MWh, m³ ↔ litre, % ↔ oran).",
        "ornek": "Soru TL isterken 'bin TL' sütunu doğrudan döndürüldü (1000 kat fark).",
        "cozum": "Sütun birimlerini meta veride sakla, şemada göster; sonuçta birim etiketi kullan.",
    },
    "TARIH_BICIMI": {
        "ad": "Tarih biçimi hatası",
        "tanim": "Türkçe tarih yazımı ('15.03.2025', 'Ocak 2024') ile veritabanındaki ISO biçim ('2025-03-15', '2024-01') uyuşmadı.",
        "ornek": "WHERE yururluk_tarihi = '15.03.2025'",
        "cozum": "Tarihleri aktarımda ISO'ya çevir; SQL'deki Türkçe tarih literallerini yakala ve dönüştürmeyi öner.",
    },
    "SAYI_BICIMI": {
        "ad": "Türkçe sayı biçimi hatası",
        "tanim": "'1.234,56' biçimi yanlış yorumlandı: sayı metin kaldı, nokta ondalık sanıldı (1000 kat hata) "
                 "veya virgül binlik sanıldı (10/100 kat hata); SQL'e Türkçe sayı literali yazıldı.",
        "ornek": "SUM('1.234,56') = 1.234  |  '4,5' -> 45",
        "cozum": "Aktarımda TR yerel ayarıyla ayrıştır; SQL literallerini denetle.",
    },
    "SAYISAL_AKIL": {
        "ad": "Sayısal akıl yürütme hatası",
        "tanim": "Toplama/ortalama/yüzde formülü yanlış; belgedeki 'Toplam' satırı tekrar sayıldı; "
                 "işaret (negatif gider) veya tamsayı bölmesi hatası.",
        "ornek": "SUM(...) toplam satırını da içerdi (2 kat sonuç)",
        "cozum": "toplam_satiri bayrağı; formül kanıtları; sonuç makullük kontrolleri.",
    },
    "SOZDIZIMI": {
        "ad": "SQL sözdizimi hatası",
        "tanim": "Üretilen sorgu ayrıştırılamadı veya SQLite lehçesine uymuyor.",
        "ornek": "near \"TOP\": syntax error",
        "cozum": "Hata mesajını geri besle; yalnızca SQLite sözdizimi iste.",
    },
    "BOS_SONUC": {
        "ad": "Boş sonuç",
        "tanim": "Sorgu çalıştı ama satır dönmedi; çoğunlukla değer/tarih eşleme sorunudur.",
        "ornek": "WHERE donem = 'Mart 2025' -> 0 satır",
        "cozum": "WHERE literallerini tablodaki gerçek değerlerle karşılaştır, en yakınlarını öner.",
    },
    "CEVAPSIZLIK": {
        "ad": "Cevaplanabilirlik hatası",
        "tanim": "Cevabı tablolarda olmayan soruya uydurma cevap verildi ya da cevaplanabilir soru reddedildi.",
        "ornek": "'Mersin'in tüketimi' sorusuna başka bir ilin değeri döndü.",
        "cozum": "Kapsam dışı yıl/varlık denetimi; emin değilse açıkça 'cevap bulunamadı' de.",
    },
    "YURUTME": {
        "ad": "Diğer yürütme hatası",
        "tanim": "Bilinmeyen fonksiyon, tip uyuşmazlığı, zaman aşımı vb.",
        "ornek": "no such function: YEAR",
        "cozum": "SQLite fonksiyonlarını kullan (strftime, substr).",
    },
    "YANLIS_SONUC": {
        "ad": "Yanlış sonuç (sınıflandırılamadı)",
        "tanim": "Sorgu çalıştı fakat altın sonuçla uyuşmadı ve belirgin bir alt tipe atanamadı.",
        "ornek": "-",
        "cozum": "Elle inceleme.",
    },
}


@dataclass
class ErrorEvent:
    tip: str
    mesaj: str
    sql: str = ""
    oneriler: list[str] = field(default_factory=list)
    asama: str = "yurutme"           # statik | yurutme | sonuc | degerlendirme
    soru: str = ""
    zaman: str = field(default_factory=lambda: datetime.now().isoformat(timespec="seconds"))

    def feedback(self) -> str:
        info = CATALOG.get(self.tip, {})
        s = f"[{self.tip} – {info.get('ad', '')}] {self.mesaj}"
        if self.oneriler:
            s += "\nÖneriler: " + " | ".join(self.oneriler)
        return s


# --------------------------------------------------------------------------- yardımcılar
_STR_LIT = re.compile(r"'((?:[^']|'')*)'")
_TR_NUM_LIT = re.compile(r"^\s*-?\d{1,3}(\.\d{3})+(,\d+)?\s*$|^\s*-?\d+,\d+\s*$")
_BARE_TR_NUM = re.compile(r"(?<![\w.'])\d{1,3}(?:\.\d{3}){2,}(?:,\d+)?(?![\w.])")
_DMY = re.compile(r"^\d{1,2}[./]\d{1,2}[./]\d{4}$")


def string_literals(sql: str) -> list[str]:
    return [m.group(1).replace("''", "'") for m in _STR_LIT.finditer(sql or "")]


def _all_columns(con) -> list[tuple[str, str]]:
    try:
        return con.execute("SELECT tablo, sutun FROM _sutunlar").fetchall()
    except sqlite3.Error:
        return []


def _close(name: str, options: list[str], n: int = 3) -> list[str]:
    try:
        from rapidfuzz import process
        return [m[0] for m in process.extract(name, options, limit=n, score_cutoff=45)]
    except Exception:
        import difflib
        return difflib.get_close_matches(name, options, n=n, cutoff=0.4)


def static_checks(sql: str, con) -> list[ErrorEvent]:
    """Yürütme öncesi denetimler: sözdizimi, salt-okuma, TR sayı/tarih literalleri, toplam satırı."""
    events = []
    s = (sql or "").strip().rstrip(";")
    if not s:
        return [ErrorEvent("SOZDIZIMI", "Boş sorgu üretildi.", sql, asama="statik")]
    # yazma komutları: anahtar sözcük denetimi; REPLACE(...) metin işlevi serbesttir (REPLACE INTO değil)
    if not re.match(r"^\s*(select|with)\b", s, re.I) or re.search(
            r"\b(insert|update|delete|drop|alter|create|attach|detach|pragma|vacuum|reindex)\b|\breplace\b(?!\s*\()",
            s, re.I):
        return [ErrorEvent("SOZDIZIMI", "Yalnızca SELECT/WITH sorgularına izin verilir.", sql, asama="statik")]
    if s.count(";") > 0:
        return [ErrorEvent("SOZDIZIMI", "Tek bir SQL ifadesi üretin (';' ile birden çok ifade yok).", sql,
                           asama="statik")]
    tree = None
    try:
        import sqlglot
        from sqlglot import exp
        tree = sqlglot.parse_one(s, read="sqlite")
        # ifade tipi denetimi: kök sorgu SELECT/birleşim olmalı, içinde hiçbir yazma düğümü bulunmamalı
        write_nodes = tuple(getattr(exp, n) for n in ("Insert", "Update", "Delete", "Drop", "Create", "Alter",
                                                     "Merge", "Command") if hasattr(exp, n))
        if not isinstance(tree, (exp.Select, exp.Union, exp.Intersect, exp.Except, exp.Subquery)) or \
                any(isinstance(n, write_nodes) for n in tree.walk()):
            return [ErrorEvent("SOZDIZIMI", "Yalnızca SELECT/WITH sorgularına izin verilir.", sql, asama="statik")]
    except Exception as e:  # sqlglot yoksa veya hata
        if e.__class__.__name__ in ("ParseError", "TokenError"):
            events.append(ErrorEvent("SOZDIZIMI", f"SQL ayrıştırılamadı: {str(e).splitlines()[0][:200]}",
                                     sql, asama="statik"))
            return events
    for lit in string_literals(s):
        if _TR_NUM_LIT.match(lit):
            v = lit.replace(".", "").replace(",", ".")
            events.append(ErrorEvent("SAYI_BICIMI", f"SQL'de Türkçe biçimli sayı metni var: '{lit}'.",
                                     sql, [f"Sayıları tırnaksız ve nokta ondalıkla yazın: {v}"], "statik"))
        elif _DMY.match(lit.strip()):
            d, m, y = re.split(r"[./]", lit.strip())
            events.append(ErrorEvent("TARIH_BICIMI", f"SQL'de Türkçe tarih yazımı var: '{lit}'.", sql,
                                     [f"Tarihler ISO biçimde saklanır: '{y}-{int(m):02d}-{int(d):02d}'"], "statik"))
        else:
            parts = fold(lit).split()
            if len(parts) == 2 and parts[0] in AYLAR and parts[1].isdigit():
                events.append(ErrorEvent("TARIH_BICIMI", f"SQL'de ay adı ile tarih var: '{lit}'.", sql,
                                         [f"Dönemler 'YYYY-AA' biçimindedir: '{parts[1]}-{AYLAR[parts[0]]:02d}'"],
                                         "statik"))
    for m in _BARE_TR_NUM.finditer(s):
        events.append(ErrorEvent("SAYI_BICIMI", f"SQL'de binlik ayraçlı sayı var: {m.group(0)}", sql,
                                 [f"{m.group(0).replace('.', '').replace(',', '.')} yazın"], "statik"))
    # toplam satırı tuzağı
    try:
        flagged = {t for (t,) in con.execute("SELECT tablo FROM _sutunlar WHERE sutun='toplam_satiri'")}
    except sqlite3.Error:
        flagged = set()
    low = s.lower()
    uses_agg = re.search(r"\b(sum|avg|count|max|min)\s*\(|\border\s+by\b|\bgroup\s+by\b", low)
    for t in flagged:
        if re.search(rf'["`]?\b{re.escape(t)}\b["`]?', low) and uses_agg and "toplam_satiri" not in low \
                and _total_row_reachable(tree, t, con):
            events.append(ErrorEvent(
                "SAYISAL_AKIL",
                f"'{t}' tablosunda belgenin kendi Toplam satırı var (toplam_satiri=1); toplama/sıralama "
                "yaparken bu satır sonucu bozabilir.", sql,
                ["WHERE koşuluna 'toplam_satiri = 0' ekleyin (yalnızca toplam değeri isteniyorsa "
                 "'toplam_satiri = 1' satırını seçin)."], "statik"))
    return events


def _total_row_reachable(tree, table: str, con) -> bool:
    """Sorgunun WHERE koşulu belgenin Toplam satırını zaten dışlıyor mu? Dışlıyorsa uyarı gereksizdir.

    Tabloyu okuyan her SELECT için 'SELECT COUNT(*) FROM tablo WHERE (koşul) AND toplam_satiri = 1' çalıştırılır.
    Koşul çıkarılamazsa ya da sorgu çalışmazsa (takma ad, birleştirme vb.) temkinli davranılır: uyarı korunur."""
    if tree is None:
        return True
    try:
        from sqlglot import exp
        selects = [sel for sel in tree.find_all(exp.Select)
                   if any(tb.name.lower() == table.lower() for tb in sel.find_all(exp.Table))]
        if not selects:
            return True
        for sel in selects:
            if len({tb.name.lower() for tb in sel.find_all(exp.Table)}) != 1 or any(
                    tb.alias for tb in sel.find_all(exp.Table)):
                return True
            where = sel.args.get("where")
            cond = f"({where.this.sql(dialect='sqlite')}) AND " if where is not None else ""
            n = con.execute(f'SELECT COUNT(*) FROM "{table}" WHERE {cond}toplam_satiri = 1').fetchone()[0]
            if n:
                return True
        return False
    except Exception:  # noqa: BLE001
        return True


def classify_exception(exc: Exception, sql: str, con) -> ErrorEvent:
    msg = str(exc)
    low = msg.lower()
    cols = _all_columns(con)
    if "no such column" in low:
        bad = msg.split(":", 1)[-1].strip().split(".")[-1]
        sugg = _close(bad, sorted({c for _, c in cols}))
        return ErrorEvent("SEMA_ESLEME", msg, sql, [f"Benzer sütunlar: {', '.join(sugg)}"] if sugg else [])
    if "no such table" in low:
        bad = msg.split(":", 1)[-1].strip()
        sugg = _close(bad, sorted({t for t, _ in cols}))
        return ErrorEvent("SEMA_ESLEME", msg, sql, [f"Benzer tablolar: {', '.join(sugg)}"] if sugg else [])
    if "ambiguous column" in low:
        return ErrorEvent("SEMA_ESLEME", msg, sql, ["Sütunu tablo adıyla niteleyin (tablo.sutun)."])
    if "syntax error" in low or "incomplete input" in low or "unrecognized token" in low:
        return ErrorEvent("SOZDIZIMI", msg, sql, ["Yalnızca SQLite sözdizimi kullanın."])
    if "no such function" in low:
        return ErrorEvent("YURUTME", msg, sql,
                          ["SQLite fonksiyonları: strftime, substr, ROUND, ABS, TR_NORM, TR_LOWER"])
    return ErrorEvent("YURUTME", msg, sql)


def check_result(sql: str, rows: list, con) -> list[ErrorEvent]:
    """Sonuç denetimi: boş sonuçta WHERE literallerini tablodaki değerlerle karşılaştırır."""
    if rows and not all(all(v is None for v in r) for r in rows):
        return []
    events = []
    lits = [l for l in string_literals(sql) if not l.endswith("%")]
    suggestions = []
    for lit in lits:
        hits = []
        for t, c in _all_columns(con):
            try:
                vals = [v for (v,) in con.execute(f'SELECT DISTINCT "{c}" FROM "{t}" WHERE typeof("{c}")="text"')]
            except sqlite3.Error:
                continue
            exact_fold = [v for v in vals if fold(v) == fold(lit) and v != lit]
            if exact_fold:
                hits.append(f"{t}.{c} = '{exact_fold[0]}' (büyük/küçük harf veya Türkçe karakter farkı)")
                continue
            close = [v for v in _close(lit, vals, 2)]
            if close and len(lit) > 2:
                hits.append(f"{t}.{c} içinde yakın değerler: " + ", ".join(repr(v) for v in close))
        if hits:
            suggestions.extend(hits[:2])
    if rows and all(all(v is None for v in r) for r in rows):
        return [ErrorEvent("BOS_SONUC", "Sorgu yalnızca NULL döndürdü (eşleşen satır yok veya sütun metin "
                           "olarak saklanmış olabilir).", sql, suggestions or ["WHERE koşullarını gözden geçirin."],
                           "sonuc")]
    tip = "DEGER_ESLEME" if any("harf" in s for s in suggestions) else "BOS_SONUC"
    events.append(ErrorEvent(tip, "Sorgu 0 satır döndürdü.", sql,
                             suggestions or ["WHERE koşullarını gevşetin veya değerleri kontrol edin; "
                                             "metin karşılaştırmalarında TR_NORM(sutun)=TR_NORM('değer') kullanın."],
                             "sonuc"))
    return events


# --------------------------------------------------------------------------- günlük
class ErrorLog:
    """Hataları JSONL dosyasına yazar (arayüzdeki Hata Kataloğu sayfası okur)."""

    def __init__(self, path: str | Path):
        self.path = Path(path)
        self.path.parent.mkdir(parents=True, exist_ok=True)

    def add(self, ev: ErrorEvent, kaynak: str = "arayuz", **extra):
        rec = asdict(ev) | {"kaynak": kaynak} | extra
        with self.path.open("a", encoding="utf-8") as f:
            f.write(json.dumps(rec, ensure_ascii=False) + "\n")

    def read(self) -> list[dict]:
        if not self.path.exists():
            return []
        return [json.loads(l) for l in self.path.read_text(encoding="utf-8").splitlines() if l.strip()]


# --------------------------------------------------------------------------- değerlendirme analizi
def _num(x):
    return x if isinstance(x, (int, float)) and not isinstance(x, bool) else None


def diagnose_failure(item: dict, pred_sql: str | None, pred_rows, gold_rows, abstained: bool,
                     exec_error: str | None, db_mode: str) -> str:
    """Değerlendirmede başarısız bir örneğin hata tipini kestirir."""
    cats = set(item.get("kategoriler", []))
    if not item["cevaplanabilir"]:
        return "CEVAPSIZLIK"
    if abstained and not pred_sql:
        return "CEVAPSIZLIK"
    if exec_error:
        e = exec_error.lower()
        if "no such column" in e or "no such table" in e or "ambiguous" in e:
            return "SEMA_ESLEME"
        if "syntax" in e or "incomplete" in e:
            return "SOZDIZIMI"
        return "YURUTME"
    lits = string_literals(pred_sql or "")
    if any(_DMY.match(l) for l in lits) or any(
            len(fold(l).split()) == 2 and fold(l).split()[0] in AYLAR for l in lits):
        return "TARIH_BICIMI"
    if any(_TR_NUM_LIT.match(l) for l in lits):
        return "SAYI_BICIMI"
    if not pred_rows:
        return "DEGER_ESLEME" if "deger_esleme" in cats else ("TARIH_BICIMI" if "tarih" in cats else "BOS_SONUC")
    # sayısal oranlar
    p = [_num(v) for r in pred_rows for v in r]
    g = [_num(v) for r in gold_rows for v in r]
    p = [v for v in p if v is not None]
    g = [v for v in g if v is not None]
    if db_mode != "tr" and pred_rows and any(isinstance(v, str) and re.search(r"\d[.,]\d", v)
                                             for r in pred_rows for v in r):
        return "SAYI_BICIMI"
    if len(p) == 1 and len(g) == 1 and g[0] and p[0]:
        ratio = abs(p[0] / g[0])
        for k in (1e3, 1e-3, 1e6, 1e-6):
            if abs(ratio - k) / k < 0.01:
                return "SAYI_BICIMI" if db_mode != "tr" else "BIRIM"
        if any(abs(ratio - k) / k < 0.02 for k in (10, 100, 0.1, 0.01)):
            return "SAYI_BICIMI" if db_mode != "tr" else "SAYISAL_AKIL"
        if abs(ratio - 2) < 0.02:
            return "SAYISAL_AKIL"   # toplam satırı çift sayım
        return "SAYISAL_AKIL"
    if db_mode != "tr" and "tr_sayi" in cats:
        return "SAYI_BICIMI"
    if "toplam_satiri" in cats and pred_sql and "toplam_satiri" not in pred_sql.lower():
        return "SAYISAL_AKIL"
    if "birim" in cats:
        return "BIRIM"
    if "deger_esleme" in cats:
        return "DEGER_ESLEME"
    if "tarih" in cats:
        return "TARIH_BICIMI"
    if p and g:
        return "SAYISAL_AKIL"
    return "SEMA_ESLEME"