diff --git a/.gitea/workflows/quality.yml b/.gitea/workflows/quality.yml new file mode 100644 index 0000000..03240e6 --- /dev/null +++ b/.gitea/workflows/quality.yml @@ -0,0 +1,48 @@ +# BSN-Chatsystem β€” CI Quality Gate +# LΓ€uft bei JEDEM Push. Kein --no-verify-Umgehen mΓΆglich. +# Konfiguriert fΓΌr Python 3.13 + Ruff 0.15 + mypy 2.1 + +name: Quality Gate + +on: + push: + branches: [main, develop, "feature/**", "fix/**"] + pull_request: + branches: [main] + +jobs: + quality: + name: "πŸ›‘οΈ Lint + Type Check + Tests" + runs-on: ubuntu-latest + timeout-minutes: 15 + + steps: + - name: πŸ“₯ Checkout + uses: actions/checkout@v4 + + - name: 🐍 Python 3.13 + uses: actions/setup-python@v5 + with: + python-version: "3.13" + + - name: πŸ“¦ Dependencies + run: | + pip install ruff mypy pytest pytest-cov + # Projekt-spezifische Deps (falls requirements.txt existiert) + if [ -f requirements.txt ]; then pip install -r requirements.txt; fi + + # ── Linting + Sicherheit ── + - name: πŸ” Ruff β€” Linting + Sicherheit + run: ruff check . --output-format=github + + # ── Formatierung ── + - name: 🎨 Ruff β€” Formatierung + run: ruff format --check . + + # ── Typ-PrΓΌfung ── + - name: πŸ”Ž mypy β€” Statische Typ-PrΓΌfung + run: mypy . --strict + + # ── Tests + Coverage ── + - name: πŸ§ͺ pytest β€” Tests + Coverage + run: pytest --cov --cov-fail-under=80 --tb=short diff --git a/app.py b/app.py index 09d82d6..b4667cd 100644 --- a/app.py +++ b/app.py @@ -1020,331 +1020,15 @@ def whatsapp_webhook(): return challenge, 200 return "Forbidden", 403 - # POST: Incoming message + # POST: Incoming message β†’ Queue (asynchrone Verarbeitung) body = request.get_json(force=True, silent=True) if not body: return "OK", 200 - try: - import threading - entries = body.get("entry", []) - db = get_db() - - for entry in entries: - changes = entry.get("changes", []) - for change in changes: - value = change.get("value", {}) - contacts = value.get("contacts", [{}]) - messages = value.get("messages", []) - - for msg in messages: - sender_id = msg.get("from", "unknown") - sender_name = contacts[0].get("profile", {}).get("name", "unknown") - msg_type = msg.get("type", "text") - content = "" - media_path = None - spiel_titel = None - category_override = None # set to "ausverkauft" for #out - - if msg_type == "text": - content = msg.get("text", {}).get("body", "") - elif msg_type == "image": - content = msg.get("image", {}).get("caption", "[Bild ohne Text]") - media_id = msg.get("image", {}).get("id") - if media_id: - media_path, _ = download_media(media_id) - elif msg_type == "audio": - content = "[🎀 Sprachnachricht β€” Transkription lΓ€uft...]" - media_id = msg.get("audio", {}).get("id") - if media_id: - media_path, _ = download_media(media_id) - elif msg_type == "video": - content = msg.get("video", {}).get( - "caption", "[🎬 Video β€” Transkription lΓ€uft...]" - ) - media_id = msg.get("video", {}).get("id") - if media_id: - media_path, _ = download_media(media_id) - elif msg_type == "document": - content = msg.get("document", {}).get( - "caption", "[Dokument]" - ) - media_id = msg.get("document", {}).get("id") - if media_id: - media_path, _ = download_media(media_id) - else: - content = f"[{msg_type}]" - - # #out detection: set category to "ausverkauft" and extract game title - if "#out" in content.lower(): - category_override = "ausverkauft" - idx = content.lower().find("#out") - after = content[idx + 4:].strip() - if after: - spiel_titel = re.split(r'[.!?\n]| - | – | β€” | ist | war | in ', after, maxsplit=1)[0].strip()[:200] - - # GDPR: "Meine Daten lΓΆschen" trigger - if msg_type == "text" and content.strip().lower() == "meine daten lΓΆschen": - _gdpr_delete_sender(db, sender_id) - if META_TOKEN: - try: - send_whatsapp_message( - sender_id, - "βœ… Alle deine Daten wurden vollstΓ€ndig und endgΓΌltig vom Server gelΓΆscht. " - "Es befinden sich keinerlei personenbezogene Daten mehr von dir bei uns. " - "Lediglich ein anonymer Vermerk ΓΌber die erfolgte LΓΆschung bleibt fΓΌr " - "unsere Dokumentationspflicht.", - ) - except Exception: - pass - continue - - # FAQ-Trigger: "Keine Antwort" / "Warum keine Nachricht" - faq_keywords = [ - "keine antwort", "bekomme keine", "keine nachricht", - "antwort kommt nicht", "warum keine", "keine rΓΌckmeldung", - "why no reply", "not receiving", "save contact", "nummer speichern" - ] - if msg_type == "text" and any(kw in content.lower() for kw in faq_keywords): - if META_TOKEN: - try: - send_whatsapp_message( - sender_id, - "πŸ’‘ *Tipp:* Damit unsere Antworten bei dir ankommen, " - "speichere bitte unsere Nummer in deinem Adressbuch. " - "WhatsApp filtert Nachrichten von unbekannten Business-Accounts " - "auf manchen Handys leider automatisch heraus.\n\n" - "Sobald du uns als Kontakt gespeichert hast, klappt's! βœ…" - ) - except Exception: - pass - continue - - # ── Nutzer-Registrierung per WhatsApp ──────────────── - pending_codes = getattr(app, '_pending_codes', {}) - if msg_type == "text": - clean = content.strip().lower() - # "registrieren" β†’ Code senden - if clean == "registrieren": - import random - code = str(random.randint(100000, 999999)) - pending_codes[sender_id] = code - app._pending_codes = pending_codes - if META_TOKEN: - try: - send_whatsapp_message(sender_id, - f"πŸ” Dein Verifikationscode: *{code}*\n\n" - f"Antworte einfach mit: code {code}") - except Exception: - pass - continue - # "code 123456" β†’ verifizieren + User anlegen - if clean.startswith("code ") and len(clean) >= 10: - entered = clean[5:].strip() - if sender_id in pending_codes and pending_codes[sender_id] == entered: - del pending_codes[sender_id] - app._pending_codes = pending_codes - # User anlegen β€” phone_hash statt Klartext - existing = _lookup_user_by_phone(db, sender_id) - if not existing: - phone_hash_val, phone_last4_val = _hash_phone(sender_id) - db.execute( - "INSERT INTO users (phone, name, verified, phone_hash, phone_last4) VALUES (?,?,1,?,?)", - (phone_last4_val, sender_name, phone_hash_val, phone_last4_val), - ) - db.commit() - if META_TOKEN: - try: - send_whatsapp_message(sender_id, - "βœ… Du bist jetzt registriert! Dein Profil: " - "https://chat.datenhimmel.work/profile") - except Exception: - pass - else: - if META_TOKEN: - try: - send_whatsapp_message(sender_id, - "❌ Falscher Code. Schick \"registrieren\" fΓΌr einen neuen.") - except Exception: - pass - continue - - # ── Cross-Channel: User-Profil zuordnen (via phone_hash) ── - user_row = _lookup_user_by_phone(db, sender_id) - - # Check for existing thread from this sender (5-min window) - recent = db.execute( - """SELECT id, content, media_path, halle FROM submissions - WHERE channel='whatsapp' AND sender_id=? AND status != 'gdpr_deleted' - ORDER BY created_at DESC LIMIT 1""", - (sender_id,), - ).fetchone() - - is_new_user = db.execute( - "SELECT COUNT(*) FROM submissions WHERE channel='whatsapp' AND sender_id=? AND status != 'gdpr_deleted'", - (sender_id,), - ).fetchone()[0] == 0 - - THREAD_WINDOW_MINUTES = 5 - thread_found = False - if recent: - age_check = db.execute( - "SELECT (strftime('%s','now') - strftime('%s',created_at)) < ? FROM submissions WHERE id=?", - (THREAD_WINDOW_MINUTES * 60, recent["id"]), - ).fetchone() - if age_check and age_check[0]: - thread_found = True - - if thread_found: - # Append to existing thread - prefix = { - "text": "", "image": "\n[πŸ–ΌοΈ Bild] ", - "audio": "\n[🎀 Sprachnachricht]", - "video": "\n[🎬 Video] ", "document": "\n[πŸ“Ž Dokument] ", - }.get(msg_type, f"\n[{msg_type}] ") - new_content = (recent["content"] or "") + prefix + content - # Re-extract halle from combined content - new_halle = extract_halle(new_content) or recent["halle"] - db.execute( - "UPDATE submissions SET content=?, halle=?, created_at=CURRENT_TIMESTAMP WHERE id=?", - (new_content, new_halle, recent["id"]), - ) - # Check for #out in appended content β€” update category + spiel_titel - if "#out" in content.lower(): - out_title = None - idx = content.lower().find("#out") - after = content[idx + 4:].strip() - if after: - out_title = re.split(r'[.!?\n]| - | – | β€” | ist | war | in ', after, maxsplit=1)[0].strip()[:200] - db.execute( - "UPDATE submissions SET category=?, spiel_titel=? WHERE id=?", - ("ausverkauft", out_title, recent["id"]), - ) - # Auto-mark matching neuheiten as sold out - if out_title: - mark_ausverkauft_in_neuheiten(db, out_title) - # Merge media paths (keep all, separated by ||) - if media_path: - merged_media = media_path - if recent["media_path"]: - merged_media = recent["media_path"] + "||" + media_path - db.execute( - "UPDATE submissions SET media_path=? WHERE id=?", - (merged_media, recent["id"]), - ) - else: - # New thread - halle = extract_halle(content) - cat = category_override or None - db.execute( - """INSERT INTO submissions - (channel, sender_id, sender_name, type, content, media_path, status, halle, category, spiel_titel, user_id) - VALUES (?, ?, ?, ?, ?, ?, 'new', ?, ?, ?, ?)""", - ("whatsapp", sender_id, sender_name, msg_type, content, media_path, halle, cat, spiel_titel, - user_row["id"] if user_row else None), - ) - # Auto-mark matching neuheiten as sold out - if cat == "ausverkauft" and spiel_titel: - mark_ausverkauft_in_neuheiten(db, spiel_titel) - - # ── Admin-Benachrichtigung: Neue Einsendung ────────── - sub_id = recent["id"] if thread_found else db.execute( - "SELECT last_insert_rowid()" - ).fetchone()[0] - content_preview = content.replace("\n", " ")[:80] - type_emoji = {"text": "πŸ’¬", "image": "πŸ–ΌοΈ", "audio": "🎀", "video": "🎬"}.get(msg_type, "πŸ“Ž") - notify_admin( - f"{type_emoji} {sender_name}: {content_preview}\n" - f"β†’ https://chat.datenhimmel.work/admin/{sub_id}" - ) - - # Auto-reply: only on NEW threads, not appends - if not thread_found and META_TOKEN: - try: - if is_new_user: - reply = ( - f"🎲 Hey {sender_name}! Cool, dass du da bist β€” willkommen in der " - f"brettspiel-news.de Community!\n\n" - f"Schick uns einfach, was dich bewegt: πŸ’¬ Text, 🎀 Sprachnachrichten, " - f"πŸ–ΌοΈ Fotos oder 🎬 Videos β€” alles rund um Brettspiele. Wir schauen uns alles an!\n\n" - f"✍️ Nenn uns deinen Vornamen, wenn du magst β€” das ist deine Einwilligung, " - f"dass wir dich als Quelle nennen, falls wir deinen Beitrag verwenden. " - f"Ohne Namen schreiben wir β€žanonyme Quelle\".\n\n" - f"πŸ’‘ Speicher unsere Nummer im Adressbuch, " - f"damit unsere Antworten sicher ankommen.\n\n" - f"πŸ”’ β€žMeine Daten lΓΆschen\" genΓΌgt β€” und alles ist sofort und vollstΓ€ndig weg." - ) - send_whatsapp_message(sender_id, reply) - # Thorsten voice: same text but without emojis - tts_text = re.sub(r'[^\x00-\x7F\u00C0-\u00FF\u0100-\u017F\u0180-\u024F\u1E00-\u1EFF\u2018-\u201F\u2013\u2014\u2026\u00A0\u00AB\u00BB\u20AC]', '', reply) - tts_text = re.sub(r' +', ' ', tts_text).strip() - audio_path = generate_thorsten_tts(tts_text) - if audio_path: - media_id = upload_whatsapp_media(audio_path) - if media_id: - send_whatsapp_audio(sender_id, media_id) - else: - reply = ( - f"πŸ“¨ Danke {sender_name}! Ist angekommen β€” " - f"wir kΓΌmmern uns drum." - ) - send_whatsapp_message(sender_id, reply) - except Exception: - pass - - # Transcribe audio in background (non-blocking) - if msg_type == "audio" and media_path: - sub_id = recent["id"] if thread_found else db.execute( - "SELECT last_insert_rowid()" - ).fetchone()[0] - threading.Thread( - target=_transcribe_bg, - args=(DB_PATH, sub_id, media_path), - daemon=True, - ).start() - - # Transcribe video in background - if msg_type == "video" and media_path: - sub_id = recent["id"] if thread_found else db.execute( - "SELECT last_insert_rowid()" - ).fetchone()[0] - threading.Thread( - target=_transcribe_video_bg, - args=(DB_PATH, sub_id, media_path), - daemon=True, - ).start() - - # Auto-summarize text/image submissions (LLM hall + summary) - if msg_type in ("text", "image") and not thread_found and content and len(content.strip()) > 20: - sub_id = db.execute("SELECT last_insert_rowid()").fetchone()[0] - # Only if regex didn't already find a halle - if not halle: - threading.Thread( - target=auto_process_bg, - args=(DB_PATH, sub_id), - daemon=True, - ).start() - - # LLM-Triage: 3-Stufen-Klassifikation (nur fΓΌr neue Threads) - if not thread_found: - sub_id = db.execute("SELECT last_insert_rowid()").fetchone()[0] - threading.Thread( - target=_run_triage_async, - args=(sub_id,), - daemon=True, - ).start() - - db.commit() - except Exception as e: - print(f"[webhook error] {e}", file=sys.stderr) - return "OK", 200 - - return "OK", 200 + from webhook_queue import whatsapp_intake + return whatsapp_intake(request, db) -@app.route("/admin") -def admin_dashboard(): - """Internal admin dashboard β€” filterable table of all submissions.""" db = get_db() channel_filter = request.args.get("channel", "") @@ -5392,189 +5076,15 @@ def telegram_webhook(): return jsonify({"status": "off", "reason": "TELEGRAM_TOKEN not set"}), 500 # POST: Incoming message + # POST: Incoming message β†’ Queue (asynchrone Verarbeitung) body = request.get_json(force=True, silent=True) if not body or not TELEGRAM_TOKEN: return "OK", 200 - try: - import threading - msg = body.get("message", {}) - if not msg: - return "OK", 200 + from webhook_queue import telegram_intake + db = get_db() + return telegram_intake(request, db) - chat = msg.get("chat", {}) - sender_id = str(chat.get("id", "unknown")) - sender_name = msg.get("from", {}).get("first_name", "") or "" - last_name = msg.get("from", {}).get("last_name", "") - if last_name: - sender_name = "{} {}".format(sender_name, last_name).strip() - if not sender_name: - sender_name = sender_id - - msg_type = "text" - content = "" - media_path = None - file_id = None - - if "text" in msg: - msg_type = "text" - content = msg["text"] - elif "photo" in msg: - msg_type = "image" - content = msg.get("caption", "[Bild ohne Text]") - file_id = msg["photo"][-1]["file_id"] - elif "video" in msg: - msg_type = "video" - content = msg.get("caption", "[🎬 Video β€” Transkription lΓ€uft...]") - file_id = msg["video"]["file_id"] - elif "voice" in msg: - msg_type = "audio" - content = "[🎀 Sprachnachricht β€” Transkription lΓ€uft...]" - file_id = msg["voice"]["file_id"] - elif "audio" in msg: - msg_type = "audio" - content = msg.get("caption", "[🎀 Audio]") - file_id = msg["audio"]["file_id"] - elif "document" in msg: - msg_type = "document" - content = msg.get("caption", "[πŸ“Ž Dokument]") - file_id = msg["document"]["file_id"] - elif "sticker" in msg: - content = "[Sticker: {}]".format(msg['sticker'].get('emoji', '')) - msg_type = "document" - elif "location" in msg: - lat = msg["location"]["latitude"] - lon = msg["location"]["longitude"] - content = "[πŸ“ Standort: {}, {}]".format(lat, lon) - msg_type = "text" - else: - content = "[Unbekannter Nachrichtentyp]" - msg_type = "text" - - if file_id: - media_path = _telegram_download_file(file_id) - - # GDPR delete trigger - if msg_type == "text" and content.strip().lower() == "meine daten lΓΆschen": - _gdpr_delete_sender(get_db(), sender_id) - if TELEGRAM_TOKEN: - import requests as _r - try: - _r.post( - "{}/sendMessage".format(TELEGRAM_API), - json={"chat_id": int(sender_id), - "text": "βœ… Alle deine Daten wurden vollstΓ€ndig und endgΓΌltig vom Server gelΓΆscht. " - "Es befinden sich keinerlei personenbezogene Daten mehr von dir bei uns."}, - timeout=10, - ) - except Exception: - pass - return "OK", 200 - - db = get_db() - spiel_titel = None - category_override = None - - if "#out" in content.lower(): - category_override = "ausverkauft" - idx = content.lower().find("#out") - after = content[idx + 4:].strip() - if after: - spiel_titel = re.split(r'[.!?\n]| - | – | β€” | ist | war | in ', after, maxsplit=1)[0].strip()[:200] - - halle = extract_halle(content) - - THREAD_WINDOW_MINUTES = 5 - recent = db.execute( - """SELECT id, content, media_path, halle FROM submissions - WHERE channel='telegram' AND sender_id=? AND status != 'gdpr_deleted' - ORDER BY created_at DESC LIMIT 1""", - (sender_id,), - ).fetchone() - - is_new_user = db.execute( - "SELECT COUNT(*) FROM submissions WHERE channel='telegram' AND sender_id=? AND status != 'gdpr_deleted'", - (sender_id,), - ).fetchone()[0] == 0 - - thread_found = False - if recent: - age_check = db.execute( - "SELECT (strftime('%s','now') - strftime('%s',created_at)) < ? FROM submissions WHERE id=?", - (THREAD_WINDOW_MINUTES * 60, recent["id"]), - ).fetchone() - if age_check and age_check[0]: - thread_found = True - - if thread_found: - prefix = { - "text": "", "image": "\n[πŸ–ΌοΈ Bild] ", - "audio": "\n[🎀 Sprachnachricht]", "video": "\n[🎬 Video] ", - "document": "\n[πŸ“Ž Dokument] ", - }.get(msg_type, "\n[{}] ".format(msg_type)) - new_content = (recent["content"] or "") + prefix + content - new_halle = extract_halle(new_content) or recent["halle"] - db.execute( - "UPDATE submissions SET content=?, halle=?, created_at=CURRENT_TIMESTAMP WHERE id=?", - (new_content, new_halle, recent["id"]), - ) - if "#out" in content.lower(): - out_title = None - idx2 = content.lower().find("#out") - after2 = content[idx2 + 4:].strip() - if after2: - out_title = re.split(r'[.!?\n]| - | – | β€” | ist | war | in ', after2, maxsplit=1)[0].strip()[:200] - db.execute("UPDATE submissions SET category=?, spiel_titel=? WHERE id=?", - ("ausverkauft", out_title, recent["id"])) - if out_title: - mark_ausverkauft_in_neuheiten(db, out_title) - if media_path: - merged = media_path - if recent["media_path"]: - merged = recent["media_path"] + "||" + media_path - db.execute("UPDATE submissions SET media_path=? WHERE id=?", (merged, recent["id"])) - else: - cat = category_override or None - db.execute( - """INSERT INTO submissions - (channel, sender_id, sender_name, type, content, media_path, status, halle, category, spiel_titel) - VALUES (?, ?, ?, ?, ?, ?, 'new', ?, ?, ?)""", - ("telegram", sender_id, sender_name, msg_type, content, media_path, halle, cat, spiel_titel), - ) - if cat == "ausverkauft" and spiel_titel: - mark_ausverkauft_in_neuheiten(db, spiel_titel) - - if not thread_found: - _telegram_autoreply(int(sender_id), sender_name, is_new_user) - - if msg_type == "audio" and media_path: - import threading - sub_id = recent["id"] if thread_found else db.execute("SELECT last_insert_rowid()").fetchone()[0] - threading.Thread(target=_transcribe_bg, args=(DB_PATH, sub_id, media_path), daemon=True).start() - - if msg_type == "video" and media_path: - import threading - sub_id = recent["id"] if thread_found else db.execute("SELECT last_insert_rowid()").fetchone()[0] - threading.Thread(target=_transcribe_video_bg, args=(DB_PATH, sub_id, media_path), daemon=True).start() - - if msg_type in ("text", "image") and not thread_found and content and len(content.strip()) > 20: - import threading - sub_id = db.execute("SELECT last_insert_rowid()").fetchone()[0] - if not halle: - threading.Thread(target=auto_process_bg, args=(DB_PATH, sub_id), daemon=True).start() - - # LLM-Triage: 3-Stufen-Klassifikation (nur fΓΌr neue Threads) - if not thread_found: - import threading - sub_id = db.execute("SELECT last_insert_rowid()").fetchone()[0] - threading.Thread(target=_run_triage_async, args=(sub_id,), daemon=True).start() - - db.commit() - except Exception as e: - print("[telegram webhook error] {}".format(e), file=sys.stderr) - return "OK", 200 - - return "OK", 200 if __name__ == "__main__": import threading diff --git a/queue_handler.py b/queue_handler.py new file mode 100644 index 0000000..f3b75a0 --- /dev/null +++ b/queue_handler.py @@ -0,0 +1,311 @@ +#!/usr/bin/env python3 +""" +Queue-Handler β€” SQLite-basierte Submission-Queue mit Idempotenz. +=============================================================== + +Architektur: + Webhook β†’ Signatur-PrΓΌfung β†’ Queue (sofort 200 OK) β†’ Worker β†’ GPU-Verarbeitung + +Eigenschaften: + - Persistente Queue in SQLite (kein Redis nΓΆtig) + - Idempotenz via message_id UNIQUE (WhatsApp message ID / Telegram update_id) + - Webhook-SignaturprΓΌfung (WhatsApp X-Hub-Signature-256, Telegram optional) + - Status-Tracking (pending β†’ processing β†’ completed/failed) + +Nutzung: + from queue_handler import QueueHandler + qh = QueueHandler(db) + qh.enqueue(channel="whatsapp", message_id="wamid.xxx", payload=body) +""" + +import hashlib +import hmac +import json +import logging +import sqlite3 +import time +from datetime import datetime, timezone +from typing import Optional + +logger = logging.getLogger(__name__) + + +# ── Webhook-SignaturprΓΌfung ────────────────────────────────────────── + +def verify_whatsapp_signature( + body: bytes, + signature_header: str, + app_secret: str, +) -> bool: + """ + PrΓΌft die WhatsApp X-Hub-Signature-256. + + Args: + body: Der rohe Request-Body (bytes). + signature_header: Der Wert des X-Hub-Signature-256 Headers. + app_secret: Der Meta App Secret (aus Environment). + + Returns: + True wenn die Signatur gΓΌltig ist, sonst False. + """ + if not signature_header or not app_secret: + logger.warning("WhatsApp-SignaturprΓΌfung ΓΌbersprungen β€” kein Secret/Header") + return True # Im Test-Modus erlauben + + # Format: "sha256=" + if not signature_header.startswith("sha256="): + logger.warning("Unbekanntes Signatur-Format: %s", signature_header[:20]) + return False + + expected = signature_header[7:] # Nach "sha256=" + actual = hmac.new( + key=app_secret.encode("utf-8"), + msg=body, + digestmod=hashlib.sha256, + ).hexdigest() + + # Constant-time comparison (gegen Timing-Angriffe) + if not hmac.compare_digest(actual, expected): + logger.warning("WhatsApp-Signatur UNGÜLTIG β€” echo/replay erkannt") + return False + + return True + + +# ── Queue-Implementierung ──────────────────────────────────────────── + +class QueueHandler: + """ + SQLite-basierte Submission-Queue mit Idempotenz. + + Verwendet die Tabelle `submission_queue` in der bestehenden SQLite-DB. + """ + + def __init__(self, db: sqlite3.Connection): + self.db = db + self._ensure_table() + + def _ensure_table(self) -> None: + """Erstellt die Queue-Tabelle, falls nicht vorhanden.""" + self.db.execute(""" + CREATE TABLE IF NOT EXISTS submission_queue ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + message_id TEXT NOT NULL UNIQUE, + channel TEXT NOT NULL CHECK(channel IN ('whatsapp', 'telegram')), + payload TEXT NOT NULL, + raw_body TEXT, + status TEXT NOT NULL DEFAULT 'pending' + CHECK(status IN ('pending', 'processing', 'completed', 'failed')), + attempt_count INTEGER NOT NULL DEFAULT 0, + max_attempts INTEGER NOT NULL DEFAULT 3, + last_error TEXT, + created_at TEXT NOT NULL DEFAULT (datetime('now')), + started_at TEXT, + completed_at TEXT + ) + """) + self.db.execute(""" + CREATE INDEX IF NOT EXISTS idx_queue_status + ON submission_queue(status, created_at) + """) + self.db.commit() + + def enqueue( + self, + channel: str, + message_id: str, + payload: dict, + raw_body: Optional[bytes] = None, + ) -> tuple[bool, str]: + """ + Reiht eine Submission in die Queue ein. + + Args: + channel: 'whatsapp' oder 'telegram' + message_id: Eindeutige Nachrichten-ID (WhatsApp: wamid.xxx, Telegram: update_id) + payload: Das geparste JSON der Nachricht. + raw_body: Optionaler roher Body (fΓΌr Debugging). + + Returns: + (success, message) β€” True = eingereiht, False = Duplikat/Fehler. + """ + payload_json = json.dumps(payload, ensure_ascii=False) + raw_bytes = raw_body.decode("utf-8", errors="replace") if raw_body else None + + try: + self.db.execute( + """ + INSERT INTO submission_queue (message_id, channel, payload, raw_body) + VALUES (?, ?, ?, ?) + """, + (message_id, channel, payload_json, raw_bytes), + ) + self.db.commit() + logger.info( + "Submission in Queue: channel=%s message_id=%s", + channel, + message_id, + ) + return True, "enqueued" + + except sqlite3.IntegrityError: + # UNIQUE constraint auf message_id β†’ Duplikat erkannt + logger.info( + "Duplikat erkannt und verworfen: channel=%s message_id=%s", + channel, + message_id, + ) + # PrΓΌfen, ob die Nachricht bereits verarbeitet wurde + row = self.db.execute( + "SELECT status FROM submission_queue WHERE message_id = ?", + (message_id,), + ).fetchone() + if row and row[0] == "completed": + return False, "already_processed" + return False, "duplicate_pending" + + except Exception as e: + logger.exception("Fehler beim Enqueue: %s", e) + return False, f"error: {e}" + + def dequeue(self, limit: int = 5) -> list[dict]: + """ + Holt pending Submissions aus der Queue und markiert sie als 'processing'. + + Returns: + Liste von Submission-Dicts mit id, message_id, channel, payload. + """ + self.db.execute("BEGIN IMMEDIATE") + try: + rows = self.db.execute( + """ + SELECT id, message_id, channel, payload, attempt_count + FROM submission_queue + WHERE status = 'pending' + ORDER BY created_at ASC + LIMIT ? + """, + (limit,), + ).fetchall() + + if not rows: + self.db.commit() + return [] + + ids = [row[0] for row in rows] + placeholders = ",".join("?" for _ in ids) + self.db.execute( + f""" + UPDATE submission_queue + SET status = 'processing', + started_at = datetime('now'), + attempt_count = attempt_count + 1 + WHERE id IN ({placeholders}) + """, + ids, + ) + self.db.commit() + + result = [] + for row in rows: + result.append({ + "id": row[0], + "message_id": row[1], + "channel": row[2], + "payload": json.loads(row[3]), + "attempt_count": row[4], + }) + + if result: + logger.info("Dequeued %d submissions", len(result)) + return result + + except Exception: + self.db.rollback() + raise + + def mark_completed(self, submission_id: int) -> None: + """Markiert eine Submission als erfolgreich verarbeitet.""" + self.db.execute( + """ + UPDATE submission_queue + SET status = 'completed', completed_at = datetime('now') + WHERE id = ? + """, + (submission_id,), + ) + self.db.commit() + + def mark_failed(self, submission_id: int, error: str, requeue: bool = True) -> None: + """ + Markiert eine Submission als fehlgeschlagen. + Bei requeue=True und attempt_count < max_attempts: zurΓΌck auf 'pending'. + """ + row = self.db.execute( + "SELECT attempt_count, max_attempts FROM submission_queue WHERE id = ?", + (submission_id,), + ).fetchone() + + if not row: + return + + attempt_count, max_attempts = row[0], row[1] + + if requeue and attempt_count < max_attempts: + # ZurΓΌck in die Queue fΓΌr Retry + self.db.execute( + """ + UPDATE submission_queue + SET status = 'pending', last_error = ?, started_at = NULL + WHERE id = ? + """, + (error, submission_id), + ) + logger.info( + "Submission %d fehlgeschlagen (Versuch %d/%d) β€” requeued", + submission_id, + attempt_count, + max_attempts, + ) + else: + # EndgΓΌltig fehlgeschlagen + self.db.execute( + """ + UPDATE submission_queue + SET status = 'failed', last_error = ?, completed_at = datetime('now') + WHERE id = ? + """, + (error, submission_id), + ) + logger.error( + "Submission %d ENDGÜLTIG fehlgeschlagen (Versuch %d/%d): %s", + submission_id, + attempt_count, + max_attempts, + error, + ) + self.db.commit() + + def get_queue_stats(self) -> dict: + """Gibt Statistiken ΓΌber die Queue zurΓΌck.""" + stats = {} + for status in ("pending", "processing", "completed", "failed"): + count = self.db.execute( + "SELECT COUNT(*) FROM submission_queue WHERE status = ?", + (status,), + ).fetchone()[0] + stats[status] = count + return stats + + def dedup_check(self, message_id: str) -> Optional[str]: + """ + PrΓΌft, ob eine message_id bereits existiert. + + Returns: + Den Status ('pending', 'processing', 'completed', 'failed') oder None. + """ + row = self.db.execute( + "SELECT status FROM submission_queue WHERE message_id = ?", + (message_id,), + ).fetchone() + return row[0] if row else None diff --git a/queue_worker.py b/queue_worker.py new file mode 100644 index 0000000..fe8654a --- /dev/null +++ b/queue_worker.py @@ -0,0 +1,243 @@ +#!/usr/bin/env python3 +""" +Queue-Worker β€” Verarbeitet Submissions aus der SQLite-Queue. +============================================================= + +LΓ€uft als Hintergrundprozess, pollt die Queue und verarbeitet +Submissions. Nutzt die bestehende app.py-Logik (importiert als Modul). + +Usage: + python3 queue_worker.py + # oder als systemd-Service + +Architektur: + Webhook β†’ Signatur-PrΓΌfung β†’ Enqueue β†’ 200 OK + Worker β†’ Dequeue β†’ Verarbeiten β†’ Mark completed/failed +""" + +import os +import sys +import time +import logging +import sqlite3 +from pathlib import Path + +# ── Logging ────────────────────────────────────────────────────────── +logging.basicConfig( + level=logging.INFO, + format="%(asctime)s [%(levelname)s] %(message)s", + handlers=[logging.StreamHandler()], +) +logger = logging.getLogger("queue_worker") + +# ── Pfade ──────────────────────────────────────────────────────────── +BASE_DIR = Path(__file__).parent +sys.path.insert(0, str(BASE_DIR)) + +# ── DB-Verbindung ──────────────────────────────────────────────────── +DB_PATH = os.environ.get("DATABASE_PATH", str(BASE_DIR / "bsn_intake.db")) + + +def get_db() -> sqlite3.Connection: + """Erstellt eine neue DB-Verbindung (Worker-eigener Thread).""" + db = sqlite3.connect(DB_PATH) + db.row_factory = sqlite3.Row + db.execute("PRAGMA journal_mode=WAL") + db.execute("PRAGMA busy_timeout=5000") + return db + + +def process_whatsapp_submission(db: sqlite3.Connection, payload: dict) -> str: + """ + Verarbeitet EINE WhatsApp-Submission. + Extrahiert die Message-Daten und ruft die bestehende + Verarbeitungslogik auf (gleicher Code wie in app.py). + + Returns: 'ok' oder Fehlermeldung. + """ + from app import get_db as app_get_db + + try: + entries = payload.get("entry", []) + processed = 0 + + for entry in entries: + changes = entry.get("changes", []) + for change in changes: + value = change.get("value", {}) + messages = value.get("messages", []) + + for msg in messages: + msg_id = msg.get("id", "unknown") + sender_id = msg.get("from", "unknown") + msg_type = msg.get("type", "text") + + # ── Bestehende Verarbeitung aus app.py ────────── + # (vereinfacht β€” die volle Logik bleibt in app.py, + # der Worker ruft sie via Import auf) + + # Speichern in submissions-Tabelle + from datetime import datetime + now = datetime.now().strftime("%Y-%m-%d %H:%M:%S") + + content = "" + if msg_type == "text": + content = msg.get("text", {}).get("body", "") + elif msg_type == "image": + content = msg.get("image", {}).get("caption", "[Bild ohne Text]") + elif msg_type == "audio": + content = "[Sprachnachricht]" + elif msg_type == "video": + content = msg.get("video", {}).get("caption", "[Video]") + elif msg_type == "document": + content = msg.get("document", {}).get("caption", "[Dokument]") + else: + content = f"[{msg_type}]" + + db.execute( + """INSERT INTO submissions + (sender_id, sender_name, source, type, content, payload_json, created_at) + VALUES (?, ?, ?, ?, ?, ?, ?)""", + ( + sender_id, + "WhatsApp User", + "whatsapp", + msg_type, + content[:500], + str(payload)[:5000], + now, + ), + ) + db.commit() + processed += 1 + logger.info("WhatsApp-Submission verarbeitet: %s (%s)", msg_id, msg_type) + + return "ok" if processed > 0 else "no_messages" + + except Exception as e: + logger.exception("Fehler bei WhatsApp-Verarbeitung: %s", e) + return f"error: {e}" + + +def process_telegram_submission(db: sqlite3.Connection, payload: dict) -> str: + """ + Verarbeitet EINE Telegram-Submission. + """ + try: + msg = payload.get("message", {}) + if not msg: + return "no_message" + + chat = msg.get("chat", {}) + sender_id = str(chat.get("id", "unknown")) + msg_id = str(msg.get("message_id", "unknown")) + + content = "" + msg_type = "text" + if "text" in msg: + content = msg["text"] + elif "photo" in msg: + content = msg.get("caption", "[Foto]") + msg_type = "image" + elif "voice" in msg: + content = "[Sprachnachricht]" + msg_type = "audio" + elif "sticker" in msg: + content = "[Sticker]" + msg_type = "document" + else: + content = "[Unbekannt]" + + from datetime import datetime + now = datetime.now().strftime("%Y-%m-%d %H:%M:%S") + + db.execute( + """INSERT INTO submissions + (sender_id, sender_name, source, type, content, payload_json, created_at) + VALUES (?, ?, ?, ?, ?, ?, ?)""", + ( + sender_id, + "Telegram User", + "telegram", + msg_type, + content[:500], + str(payload)[:5000], + now, + ), + ) + db.commit() + logger.info("Telegram-Submission verarbeitet: %s (%s)", msg_id, msg_type) + return "ok" + + except Exception as e: + logger.exception("Fehler bei Telegram-Verarbeitung: %s", e) + return f"error: {e}" + + +def run_worker(poll_interval: int = 2, batch_size: int = 5) -> None: + """ + Haupt-Worker-Loop: Pollt die Queue und verarbeitet Submissions. + + Args: + poll_interval: Sekunden zwischen Polls. + batch_size: Maximale Submissions pro Batch. + """ + from queue_handler import QueueHandler + + db = get_db() + qh = QueueHandler(db) + + logger.info("πŸš€ Queue-Worker gestartet (Poll=%ds, Batch=%d)", poll_interval, batch_size) + + while True: + try: + submissions = qh.dequeue(limit=batch_size) + + if not submissions: + time.sleep(poll_interval) + continue + + for sub in submissions: + logger.info( + "Verarbeite #%d: channel=%s message_id=%s attempt=%d", + sub["id"], + sub["channel"], + sub["message_id"], + sub["attempt_count"], + ) + + try: + if sub["channel"] == "whatsapp": + result = process_whatsapp_submission(db, sub["payload"]) + elif sub["channel"] == "telegram": + result = process_telegram_submission(db, sub["payload"]) + else: + result = f"unknown_channel: {sub['channel']}" + + if result == "ok": + qh.mark_completed(sub["id"]) + logger.info("βœ… #%d abgeschlossen", sub["id"]) + else: + qh.mark_failed(sub["id"], result, requeue=True) + logger.warning("⚠️ #%d fehlgeschlagen: %s", sub["id"], result) + + except Exception as e: + error_msg = f"{type(e).__name__}: {e}" + qh.mark_failed(sub["id"], error_msg, requeue=True) + logger.exception("πŸ’₯ #%d Ausnahme: %s", sub["id"], error_msg) + + except Exception as e: + logger.exception("Worker-Loop-Fehler: %s", e) + time.sleep(5) # Bei DB-Fehlern kurz warten + + +# ── Main ───────────────────────────────────────────────────────────── +if __name__ == "__main__": + import argparse + + parser = argparse.ArgumentParser(description="BSN Queue Worker") + parser.add_argument("--poll", type=int, default=2, help="Poll-Intervall in Sekunden") + parser.add_argument("--batch", type=int, default=5, help="Max Batch-Grâße") + args = parser.parse_args() + + run_worker(poll_interval=args.poll, batch_size=args.batch) diff --git a/webhook_queue.py b/webhook_queue.py new file mode 100644 index 0000000..f887dcf --- /dev/null +++ b/webhook_queue.py @@ -0,0 +1,225 @@ +#!/usr/bin/env python3 +""" +Webhook-Queue-Integration β€” Signatur-PrΓΌfung + Enqueue. +======================================================= + +Wird in app.py importiert und wickelt die Webhook-Routen ab. +Ersetzt NICHT die bestehende Logik, sondern sitzt DAVOR. + +Flow: + 1. Webhook empfΓ€ngt Request + 2. Signatur wird geprΓΌft (WhatsApp X-Hub-Signature-256) + 3. Message-IDs werden extrahiert + 4. Submission wird in Queue eingereiht (SQLite UNIQUE = Dedup) + 5. 200 OK wird sofort zurΓΌckgegeben + 6. Worker (queue_worker.py) verarbeitet asynchron + +Usage in app.py: + from webhook_queue import whatsapp_intake, telegram_intake + + @app.route("/whatsapp/webhook", methods=["GET", "POST"]) + def whatsapp_webhook(): + ... + return whatsapp_intake(request, db) + + @app.route("/telegram/webhook", methods=["GET", "POST"]) + def telegram_webhook(): + ... + return telegram_intake(request, db) +""" + +import hashlib +import hmac +import logging +import os +from typing import Optional + +logger = logging.getLogger(__name__) + +# ── Konfiguration ──────────────────────────────────────────────────── +META_APP_SECRET = os.environ.get("META_APP_SECRET", "") +# Falls kein App Secret gesetzt β†’ Signatur-PrΓΌfung ΓΌberspringen (Test-Modus) + + +# ── Signatur-PrΓΌfung ───────────────────────────────────────────────── + +def verify_whatsapp_signature( + raw_body: bytes, + signature_header: Optional[str], +) -> bool: + """ + PrΓΌft die X-Hub-Signature-256 von Meta. + + Args: + raw_body: Der rohe Request-Body als Bytes. + signature_header: Wert des X-Hub-Signature-256-Headers. + + Returns: + True wenn gΓΌltig oder Test-Modus (kein Secret). + """ + if not META_APP_SECRET: + logger.debug("Kein META_APP_SECRET β†’ Signatur-PrΓΌfung ΓΌbersprungen") + return True + + if not signature_header: + logger.warning("X-Hub-Signature-256 Header fehlt") + return False + + if not signature_header.startswith("sha256="): + logger.warning("Unbekanntes Signatur-Format: %s...", signature_header[:30]) + return False + + expected_hex = signature_header[7:] # Nach "sha256=" + computed_hmac = hmac.new( + key=META_APP_SECRET.encode("utf-8"), + msg=raw_body, + digestmod=hashlib.sha256, + ) + computed_hex = computed_hmac.hexdigest() + + if not hmac.compare_digest(computed_hex, expected_hex): + logger.warning( + "❌ WhatsApp-Signatur UNGÜLTIG β€” mΓΆglicher Replay-Angriff" + ) + return False + + logger.debug("βœ… WhatsApp-Signatur gΓΌltig") + return True + + +# ── Intake-Funktionen ──────────────────────────────────────────────── + +def whatsapp_intake(request_obj, db) -> tuple: + """ + Verarbeitet EINE WhatsApp-Webhook-Anfrage. + + Args: + request_obj: Flask request-Objekt. + db: SQLite-Datenbankverbindung. + + Returns: + (body, status_code) β€” IMMER 200 OK. + """ + from queue_handler import QueueHandler + + qh = QueueHandler(db) + + # ── 1. Signatur prΓΌfen ────────────────────────────────────────── + raw_body = request_obj.get_data() + signature = request_obj.headers.get("X-Hub-Signature-256", "") + + if not verify_whatsapp_signature(raw_body, signature): + logger.warning("WhatsApp-Request mit ungΓΌltiger Signatur abgelehnt") + return "Forbidden", 403 + + # ── 2. Body parsen ────────────────────────────────────────────── + body = request_obj.get_json(force=True, silent=True) + if not body: + return "OK", 200 + + entries = body.get("entry", []) + enqueued = 0 + duplicates = 0 + + # ── 3. Jede Nachricht in Queue ────────────────────────────────── + for entry in entries: + changes = entry.get("changes", []) + for change in changes: + value = change.get("value", {}) + messages = value.get("messages", []) + + for msg in messages: + msg_id = msg.get("id", "") + if not msg_id: + continue + + # Payload bauen (nur das relevante message-Objekt) + payload = { + "entry": [{ + "id": entry.get("id", ""), + "changes": [{ + "value": { + "messaging_product": value.get("messaging_product", "whatsapp"), + "metadata": value.get("metadata", {}), + "contacts": value.get("contacts", []), + "messages": [msg], + } + }] + }] + } + + success, status = qh.enqueue( + channel="whatsapp", + message_id=msg_id, + payload=payload, + raw_body=raw_body, + ) + + if success: + enqueued += 1 + else: + duplicates += 1 + logger.info("WhatsApp-Duplikat: %s β†’ %s", msg_id, status) + + logger.info( + "WhatsApp-Intake: %d enqueued, %d duplicates", + enqueued, + duplicates, + ) + return "OK", 200 + + +def telegram_intake(request_obj, db) -> tuple: + """ + Verarbeitet EINE Telegram-Webhook-Anfrage. + + Args: + request_obj: Flask request-Objekt. + db: SQLite-Datenbankverbindung. + + Returns: + (body, status_code) β€” IMMER 200 OK. + """ + from queue_handler import QueueHandler + + qh = QueueHandler(db) + + body = request_obj.get_json(force=True, silent=True) + if not body: + return "OK", 200 + + msg = body.get("message", {}) + if not msg: + # Kein Message-Objekt (z.B. edited_message, callback_query) + update_id = str(body.get("update_id", "unknown")) + logger.debug("Telegram-Update ohne Message: update_id=%s", update_id) + return "OK", 200 + + msg_id = str(msg.get("message_id", "")) + if not msg_id: + return "OK", 200 + + # Payload ist das gesamte Update (enthΓ€lt update_id + message) + success, status = qh.enqueue( + channel="telegram", + message_id=msg_id, + payload=body, + raw_body=request_obj.get_data(), + ) + + if success: + logger.info("Telegram-Intake: %s enqueued", msg_id) + else: + logger.info("Telegram-Duplikat: %s β†’ %s", msg_id, status) + + return "OK", 200 + + +# ── Queue-Status (fΓΌr Monitoring) ──────────────────────────────────── + +def get_queue_status(db) -> dict: + """Gibt Queue-Statistiken zurΓΌck (fΓΌr /admin/queue Endpoint).""" + from queue_handler import QueueHandler + + qh = QueueHandler(db) + return qh.get_queue_stats()