| """Çözümlenmiş tabloları tip çıkarımı + adlandırma ile SQLite'a aktarır. |
| |
| Her veritabanı iki meta tablo içerir: |
| _tablolar : tablo adı, belge başlığı, kaynak, sayfa, birim, çıkarıcı, ... |
| _sutunlar : sütun adı, orijinal başlık, tip, birim, çarpan, örnek değerler, ... |
| Bu meta bilgiler Text-to-SQL modülüne şema bağlamı olarak verilir. |
| """ |
| from __future__ import annotations |
|
|
| import json |
| import re |
| import sqlite3 |
| from dataclasses import dataclass, field |
| from pathlib import Path |
|
|
| from .extract import ResolvedTable |
| from .trnum import (detect_unit, fold, parse_date, parse_number, slugify, strip_units, |
| tr_lower, _RE_DATE_RANGE, _RE_YEAR) |
|
|
| STOP = {"ve", "ile", "gore", "ait", "icin", "olarak", "ornek", "yili", "tablo", "olan"} |
| TOTAL_RE = re.compile(r"^(genel\s+)?toplam\b|^turkiye( toplami)?$|^toplam\s*\(|^[a-z]+(\s+[a-z]+)?\s+toplami$") |
|
|
|
|
| def table_name_from_caption(caption: str, fallback: str = "tablo") -> str: |
| cap = re.sub(r"^\s*(tablo|çizelge|cizelge|table)\s*[\d.]*\s*[:.\-–]?\s*", "", caption or "", flags=re.I) |
| cap = strip_units(cap) |
| toks = [t for t in slugify(cap, 200).split("_") if t and t not in STOP and not t.isdigit()] if cap.strip() else [] |
| name = "_".join(toks[:4]) |
| if not name: |
| name = slugify(fallback) |
| if name[0].isdigit(): |
| name = "t_" + name |
| return name |
|
|
|
|
| def column_name(parts: list[str], idx: int) -> str: |
| text = " ".join(parts) |
| years = [] |
| m = _RE_DATE_RANGE.search(text) |
| if m: |
| years.append(m.group(4)) |
| text = text[:m.start()] + " " + text[m.end():] |
| for y in re.findall(r"\(\s*((?:19|20)\d{2})\s*\)", text): |
| if y not in years: |
| years.append(y) |
| text = strip_units(text) |
| for y in _RE_YEAR.findall(text): |
| if y not in years: |
| years.append(y) |
| text = _RE_YEAR.sub(" ", text) |
| base = slugify(text, 40) if re.search(r"[A-Za-zÇĞİÖŞÜçğıöşü]", text) else "" |
| if base == "sutun": |
| base = "" |
| if not base and years: |
| base = "yil" |
| if not base: |
| base = f"sutun_{idx + 1}" |
| name = "_".join([base] + years) |
| return ("c_" + name) if name[0].isdigit() else name |
|
|
|
|
| @dataclass |
| class ColumnInfo: |
| name: str |
| original: str |
| sqltype: str |
| semantic: str |
| unit: str | None = None |
| multiplier: float = 1.0 |
| unparsed: int = 0 |
| samples: list = field(default_factory=list) |
| description: str = "" |
|
|
|
|
| @dataclass |
| class BuiltTable: |
| name: str |
| caption: str |
| columns: list[ColumnInfo] |
| rows: list[list] |
| source: str |
| page: int | None |
| extractor: str |
| unit: str | None |
| n_header_rows: int |
| merged_cells: int |
| mode: str |
| summary: str = "" |
| context_years: str = "" |
|
|
|
|
| def _infer_column(values: list[str], mode: str): |
| nonnull = [v for v in values if fold(v) not in ("", "-", "—", "...")] |
| if not nonnull or mode == "text": |
| return "TEXT", "metin", list(values), 0 |
| |
| dates = [parse_date(v) for v in nonnull] |
| nums = [parse_number(v, mode) for v in nonnull] |
| n_date = sum(d is not None for d in dates) |
| n_num = sum(p.ok and p.value is not None for p in nums) |
| if n_date / len(nonnull) >= 0.8 and n_num / len(nonnull) < 0.8: |
| return "TEXT", "tarih", [parse_date(v) for v in values], 0 |
| if n_num / len(nonnull) >= 0.8: |
| parsed = [parse_number(v, mode) for v in values] |
| out = [p.value if p.ok else None for p in parsed] |
| unparsed = sum(1 for v, p in zip(values, parsed) if not p.ok and fold(v) not in ("", "-")) |
| is_pct = any(p.is_percent for p in parsed) |
| all_int = all(x is None or float(x).is_integer() for x in out) |
| if mode == "tr": |
| is_int = all_int and not any("," in v for v in nonnull) |
| else: |
| is_int = all_int |
| if is_int: |
| out = [int(x) if x is not None else None for x in out] |
| sem = "yuzde" if is_pct else "sayi" |
| if is_int and all(x is None or 1900 <= x <= 2100 for x in out): |
| sem = "yil" |
| return ("INTEGER" if is_int else "REAL"), sem, out, unparsed |
| return "TEXT", "metin", list(values), 0 |
|
|
|
|
| def build_table(rt: ResolvedTable, mode: str = "tr", name: str | None = None) -> BuiltTable: |
| ncol = len(rt.header_parts) |
| cap_unit, cap_mult = detect_unit(rt.caption) |
| cols, data_cols = [], [] |
| used = set() |
| for j in range(ncol): |
| parts = rt.header_parts[j] |
| original = " ".join(parts) |
| vals = [r[j] if j < len(r) else "" for r in rt.body] |
| sqltype, sem, conv, unparsed = _infer_column(vals, mode) |
| nm = column_name(parts, j) if parts else ("kalem" if sqltype == "TEXT" else f"sutun_{j + 1}") |
| base, k = nm, 2 |
| while nm in used: |
| nm = f"{base}_{k}" |
| k += 1 |
| used.add(nm) |
| if sem == "sayi" and re.search(r"\b(dipnot\w*|referanslar\w*|kod|sira no)\b", fold(original)): |
| sem = "kod" |
| unit, mult = detect_unit(original) |
| if sem == "kod": |
| unit, mult = None, 1.0 |
| if sem == "yuzde" or (unit is None and "%" in original): |
| unit, mult = "%", 1.0 |
| if sem == "sayi": |
| sem = "yuzde" |
| if unit is None and sem == "sayi" and cap_unit and nm != "yil" and sem != "kod": |
| unit, mult = cap_unit, cap_mult |
| if sem == "yil": |
| unit, mult = None, 1.0 |
| samples = [v for v in conv if v not in (None, "")][:3] |
| cols.append(ColumnInfo(nm, original, sqltype, sem, unit, mult, unparsed, samples)) |
| data_cols.append(conv) |
|
|
| rows = [list(r) for r in zip(*data_cols)] if data_cols else [] |
| |
| text_idx = [i for i, c in enumerate(cols) if c.sqltype == "TEXT" and c.semantic == "metin"][:2] |
| flags = [] |
| for r, raw in zip(rows, rt.body): |
| is_tot = any(TOTAL_RE.search(fold(raw[i])) for i in text_idx if i < len(raw)) |
| if not text_idx and raw: |
| is_tot = bool(TOTAL_RE.search(fold(raw[0]))) |
| flags.append(1 if is_tot else 0) |
| if any(flags): |
| cols.append(ColumnInfo("toplam_satiri", "(otomatik) Toplam satırı mı?", "INTEGER", "bayrak", |
| samples=[0, 1])) |
| rows = [r + [f] for r, f in zip(rows, flags)] |
|
|
| fallback = (Path(rt.source).stem or "tablo") + (f"_s{rt.page}" if rt.page else "") |
| tname = name or table_name_from_caption(rt.caption, fallback) |
| own_years = re.findall(r"(?:19|20)\d{2}", (rt.caption or "") + " " + " ".join(c.name for c in cols)) |
| if not own_years and not any(c.semantic in ("yil", "tarih") for c in cols): |
| ctx = sorted(set(re.findall(r"\b((?:19|20)\d{2})\b", rt.context or ""))) |
| else: |
| ctx = [] |
| return BuiltTable(tname, rt.caption, cols, rows, rt.source, rt.page, rt.extractor, cap_unit, |
| rt.n_header_rows, rt.merged_cells, mode, context_years=",".join(ctx)) |
|
|
|
|
| |
| def register_functions(con: sqlite3.Connection): |
| """Türkçe duyarlı yardımcı fonksiyonlar: TR_LOWER, TR_NORM.""" |
| con.create_function("TR_LOWER", 1, lambda s: tr_lower(s) if s is not None else None, |
| deterministic=True) |
| con.create_function("TR_NORM", 1, lambda s: fold(s) if s is not None else None, |
| deterministic=True) |
|
|
|
|
| def connect(db_path: str | Path, readonly: bool = False) -> sqlite3.Connection: |
| if readonly: |
| con = sqlite3.connect(f"file:{Path(db_path).as_posix()}?mode=ro", uri=True, |
| check_same_thread=False) |
| else: |
| con = sqlite3.connect(str(db_path), check_same_thread=False) |
| register_functions(con) |
| return con |
|
|
|
|
| META_DDL = """ |
| CREATE TABLE IF NOT EXISTS _tablolar( |
| tablo TEXT PRIMARY KEY, baslik TEXT, kaynak TEXT, sayfa INTEGER, cikarici TEXT, |
| birim TEXT, satir_sayisi INTEGER, sutun_sayisi INTEGER, baslik_satir_sayisi INTEGER, |
| birlesik_hucre INTEGER, sayi_kipi TEXT, ozet TEXT); |
| CREATE TABLE IF NOT EXISTS _sutunlar( |
| tablo TEXT, sutun TEXT, sira INTEGER, orijinal_baslik TEXT, sql_tipi TEXT, anlam TEXT, |
| birim TEXT, carpan REAL, ayristirilamayan INTEGER, ornek_degerler TEXT, aciklama TEXT, |
| PRIMARY KEY(tablo, sutun)); |
| """ |
|
|
|
|
| def ensure_meta(con: sqlite3.Connection): |
| """Meta tabloları oluşturur; eski sürüm veritabanlarına yeni sütunları ekler.""" |
| con.executescript(META_DDL) |
| for tbl, col in (("_tablolar", "ozet"), ("_sutunlar", "aciklama"), ("_tablolar", "baglam_yillari")): |
| have = {r[1] for r in con.execute(f"PRAGMA table_info({tbl})")} |
| if col not in have: |
| con.execute(f"ALTER TABLE {tbl} ADD COLUMN {col} TEXT") |
| con.commit() |
|
|
|
|
| def drop_same_source(con: sqlite3.Connection, bt: "BuiltTable") -> list[str]: |
| """Aynı belge, sayfa ve başlıktan daha önce farklı adla aktarılmış tabloları siler (yeniden aktarımda kopya oluşmasın).""" |
| ensure_meta(con) |
| old = [r[0] for r in con.execute( |
| "SELECT tablo FROM _tablolar WHERE kaynak=? AND baslik=? AND COALESCE(sayfa,-1)=COALESCE(?,-1) AND tablo<>?", |
| (bt.source, bt.caption, bt.page, bt.name))] |
| for t in old: |
| con.execute(f'DROP TABLE IF EXISTS "{t}"') |
| con.execute("DELETE FROM _tablolar WHERE tablo=?", (t,)) |
| con.execute("DELETE FROM _sutunlar WHERE tablo=?", (t,)) |
| con.commit() |
| return old |
|
|
|
|
| def set_table_summary(con: sqlite3.Connection, table: str, summary: str): |
| ensure_meta(con) |
| con.execute("UPDATE _tablolar SET ozet=? WHERE tablo=?", (summary, table)) |
| con.commit() |
|
|
|
|
| def write_table(con: sqlite3.Connection, bt: BuiltTable, replace: bool = True) -> str: |
| ensure_meta(con) |
| name = bt.name |
| existing = {r[0] for r in con.execute("SELECT name FROM sqlite_master WHERE type='table'")} |
| |
| same = con.execute("SELECT tablo FROM _tablolar WHERE kaynak IS ? AND sayfa IS ? AND baslik IS ? " |
| "AND (tablo=? OR tablo LIKE ? ESCAPE '\\')", (bt.source, bt.page, bt.caption, name, name + "\\_%") |
| ).fetchone() if replace else None |
| if same: |
| name = same[0] |
| other_source = not same and name in existing |
| if not replace or other_source: |
| |
| base, k = name, 2 |
| while name in existing: |
| name = f"{base}_{k}" |
| k += 1 |
| con.execute(f'DROP TABLE IF EXISTS "{name}"') |
| con.execute("DELETE FROM _tablolar WHERE tablo=?", (name,)) |
| con.execute("DELETE FROM _sutunlar WHERE tablo=?", (name,)) |
| coldefs = ", ".join(f'"{c.name}" {c.sqltype}' for c in bt.columns) |
| con.execute(f'CREATE TABLE "{name}" ({coldefs})') |
| ph = ", ".join("?" for _ in bt.columns) |
| con.executemany(f'INSERT INTO "{name}" VALUES ({ph})', bt.rows) |
| con.execute("INSERT INTO _tablolar (tablo, baslik, kaynak, sayfa, cikarici, birim, satir_sayisi, sutun_sayisi, " |
| "baslik_satir_sayisi, birlesik_hucre, sayi_kipi, ozet, baglam_yillari) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?)", |
| (name, bt.caption, bt.source, bt.page, bt.extractor, bt.unit, len(bt.rows), |
| len(bt.columns), bt.n_header_rows, bt.merged_cells, bt.mode, bt.summary or None, |
| bt.context_years or None)) |
| for i, c in enumerate(bt.columns): |
| con.execute("INSERT INTO _sutunlar (tablo, sutun, sira, orijinal_baslik, sql_tipi, anlam, birim, carpan, " |
| "ayristirilamayan, ornek_degerler, aciklama) VALUES (?,?,?,?,?,?,?,?,?,?,?)", |
| (name, c.name, i, c.original, c.sqltype, c.semantic, c.unit, c.multiplier, |
| c.unparsed, json.dumps(c.samples, ensure_ascii=False, default=str), c.description or None)) |
| con.commit() |
| return name |
|
|
|
|
| def list_tables(con) -> list[str]: |
| try: |
| return [r[0] for r in con.execute("SELECT tablo FROM _tablolar ORDER BY rowid")] |
| except sqlite3.OperationalError: |
| return [r[0] for r in con.execute( |
| "SELECT name FROM sqlite_master WHERE type='table' AND name NOT LIKE '\\_%' ESCAPE '\\'")] |
|
|
|
|
| def describe_schema(con, tables: list[str] | None = None, n_samples: int = 3, |
| distinct_text: int = 25) -> str: |
| """LLM'e verilecek şema açıklaması (CREATE TABLE + Türkçe yorumlar + örnek değerler).""" |
| out = [] |
| tables = tables or list_tables(con) |
| for t in tables: |
| tcols = {r[1] for r in con.execute("PRAGMA table_info(_tablolar)")} |
| ccols = {r[1] for r in con.execute("PRAGMA table_info(_sutunlar)")} |
| ozet_sql = "ozet" if "ozet" in tcols else "NULL" |
| acik_sql = "aciklama" if "aciklama" in ccols else "NULL" |
| meta = con.execute(f"SELECT baslik, kaynak, sayfa, birim, {ozet_sql} FROM _tablolar WHERE tablo=?", (t,)).fetchone() |
| cols = con.execute(f"SELECT sutun, sql_tipi, anlam, birim, orijinal_baslik, ornek_degerler, {acik_sql} " |
| "FROM _sutunlar WHERE tablo=? ORDER BY sira", (t,)).fetchall() |
| if meta: |
| out.append(f"-- Tablo: {t} | Belgedeki başlık: \"{meta[0]}\" | Kaynak: {meta[1]}" |
| + (f" s.{meta[2]}" if meta[2] else "") + (f" | Birim: {meta[3]}" if meta[3] else "")) |
| if meta[4]: |
| out.append(f"-- Özet: {meta[4]}") |
| lines = [] |
| for sutun, tip, anlam, birim, orij, ornek, aciklama in cols: |
| note = f"orijinal: \"{orij}\"; anlam: {anlam}" + (f"; birim: {birim}" if birim else "") |
| if aciklama: |
| note += f"; açıklama: {aciklama}" |
| if tip == "TEXT" and anlam in ("metin",): |
| vals = [r[0] for r in con.execute( |
| f'SELECT DISTINCT "{sutun}" FROM "{t}" WHERE "{sutun}" IS NOT NULL LIMIT {distinct_text + 1}')] |
| if len(vals) <= distinct_text: |
| note += "; değerler: " + ", ".join(repr(v) for v in vals) |
| else: |
| note += "; örnek: " + ", ".join(repr(v) for v in vals[:n_samples]) |
| elif anlam == "tarih": |
| note += "; ISO biçim, örnek: " + ", ".join(json.loads(ornek or "[]")[:2]) |
| else: |
| note += "; örnek: " + ", ".join(str(x) for x in json.loads(ornek or "[]")[:n_samples]) |
| lines.append(f' "{sutun}" {tip}, -- {note}') |
| if lines: |
| lines[-1] = lines[-1].replace(", --", " --", 1) |
| out.append(f'CREATE TABLE "{t}" (\n' + "\n".join(lines) + "\n);") |
| return "\n".join(out) |
|
|
