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, 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