Blame for t2s/schema.py

"""Çö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):  # 'Nüfus (2025)': parantezli yıl birim değildir
        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               # INTEGER | REAL | TEXT
    semantic: str              # sayi | yuzde | tarih | metin | yil | bayrak
    unit: str | None = None
    multiplier: float = 1.0
    unparsed: int = 0          # sayısal sütunda ayrıştırılamayan hücre sayısı
    samples: list = field(default_factory=list)
    description: str = ""      # isteğe bağlı açıklama (ör. AI önerisi)


@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 = ""          # isteğe bağlı tablo özeti (ör. AI önerisi)
    context_years: str = ""    # tablo kendisi yıl içermiyorsa, hemen üstündeki belge metninde geçen yıllar


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
    # tarih?
    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":   # '258.107.640,00' gibi ondalık virgüllü yazımlar REAL kalmalı
            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"  # dipnot numarası / ekonomik kod: ölçü değildir, birimi yoktur
        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 []
    # toplam satırı işareti
    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))


# ======================================================================= SQLite
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'")}
    # aynı belge, sayfa ve başlıktan gelen tablo varsa (yeniden aktarım) onun yerine yaz
    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:
        # başka bir belgenin/sayfanın tablosunun üzerine yazma: benzersiz ad seç
        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)