| """Bilgi tabanını dışa aktarma: SQLite, Excel, CSV (zip) ve SQL dökümü. |
| |
| Tüm biçimler veri tablolarıyla birlikte meta tabloları (_tablolar, _sutunlar) da içerir; böylece orijinal |
| başlık, birim, kaynak belge/sayfa ve (varsa) AI açıklamaları dışa aktarımda kaybolmaz. |
| """ |
| from __future__ import annotations |
|
|
| import io |
| import re |
| import sqlite3 |
| import tempfile |
| import zipfile |
| from datetime import datetime |
| from pathlib import Path |
|
|
| import pandas as pd |
|
|
| from .schema import list_tables |
|
|
|
|
| def _all_tables(con: sqlite3.Connection, include_meta: bool = True) -> list[str]: |
| tabs = list_tables(con) |
| if include_meta: |
| have = {r[0] for r in con.execute("SELECT name FROM sqlite_master WHERE type='table'")} |
| tabs += [m for m in ("_tablolar", "_sutunlar") if m in have] |
| return tabs |
|
|
|
|
| def stamp() -> str: |
| return datetime.now().strftime("%Y%m%d_%H%M") |
|
|
|
|
| def sqlite_bytes(con: sqlite3.Connection) -> bytes: |
| """Açık bağlantıdan tutarlı bir anlık görüntü (SQLite backup API).""" |
| with tempfile.TemporaryDirectory() as d: |
| p = Path(d) / "bilgi_tabani.db" |
| dst = sqlite3.connect(p) |
| con.backup(dst) |
| dst.close() |
| return p.read_bytes() |
|
|
|
|
| def _sheet_name(name: str, used: set[str]) -> str: |
| base = re.sub(r"[\[\]\*\?/\\:]", "_", name)[:31] or "tablo" |
| s, k = base, 2 |
| while s.lower() in used: |
| suffix = f"_{k}" |
| s = base[:31 - len(suffix)] + suffix |
| k += 1 |
| used.add(s.lower()) |
| return s |
|
|
|
|
| def xlsx_bytes(con: sqlite3.Connection, tables: list[str] | None = None) -> bytes: |
| buf = io.BytesIO() |
| used: set[str] = set() |
| with pd.ExcelWriter(buf, engine="openpyxl") as xw: |
| for t in tables or _all_tables(con): |
| df = pd.read_sql(f'SELECT * FROM "{t}"', con) |
| sheet = _sheet_name(t, used) |
| df.to_excel(xw, sheet_name=sheet, index=False) |
| ws = xw.sheets[sheet] |
| ws.freeze_panes = "A2" |
| for i, col in enumerate(df.columns, start=1): |
| width = max([len(str(col))] + [len(str(v)) for v in df[col].head(200)]) + 2 |
| ws.column_dimensions[ws.cell(row=1, column=i).column_letter].width = min(60, max(8, width)) |
| return buf.getvalue() |
|
|
|
|
| def csv_zip_bytes(con: sqlite3.Connection, tables: list[str] | None = None) -> bytes: |
| buf = io.BytesIO() |
| with zipfile.ZipFile(buf, "w", zipfile.ZIP_DEFLATED) as z: |
| for t in tables or _all_tables(con): |
| df = pd.read_sql(f'SELECT * FROM "{t}"', con) |
| |
| z.writestr(f"{t}.csv", df.to_csv(index=False).encode("utf-8-sig")) |
| return buf.getvalue() |
|
|
|
|
| def table_csv_bytes(con: sqlite3.Connection, table: str) -> bytes: |
| return pd.read_sql(f'SELECT * FROM "{table}"', con).to_csv(index=False).encode("utf-8-sig") |
|
|
|
|
| def sql_dump_bytes(con: sqlite3.Connection) -> bytes: |
| header = (f"-- Tablo2SQL-TR bilgi tabanı dökümü · {datetime.now().isoformat(timespec='seconds')}\n" |
| "-- Yükleme: sqlite3 yeni.db < bu_dosya.sql\n") |
| return (header + "\n".join(con.iterdump()) + "\n").encode("utf-8") |
|
|
