Files
business-card-scanner/app/db.py
Lucas Orth bf0a397597
All checks were successful
Build & Deploy / deploy (push) Successful in 10m41s
Anlass, Dubletten, Rückseite, Teilen; Startseite und App-Anmutung
Vier Funktionen:

- Anlass: ein Feld auf der Startseite gilt für alle folgenden Scans und
  wird im Browser gemerkt. Ordnet den Bestand verlässlicher als der
  Ortsname aus den Koordinaten. Durchsuchbar, nachträglich änderbar.
- Dubletten: die Detailansicht zeigt Karten, die dieselbe Person meinen
  könnten - gleiche E-Mail, gleiche Telefonnummer (verglichen werden die
  letzten acht Ziffern, damit +49 511 123456 und 0511/123456 aufeinander
  passen) oder gleicher Nachname bei gleicher Firma.
- Rückseite: nachträglich ein zweites Bild zur Karte. Es füllt nur
  Felder, die die Vorderseite offen gelassen hat.
- Schnellwahl bei Erinnerungen: in 3 Tagen / 1 Woche / 2 Wochen /
  1 Monat, jeweils 9 Uhr. Das Datumsfeld braucht man damit selten.

Teilen als vCard über das System-Teilenblatt, mit Download als
Rückfallebene, wo es kein Teilenblatt gibt.

Oberfläche:

- Startseite führt jetzt: Anlass, große Aufnahmefläche mit Symbol statt
  Textzeile, leise Alternative, darunter die zuletzt erfassten Karten.
  Vorher stand dort eine Überschrift mit Zahlen und man musste raten.
- Weniger Webseite, mehr App: kein Seitenrahmen auf dem Telefon, Symbole
  in der Tableiste mit Zählerblasen, kein Tap-Highlight, keine
  Textmarkierung auf Bedienelementen, Safe-Area oben, Einblendung beim
  Ansichtswechsel, Mindesthöhe 54 px für Schaltflächen.
- Die Marken "exportiert" / "nicht exportiert" sind aus der Liste raus;
  dafür gibt es den Filter.

Neue Spalten occasion und back_image_file kommen per ALTER TABLE in
bestehende Datenbanken - CREATE TABLE IF NOT EXISTS rührt eine
vorhandene Tabelle nicht an.

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

113 lines
3.1 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,
back_image_file TEXT,
source TEXT NOT NULL DEFAULT 'batch',
occasion TEXT,
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)
_migrate(_conn)
_conn.commit()
return _conn
# Spalten, die nach dem ersten Deploy dazugekommen sind. CREATE TABLE IF NOT
# EXISTS ruehrt eine bestehende Tabelle nicht an - ohne das hier liefe die
# App auf dem Server mit dem alten Schema weiter.
LATER_COLUMNS = {
"back_image_file": "TEXT",
"occasion": "TEXT",
}
def _migrate(conn: sqlite3.Connection) -> None:
present = {row["name"] for row in conn.execute("PRAGMA table_info(cards)")}
for name, kind in LATER_COLUMNS.items():
if name not in present:
conn.execute(f"ALTER TABLE cards ADD COLUMN {name} {kind}")
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()