| """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 |
|
|
| |
| 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" |
| 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 |
|
|
|
|
| |
| _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")] |
| |
| 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") |
| |
| 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: |
| 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")) |
| |
| 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: |
| 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 |
|
|
|
|
| |
| 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()] |
|
|
|
|
| |
| 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") |
| |
| 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" |
| 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" |
|
|
