"""smb-crm back-office service. Postgres is the source of truth. This serves the operator dashboard + a small JSON API (read now; add/edit/delete and the DB->Sheets mirror layered on next). Browser access is gated by Caddy basic-auth on onboard.mivanchenko.de; the machine-to-machine ingest path (n8n) is gated by the CRM_API_TOKEN header. """ import os import re import time from datetime import datetime, date from decimal import Decimal from flask import Flask, jsonify, request, Response from waitress import serve import db app = Flask(__name__, static_folder="static", static_url_path="") CRM_TOKEN = os.environ.get("CRM_API_TOKEN", "") # Static dashboard, with the CRM token injected so the (basic-auth-gated) # operator page can call the token-protected mutation endpoints. with open(os.path.join(os.path.dirname(__file__), "static", "index.html")) as _f: INDEX_HTML = _f.read().replace("__CRM_TOKEN__", CRM_TOKEN) def authed(): return bool(CRM_TOKEN) and request.headers.get("X-CRM-Token") == CRM_TOKEN def log_activity(cur, client_id, action, detail, result="ok"): cur.execute( "INSERT INTO activity_log (ts, workflow, client_id, action, detail, result) " "VALUES (now(), %s, %s, %s, %s, %s)", ("back-office", client_id, action, detail, result)) LIST_ORDER = { "leads": "received_at DESC NULLS LAST", "clients": "client_id", "projects": "project_id", "bookings": "start_time DESC NULLS LAST", "invoices": "issued_date DESC NULLS LAST", "activity_log": "ts DESC NULLS LAST", } def jsonable(v): if isinstance(v, (datetime, date)): return v.isoformat() if isinstance(v, Decimal): return float(v) return v def rows_json(rows): return [{k: jsonable(v) for k, v in r.items()} for r in rows] @app.get("/healthz") def healthz(): try: with db.connect() as conn, conn.cursor() as cur: cur.execute("SELECT 1") cur.fetchone() return Response("ok\n", mimetype="text/plain") except Exception as e: # noqa: BLE001 return Response(f"db error: {e}\n", status=500, mimetype="text/plain") @app.get("/api/") def list_entity(entity): if entity not in db.TABLES: return jsonify({"error": "unknown entity"}), 404 order = LIST_ORDER.get(entity, db.TABLES[entity]["pk"]) with db.connect() as conn, conn.cursor() as cur: cur.execute(f"SELECT * FROM {entity} ORDER BY {order}") rows = cur.fetchall() return jsonify({"entity": entity, "count": len(rows), "rows": rows_json(rows)}) def compute_renewal(start, cycle): """Mirror the onboarding workflow: monthly -> +1 month, yearly -> +1 year.""" cycle = (cycle or "monthly").lower() if cycle == "monthly": m = start.month % 12 + 1 y = start.year + (1 if start.month == 12 else 0) d = min(start.day, [31, 29 if y % 4 == 0 and (y % 100 or not y % 400) else 28, 31, 30, 31, 30, 31, 31, 30, 31, 30, 31][m - 1]) return date(y, m, d) if cycle == "yearly": return date(start.year + 1, start.month, start.day) return None def next_client_id(cur): cur.execute("SELECT client_id FROM clients") mx = 0 for r in cur.fetchall(): m = re.match(r"C-(\d+)$", r["client_id"] or "") if m: mx = max(mx, int(m.group(1))) return "C-%04d" % (mx + 1) @app.post("/api/") def add_entity(entity): if entity not in db.TABLES: return jsonify({"error": "unknown entity"}), 404 if not authed(): return jsonify({"error": "forbidden"}), 403 spec = db.TABLES[entity] pk = spec["cols"][0] if entity == "activity_log" else spec["pk"] body = request.get_json(force=True, silent=True) or {} row = db.coerce_row(entity, body) with db.connect() as conn, conn.cursor() as cur: if entity == "leads": row["lead_id"] = row.get("lead_id") or "L-" + str(int(time.time() * 1000)) row["received_at"] = row.get("received_at") or datetime.utcnow() row["status"] = row.get("status") or "new" if row.get("notified") is None: row["notified"] = False elif entity == "clients": row["client_id"] = row.get("client_id") or next_client_id(cur) row["created_at"] = row.get("created_at") or datetime.utcnow() row["status"] = row.get("status") or "lead" if not row.get("renewal_date") and row.get("start_date"): row["renewal_date"] = compute_renewal( row["start_date"], row.get("billing_cycle")) elif not row.get(pk): return jsonify({"error": f"{pk} required"}), 400 cols = spec["cols"] ph = ", ".join(["%s"] * len(cols)) cur.execute(f"INSERT INTO {entity} ({', '.join(cols)}) VALUES ({ph})", [row.get(c) for c in cols]) log_activity(cur, row.get("client_id"), f"add {entity}", f"{pk}={row.get(pk)}") conn.commit() return jsonify({"added": row.get(pk)}), 201 @app.patch("/api//") def edit_entity(entity, ident): if entity not in db.TABLES: return jsonify({"error": "unknown entity"}), 404 if not authed(): return jsonify({"error": "forbidden"}), 403 spec = db.TABLES[entity] pk = spec["pk"] body = request.get_json(force=True, silent=True) or {} setcols = [c for c in body if c in spec["cols"] and c != pk] if not setcols: return jsonify({"error": "no editable fields"}), 400 typed = db.coerce_row(entity, body) setsql = ", ".join(f"{c} = %s" for c in setcols) vals = [typed[c] for c in setcols] + [ident] with db.connect() as conn, conn.cursor() as cur: cur.execute(f"UPDATE {entity} SET {setsql} WHERE {pk} = %s RETURNING {pk}", vals) if not cur.fetchone(): return jsonify({"error": "not found"}), 404 log_activity(cur, ident if entity == "clients" else None, f"edit {entity}", f"{pk}={ident}: {', '.join(setcols)}") conn.commit() return jsonify({"updated": ident, "fields": setcols}) @app.delete("/api//") def delete_entity(entity, ident): if entity not in db.TABLES: return jsonify({"error": "unknown entity"}), 404 if not authed(): return jsonify({"error": "forbidden"}), 403 pk = db.TABLES[entity]["pk"] with db.connect() as conn, conn.cursor() as cur: cur.execute(f"DELETE FROM {entity} WHERE {pk} = %s RETURNING {pk}", (ident,)) if not cur.fetchone(): return jsonify({"error": "not found"}), 404 log_activity(cur, ident if entity == "clients" else None, f"delete {entity}", f"{pk}={ident}") conn.commit() return jsonify({"deleted": ident}) @app.get("/") def index(): return Response(INDEX_HTML, mimetype="text/html") if __name__ == "__main__": serve(app, host="0.0.0.0", port=8080)