7262aee63d
New /settings page (linked from the header's burger menu) offers: - CSV export of the user's links, for spreadsheet apps. - SQL dump export as INSERT INTO links (...) statements, scoped to the current user only — never a raw full-database dump, since that would leak other accounts' password hashes and data. Embeddings are excluded (regenerated via "KI neu beschreiben lassen" if needed). - SQL import that re-adds a previously exported dump to the current account (additive, doesn't touch existing links). Import safety (app/backup.py): uploaded SQL is never executed against the real database. Each non-comment line is required to start with "insert into links" and is run one statement at a time against an isolated in-memory SQLite database with only a whitelisted `links` schema (no id/user_id columns) — sqlite3.execute() also rejects multiple statements per call. Only after that succeeds are rows copied into the real DB via the ORM, with user_id forced to the logged-in user. Verified this rejects DROP TABLE, ATTACH DATABASE, stacked statements, cross-table subqueries, and user_id injection. Upload is capped at 2 MB. Also adds download/upload icons and a proper file-input styling pattern (visually-hidden input + <label> trigger + filename readout), since the browser's ::file-selector-button pseudo-element didn't render reliably in testing. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
106 lines
3.5 KiB
Python
106 lines
3.5 KiB
Python
"""CSV-/SQL-Export und SQL-Import der eigenen Links (Backup & Restore)."""
|
||
|
||
import csv
|
||
import io
|
||
import sqlite3
|
||
|
||
from .models import Link
|
||
|
||
SQL_COLUMNS = [
|
||
"url", "title", "summary", "category", "manufacturer",
|
||
"tags", "status", "needs_review", "created_at",
|
||
]
|
||
|
||
|
||
def to_csv(links: list[Link]) -> str:
|
||
buf = io.StringIO()
|
||
writer = csv.writer(buf)
|
||
writer.writerow(
|
||
["url", "title", "summary", "category", "manufacturer", "tags", "status", "created_at"]
|
||
)
|
||
for link in links:
|
||
writer.writerow(
|
||
[
|
||
link.url,
|
||
link.title,
|
||
link.summary,
|
||
link.category,
|
||
link.manufacturer,
|
||
", ".join(link.tag_list),
|
||
link.status,
|
||
link.created_at.isoformat(),
|
||
]
|
||
)
|
||
return buf.getvalue()
|
||
|
||
|
||
def _sql_quote(value: str) -> str:
|
||
return "'" + value.replace("'", "''") + "'"
|
||
|
||
|
||
def to_sql_dump(links: list[Link]) -> str:
|
||
lines = [
|
||
"-- LinkVault-Export – Links des angemeldeten Kontos.",
|
||
"-- Kann über 'Einstellungen -> SQL-Import' wieder eingespielt werden.",
|
||
f"-- Zeilen: {len(links)}",
|
||
"",
|
||
]
|
||
for link in links:
|
||
values = ", ".join(
|
||
[
|
||
_sql_quote(link.url),
|
||
_sql_quote(link.title),
|
||
_sql_quote(link.summary),
|
||
_sql_quote(link.category),
|
||
_sql_quote(link.manufacturer),
|
||
_sql_quote(link.tags),
|
||
_sql_quote(link.status),
|
||
"1" if link.needs_review else "0",
|
||
_sql_quote(link.created_at.isoformat()),
|
||
]
|
||
)
|
||
lines.append(f"INSERT INTO links ({', '.join(SQL_COLUMNS)}) VALUES ({values});")
|
||
return "\n".join(lines) + "\n"
|
||
|
||
|
||
def parse_sql_dump(sql_text: str, max_rows: int = 5000) -> list[dict]:
|
||
"""Liest die von to_sql_dump() erzeugten INSERT-Zeilen sicher ein.
|
||
|
||
Läuft bewusst gegen eine isolierte In-Memory-SQLite-Datenbank statt gegen
|
||
die echte App-Datenbank: nur Zeilen, die (nach Trimmen) mit
|
||
'insert into links' beginnen, werden ausgeführt, und sqlite3.execute()
|
||
lässt ohnehin nur je eine einzelne Anweisung zu. Dadurch können weder
|
||
andere Tabellen berührt noch mehrere Anweisungen aneinandergehängt werden.
|
||
Das Ergebnis wird anschließend vom Aufrufer kontrolliert per ORM in die
|
||
echte Datenbank übernommen (mit fest gesetzter user_id).
|
||
"""
|
||
conn = sqlite3.connect(":memory:")
|
||
try:
|
||
conn.execute(
|
||
"CREATE TABLE links ("
|
||
"url TEXT, title TEXT, summary TEXT, category TEXT, manufacturer TEXT, "
|
||
"tags TEXT, status TEXT, needs_review INTEGER, created_at TEXT)"
|
||
)
|
||
statements = 0
|
||
for raw_line in sql_text.splitlines():
|
||
line = raw_line.strip()
|
||
if not line or line.startswith("--"):
|
||
continue
|
||
if not line.lower().startswith("insert into links"):
|
||
continue
|
||
if statements >= max_rows:
|
||
raise ValueError(f"Zu viele Zeilen (Limit: {max_rows}).")
|
||
try:
|
||
conn.execute(line)
|
||
except sqlite3.Error as e:
|
||
raise ValueError(f"Ungültige Zeile: {e}") from e
|
||
statements += 1
|
||
|
||
if statements == 0:
|
||
raise ValueError("Keine gültigen INSERT-Zeilen für 'links' gefunden.")
|
||
|
||
cur = conn.execute(f"SELECT {', '.join(SQL_COLUMNS)} FROM links")
|
||
return [dict(zip(SQL_COLUMNS, row)) for row in cur.fetchall()]
|
||
finally:
|
||
conn.close()
|