Text-to-SQL / scripts /hata_enjeksiyonu.py

Upload 466 files

a72d1cf verified 8 days ago

7.82 kB

"""Hata enjeksiyonu deneyi: hata kataloğunun tespit, tip ve öneri doğruluğu.
Altın SQL sorgularına, LLM tabanlı Text-to-SQL sistemlerinde sık görülen hata tipleri
kontrollü olarak enjekte edilir. Her bozuk sorgu ajanın denetim zincirinden
(statik denetim → yürütme → sonuç denetimi) geçirilir ve şunlar ölçülür:
  sessiz   : denetim olmasaydı sorgu hatasız çalışıp YANLIŞ sonuç döndürür müydü?
  tespit   : zincir en az bir hata/uyarı üretti mi?
  tip      : ilk olayın tipi beklenen katalog tipiyle aynı mı?
  oneri    : geri besleme metni doğru onarımı (doğru sütun adı / değer / ISO tarih /
             sayı / toplam_satiri koşulu) içeriyor mu?
Çalıştırma:  python scripts/hata_enjeksiyonu.py
"""
from __future__ import annotations

import json
import re
import sys
from collections import defaultdict
from pathlib import Path

ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(ROOT))

from t2s.errors import check_result, classify_exception, static_checks  # noqa: E402
from t2s.evaluate import load_benchmark, results_match  # noqa: E402
from t2s.schema import connect  # noqa: E402
from t2s.trnum import AYLAR, fold  # noqa: E402

AY_ADI = {v: k for k, v in AYLAR.items() if len(k) > 3}
AY_TR = {1: "Ocak", 2: "Şubat", 3: "Mart", 4: "Nisan", 5: "Mayıs", 6: "Haziran", 7: "Temmuz",
         8: "Ağustos", 9: "Eylül", 10: "Ekim", 11: "Kasım", 12: "Aralık"}


def tr_thousands(n: float) -> str:
    s = f"{n:,.0f}" if float(n).is_integer() else f"{n:,.2f}"
    return s.replace(",", "§").replace(".", ",").replace("§", ".")


def perturbations(sql: str, cols: set[str]):
    """(tip, bozuk_sql, onarım_belirteci) üçlüleri üretir."""
    out = []
    # 1) Şema eşleme: sütun adındaki alt çizgiyi kaldır ya da yıl önekini taşı
    used = [c for c in sorted(cols, key=len, reverse=True) if re.search(rf"\b{re.escape(c)}\b", sql) and "_" in c
            and c != "toplam_satiri"]
    if used:
        c = used[0]
        out.append(("SEMA_ESLEME", re.sub(rf"\b{re.escape(c)}\b", c.replace("_", ""), sql, count=1), c))
    # 2) Değer eşleme: metin literalini Türkçe dışı büyük harfe çevir (İzmir -> IZMIR)
    m = re.search(r"=\s*'([^'%]+)'", sql)
    if m and not re.match(r"^\d{4}-\d{2}", m.group(1)):
        v = m.group(1)
        bad = fold(v).upper()
        if bad != v:
            out.append(("DEGER_ESLEME", sql[:m.start(1)] + bad + sql[m.end(1):], v))
    # 3) Tarih biçimi: ISO -> Türkçe yazım
    m = re.search(r"'(\d{4})-(\d{2})-(\d{2})'", sql)
    if m:
        y, mo, d = m.groups()
        out.append(("TARIH_BICIMI", sql[:m.start()] + f"'{d}.{mo}.{y}'" + sql[m.end():], f"'{y}-{mo}-{d}'"))
    else:
        m = re.search(r"'(\d{4})-(\d{2})'", sql)
        if m:
            y, mo = m.groups()
            out.append(("TARIH_BICIMI", sql[:m.start()] + f"'{AY_TR[int(mo)]} {y}'" + sql[m.end():], f"'{y}-{mo}'"))
    # 4) Sayı biçimi: WHERE eşiğini Türkçe biçimli metin olarak yaz (> 2000 -> > '2.000,00')
    m = re.search(r"([<>]=?)\s*(\d{3,})(?![\d.])", sql)
    if m:
        n = int(m.group(2))
        lit = tr_thousands(n) + ",00"
        out.append(("SAYI_BICIMI", sql[:m.start(2)] + f"'{lit}'" + sql[m.end(2):], str(n)))
        if n >= 1000:  # tırnaksız binlik ayraçlı yazım (SQLite bunu ondalık sayı olarak okur)
            out.append(("SAYI_BICIMI*", sql[:m.start(2)] + tr_thousands(n) + sql[m.end(2):], str(n)))
    # 5) Sayısal akıl yürütme: toplam satırı koşulunu kaldır
    if re.search(r"toplam_satiri\s*=\s*0", sql):
        bad = re.sub(r"\s+AND\s+toplam_satiri\s*=\s*0|toplam_satiri\s*=\s*0\s+AND\s+", " ", sql)
        bad = re.sub(r"\s+WHERE\s+toplam_satiri\s*=\s*0", "", bad)
        if bad != sql:
            out.append(("SAYISAL_AKIL", bad, "toplam_satiri"))
    # 6) Sözdizimi: başka bir SQL lehçesinden alışkanlık (LIMIT 1 -> TOP 1)
    m = re.search(r"^SELECT\s+(.*)\s+LIMIT 1$", sql, re.S)
    if m:
        out.append(("SOZDIZIMI", "SELECT TOP 1 " + m.group(1), "LIMIT"))
    return out


def run_chain(sql: str, con):
    events = static_checks(sql, con)
    rows, exc = None, None
    if not any(e.tip == "SOZDIZIMI" for e in events):
        try:
            rows = [tuple(r) for r in con.execute(sql).fetchall()]
            events += check_result(sql, rows, con)
        except Exception as e:  # noqa: BLE001
            exc = e
            events.append(classify_exception(e, sql, con))
    return events, rows, exc


def main():
    sysdb = connect(ROOT / "outputs" / "db" / "sistem_tr.db", readonly=True)
    ref = connect(ROOT / "outputs" / "db" / "referans.db", readonly=True)
    cols = {c for (c,) in sysdb.execute("SELECT DISTINCT sutun FROM _sutunlar")}
    items = [it for it in load_benchmark() if it["altin_sql"]]
    stats = defaultdict(lambda: defaultdict(int))
    examples = {}
    for it in items:
        gold = [tuple(r) for r in ref.execute(it["altin_sql"]).fetchall()]
        gold_tips = {e.tip for e in run_chain(it["altin_sql"], sysdb)[0]}  # bozulmamış sorguda da çıkan uyarılar
        for tip, bad, fix in perturbations(it["altin_sql"], cols):
            ev, rows, exc = run_chain(bad, sysdb)
            s = stats[tip]
            s["n"] += 1
            silent = exc is None and rows is not None and not results_match(rows, gold) and \
                not any(e.tip == "SOZDIZIMI" for e in ev)
            s["sessiz"] += silent
            s["sessiz_tespit"] += silent and bool(ev)  # yalnızca sessiz (zararlı) hatalar üzerinde tespit
            s["tespit"] += bool(ev)
            exp = tip.rstrip("*")
            s["tip"] += bool(ev) and ev[0].tip == exp
            s["tespit_dogru_tip"] += any(e.tip == exp for e in ev)  # beklenen tipte en az bir olay
            s["tespit_ozgun"] += any(e.tip not in gold_tips for e in ev)  # altın sorguda da çıkan uyarılar hariç
            fb = " ".join(e.feedback() for e in ev)
            s["oneri"] += bool(ev) and fix.lower() in fb.lower()
            examples.setdefault(tip, {"soru": it["soru"], "bozuk_sql": bad,
                                      "geri_besleme": ev[0].feedback() if ev else "(tespit edilmedi)"})
    # yanlış alarm: bozulmamış altın sorgularda denetim zinciri olay üretiyor mu?
    fa = []
    for it in items:
        ev, rows, exc = run_chain(it["altin_sql"], sysdb)
        if ev:
            fa.append({"id": it["id"], "tipler": [e.tip for e in ev]})
    out = {k: dict(v) for k, v in stats.items()}
    tot = defaultdict(int)
    for k, v in out.items():
        for kk, vv in v.items():
            tot[kk] += vv
    out["TOPLAM"] = dict(tot)
    print(f"{'tip':14} {'n':>4} {'sessiz':>7} {'tespit':>7} {'tip':>7} {'öneri':>7}")
    for k, v in out.items():
        n = v["n"]
        print(f"{k:14} {n:4d} {100*v['sessiz']/n:6.1f}% {100*v['tespit']/n:6.1f}% {100*v['tip']/n:6.1f}% "
              f"{100*v['oneri']/n:6.1f}%")
    ci = {k: {m: wilson(v[m], v["n"]) for m in ("sessiz", "tespit", "tespit_ozgun", "tespit_dogru_tip", "tip", "oneri")}
          for k, v in out.items()}
    print(f"yanlış alarm: {len(fa)}/{len(items)} altın sorgu")
    (ROOT / "outputs" / "hata_enjeksiyonu.json").write_text(
        json.dumps({"ozet": out, "wilson_ga95": ci, "altin_sorgu": len(items),
                    "yanlis_alarm": {"n": len(fa), "toplam": len(items), "ornekler": fa},
                    "ornekler": examples}, ensure_ascii=False, indent=1), encoding="utf-8")


def wilson(k: int, n: int, z: float = 1.96):
    if not n:
        return None
    p = k / n
    d = 1 + z * z / n
    c = (p + z * z / (2 * n)) / d
    h = z * ((p * (1 - p) / n + z * z / (4 * n * n)) ** 0.5) / d
    return [round(100 * (c - h), 1), round(100 * (c + h), 1)]


if __name__ == "__main__":
    main()

	"""Hata enjeksiyonu deneyi: hata kataloğunun tespit, tip ve öneri doğruluğu.

	Altın SQL sorgularına, LLM tabanlı Text-to-SQL sistemlerinde sık görülen hata tipleri
	kontrollü olarak enjekte edilir. Her bozuk sorgu ajanın denetim zincirinden
	(statik denetim → yürütme → sonuç denetimi) geçirilir ve şunlar ölçülür:
	sessiz : denetim olmasaydı sorgu hatasız çalışıp YANLIŞ sonuç döndürür müydü?
	tespit : zincir en az bir hata/uyarı üretti mi?
	tip : ilk olayın tipi beklenen katalog tipiyle aynı mı?
	oneri : geri besleme metni doğru onarımı (doğru sütun adı / değer / ISO tarih /
	sayı / toplam_satiri koşulu) içeriyor mu?

	Çalıştırma: python scripts/hata_enjeksiyonu.py
	"""
	from __future__ import annotations

	import json
	import re
	import sys
	from collections import defaultdict
	from pathlib import Path

	ROOT = Path(__file__).resolve().parent.parent
	sys.path.insert(0, str(ROOT))

	from t2s.errors import check_result, classify_exception, static_checks # noqa: E402
	from t2s.evaluate import load_benchmark, results_match # noqa: E402
	from t2s.schema import connect # noqa: E402
	from t2s.trnum import AYLAR, fold # noqa: E402

	AY_ADI = {v: k for k, v in AYLAR.items() if len(k) > 3}
	AY_TR = {1: "Ocak", 2: "Şubat", 3: "Mart", 4: "Nisan", 5: "Mayıs", 6: "Haziran", 7: "Temmuz",
	8: "Ağustos", 9: "Eylül", 10: "Ekim", 11: "Kasım", 12: "Aralık"}


	def tr_thousands(n: float) -> str:
	s = f"{n:,.0f}" if float(n).is_integer() else f"{n:,.2f}"
	return s.replace(",", "§").replace(".", ",").replace("§", ".")


	def perturbations(sql: str, cols: set[str]):
	"""(tip, bozuk_sql, onarım_belirteci) üçlüleri üretir."""
	out = []
	# 1) Şema eşleme: sütun adındaki alt çizgiyi kaldır ya da yıl önekini taşı
	used = [c for c in sorted(cols, key=len, reverse=True) if re.search(rf"\b{re.escape(c)}\b", sql) and "_" in c
	and c != "toplam_satiri"]
	if used:
	c = used[0]
	out.append(("SEMA_ESLEME", re.sub(rf"\b{re.escape(c)}\b", c.replace("_", ""), sql, count=1), c))
	# 2) Değer eşleme: metin literalini Türkçe dışı büyük harfe çevir (İzmir -> IZMIR)
	m = re.search(r"=\s*'([^'%]+)'", sql)
	if m and not re.match(r"^\d{4}-\d{2}", m.group(1)):
	v = m.group(1)
	bad = fold(v).upper()
	if bad != v:
	out.append(("DEGER_ESLEME", sql[:m.start(1)] + bad + sql[m.end(1):], v))
	# 3) Tarih biçimi: ISO -> Türkçe yazım
	m = re.search(r"'(\d{4})-(\d{2})-(\d{2})'", sql)
	if m:
	y, mo, d = m.groups()
	out.append(("TARIH_BICIMI", sql[:m.start()] + f"'{d}.{mo}.{y}'" + sql[m.end():], f"'{y}-{mo}-{d}'"))
	else:
	m = re.search(r"'(\d{4})-(\d{2})'", sql)
	if m:
	y, mo = m.groups()
	out.append(("TARIH_BICIMI", sql[:m.start()] + f"'{AY_TR[int(mo)]} {y}'" + sql[m.end():], f"'{y}-{mo}'"))
	# 4) Sayı biçimi: WHERE eşiğini Türkçe biçimli metin olarak yaz (> 2000 -> > '2.000,00')
	m = re.search(r"([<>]=?)\s*(\d{3,})(?![\d.])", sql)
	if m:
	n = int(m.group(2))
	lit = tr_thousands(n) + ",00"
	out.append(("SAYI_BICIMI", sql[:m.start(2)] + f"'{lit}'" + sql[m.end(2):], str(n)))
	if n >= 1000: # tırnaksız binlik ayraçlı yazım (SQLite bunu ondalık sayı olarak okur)
	out.append(("SAYI_BICIMI*", sql[:m.start(2)] + tr_thousands(n) + sql[m.end(2):], str(n)))
	# 5) Sayısal akıl yürütme: toplam satırı koşulunu kaldır
	if re.search(r"toplam_satiri\s=\s0", sql):
	bad = re.sub(r"\s+AND\s+toplam_satiri\s=\s0\|toplam_satiri\s=\s0\s+AND\s+", " ", sql)
	bad = re.sub(r"\s+WHERE\s+toplam_satiri\s=\s0", "", bad)
	if bad != sql:
	out.append(("SAYISAL_AKIL", bad, "toplam_satiri"))
	# 6) Sözdizimi: başka bir SQL lehçesinden alışkanlık (LIMIT 1 -> TOP 1)
	m = re.search(r"^SELECT\s+(.*)\s+LIMIT 1$", sql, re.S)
	if m:
	out.append(("SOZDIZIMI", "SELECT TOP 1 " + m.group(1), "LIMIT"))
	return out


	def run_chain(sql: str, con):
	events = static_checks(sql, con)
	rows, exc = None, None
	if not any(e.tip == "SOZDIZIMI" for e in events):
	try:
	rows = [tuple(r) for r in con.execute(sql).fetchall()]
	events += check_result(sql, rows, con)
	except Exception as e: # noqa: BLE001
	exc = e
	events.append(classify_exception(e, sql, con))
	return events, rows, exc


	def main():
	sysdb = connect(ROOT / "outputs" / "db" / "sistem_tr.db", readonly=True)
	ref = connect(ROOT / "outputs" / "db" / "referans.db", readonly=True)
	cols = {c for (c,) in sysdb.execute("SELECT DISTINCT sutun FROM _sutunlar")}
	items = [it for it in load_benchmark() if it["altin_sql"]]
	stats = defaultdict(lambda: defaultdict(int))
	examples = {}
	for it in items:
	gold = [tuple(r) for r in ref.execute(it["altin_sql"]).fetchall()]
	gold_tips = {e.tip for e in run_chain(it["altin_sql"], sysdb)[0]} # bozulmamış sorguda da çıkan uyarılar
	for tip, bad, fix in perturbations(it["altin_sql"], cols):
	ev, rows, exc = run_chain(bad, sysdb)
	s = stats[tip]
	s["n"] += 1
	silent = exc is None and rows is not None and not results_match(rows, gold) and \
	not any(e.tip == "SOZDIZIMI" for e in ev)
	s["sessiz"] += silent
	s["sessiz_tespit"] += silent and bool(ev) # yalnızca sessiz (zararlı) hatalar üzerinde tespit
	s["tespit"] += bool(ev)
	exp = tip.rstrip("*")
	s["tip"] += bool(ev) and ev[0].tip == exp
	s["tespit_dogru_tip"] += any(e.tip == exp for e in ev) # beklenen tipte en az bir olay
	s["tespit_ozgun"] += any(e.tip not in gold_tips for e in ev) # altın sorguda da çıkan uyarılar hariç
	fb = " ".join(e.feedback() for e in ev)
	s["oneri"] += bool(ev) and fix.lower() in fb.lower()
	examples.setdefault(tip, {"soru": it["soru"], "bozuk_sql": bad,
	"geri_besleme": ev[0].feedback() if ev else "(tespit edilmedi)"})
	# yanlış alarm: bozulmamış altın sorgularda denetim zinciri olay üretiyor mu?
	fa = []
	for it in items:
	ev, rows, exc = run_chain(it["altin_sql"], sysdb)
	if ev:
	fa.append({"id": it["id"], "tipler": [e.tip for e in ev]})
	out = {k: dict(v) for k, v in stats.items()}
	tot = defaultdict(int)
	for k, v in out.items():
	for kk, vv in v.items():
	tot[kk] += vv
	out["TOPLAM"] = dict(tot)
	print(f"{'tip':14} {'n':>4} {'sessiz':>7} {'tespit':>7} {'tip':>7} {'öneri':>7}")
	for k, v in out.items():
	n = v["n"]
	print(f"{k:14} {n:4d} {100v['sessiz']/n:6.1f}% {100v['tespit']/n:6.1f}% {100*v['tip']/n:6.1f}% "
	f"{100*v['oneri']/n:6.1f}%")
	ci = {k: {m: wilson(v[m], v["n"]) for m in ("sessiz", "tespit", "tespit_ozgun", "tespit_dogru_tip", "tip", "oneri")}
	for k, v in out.items()}
	print(f"yanlış alarm: {len(fa)}/{len(items)} altın sorgu")
	(ROOT / "outputs" / "hata_enjeksiyonu.json").write_text(
	json.dumps({"ozet": out, "wilson_ga95": ci, "altin_sorgu": len(items),
	"yanlis_alarm": {"n": len(fa), "toplam": len(items), "ornekler": fa},
	"ornekler": examples}, ensure_ascii=False, indent=1), encoding="utf-8")


	def wilson(k: int, n: int, z: float = 1.96):
	if not n:
	return None
	p = k / n
	d = 1 + z * z / n
	c = (p + z * z / (2 * n)) / d
	h = z * ((p * (1 - p) / n + z * z / (4 * n * n)) ** 0.5) / d
	return [round(100 * (c - h), 1), round(100 * (c + h), 1)]


	if __name__ == "__main__":
	main()