"""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 booking_db as bdb import db import owner_mail from sheets import Sheets from booking_api import bp as booking_bp from public_booking import bp as public_booking_bp from manage_booking import bp as manage_booking_bp from owner_auth import bp as owner_auth_bp from owner_booking import bp as owner_booking_bp from owner_settings import bp as owner_settings_bp app = Flask(__name__, static_folder="static", static_url_path="") app.register_blueprint(booking_bp) app.register_blueprint(public_booking_bp) app.register_blueprint(manage_booking_bp) app.register_blueprint(owner_auth_bp) app.register_blueprint(owner_booking_bp) app.register_blueprint(owner_settings_bp) # Dedicated secret for the owner-login session cookie -- deliberately not # shared with CRM_API_TOKEN or BOOKING_TOKEN_SECRET (#19), same reasoning as # booking_api.py's own token secret: rotating one must never silently affect # the others. app.secret_key = os.environ.get("SESSION_SECRET_KEY", "") 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.get("/api/owner_users") def list_owner_users(): """Owner-accounts list for the CRM dashboard's new tab (#19). Requires the CRM token to read, same as credentials -- these rows identify who can log into a client's owner portal.""" if not authed(): return jsonify({"error": "forbidden"}), 403 rows = bdb.list_users() return jsonify({"entity": "owner_users", "count": len(rows), "rows": rows_json(rows)}) @app.post("/api/owner_users//reset-password") def trigger_owner_password_reset(user_id): """Operator-triggered reset (#19): sends the same reset email an owner would send themselves via /owner/forgot-password.""" if not authed(): return jsonify({"error": "forbidden"}), 403 user = bdb.get_user(user_id) if user is None: return jsonify({"error": "not found"}), 404 client = bdb.get_client(user["client_id"]) tok = bdb.create_password_reset_token(user["user_id"]) owner_mail.send_password_reset(client, user, tok["token"]) with db.connect() as conn, conn.cursor() as cur: log_activity(cur, user["client_id"], "owner password reset", f"user_id={user_id} (operator-triggered)") conn.commit() mirror_async("activity_log") return jsonify({"sent": user_id}) @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)