| """Belge -> tablo -> SQLite hattı; referans veritabanı ve çıkarım doğruluğu ölçümü.""" |
| from __future__ import annotations |
|
|
| import json |
| import re |
| from dataclasses import dataclass |
| from pathlib import Path |
|
|
| from .extract import ResolvedTable, extract_any, resolve |
| from .schema import BuiltTable, build_table, connect, write_table |
| from .trnum import fold, parse_date, parse_number |
|
|
| ROOT = Path(__file__).resolve().parent.parent |
| DOCS_DIR = ROOT / "data" / "docs" |
| GT_DIR = ROOT / "data" / "ground_truth" |
| DB_DIR = ROOT / "outputs" / "db" |
|
|
|
|
| @dataclass |
| class DocResult: |
| path: str |
| resolved: list[ResolvedTable] |
| built: list[BuiltTable] |
| written: list[str] |
|
|
|
|
| def process_document(path, con=None, method: str = "auto", mode: str = "tr", |
| min_rows: int = 1) -> DocResult: |
| raws = extract_any(path, method) |
| resolved, built, written = [], [], [] |
| rts = merge_continuations([resolve(raw) for raw in raws]) |
| for rt in rts: |
| if len(rt.body) < min_rows or len(rt.header_parts) < 2: |
| continue |
| bt = build_table(rt, mode=mode) |
| resolved.append(rt) |
| built.append(bt) |
| if con is not None: |
| written.append(write_table(con, bt)) |
| return DocResult(str(path), resolved, built, written) |
|
|
|
|
| def _is_title(caption: str) -> bool: |
| return bool(re.match(r"^(tablo|cizelge|table)\s*[\d.:-]", fold(caption or ""))) |
|
|
|
|
| def merge_continuations(rts: list) -> list: |
| """Sayfa sonunda bölünüp sonraki sayfada aynı başlıkla süren tabloları birleştirir. |
| |
| Koşullar: aynı belge, ardışık sayfa, aynı sütun sayısı ve aynı (Türkçe katlanmış) başlıklar; devam parçasının |
| ya hiç başlığı yoktur ya da başlığı öncekiyle aynıdır (başlığı farklıysa ayrı bir tablodur).""" |
| out = [] |
| for rt in rts: |
| prev = out[-1] if out else None |
| last_page = getattr(prev, "last_page", None) or (prev.page if prev is not None else None) |
| if (prev is not None and rt.page and last_page and rt.page == last_page + 1 and rt.source == prev.source |
| and (not (rt.caption or "").strip() or fold(rt.caption) == fold(prev.caption or "")) |
| and len(rt.header_parts) == len(prev.header_parts) and rt.header_parts |
| and [fold(" ".join(h)) for h in rt.header_parts] == [fold(" ".join(h)) for h in prev.header_parts]): |
| prev.body.extend(rt.body) |
| prev.merged_cells += rt.merged_cells |
| prev.last_page = rt.page |
| continue |
| out.append(rt) |
| return out |
|
|
|
|
| def build_database(db_path, docs: list | None = None, method: str = "auto", mode: str = "tr", |
| verbose: bool = False) -> list[DocResult]: |
| db_path = Path(db_path) |
| db_path.parent.mkdir(parents=True, exist_ok=True) |
| if db_path.exists(): |
| db_path.unlink() |
| con = connect(db_path) |
| docs = docs or sorted(p for p in DOCS_DIR.iterdir() if p.suffix.lower() in (".pdf", ".html", ".htm")) |
| results = [] |
| for d in docs: |
| r = process_document(d, con, method=method, mode=mode) |
| results.append(r) |
| if verbose: |
| print(f" {Path(d).name}: {len(r.written)} tablo -> {', '.join(r.written)}") |
| con.close() |
| return results |
|
|
|
|
| |
| def load_ground_truth() -> list[dict]: |
| return [json.loads(p.read_text(encoding="utf-8")) for p in sorted(GT_DIR.glob("*.json"))] |
|
|
|
|
| def gt_to_resolved(gt: dict) -> ResolvedTable: |
| hdr = gt["baslik_satirlari"] |
| ncols = len(gt["govde"][0]) |
| parts = [[] for _ in range(ncols)] |
| occupied = {} |
| for r, row in enumerate(hdr): |
| c = 0 |
| for text, cs, rs in row: |
| while (r, c) in occupied: |
| c += 1 |
| for dr in range(rs): |
| for dc in range(cs): |
| occupied[(r + dr, c + dc)] = text |
| c += cs |
| for j in range(ncols): |
| for r in range(len(hdr)): |
| t = occupied.get((r, j)) |
| if t and (not parts[j] or parts[j][-1] != t): |
| parts[j].append(t) |
| return ResolvedTable(gt["baslik"], parts, gt["govde"], gt["belge"], gt.get("sayfa"), |
| "altin-veri", n_header_rows=len(hdr)) |
|
|
|
|
| def build_reference_db(db_path) -> list[str]: |
| db_path = Path(db_path) |
| db_path.parent.mkdir(parents=True, exist_ok=True) |
| if db_path.exists(): |
| db_path.unlink() |
| con = connect(db_path) |
| names = [write_table(con, build_table(gt_to_resolved(gt), mode="tr")) for gt in load_ground_truth()] |
| con.close() |
| return names |
|
|
|
|
| |
| def _cell_eq(a: str, b: str) -> bool: |
| if fold(a) == fold(b): |
| return True |
| pa, pb = parse_number(a), parse_number(b) |
| if pa.ok and pb.ok and pa.value is not None and pb.value is not None: |
| return abs(pa.value - pb.value) <= 1e-6 * max(1, abs(pb.value)) |
| da, db = parse_date(a), parse_date(b) |
| return da is not None and da == db |
|
|
|
|
| def extraction_accuracy(method: str = "auto") -> list[dict]: |
| """Her altın tablo için: tablo bulundu mu, başlık doğru mu, hücre doğruluğu.""" |
| gts = load_ground_truth() |
| cache: dict[str, list[ResolvedTable]] = {} |
| report = [] |
| for gt in gts: |
| doc = DOCS_DIR / gt["belge"] |
| if doc.name not in cache: |
| try: |
| cache[doc.name] = [resolve(r) for r in extract_any(doc, method)] |
| except Exception as e: |
| cache[doc.name] = [] |
| print(" ! çıkarım hatası", doc.name, e) |
| ref = gt_to_resolved(gt) |
| cands = [rt for rt in cache[doc.name] if fold(rt.caption) == fold(gt["baslik"])] or \ |
| [rt for rt in cache[doc.name] if gt.get("sayfa") in (None, rt.page)] |
| rec = {"tablo": gt["tablo_id"], "belge": gt["belge"], "cikarici": method, "bulundu": bool(cands), |
| "baslik_dogru": False, "hucre_dogrulugu": 0.0, "satir_ok": False, "sutun_ok": False} |
| if cands: |
| rt = cands[0] |
| rec["sutun_ok"] = len(rt.header_parts) == len(ref.header_parts) |
| rec["satir_ok"] = len(rt.body) == len(ref.body) |
| rec["baslik_dogru"] = [fold(" ".join(p)) for p in rt.header_parts] == \ |
| [fold(" ".join(p)) for p in ref.header_parts] |
| tot = ok = 0 |
| for i, row in enumerate(ref.body): |
| for j, v in enumerate(row): |
| tot += 1 |
| if i < len(rt.body) and j < len(rt.body[i]) and _cell_eq(rt.body[i][j], v): |
| ok += 1 |
| rec["hucre_dogrulugu"] = ok / tot if tot else 0.0 |
| rec["birlesik_hucre"] = rt.merged_cells |
| rec["baslik_satiri"] = rt.n_header_rows |
| report.append(rec) |
| return report |
|
|
|
|
| def number_format_stats(db_paths: dict[str, Path]) -> list[dict]: |
| """Kip başına sayısal sütun sayısı, ayrıştırılamayan hücre ve TEXT'e düşen sütun sayısı.""" |
| out = [] |
| for mode, p in db_paths.items(): |
| con = connect(p, readonly=True) |
| rows = con.execute("SELECT sql_tipi, anlam, ayristirilamayan FROM _sutunlar").fetchall() |
| con.close() |
| out.append({ |
| "kip": mode, |
| "sayisal_sutun": sum(1 for t, a, _ in rows if t in ("INTEGER", "REAL")), |
| "metin_sutun": sum(1 for t, a, _ in rows if t == "TEXT"), |
| "ayristirilamayan_hucre": sum(u or 0 for *_, u in rows), |
| }) |
| return out |
|
|
