business-card-scanner/app/db.py

"""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}")
    _split_legacy_names(conn)


def _split_legacy_names(conn: sqlite3.Connection) -> None:
    """full_name ist als Feld entfallen. Aeltere Datensaetze haben nur diesen
    einen Namen - am letzten Leerzeichen trennen, damit sie nicht namenlos
    dastehen. Die Spalte bleibt: SQLite baut zum Loeschen die Tabelle neu,
    und der Gewinn waere ein paar Byte."""
    rows = conn.execute(
        "SELECT id, full_name FROM cards "
        "WHERE full_name IS NOT NULL AND first_name IS NULL AND last_name IS NULL"
    ).fetchall()
    for row in rows:
        first, _, last = row["full_name"].strip().rpartition(" ")
        conn.execute(
            "UPDATE cards SET first_name = ?, last_name = ? WHERE id = ?",
            (first or None, last or None, row["id"]),
        )


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()