Back office: Postgres source-of-truth + read dashboard

Stand up the DB-first CRM backbone (architecture pivot: Postgres is the
source of truth, Google Sheets becomes a one-way downstream mirror).

- backoffice/ stack: smb-db (Postgres 16) + smb-crm (Flask/waitress service).
- Schema mirrors the six Sheet tabs (clients, leads, projects, activity_log,
  bookings, invoices) with typed columns + updated_at triggers.
- Service-account Sheets client (PyJWT) for the one-time import + future mirror.
- import_from_sheets.py: idempotent seed of Postgres from the live Sheets.
- Read dashboard (Leads & Clients tables) at onboard.mivanchenko.de/crm,
  behind the existing Caddy basic-auth; JSON API reads straight from Postgres.

Deployed + verified: import seeded DB, dashboard/API live, no-auth blocked,
onboarding form unaffected. Add/edit/delete + DB->Sheets sync + n8n ingest
swap are the next steps.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

backoffice/app/db.py

@@ -0,0 +1,148 @@
+"""Postgres access + the table/column contract shared across the app.
+
+The TABLES map is the single definition of which entities exist, their columns,
+and their primary key. Read/CRUD endpoints, the importer and the Sheets mirror
+all derive from it so they can never drift apart.
+"""
+import os
+import re
+from datetime import datetime, date
+
+import psycopg
+from psycopg.rows import dict_row
+
+DATABASE_URL = os.environ["DATABASE_URL"]
+
+# entity -> (sheet tab, primary key, ordered columns)
+TABLES = {
+    "clients": {
+        "tab": "Clients",
+        "pk": "client_id",
+        "cols": ["client_id", "business_name", "owner_name", "email", "phone",
+                 "niche", "tier", "status", "domain", "stack_notes", "vault_ref",
+                 "services", "billing_cycle", "monthly_fee_eur", "start_date",
+                 "renewal_date", "created_at", "notes"],
+        "dates": ["start_date", "renewal_date"],
+        "timestamps": ["created_at"],
+        "numbers": ["monthly_fee_eur"],
+        "bools": [],
+    },
+    "leads": {
+        "tab": "Leads",
+        "pk": "lead_id",
+        "cols": ["lead_id", "received_at", "client_id", "source", "name",
+                 "contact", "service_interest", "message", "status", "notified"],
+        "dates": [],
+        "timestamps": ["received_at"],
+        "numbers": [],
+        "bools": ["notified"],
+    },
+    "projects": {
+        "tab": "Projects",
+        "pk": "project_id",
+        "cols": ["project_id", "client_id", "deliverable", "tier", "checklist",
+                 "go_live_date", "status"],
+        "dates": ["go_live_date"],
+        "timestamps": [],
+        "numbers": [],
+        "bools": [],
+    },
+    "bookings": {
+        "tab": "Bookings",
+        "pk": "booking_id",
+        "cols": ["booking_id", "created_at", "client_id", "customer_name",
+                 "customer_contact", "service", "start_time", "end_time",
+                 "source", "status"],
+        "dates": [],
+        "timestamps": ["created_at", "start_time", "end_time"],
+        "numbers": [],
+        "bools": [],
+    },
+    "invoices": {
+        "tab": "Invoices",
+        "pk": "invoice_id",
+        "cols": ["invoice_id", "client_id", "issued_date", "due_date",
+                 "amount_eur", "period", "status", "paid_date"],
+        "dates": ["issued_date", "due_date", "paid_date"],
+        "timestamps": [],
+        "numbers": ["amount_eur"],
+        "bools": [],
+    },
+    "activity_log": {
+        "tab": "Activity Log",
+        "pk": "id",
+        "cols": ["ts", "workflow", "client_id", "action", "detail", "result"],
+        "dates": [],
+        "timestamps": ["ts"],
+        "numbers": [],
+        "bools": [],
+    },
+}
+
+
+def connect():
+    return psycopg.connect(DATABASE_URL, row_factory=dict_row)
+
+
+# ---- coercion: Sheet strings / JSON values -> typed Python for Postgres ----
+
+def parse_date(v):
+    if v in (None, ""):
+        return None
+    if isinstance(v, date):
+        return v
+    m = re.match(r"(\d{4})-(\d{2})-(\d{2})", str(v))
+    return date(int(m.group(1)), int(m.group(2)), int(m.group(3))) if m else None
+
+
+def parse_ts(v):
+    if v in (None, ""):
+        return None
+    if isinstance(v, datetime):
+        return v
+    s = str(v).strip()
+    for fmt in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%d %H:%M", "%Y-%m-%dT%H:%M:%S",
+                "%Y-%m-%dT%H:%M", "%Y-%m-%d"):
+        try:
+            return datetime.strptime(s, fmt)
+        except ValueError:
+            continue
+    return None
+
+
+def parse_num(v):
+    if v in (None, ""):
+        return None
+    s = str(v).replace("€", "").replace(",", ".").strip()
+    try:
+        return float(s)
+    except ValueError:
+        return None
+
+
+def parse_bool(v):
+    if v in (None, ""):
+        return None
+    if isinstance(v, bool):
+        return v
+    return str(v).strip().lower() in ("true", "1", "yes", "ja", "wahr")
+
+
+def coerce_row(entity, rec):
+    """Return a dict of column -> typed value for the given entity."""
+    spec = TABLES[entity]
+    out = {}
+    for col in spec["cols"]:
+        v = rec.get(col, None)
+        if isinstance(v, str):
+            v = v.strip() or None
+        if col in spec["dates"]:
+            v = parse_date(v)
+        elif col in spec["timestamps"]:
+            v = parse_ts(v)
+        elif col in spec["numbers"]:
+            v = parse_num(v)
+        elif col in spec["bools"]:
+            v = parse_bool(v)
+        out[col] = v
+    return out