"""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 import queue import threading import traceback from datetime import datetime, date, timezone from decimal import Decimal from flask import Flask, jsonify, request, Response from waitress import serve import db from sheets import Sheets from booking_api import bp as booking_bp from public_booking import bp as public_booking_bp app = Flask(__name__, static_folder="static", static_url_path="") app.register_blueprint(booking_bp) app.register_blueprint(public_booking_bp) CRM_TOKEN = os.environ.get("CRM_API_TOKEN", "") # Separate read-only token for the public iCal feed (calendar apps can't send # the basic-auth header, so the feed is gated by this query token instead). ICS_TOKEN = os.environ.get("ICS_TOKEN", "") # DB -> Sheets one-way mirror. Postgres is the source of truth; the Sheet is a # best-effort projection. A mirror failure never fails the DB write. SH = None try: SH = Sheets(os.environ["GOOGLE_SA_JSON"], os.environ["SHEET_ID"]) except Exception: # noqa: BLE001 traceback.print_exc() def _cell(v): if v is None: return "" if isinstance(v, bool): return "TRUE" if v else "FALSE" if isinstance(v, datetime): return v.strftime("%Y-%m-%d %H:%M") if isinstance(v, date): return v.strftime("%Y-%m-%d") if isinstance(v, Decimal): f = float(v) return str(int(f)) if f == int(f) else str(f) s = str(v) return ("'" + s) if s[:1] in "=+-@" else s # neutralise formula injection def mirror_entity(entity): spec = db.TABLES[entity] if SH is None or spec.get("mirror") is False: return 0 cols = spec["cols"] order = LIST_ORDER.get(entity, spec["pk"]) with db.connect() as conn, conn.cursor() as cur: cur.execute(f"SELECT {', '.join(cols)} FROM {entity} ORDER BY {order}") rows = cur.fetchall() grid = [[_cell(r[c]) for c in cols] for r in rows] SH.overwrite(spec["tab"], cols, grid) return len(grid) # Serialize all mirror writes through one worker so concurrent mutations can't # race on the shared Sheets client / token. _mirror_q = queue.Queue() def _mirror_worker(): while True: entity = _mirror_q.get() try: mirror_entity(entity) except Exception: # noqa: BLE001 print(f"[mirror] {entity} sync failed:", flush=True) traceback.print_exc() finally: _mirror_q.task_done() threading.Thread(target=_mirror_worker, daemon=True).start() def mirror_async(entity): if SH is not None and db.TABLES[entity].get("mirror") is not False: _mirror_q.put(entity) # 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", "credentials": "client_id", } 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 # Credentials carry secrets, not just business metadata — require the # mutation-level token even to read, on top of the Caddy basic-auth every # other entity's (read-only) list relies on alone. if entity == "credentials" and not authed(): return jsonify({"error": "forbidden"}), 403 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.now(timezone.utc) 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.now(timezone.utc) 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 entity == "credentials": row["cred_id"] = row.get("cred_id") or "CR-" + str(int(time.time() * 1000)) row["created_at"] = row.get("created_at") or datetime.now(timezone.utc) 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() mirror_async(entity) mirror_async("activity_log") 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() mirror_async(entity) mirror_async("activity_log") 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() mirror_async(entity) mirror_async("activity_log") return jsonify({"deleted": ident}) @app.post("/api/sync") def sync_all(): """Full DB -> Sheets resync of every tab (manual / alignment).""" if not authed(): return jsonify({"error": "forbidden"}), 403 if SH is None: return jsonify({"error": "sheets unavailable"}), 503 out = {} for entity in db.TABLES: try: out[entity] = mirror_entity(entity) except Exception as e: # noqa: BLE001 out[entity] = f"error: {e}" return jsonify({"synced": out}) def _ics_dt(v): if isinstance(v, datetime): u = v if v.tzinfo else v.replace(tzinfo=timezone.utc) return u.astimezone(timezone.utc).strftime("%Y%m%dT%H%M%SZ") return None def _ics_esc(t): return (str(t or "").replace("\\", "\\\\").replace(";", "\\;") .replace(",", "\\,").replace("\n", "\\n")) @app.get("/api/bookings.ics") def bookings_ics(): """Read-only iCal feed for Apple/Google Calendar subscription. Gated by the ICS_TOKEN query param (no header auth, so calendar apps can fetch it).""" if not ICS_TOKEN or request.args.get("token") != ICS_TOKEN: return Response("forbidden\n", status=403, mimetype="text/plain") cid = request.args.get("client_id") with db.connect() as conn, conn.cursor() as cur: if cid: cur.execute("SELECT * FROM bookings WHERE client_id = %s ORDER BY start_time", (cid,)) else: cur.execute("SELECT * FROM bookings ORDER BY start_time") rows = cur.fetchall() now = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ") out = ["BEGIN:VCALENDAR", "VERSION:2.0", "PRODID:-//smb-crm//bookings//DE", "CALSCALE:GREGORIAN", "METHOD:PUBLISH", "X-WR-CALNAME:" + _ics_esc("Buchungen " + cid if cid else "Buchungen")] for r in rows: st = _ics_dt(r.get("start_time")) if not st: continue summary = r.get("service") or "Termin" if r.get("customer_name"): summary += " – " + r["customer_name"] out += ["BEGIN:VEVENT", "UID:%s@smb-crm" % (r.get("booking_id") or now), "DTSTAMP:%s" % (_ics_dt(r.get("created_at")) or now), "DTSTART:%s" % st] en = _ics_dt(r.get("end_time")) if en: out.append("DTEND:%s" % en) out.append("SUMMARY:" + _ics_esc(summary)) if r.get("customer_contact"): out.append("DESCRIPTION:" + _ics_esc("Kontakt: " + r["customer_contact"])) out += ["STATUS:CONFIRMED", "END:VEVENT"] out.append("END:VCALENDAR") return Response("\r\n".join(out) + "\r\n", mimetype="text/calendar") @app.get("/") def index(): return Response(INDEX_HTML, mimetype="text/html") if __name__ == "__main__": serve(app, host="0.0.0.0", port=8080)