#!/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, channel, 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, channel, 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)