File size: 3,174 Bytes
a72d1cf | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 | """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)
# utf-8-sig: Excel'in Türkçe karakterleri doğru açması için
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")
|