Files
business-card-scanner/app/db.py
Lucas Orth ab993e98e1 Visitenkarten-Scanner: Stapelscan, Extraktion, Übersicht, vCard-Export
Selbst gehostete PWA, die Visitenkarten von einem Foto freistellt, ausliest
und als Kontakt bereitstellt.

- Stapelscan: OpenCV findet die Kartenrechtecke über mehrere Binärmasken,
  entzerrt sie perspektivisch und schneidet sie einzeln aus. Ohne Fund gilt
  das ganze Foto als eine Karte.
- Extraktion: ein Aufruf je Zuschnitt an das Vision-Modell mit
  JSON-Schema. Kein vorgeschaltetes OCR - das würde Layout und
  Schriftgrößen wegwerfen, aus denen die Feldzuordnung entsteht.
- Metadaten: Aufnahmezeit und GPS aus den EXIF-Daten des Fotos, Ortsname
  über Nominatim, Browserstandort nur als Rückfallebene.
- Übersicht mit Volltextsuche und Filtern, Detailansicht mit Korrekturmaske.
- Notizfeld je Karte, Erinnerungen per Mail inklusive Nachholen verpasster
  Termine nach einem Neustart.
- vCard 3.0 einzeln und als Sammeldatei, Karte gilt danach als exportiert.
- Anmeldung über ein Passwort, Sitzung als signiertes Cookie.
- Deployment per Tag-Push nach den Konventionen in DEPLOY.md.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-09-06 11:45:56 +02:00

101 lines
2.7 KiB
Python

"""SQLite-Zugriff. Eine Verbindung, Schreibzugriffe unter einem Lock."""
import sqlite3
import threading
from typing import Any, Iterable
from . import config
_lock = threading.Lock()
_conn: sqlite3.Connection | None = None
SCHEMA = """
CREATE TABLE IF NOT EXISTS cards (
id TEXT PRIMARY KEY,
created_at TEXT NOT NULL,
captured_at TEXT,
batch_id TEXT,
image_file TEXT,
source TEXT NOT NULL DEFAULT 'batch',
full_name TEXT,
first_name TEXT,
last_name TEXT,
job_title TEXT,
company TEXT,
department TEXT,
email TEXT,
phone TEXT,
mobile TEXT,
fax TEXT,
website TEXT,
street TEXT,
postal_code TEXT,
city TEXT,
country TEXT,
notes TEXT NOT NULL DEFAULT '',
lat REAL,
lon REAL,
place TEXT,
geo_source TEXT,
extract_status TEXT NOT NULL DEFAULT 'ok',
extract_error TEXT,
exported_at TEXT
);
CREATE TABLE IF NOT EXISTS reminders (
id TEXT PRIMARY KEY,
card_id TEXT NOT NULL REFERENCES cards(id) ON DELETE CASCADE,
text TEXT NOT NULL,
due_at TEXT NOT NULL,
created_at TEXT NOT NULL,
sent_at TEXT,
send_error TEXT
);
CREATE INDEX IF NOT EXISTS idx_cards_created ON cards(created_at DESC);
CREATE INDEX IF NOT EXISTS idx_reminders_due ON reminders(due_at) WHERE sent_at IS NULL;
CREATE TABLE IF NOT EXISTS geocache (
key TEXT PRIMARY KEY,
place TEXT,
fetched_at TEXT NOT NULL
);
"""
def connect() -> sqlite3.Connection:
global _conn
if _conn is None:
config.ensure_dirs()
_conn = sqlite3.connect(config.DB_PATH, check_same_thread=False)
_conn.row_factory = sqlite3.Row
_conn.execute("PRAGMA journal_mode=WAL")
_conn.execute("PRAGMA foreign_keys=ON")
_conn.executescript(SCHEMA)
_conn.commit()
return _conn
def query(sql: str, params: Iterable[Any] = ()) -> list[sqlite3.Row]:
with _lock:
return connect().execute(sql, tuple(params)).fetchall()
def query_one(sql: str, params: Iterable[Any] = ()) -> sqlite3.Row | None:
rows = query(sql, params)
return rows[0] if rows else None
def execute(sql: str, params: Iterable[Any] = ()) -> None:
with _lock:
conn = connect()
conn.execute(sql, tuple(params))
conn.commit()
def execute_many(statements: list[tuple[str, Iterable[Any]]]) -> None:
with _lock:
conn = connect()
for sql, params in statements:
conn.execute(sql, tuple(params))
conn.commit()