From e59996a3859592a89dee8c2494b97dee62243784 Mon Sep 17 00:00:00 2001 From: rogalik27 Date: Thu, 25 Jun 2026 09:26:42 +0200 Subject: [PATCH] Back office: one-way DB -> Sheets mirror - After each mutation, the affected tab + activity_log are pushed to Sheets by a single serialized background worker (Postgres stays source of truth; a mirror failure never fails the DB write). Concurrency race fixed by serializing through one worker; PYTHONUNBUFFERED for visible logs. - _cell() formats dates/timestamps/numbers/bools and neutralises formula injection. POST /api/sync does a full DB->Sheets resync of every tab. Verified: add/edit/delete propagate to the Sheet; full resync aligns all tabs; DB and Sheet consistent after cleanup. Co-Authored-By: Claude Opus 4.8 --- backoffice/app/Dockerfile | 1 + backoffice/app/app.py | 95 +++++++++++++++++++++++++++++++++++++-- 2 files changed, 93 insertions(+), 3 deletions(-) diff --git a/backoffice/app/Dockerfile b/backoffice/app/Dockerfile index 60524c3..5194a2e 100644 --- a/backoffice/app/Dockerfile +++ b/backoffice/app/Dockerfile @@ -1,5 +1,6 @@ FROM python:3.12-slim +ENV PYTHONUNBUFFERED=1 WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt diff --git a/backoffice/app/app.py b/backoffice/app/app.py index 52616ad..54ccd3c 100644 --- a/backoffice/app/app.py +++ b/backoffice/app/app.py @@ -8,18 +8,85 @@ 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 +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 app = Flask(__name__, static_folder="static", static_url_path="") CRM_TOKEN = os.environ.get("CRM_API_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): + if SH is None: + return 0 + spec = db.TABLES[entity] + 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: + _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: @@ -117,13 +184,13 @@ def add_entity(entity): 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["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.utcnow() + 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( @@ -136,6 +203,8 @@ def add_entity(entity): [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 @@ -161,6 +230,8 @@ def edit_entity(entity, ident): 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}) @@ -178,9 +249,27 @@ def delete_entity(entity, ident): 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}) + + @app.get("/") def index(): return Response(INDEX_HTML, mimetype="text/html")