smb-online/backoffice/app/db.py

"""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 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 -> (primary key, ordered columns)
TABLES = {
    "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", "notify_channel", "slug"],
        "dates": ["start_date", "renewal_date"],
        "timestamps": ["created_at"],
        "numbers": ["monthly_fee_eur"],
        "bools": [],
    },
    "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": {
        "pk": "project_id",
        "cols": ["project_id", "client_id", "deliverable", "tier", "checklist",
                 "go_live_date", "status"],
        "dates": ["go_live_date"],
        "timestamps": [],
        "numbers": [],
        "bools": [],
    },
    "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": {
        "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": {
        "pk": "id",
        "cols": ["ts", "workflow", "client_id", "action", "detail", "result"],
        "dates": [],
        "timestamps": ["ts"],
        "numbers": [],
        "bools": [],
    },
    "credentials": {
        # Secrets stay in Postgres only.
        "pk": "cred_id",
        "cols": ["cred_id", "client_id", "label", "username", "secret", "notes", "created_at"],
        "dates": [],
        "timestamps": ["created_at"],
        "numbers": [],
        "bools": [],
    },
}


def connect():
    return psycopg.connect(DATABASE_URL, row_factory=dict_row)


# ---- coercion: JSON request 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()
    # ISO-8601 / RFC3339 (incl. trailing Z and +hh:mm offsets — what Google
    # Calendar / booking tools emit). fromisoformat handles Z on Python 3.11+.
    try:
        return datetime.fromisoformat(s.replace("Z", "+00:00"))
    except ValueError:
        pass
    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