"""Kural tabanlı Türkçe Text-to-SQL taban çizgisi (LLM gerektirmez). Amaç: API anahtarı olmadan da sistemin uçtan uca çalışması ve LLM yaklaşımları için karşılaştırma tabanı oluşturmak. Adımlar: 1) Soru normalizasyonu (Türkçe küçük harf, ek ayırma, TR sayı/tarih yakalama) 2) Değer eşleme : soru içindeki varlıkları tablo hücre değerlerine bağlama 3) Şema eşleme : tablo ve ölçü sütunu seçimi (kök eşleme + yıl eşleme + eş anlamlılar) 4) Niyet : seçim / toplam / ortalama / sayım / en-yüksek / listeleme / gruplama / fark / yüzde değişim / pay 5) Kapsam denetimi: tabloda olmayan yıl/dönem/varlık -> açıkça "cevapsız" Her soru için sıralı aday SQL listesi döner; ajan, hata durumunda sonraki adaya geçer. """ from __future__ import annotations import json import re from dataclasses import dataclass, field from .trnum import AYLAR, fold Q_STOP = {"ne", "kadar", "kac", "nedir", "hangi", "hangisi", "hangisidir", "hangileri", "hangileridir", "olan", "olarak", "icin", "ile", "ve", "bir", "bu", "da", "de", "mi", "mu", "midir", "var", "yil", "yili", "yilinda", "yilindaki", "yilinin", "tum", "olan", "nin", "nun", "in", "un", "dir", "tir", "dur", "tur", "misin", "listeler", "listele", "gore", "ki", "en", "toplam", "ortalama", "yuzde", "arttı", "artti", "azaldi", "degisti", "kac", "tane", "adet"} VAL_STOP = {"ve", "ile", "-", "(-)", "de", "da"} SYN = { # soru kökü -> şema kökü (alan sözlüğü) "harca": "gerceklesme", "harcan": "gerceklesme", "odene": "butce", "tuketen": "tuketim", "tuket": "tuketim", "pahal": "ucret", "ucuz": "ucret", "fiyat": "ucret", "kazanc": "kar", "enflasyon": "degisim", "calisan": "personel", "istihdam": "personel", } CHANGE_PCT = re.compile(r"yuzde\s*kac\s*(art|azal|degis|dus)|(artis|degisim|azalis)\s*(orani|yuzdesi)") CHANGE_ABS = re.compile(r"(ne\s*kadar|kac\s*\w+)\s*(artt|azald|degist|dust)|\bfark(?!li)") SHARE = re.compile(r"\bpay[ıi]?\b.*yuzde|yuzde.*\bpay") ARGMAX = re.compile(r"\ben\s*(yuksek|fazla|cok|buyuk|pahali)") ARGMIN = re.compile(r"\ben\s*(dusuk|az|kucuk|ucuz)") WHICH = re.compile(r"\bhangi|\bne zaman") LIST = re.compile(r"hangileri|listele|nelerdir|\bhangi\w*\s+\w+(ler|lar)\b") COUNT = re.compile(r"\bkac\s+(tane|adet|farkli)|\bkac\s+\w+\s+var\b|\bkac\s+\w+\s+\w+\s+var\b|" r"\bsayisi\s+(nedir|kactir)|\bkac\s+(ay|il|yil|hizmet|gider)\b") GT = re.compile(r"(fazla|uzerinde|ustunde|asan|astig|buyuk|yuksek)") LT = re.compile(r"(az\b|altinda|altina|dusuk|kucuk)") def _stem_match(a: str, b: str) -> bool: if not a or not b: return False if len(a) <= 3 or len(b) <= 3: return a == b or (len(a) <= 3 and b.startswith(a) and len(b) - len(a) <= 4 and len(a) >= 3) \ or (len(b) <= 3 and a.startswith(b) and len(a) - len(b) <= 4 and len(b) >= 3) cp = 0 for x, y in zip(a, b): if x != y: break cp += 1 return cp >= min(len(a), len(b), 5) or cp >= 4 def tokenize(text: str) -> list[str]: s = fold(text) s = re.sub(r"['’]\w*", " ", s) # İzmir'in -> izmir return [t for t in re.split(r"[^a-z0-9²³]+", s) if t] @dataclass class Col: name: str orig: str sqltype: str sem: str unit: str | None mult: float toks: list[str] = field(default_factory=list) years: list[str] = field(default_factory=list) @dataclass class TableInfo: name: str caption: str cols: list[Col] values: dict[str, list[str]] has_total: bool cap_toks: list[str] years_in_data: set[str] def load_meta(con) -> dict[str, TableInfo]: tabs = {} have = {r[1] for r in con.execute("PRAGMA table_info(_tablolar)")} ctx_sql = "baglam_yillari" if "baglam_yillari" in have else "NULL" for t, cap, ctx_years in con.execute(f"SELECT tablo, baslik, {ctx_sql} FROM _tablolar"): cols = [] for nm, orig, typ, sem, unit, mult in con.execute( "SELECT sutun, orijinal_baslik, sql_tipi, anlam, birim, carpan FROM _sutunlar " "WHERE tablo=? ORDER BY sira", (t,)): years = re.findall(r"(?:19|20)\d{2}", nm) toks = [x for x in re.split(r"_", nm) if x and not x.isdigit() and x != "yil"] toks += [x for x in tokenize(orig or "") if not x.isdigit() and x not in toks] cols.append(Col(nm, orig or nm, typ, sem, unit, mult or 1.0, toks, years)) values, years = {}, set() for c in cols: if c.sqltype == "TEXT": vals = [v for (v,) in con.execute(f'SELECT DISTINCT "{c.name}" FROM "{t}" WHERE "{c.name}" IS NOT NULL')] if c.sem == "metin": values[c.name] = vals for v in vals: years.update(re.findall(r"(?:19|20)\d{2}", str(v))) elif c.sem == "yil": years.update(str(v) for (v,) in con.execute(f'SELECT DISTINCT "{c.name}" FROM "{t}"')) years.update(c.years) cap_toks = [x for x in tokenize(re.sub(r"^\s*tablo\s*\d+\s*:", "", cap or "", flags=re.I)) if x not in Q_STOP and not x.isdigit()] years.update(re.findall(r"(?:19|20)\d{2}", cap or "")) years.update(y for y in (ctx_years or "").split(",") if y) # tablonun dönemi belge metninden tabs[t] = TableInfo(t, cap or "", cols, values, any(c.name == "toplam_satiri" for c in cols), cap_toks, years) return tabs # --------------------------------------------------------------------------- yakalayıcılar _NUM_TOKEN = re.compile(r"(? list[tuple[float, bool, str]]: """Sorudaki sayıları (değer, yüzde mi, ham) döndürür; yılları ve tarih parçalarını atlar.""" out = [] qq = re.sub(r"\d{1,2}[./]\d{1,2}[./]\d{4}", " ", q) for m in _NUM_TOKEN.finditer(qq): raw = m.group(2) if re.fullmatch(r"(19|20)\d{2}", raw): continue # tarihteki gün: '1 Temmuz 2025' after = qq[m.end():m.end() + 12] if re.match(r"\s*[A-Za-zÇĞİÖŞÜçğıöşü]+\s+(19|20)\d{2}", after) and fold(after.split()[0]) in AYLAR: continue val = float(raw.replace(".", "").replace(",", ".")) mult = {"bin": 1e3, "milyon": 1e6, "milyar": 1e9}.get((m.group(4) or "").lower(), 1.0) pct = bool(m.group(1) or m.group(3)) or bool(re.search(r"yüzde\s*$|yuzde\s*$", fold(qq[:m.start()]))) if re.match(r"\s*(il|ilin|kisi|ay)\b", fold(qq[m.end():m.end() + 6])) and val < 100: continue # '17 ilin' gibi niceleyiciler out.append((val * mult, pct, raw)) return out def find_dates(q: str) -> dict: """{'full': ['2025-03-15'], 'month': ['2025-03'], 'years': ['2024']}""" f = fold(q) res = {"full": [], "month": [], "years": []} for d, m, y in re.findall(r"\b(\d{1,2})[./](\d{1,2})[./](\d{4})\b", f): res["full"].append(f"{y}-{int(m):02d}-{int(d):02d}") for d, mon, y in re.findall(r"\b(\d{1,2})\s+([a-z]+)\s+(\d{4})", f): if mon in AYLAR: res["full"].append(f"{y}-{AYLAR[mon]:02d}-{int(d):02d}") for mon, y in re.findall(r"\b([a-z]+)\s+(\d{4})", f): if mon in AYLAR: ym = f"{y}-{AYLAR[mon]:02d}" if not any(x.startswith(ym) for x in res["full"]): res["month"].append(ym) res["years"] = re.findall(r"\b((?:19|20)\d{2})\b", re.sub(r"\d{1,2}[./]\d{1,2}[./]\d{4}", " ", f)) return res def q_like(s: str) -> str: return "'" + str(s).replace("'", "''") + "'" # --------------------------------------------------------------------------- ana sınıf @dataclass class RuleResult: candidates: list[str] answerable: bool reason: str trace: dict class RuleBasedTextToSQL: name = "kural-tabanli" def __init__(self, con): self.con = con self.meta = load_meta(con) self._build_df() def _build_df(self): """Sütun kökü belge sıklığı: birçok sütunda geçen kökler ('donem') tablo seçiminde daha az ağırlık alır.""" from collections import Counter cnt = Counter() for t in self.meta.values(): for c in t.cols: for tok in set(c.toks): cnt[tok] += 1 self._df = dict(cnt) def refresh(self): self.meta = load_meta(self.con) self._build_df() # ------------------------------------------------------------ eşleme def _value_matches(self, t: TableInfo, qt: list[str]): found = [] for col, vals in t.values.items(): best = [] for v in vals: vt = [x for x in tokenize(v) if x not in VAL_STOP and not x.isdigit()] if not vt: continue pos = [min((i for i, y in enumerate(qt) if _stem_match(x, y)), default=None) for x in vt] pos = [p_ for p_ in pos if p_ is not None] hit = len(pos) sc = hit / len(vt) need = 1.0 if len(vt) <= 1 else 0.6 if sc >= need: span = (max(pos) - min(pos)) if pos else 0 generic = sum(1 for x in vt if any(_stem_match(x, y) for y in qt) and any(_stem_match(x, tok) for c in t.cols for tok in c.toks)) best.append((sc, hit, -span, -generic, len(v), v)) if best: best.sort(reverse=True) top = [b for b in best if b[:4] == best[0][:4]] weight = 1.0 if max(len(b[5]) for b in top) > 3 else 0.35 found.append((col, [b[5] for b in top], weight * (best[0][0] + 0.1 * best[0][1]))) return found def _mention_pair(self, ranked, qt): """Soruda geçen ilk iki ölçü sütununu (aynı yıla ait) geçiş sırasıyla döndürür.""" def first_pos(c): ps = [i for i, y in enumerate(qt) for tok in c.toks if _stem_match(tok, y)] return min(ps) if ps else None def ratio(c): return sum(1 for tok in set(c.toks) if any(_stem_match(tok, y) for y in qt)) / max(1, len(set(c.toks))) cands = [c for s_, _, c in ranked if s_ > 0] full = [c for c in cands if ratio(c) >= 0.99] pos = [(first_pos(c), c) for c in (full if len(full) >= 2 else cands)] pos = [(p_, c) for p_, c in pos if p_ is not None] pos.sort(key=lambda x: x[0]) for i in range(len(pos)): for j in range(i + 1, len(pos)): a, b = pos[i][1], pos[j][1] if a.years == b.years and a.toks != b.toks: return a, b return None def _col_score(self, c: Col, qt: list[str], years: list[str], idf: bool = False) -> float: sc = 0.0 for tok in c.toks: for y in qt: if _stem_match(tok, y) or SYN.get(y[:5]) and _stem_match(tok, SYN[y[:5]]) or \ any(y.startswith(k) and _stem_match(tok, v) for k, v in SYN.items()): sc += (1.0 / self._df.get(tok, 1)) if idf else 1.0 break if c.years: hit = sum(1 for y in c.years if y in years) sc += 1.5 * hit if years and hit == 0: sc -= 1.5 return sc def _table_score(self, t: TableInfo, qt, years, vmatches) -> float: sc = sum(m[2] * 3 for m in vmatches) cs = sorted((max(0.0, self._col_score(c, qt, [], idf=True)) for c in t.cols), reverse=True) if cs: sc += sum(cs[:2]) + 0.3 * sum(cs[2:]) sc += 0.5 * sum(1.0 for x in t.cap_toks if any(_stem_match(x, y) for y in qt)) if years and any(y in t.years_in_data for y in years): sc += 1.0 # sorulan varlık türü: '... il hangisidir', 'hangi mahalle' asked = re.findall(r"(\w+)\s+hangi\w*|hangi\s+(\w+)|(\w+)\s+gore\b", fold(" ".join(qt_raw)) if (qt_raw := getattr(self, "_qraw", None)) else "") for a, b, g in asked: w = a or b or g if any(_stem_match(tok, w) for c in t.cols if c.sqltype == "TEXT" or c.sem == "yil" for tok in c.toks + [c.name]): sc += 2.0 return sc # ------------------------------------------------------------ üretim def generate(self, question: str) -> RuleResult: q = question f = fold(q) qt_all = tokenize(q) self._qraw = qt_all qt = [x for x in qt_all if x not in Q_STOP and len(x) >= 2 and not x.isdigit()] dates = find_dates(q) years = dates["years"] nums = find_numbers(q) trace = {"tokenler": qt, "tarihler": dates, "sayilar": nums} scored = [] for t in self.meta.values(): vm = self._value_matches(t, qt) scored.append((self._table_score(t, qt, years, vm), t.name, vm)) scored.sort(reverse=True) trace["tablo_puanlari"] = [(round(s, 2), n) for s, n, _ in scored] if not scored or scored[0][0] < 1.0: return RuleResult([], False, "Soru yüklenen tabloların hiçbiriyle eşleşmedi.", trace) cands, reasons = [], [] top_score = scored[0][0] for rank, (sc, tname, vm) in enumerate(scored[:2]): if sc < 1.0 or (rank > 0 and sc < 0.6 * top_score): continue res = self._build_for_table(self.meta[tname], q, f, qt, dates, nums, vm, trace) if res[0] is None: reasons.append(res[1]) continue cands.append(res[0]) if not cands: return RuleResult([], False, reasons[0] if reasons else "Uygun sorgu kurulamadı.", trace) return RuleResult(cands, True, "", trace) def _build_for_table(self, t: TableInfo, q, f, qt, dates, nums, vm, trace): years = dates["years"] num_cols = [c for c in t.cols if c.sqltype in ("INTEGER", "REAL") and c.sem in ("sayi", "yuzde")] year_col = next((c for c in t.cols if c.sem == "yil"), None) date_col = next((c for c in t.cols if c.sem == "tarih"), None) text_cols = [c for c in t.cols if c.sqltype == "TEXT" and c.sem == "metin"] T = f'"{t.name}"' where = [] # ---------- kapsam denetimi (cevapsız sorular) mentioned_years = set(years) for d in dates["full"] + dates["month"]: mentioned_years.add(d[:4]) missing = [y for y in mentioned_years if y not in t.years_in_data] range_q = re.search(r"(\d{4})\s*-\s*(\d{4})\s*(donem|arasi|yillari)", f) # yalnızca tabloda (ya da belge bağlamında) yıl bilgisi varsa red; hiç yıl bilgisi yoksa kanıt yokluğudur if missing and t.years_in_data and not range_q: return None, (f"'{t.caption}' tablosunda {', '.join(sorted(missing))} yılına ait veri yok; " "bu soru mevcut tablolardan cevaplanamaz.") if date_col is not None and dates["month"] and date_col.sem == "tarih": vals = {str(v) for (v,) in self.con.execute(f'SELECT DISTINCT "{date_col.name}" FROM {T}')} month_vals = {v[:7] for v in vals} for m in dates["month"]: if m not in month_vals and not re.search(r"(once|sonra|itibaren)", f): return None, f"Tabloda {m} dönemine ait kayıt bulunmuyor; veri bu dönemi kapsamıyor." # ---------- değer filtreleri used_val_cols = set() for col, vals, sc in vm: used_val_cols.add(col) if len(vals) == 1: where.append(f'"{col}" = {q_like(vals[0])}') else: where.append(f'"{col}" IN ({", ".join(q_like(v) for v in vals)})') # ---------- niyet pct_change = bool(CHANGE_PCT.search(f)) if pct_change and not re.search(r"yuzde\s*kac", f) and any("degisim" in c.toks for c in num_cols): pct_change = False # tabloda hazır 'değişim oranı' sütunu var abs_change = bool(CHANGE_ABS.search(f)) and not pct_change share = bool(SHARE.search(f)) is_avg = "ortalama" in f is_sum = bool(re.search(r"\btoplam\b", f)) is_count = bool(COUNT.search(f)) and not is_avg argmax, argmin = bool(ARGMAX.search(f)), bool(ARGMIN.search(f)) which = bool(WHICH.search(f)) is_list = bool(LIST.search(f)) first = bool(re.search(r"\bilk\b", f)) group_col = None mg = re.search(r"(\w+)\s+(\w+\s+)?gore\b", f) if mg and ("listele" in f or is_avg or is_sum): words = [w for w in (mg.group(1), (mg.group(2) or "").strip()) if w] gbest = 0 for c in text_cols: n_ = sum(1 for tok in c.toks if any(_stem_match(tok, w) for w in words)) if n_ > gbest: group_col, gbest = c, n_ if group_col is None and mg.group(2): w = mg.group(2).strip() group_col = next((c for c in text_cols if any(_stem_match(tok, w) for tok in c.toks)), None) # ---------- ölçü sütunu def best_measure(exclude=()): ranked = sorted(((self._col_score(c, qt, years), i, c) for i, c in enumerate(num_cols) if c.name not in exclude), key=lambda x: (-x[0], x[1])) return ranked ranked = best_measure() measure = ranked[0][2] if ranked and ranked[0][0] > 0 else (num_cols[0] if len(num_cols) == 1 else None) if measure is None and ranked: # yalnızca yıl ile seçilen sütunlar (yil_2024 gibi) yr = [c for c in num_cols if c.years and all(y in years for y in c.years)] measure = yr[0] if yr else ranked[0][2] # yüzde soruları yüzde sütununu tercih etsin if measure is not None and ("yuzde" in f or "%" in q or "oran" in f) and not (pct_change or share): pcols = [c for c in num_cols if c.unit == "%" or c.sem == "yuzde"] if pcols and measure not in pcols: pr = sorted(pcols, key=lambda c: -self._col_score(c, qt, years)) if self._col_score(pr[0], qt, years) > 0 or len(pcols) == 1: measure = pr[0] if not nums or not any(n[1] for n in nums) or True else measure trace["olcu"] = measure.name if measure else None # ---------- yıl / tarih satır filtreleri if year_col is not None and years and not (pct_change or abs_change): if range_q: where.append(f'"{year_col.name}" BETWEEN {range_q.group(1)} AND {range_q.group(2)}') elif len(years) == 1 and not (measure and measure.years): where.append(f'"{year_col.name}" = {years[0]}') if date_col is not None: cmp = "<" if re.search(r"\bonce", f) else (">" if re.search(r"\bsonra", f) else (">=" if "itibaren" in f else "=")) if dates["full"]: where.append(f'"{date_col.name}" {cmp} {q_like(dates["full"][0])}') elif dates["month"] and not (pct_change or abs_change): m = dates["month"][0] is_full = any(len(str(v)) == 10 for (v,) in self.con.execute( f'SELECT "{date_col.name}" FROM {T} LIMIT 3')) if is_full: where.append(f'"{date_col.name}" {cmp if cmp != "=" else ">="} {q_like(m + "-01")}' if cmp != "=" else f'substr("{date_col.name}",1,7) = {q_like(m)}') else: where.append(f'"{date_col.name}" {cmp} {q_like(m)}') elif years and not (pct_change or abs_change): where.append(f'"{date_col.name}" LIKE {q_like(years[0] + "-%")}') # ---------- sayısal eşik thr = [n for n in nums] if thr and measure is not None and not (pct_change or abs_change or share): val, is_pct, raw = thr[0] tail = f[f.find(fold(raw)) + len(fold(raw)):] if fold(raw) in f else f op = ">" if GT.search(tail[:40]) else ("<" if LT.search(tail[:40]) else None) if op: target = measure if is_pct: pc = [c for c in num_cols if c.unit == "%" or c.sem == "yuzde"] if pc: target = max(pc, key=lambda c: self._col_score(c, qt, years)) v = val / (target.mult if target.unit and target.unit.startswith("bin") else 1.0) where.append(f'"{target.name}" {op} {v:g}') measure = target # sütunlar arası karşılaştırma: "gerçekleşmesi bütçesini aştı" if re.search(r"(asti|astig|gecti)", f) and not thr and measure is not None and len(ranked) > 1: pair = self._mention_pair(ranked, qt) if pair: where.append(f'"{pair[0].name}" > "{pair[1].name}"') # iki yıl arasında artan/azalan satırlar: 'tüketimi 2023'ten 2024'e azalan' if not (pct_change or abs_change) and len(set(years)) == 2 and re.search(r"(azal|dus|art)", f): y0, y1 = sorted(set(years)) ca = next((c for c in num_cols if c.years == [y0]), None) cb = next((c for c in num_cols if c.years == [y1] and ca is not None and [x for x in c.toks] == [x for x in ca.toks]), None) if ca is not None and cb is not None: op = "<" if re.search(r"(azal|dus)", f) else ">" where.append(f'"{cb.name}" {op} "{ca.name}"') measure = cb total_filter = [] if t.has_total: total_filter = ["toplam_satiri = 0"] if measure is not None and measure.unit == "%" and re.search(r"\b(genel|toplam)\b", f) \ and not any(c in used_val_cols for c in t.values): total_filter = ["toplam_satiri = 1"] def W(extra=()): conds = list(where) + list(extra) return (" WHERE " + " AND ".join(conds)) if conds else "" def label_col(): in_where = {c.name for c in t.cols if any(f'"{c.name}"' in w for w in where)} qtoks = tokenize(q) best_c, best_r = None, 0.0 for c in [year_col, date_col] + text_cols: if c is None or c.name in in_where or c.name in used_val_cols or not c.toks: continue r_ = sum(1 for tok in c.toks if any(_stem_match(tok, y) for y in qtoks)) / len(c.toks) if r_ > best_r: best_c, best_r = c, r_ if best_c is not None: return best_c if re.search(r"\b(ay|ayi|donem|donemi|ne zaman)\b", f) and date_col is not None: return date_col if re.search(r"\byil\b", f) and year_col is not None: return year_col for c in text_cols: if c.name not in used_val_cols: return c return text_cols[0] if text_cols else (year_col or date_col) # birim dönüşümü def conv(expr: str, col: Col | None) -> str: if col is None or not col.unit: return expr u = col.unit.lower() if u.startswith("bin ") and re.search(r"\bkac\s+(tl|mwh)\b|\b(tl|mwh)\s*(olarak|cinsinden)", f) \ and not re.search(r"\bbin\s+(tl|mwh)", f): return f"{expr}*1000" if u in ("m³",) and re.search(r"\blitre", f): return f"{expr}*1000" return expr if measure is None and not is_count: return None, "Soruda tabloya ait bir ölçü (sütun) bulunamadı." # ---------- iki ölçü arasındaki fark ('bütçe ile gerçekleşme arasındaki fark') if abs_change and re.search(r"\bfark", f): pair = self._mention_pair(ranked, qt) if pair: return f'SELECT "{pair[0].name}"-"{pair[1].name}" FROM {T}{W(() if vm else total_filter)}', "" # ---------- yüzde değişim / fark if pct_change or abs_change: a = b = None if measure is not None and measure.years: base_toks = [x for x in measure.toks if not x.isdigit()] pool = [c for c in num_cols if c.years and [x for x in c.toks if not x.isdigit()] == base_toks] ys = sorted(set(years)) if len(ys) >= 2: a = next((c for c in pool if ys[0] in c.years), None) b = next((c for c in pool if ys[-1] in c.years), None) if (a is None or b is None): yc = sorted([c for c in num_cols if c.years and c.sem == "sayi"], key=lambda c: c.years[-1]) if len(yc) == 2: a, b = yc elif len(ys) == 1 and len(pool) >= 2: a = next((c for c in pool if ys[0] in c.years), None) b = max(pool, key=lambda c: c.years[-1]) if a is not None and b is not None and a is not b: neg = any(re.search(r"gider|\(-\)", fold(v)) for _, vs, _ in vm for v in vs) A_, B_ = (f'ABS("{a.name}")', f'ABS("{b.name}")') if neg and abs_change else (f'"{a.name}"', f'"{b.name}"') expr = f"({B_}-{A_})*100.0/{A_}" if pct_change else f"{B_}-{A_}" return f"SELECT {expr} FROM {T}{W(total_filter if not vm else ())}", "" # satır bazlı (yıl sütunu veya tarih sütunu) key = year_col or date_col pts = sorted(set(years)) if year_col else (dates["month"] or dates["full"]) if key is not None and len(pts) >= 2 and measure is not None: def sub(p): lit = p if year_col else q_like(p) return f'(SELECT "{measure.name}" FROM {T} WHERE "{key.name}" = {lit})' s0, s1 = sub(pts[0]), sub(pts[-1]) expr = f"({s1}-{s0})*100.0/{s0}" if pct_change else f"{s1}-{s0}" return f"SELECT {expr}", "" return None, "Değişim hesaplamak için iki dönem/sütun belirlenemedi." # ---------- pay if share and measure is not None: denom = next((c for c in num_cols if c.name in ("toplam",) or "toplam" in c.toks), None) if denom is not None and denom is not measure: return f'SELECT "{measure.name}"*100.0/"{denom.name}" FROM {T}{W()}', "" # ---------- gruplama if group_col is not None and measure is not None: agg = "AVG" if is_avg else "SUM" return (f'SELECT "{group_col.name}", {agg}("{measure.name}") FROM {T}{W(total_filter)} ' f'GROUP BY "{group_col.name}"'), "" # ---------- sayım measure_named = measure is not None and self._col_score(measure, qt, []) > 0 strong_count = bool(re.search(r"\bkac\s+(tane|adet|farkli)|\bvar\b", f)) mk = re.search(r"\bkac\s+(\w+)", f) if mk and any(_stem_match(tok, mk.group(1)) for c in text_cols for tok in c.toks): strong_count = is_count = True # 'kaç gider türünün', 'kaç il if is_count and (strong_count or not measure_named): cc = None if "farkli" in f: cc = next((c for c in text_cols if any(_stem_match(tok, y) for tok in c.toks for y in qt) and c.name not in used_val_cols), None) sel = f'COUNT(DISTINCT "{cc.name}")' if cc else "COUNT(*)" return f"SELECT {sel} FROM {T}{W(total_filter)}", "" # ---------- en yüksek / en düşük if (argmax or argmin) and measure is not None: order = "DESC" if argmax else "ASC" if which or not re.search(r"(nedir|ne kadar|kac)\b", f): lc = label_col() nn = total_filter + [f'"{measure.name}" IS NOT NULL'] # tanımsız ('-') değerler sıralamaya girmesin return f'SELECT "{lc.name}" FROM {T}{W(nn)} ORDER BY "{measure.name}" {order} LIMIT 1', "" return f'SELECT {"MAX" if argmax else "MIN"}({conv(chr(34) + measure.name + chr(34), measure)}) FROM {T}{W(total_filter)}', "" if first and (which or "donem" in f): lc = label_col() return f'SELECT MIN("{lc.name}") FROM {T}{W(total_filter)}', "" if is_avg and measure is not None: return f'SELECT AVG({conv(chr(34) + measure.name + chr(34), measure)}) FROM {T}{W(total_filter)}', "" if is_list or (which and len(where) > len(vm)): lc = label_col() return f'SELECT "{lc.name}" FROM {T}{W(total_filter)}', "" if is_sum and measure is not None and "toplam" not in measure.toks and not total_filter == ["toplam_satiri = 1"]: if not vm or len(vm) and any(len(v) > 1 for _, v, _ in vm) or any( c not in [x.name for x in text_cols[:1]] for c in used_val_cols): return f'SELECT SUM({conv(chr(34) + measure.name + chr(34), measure)}) FROM {T}{W(total_filter)}', "" # ---------- tekil seçim row_filtered = bool(vm) or any("toplam_satiri = 1" in x for x in total_filter) or \ any(year_col is not None and f'"{year_col.name}"' in w for w in where) or \ any(date_col is not None and f'"{date_col.name}"' in w for w in where) if not row_filtered and text_cols: return None, ("Soruda geçen varlık (ör. il, kalem, hizmet adı) tablodaki değerlerle eşleşmedi; " "bu bilgi belgelerde yer almıyor olabilir.") extra = total_filter if total_filter == ["toplam_satiri = 1"] else [] return f'SELECT {conv(chr(34) + measure.name + chr(34), measure)} FROM {T}{W(extra)}', "" def explain(res: RuleResult) -> str: return json.dumps(res.trace, ensure_ascii=False, default=str)