Hermes Agent
a5eb0e7d27
Quality Gate / 🛡️ Lint + Type Check + Tests (push) Failing after 11m21s
- webhook_queue.py: WhatsApp X-Hub-Signature-256 + Telegram Webhook-Intake - queue_handler.py: SQLite-basierte Queue mit Idempotenz (UNIQUE message_id) - queue_worker.py: Asynchroner Worker für GPU-Verarbeitung - app.py: Webhooks umgebaut → sofort 200 OK, Worker verarbeitet später - Redis 8.0 läuft als Infrastruktur (Queue ist SQLite für Persistenz)
244 lines
8.7 KiB
Python
244 lines
8.7 KiB
Python
#!/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)
|