| """Belgelerden tablo çıkarma: PDF (pdfplumber, Camelot), HTML, CSV/XLSX. |
| |
| Tüm çıkarıcılar ortak bir ara gösterime dönüşür: |
| RawTable -> hücreler + satır/sütun birleşimleri (rowspan/colspan) |
| resolve() -> çok satırlı başlıkları düzleştirir, birleşik hücreleri yayar |
| ve ResolvedTable (başlık parçaları + gövde metinleri) üretir. |
| """ |
| from __future__ import annotations |
|
|
| import re |
| from dataclasses import dataclass, field |
| from pathlib import Path |
|
|
| from .trnum import clean_cell, fold, parse_date, parse_number |
|
|
|
|
| @dataclass |
| class Cell: |
| row: int |
| col: int |
| text: str |
| rowspan: int = 1 |
| colspan: int = 1 |
|
|
|
|
| @dataclass |
| class RawTable: |
| cells: list[Cell] |
| n_rows: int |
| n_cols: int |
| caption: str = "" |
| source: str = "" |
| page: int | None = None |
| extractor: str = "" |
| context: str = "" |
|
|
| def grid(self, fill_spans: bool = True) -> tuple[list[list[str]], list[list[str]]]: |
| """(metin ızgarası, birleşim işaret ızgarası). İşaretler: '' tek hücre, |
| 'o' birleşimin kökü, 'h' yatay kopya, 'v' dikey kopya.""" |
| g = [["" for _ in range(self.n_cols)] for _ in range(self.n_rows)] |
| m = [["" for _ in range(self.n_cols)] for _ in range(self.n_rows)] |
| for c in self.cells: |
| for dr in range(c.rowspan): |
| for dc in range(c.colspan): |
| r, k = c.row + dr, c.col + dc |
| if r >= self.n_rows or k >= self.n_cols: |
| continue |
| if dr == 0 and dc == 0: |
| g[r][k] = c.text |
| m[r][k] = "o" if (c.rowspan > 1 or c.colspan > 1) else "" |
| else: |
| g[r][k] = c.text if fill_spans else "" |
| m[r][k] = "v" if dr > 0 else "h" |
| return g, m |
|
|
|
|
| @dataclass |
| class ResolvedTable: |
| caption: str |
| header_parts: list[list[str]] |
| body: list[list[str]] |
| source: str = "" |
| page: int | None = None |
| extractor: str = "" |
| n_header_rows: int = 1 |
| merged_cells: int = 0 |
| context: str = "" |
| raw: RawTable | None = field(default=None, repr=False) |
|
|
| @property |
| def headers(self) -> list[str]: |
| return [" ".join(p) for p in self.header_parts] |
|
|
|
|
| |
| def _caption_from_text(text: str) -> str: |
| lines = [l.strip() for l in (text or "").splitlines() if l.strip()] |
| for l in reversed(lines[-4:]): |
| if re.match(r"^(tablo|çizelge|cizelge|table)\s*[\d.:-]", fold(l)): |
| return l |
| |
| for l in reversed(lines[-4:]): |
| if not l.startswith("(") and re.search(r"\b(tablosu|cizelgesi|bilanco)\b", fold(l)): |
| return l |
| plain = [l for l in lines[-4:] if not l.startswith("(")] |
| return plain[-1] if plain else (lines[-1] if lines else "") |
|
|
|
|
| def _spans_from_bboxes(cell_boxes, tol: float = 1.5): |
| """pdfplumber hücre kutularından satır/sütun sınırları ve birleşimleri hesaplar.""" |
| def uniq(vals): |
| out = [] |
| for v in sorted(vals): |
| if not out or abs(v - out[-1]) > tol: |
| out.append(v) |
| return out |
| xs = uniq([b[0] for b in cell_boxes] + [b[2] for b in cell_boxes]) |
| ys = uniq([b[1] for b in cell_boxes] + [b[3] for b in cell_boxes]) |
| def idx(arr, v): |
| return min(range(len(arr)), key=lambda i: abs(arr[i] - v)) |
| out = [] |
| for b in cell_boxes: |
| c0, c1 = idx(xs, b[0]), idx(xs, b[2]) |
| r0, r1 = idx(ys, b[1]), idx(ys, b[3]) |
| out.append((r0, c0, max(1, r1 - r0), max(1, c1 - c0), b)) |
| return out, len(ys) - 1, len(xs) - 1 |
|
|
|
|
| def extract_pdf_pdfplumber(path: str | Path) -> list[RawTable]: |
| import pdfplumber |
| tables = [] |
| with pdfplumber.open(str(path)) as pdf: |
| for pno, page in enumerate(pdf.pages, start=1): |
| for t in page.find_tables(): |
| boxes = [c for c in t.cells if c] |
| if not boxes: |
| continue |
| spans, nr, nc = _spans_from_bboxes(boxes) |
| cells = [] |
| for r0, c0, rs, cs, b in spans: |
| try: |
| txt = page.crop(b).extract_text() or "" |
| except ValueError: |
| txt = "" |
| cells.append(Cell(r0, c0, clean_cell(txt), rs, cs)) |
| above = page.crop((0, 0, page.width, max(1, t.bbox[1]))).extract_text() or "" |
| tables.append(RawTable(cells, nr, nc, caption=_caption_from_text(above), |
| source=Path(path).name, page=pno, extractor="pdfplumber", |
| context=above[-600:])) |
| return tables |
|
|
|
|
| def extract_pdf_camelot(path: str | Path, flavor: str = "lattice") -> list[RawTable]: |
| """Camelot: birleşik hücre metnini copy_text ile yayar; birleşim bilgisi kaybolur.""" |
| import camelot |
| import pdfplumber |
| kw = {"copy_text": ["h", "v"]} if flavor == "lattice" else {} |
| res = camelot.read_pdf(str(path), flavor=flavor, pages="all", **kw) |
| out = [] |
| with pdfplumber.open(str(path)) as pdf: |
| for t in res: |
| df = t.df |
| cells = [Cell(r, c, clean_cell(df.iat[r, c])) for r in range(df.shape[0]) |
| for c in range(df.shape[1])] |
| pno = int(t.page) |
| page = pdf.pages[pno - 1] |
| |
| try: |
| x0, y0, x1, y1 = t._bbox |
| top = page.height - y1 |
| above = page.crop((0, 0, page.width, max(1, top))).extract_text() or "" |
| except Exception: |
| above = "" |
| out.append(RawTable(cells, df.shape[0], df.shape[1], caption=_caption_from_text(above), |
| source=Path(path).name, page=pno, extractor=f"camelot-{flavor}", |
| context=above[-600:])) |
| return out |
|
|
|
|
| def extract_pdf(path, method: str = "auto") -> list[RawTable]: |
| if method == "pdfplumber": |
| return extract_pdf_pdfplumber(path) |
| if method in ("camelot", "camelot-lattice"): |
| return extract_pdf_camelot(path, "lattice") |
| if method == "camelot-stream": |
| return extract_pdf_camelot(path, "stream") |
| tabs = extract_pdf_pdfplumber(path) |
| if not tabs: |
| try: |
| tabs = extract_pdf_camelot(path, "stream") |
| except Exception: |
| tabs = [] |
| return tabs |
|
|
|
|
| |
| def extract_html(path_or_html: str | Path) -> list[RawTable]: |
| from bs4 import BeautifulSoup |
| p = Path(str(path_or_html)) |
| is_file = len(str(path_or_html)) < 500 and p.exists() |
| html = p.read_text(encoding="utf-8", errors="replace") if is_file else str(path_or_html) |
| soup = BeautifulSoup(html, "lxml") |
| out = [] |
| for ti, table in enumerate(soup.find_all("table")): |
| if table.find_parent("table") is not None: |
| continue |
| cap_el = table.find("caption") |
| caption = clean_cell(cap_el.get_text(" ")) if cap_el else "" |
| prev_txt = [] |
| for sib in table.find_all_previous(["p", "h1", "h2", "h3", "h4", "div"], limit=4): |
| prev_txt.insert(0, clean_cell(sib.get_text(" "))) |
| context = "\n".join(prev_txt) |
| if not caption: |
| caption = _caption_from_text(context) |
| occupied, cells, max_c = set(), [], 0 |
| rows = [tr for tr in table.find_all("tr") if tr.find_parent("table") is table |
| and tr.find_parent("tfoot") is None] |
| for r, tr in enumerate(rows): |
| c = 0 |
| for td in tr.find_all(["td", "th"], recursive=False): |
| while (r, c) in occupied: |
| c += 1 |
| rs = int(td.get("rowspan", 1) or 1) |
| cs = int(td.get("colspan", 1) or 1) |
| cells.append(Cell(r, c, clean_cell(td.get_text(" ")), rs, cs)) |
| for dr in range(rs): |
| for dc in range(cs): |
| occupied.add((r + dr, c + dc)) |
| c += cs |
| max_c = max(max_c, c) |
| if not cells: |
| continue |
| n_rows = max(c.row + c.rowspan for c in cells) |
| name = p.name if is_file else f"html_girdi_{ti + 1}" |
| out.append(RawTable(cells, n_rows, max_c, caption=caption, source=name, |
| extractor="html-bs4", context=context[-600:])) |
| return out |
|
|
|
|
| |
| def extract_tabular(path: str | Path) -> list[RawTable]: |
| import pandas as pd |
| p = Path(path) |
| if p.suffix.lower() == ".csv": |
| sample = p.read_text(encoding="utf-8", errors="replace")[:2000] |
| sep = ";" if sample.count(";") > sample.count(",") else "," |
| dfs = {p.stem: pd.read_csv(p, sep=sep, dtype=str, header=None, keep_default_na=False)} |
| else: |
| dfs = pd.read_excel(p, sheet_name=None, dtype=str, header=None) |
| out = [] |
| for name, df in dfs.items(): |
| df = df.fillna("") |
| cells = [Cell(r, c, clean_cell(df.iat[r, c])) for r in range(df.shape[0]) for c in range(df.shape[1])] |
| out.append(RawTable(cells, df.shape[0], df.shape[1], caption=str(name), source=p.name, |
| extractor="pandas")) |
| return out |
|
|
|
|
| def extract_any(path: str | Path, method: str = "auto") -> list[RawTable]: |
| suf = Path(path).suffix.lower() |
| if suf == ".pdf": |
| return extract_pdf(path, method) |
| if suf in (".html", ".htm", ".xhtml"): |
| return extract_html(path) |
| if suf in (".csv", ".xlsx", ".xls"): |
| return extract_tabular(path) |
| raise ValueError(f"Desteklenmeyen biçim: {suf}") |
|
|
|
|
| |
| def _is_data_like(s: str) -> bool: |
| if not s: |
| return False |
| pn = parse_number(s, "tr") |
| return (pn.ok and pn.value is not None) or parse_date(s) is not None |
|
|
|
|
| def _header_row_count(grid, marks, max_h: int = 3) -> int: |
| """Üstten, veri hücresi oranı düşük olan satırları başlık sayar. |
| Yıl gibi sayısal başlıkları ('2023') veri saymamak için birleşim ve tekrar ipuçları kullanılır.""" |
| n = 0 |
| period_seen = False |
| for r, row in enumerate(grid[:max_h + 1]): |
| vals = row[1:] if len(row) > 1 else row |
| nonempty = [v for v in vals if v] |
| if not nonempty: |
| if period_seen: |
| break |
| n += 1 |
| continue |
| |
| if not row[0] and nonempty and all(parse_date(v) and not parse_number(v, "tr").ok for v in nonempty): |
| n += 1 |
| period_seen = True |
| if n >= max_h: |
| break |
| continue |
| data_like = [v for v in nonempty if _is_data_like(v) and not re.fullmatch(r"(19|20)\d{2}", v)] |
| has_merge = any(m in ("o", "h") for m in marks[r]) |
| if len(data_like) / len(nonempty) >= 0.5 and not (has_merge and r == 0): |
| break |
| |
| n += 1 |
| if n >= max_h: |
| break |
| return max(1, n) |
|
|
|
|
| def resolve(raw: RawTable) -> ResolvedTable: |
| grid, marks = raw.grid(fill_spans=True) |
| |
| keep_rows = [i for i, r in enumerate(grid) if any(x for x in r)] |
| grid = [grid[i] for i in keep_rows] |
| marks = [marks[i] for i in keep_rows] |
| keep_cols = [j for j in range(raw.n_cols) if any(r[j] for r in grid)] |
| grid = [[r[j] for j in keep_cols] for r in grid] |
| marks = [[r[j] for j in keep_cols] for r in marks] |
| if not grid: |
| return ResolvedTable(raw.caption, [], [], raw.source, raw.page, raw.extractor, raw=raw) |
|
|
| |
| |
| caption, context = raw.caption, raw.context |
| while len(grid) > 2 and len(grid[0]) > 2: |
| filled = [x for x, m in zip(grid[0], marks[0]) if x and m != "h"] |
| if len(set(filled)) != 1: |
| break |
| line = filled[0] |
| if re.match(r"^(tablo|cizelge|table)\s*[\d.:-]", fold(line)) or ( |
| not re.match(r"^(tablo|cizelge|table)\s*[\d.:-]", fold(caption or "")) |
| and not line.startswith("(") and re.search(r"\b(tablosu|cizelgesi|bilanco)\b", fold(line))): |
| caption = line |
| else: |
| context = (context + "\n" + line).strip() |
| grid, marks = grid[1:], marks[1:] |
| raw_caption_ok = re.match(r"^(tablo|cizelge|table)\s*[\d.:-]", fold(caption or "")) |
| if not raw_caption_ok: |
| m = re.search(r"((Tablo|Çizelge|Table)\s*[\d.]+\s*[:.\-–].*)", context or "") |
| if m: |
| caption = m.group(1).strip() |
|
|
| nh = _header_row_count(grid, marks, max_h=5) |
| header_parts = [] |
| for j in range(len(keep_cols)): |
| parts = [] |
| for i in range(nh): |
| t = grid[i][j] |
| if t and (not parts or fold(parts[-1]) != fold(t)): |
| parts.append(t) |
| header_parts.append(parts) |
|
|
| body = [row[:] for row in grid[nh:]] |
| |
| |
| j = 1 |
| while j < len(header_parts): |
| if header_parts[j] and body and not any(r[j] for r in body): |
| header_parts[j - 1] = header_parts[j - 1] + [p for p in header_parts[j] if p not in header_parts[j - 1]] |
| del header_parts[j] |
| for r in body: |
| del r[j] |
| keep_cols = keep_cols[:j] + keep_cols[j + 1:] |
| else: |
| j += 1 |
| |
| while body and len({x for x in body[-1] if x}) == 1 and re.match( |
| r"^(kaynak|not|notlar|aciklama|\(\*+\)|\*)\b", fold(next(x for x in body[-1] if x))): |
| body.pop() |
| |
| |
| for j in range(min(2, len(keep_cols))): |
| col = [r[j] for r in body] |
| text_like = sum(1 for v in col if v and not _is_data_like(v)) |
| if text_like >= max(1, 0.5 * len([v for v in col if v])): |
| for i in range(1, len(body)): |
| if not body[i][j] and any(body[i][k] for k in range(len(keep_cols)) if k != j): |
| body[i][j] = body[i - 1][j] |
| merged = sum(1 for c in raw.cells if c.rowspan > 1 or c.colspan > 1) |
| return ResolvedTable(caption, header_parts, body, raw.source, raw.page, raw.extractor, |
| n_header_rows=nh, merged_cells=merged, context=context, raw=raw) |
|
|
