diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..50679b1 --- /dev/null +++ b/.gitignore @@ -0,0 +1,46 @@ +# Python +__pycache__/ +*.py[cod] +*$py.class +*.so +.Python +build/ +develop-eggs/ +dist/ +downloads/ +eggs/ +.eggs/ +lib/ +lib64/ +parts/ +sdist/ +var/ +wheels/ +*.egg-info/ +.installed.cfg +*.egg + +# Virtual Environment +venv/ +ENV/ +env/ + +# IDEs +.idea/ +.vscode/ +*.swp +*.swo + +# Testing +.pytest_cache/ + coverage/ +.coverage +htmlcov/ + +# Linting +.ruff_cache/ +.mypy_cache/ + +# Environment variables +.env +.env.local diff --git a/pyproject.toml b/pyproject.toml index 2ddb9e0..e605e3f 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,17 +1,21 @@ [project] name = "bsn-livesystem" version = "0.1.0" +description = "BSN Livesystem - WhatsApp & Telegram Community Integration" requires-python = ">=3.13" dependencies = [ - "flask", - "pydantic", - "redis", - "structlog", - "pytest", - "pytest-cov", - "mypy", + "flask>=3.0.0", + "pydantic>=2.0.0", + "redis>=5.0.0", + "structlog>=24.0.0", ] +[project.group.dev.dependencies] +pytest = ">=8.0.0" +pytest-cov = ">=5.0.0" +mypy = ">=1.0.0" +fakeredis = ">=2.0.0" + [tool.ruff] line-length = 100 target-version = "py313" diff --git a/src/__init__.py b/src/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/models/__init__.py b/src/models/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/models/submission.py b/src/models/submission.py new file mode 100644 index 0000000..f786f98 --- /dev/null +++ b/src/models/submission.py @@ -0,0 +1,16 @@ +"""Submission-Modell für eingehende Community-Nachrichten.""" + +from datetime import datetime + +from pydantic import BaseModel, Field + + +class Submission(BaseModel): + """Eine Community-Einreichung (WhatsApp/Telegram).""" + + message_id: str = Field(..., description="Eindeutige Nachrichten-ID") + plattform: str = Field(..., description="whatsapp oder telegram") + absender: str = Field(..., description="Absender-ID oder Telefonnummer") + text: str = Field(..., min_length=1, max_length=5000, description="Nachrichtentext") + zeitstempel: datetime = Field(default_factory=datetime.now, description="Eingangszeitpunkt") + status: str = Field(default="neu", description="neu, in_bearbeitung, erledigt, fehler") diff --git a/src/queue/__init__.py b/src/queue/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/queue/handler.py b/src/queue/handler.py new file mode 100644 index 0000000..1090b69 --- /dev/null +++ b/src/queue/handler.py @@ -0,0 +1,130 @@ +"""Queue-Handler für Redis-basierte Message-Queue mit Idempotenz-Garantie.""" +from datetime import datetime +from typing import Any + +import redis + +from src.models.submission import Submission + + +class QueueHandler: + """Redis-basierte Message-Queue mit Idempotenz-Garantie. + + Verwendet Redis-Streams für persistente Pufferung und ein Redis-Set + für Deduplizierung mit 24h TTL. + """ + + def __init__(self, redis_host: str = "localhost", redis_port: int = 6379) -> None: + """Initialisiert den Queue-Handler mit Redis-Verbindung. + + Args: + redis_host: Redis-Server-Host (Standard: localhost) + redis_port: Redis-Server-Port (Standard: 6379) + """ + self.redis = redis.Redis(host=redis_host, port=redis_port, decode_responses=True) + self.stream_key = "bsn:submissions" + self.processed_key = "bsn:processed" + self.consumer_group = "bsn-workers" + + # Consumer-Group initialisieren (falls noch nicht vorhanden) + try: + self.redis.xgroup_create(self.stream_key, self.consumer_group, id="0") + except redis.exceptions.ResponseError: + pass # Group existiert bereits + + def enqueue(self, submission: Submission) -> str | None: + """Legt eine Submission in die Queue. Überspringt Duplikate. + + Prüft ob message_id bereits verarbeitet wurde. + Wenn ja → return None (Idempotenz) + Wenn nein → XADD in Stream, merke ID in processed_set (TTL 24h) + + Args: + submission: Die zu enqueuende Submission. + + Returns: + Die Redis-Stream-ID bei Erfolg, None bei Duplikat. + """ + # Prüfe ob message_id bereits verarbeitet wurde + if self.redis.sismember(self.processed_key, submission.message_id): + return None + + # Submission als dict serialisieren + data: dict[str, Any] = { + "message_id": submission.message_id, + "plattform": submission.plattform, + "absender": submission.absender, + "text": submission.text, + "zeitstempel": submission.zeitstempel.isoformat(), + "status": submission.status, + } + + # XADD in Stream + stream_id = self.redis.xadd(self.stream_key, data) + + # In processed_set merken mit 24h TTL (86400 Sekunden) + self.redis.sadd(self.processed_key, submission.message_id) + self.redis.expire(self.processed_key, 86400) + + # stream_id kann bytes oder str sein, aber mit decode_responses=True ist es str + return str(stream_id) if stream_id else None + + def dequeue(self, count: int = 5) -> list[Submission]: + """Holt bis zu `count` Submissions aus der Queue (FIFO). + + Nutzt XREADGROUP für Consumer-Group-Verarbeitung. + + Args: + count: Maximale Anzahl zu holender Submissions (Standard: 5). + + Returns: + Liste von Submission-Objekten. + """ + messages: list[Any] = self.redis.xreadgroup( + groupname=self.consumer_group, + consumername="worker-1", # Einfacher Einzel-Consumer + streams={self.stream_key: ">"}, # ">" = nur neue Nachrichten + count=count, + block=0, # Nicht blockierend + ) + + submissions: list[Submission] = [] + if messages: + for stream_name, stream_messages in messages: + for message_tuple in stream_messages: + # message_tuple ist (id, {field: value, ...}) + message_data = message_tuple[1] + try: + zeitstempel_str = str(message_data.get("zeitstempel", "")) + submission = Submission( + message_id=str(message_data.get("message_id", "")), + plattform=str(message_data.get("plattform", "")), + absender=str(message_data.get("absender", "")), + text=str(message_data.get("text", "")), + zeitstempel=datetime.fromisoformat(zeitstempel_str), + status=str(message_data.get("status", "neu")), + ) + submissions.append(submission) + except (KeyError, ValueError, TypeError): + continue # Ungültige Nachricht überspringen + + return submissions + + def mark_done(self, submission: Submission) -> None: + """Markiert eine Submission als erfolgreich verarbeitet. + + Entfernt die Nachricht aus dem Stream-Aktivitäts-Tracking. + + Args: + submission: Die abzuschließende Submission. + """ + # Nachricht im Stream ack-en (für Consumer-Group) + self.redis.xack(self.stream_key, self.consumer_group, submission.message_id) + + def queue_size(self) -> int: + """Gibt die aktuelle Queue-Größe zurück. + + Returns: + Anzahl wartender Nachrichten im Stream. + """ + return self.redis.xlen(self.stream_key) diff --git a/src/utils/__init__.py b/src/utils/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/utils/validators.py b/src/utils/validators.py new file mode 100644 index 0000000..1a69f22 --- /dev/null +++ b/src/utils/validators.py @@ -0,0 +1,97 @@ +"""Hilfsfunktionen zur Validierung von Benutzereingaben.""" + +import re + + +def validiere_telefonnummer(telefon: str) -> str: + """Validiert und normalisiert eine Telefonnummer. + + Akzeptiert: "+491****7890", "+49 123 4567890", "00491234567890" + Entfernt Leerzeichen, Bindestriche, Klammern + Konvertiert "00" am Anfang zu "+" + Wirft ValueError bei: weniger als 10 Ziffern, kein + nach Bereinigung + + Args: + telefon: Die zu validierende Telefonnummer als String. + + Returns: + Die bereinigte Telefonnummer mit führendem "+". + + Raises: + ValueError: Wenn die Nummer ungültig ist. + """ + # Entferne alles außer + und Ziffern + bereinigt = re.sub(r"[^\d+]", "", telefon) + + # Konvertiere "00" am Anfang zu "+" + if bereinigt.startswith("00"): + bereinigt = "+" + bereinigt[2:] + + # Muss mit + beginnen + if not bereinigt.startswith("+"): + raise ValueError(f"Telefonnummer muss mit '+' beginnen: {telefon}") + + # Zähle Ziffern (ohne das +) + ziffern = re.sub(r"[^\d]", "", bereinigt) + + if len(ziffern) < 10: + raise ValueError(f"Telefonnummer muss mindestens 10 Ziffern haben: {telefon}") + + return bereinigt + + +def validiere_spiel_name(name: str) -> str: + """Validiert einen Spielnamen. + + Entfernt HTML-Tags mit Regex + Strip Whitespace an beiden Enden + Wirft ValueError bei: < 2 Zeichen, > 200 Zeichen + + Args: + name: Der zu validierende Spielname. + + Returns: + Der bereinigte Spielname. + + Raises: + ValueError: Wenn der Name ungültig ist. + """ + # Entferne HTML-Tags + bereinigt = re.sub(r"<[^>]*>", "", name) + + # Strip Whitespace + bereinigt = bereinigt.strip() + + if len(bereinigt) < 2: + raise ValueError(f"Spielname muss mindestens 2 Zeichen haben: {name}") + + if len(bereinigt) > 200: + raise ValueError(f"Spielname darf maximal 200 Zeichen haben: {name}") + + return bereinigt + + +def validiere_spieleranzahl(anzahl: int) -> int: + """Validiert eine Spieleranzahl. + + Wirft ValueError bei: < 1, > 20, kein Integer + + Args: + anzahl: Die zu valideinde Spieleranzahl. + + Returns: + Die ungeänderte Spieleranzahl. + + Raises: + ValueError: Wenn die Anzahl ungültig ist. + """ + if not isinstance(anzahl, int): + raise ValueError(f"Spieleranzahl muss ein Integer sein: {anzahl}") + + if anzahl < 1: + raise ValueError(f"Spieleranzahl muss mindestens 1 sein: {anzahl}") + + if anzahl > 20: + raise ValueError(f"Spieleranzahl darf maximal 20 sein: {anzahl}") + + return anzahl diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/test_queue_handler.py b/tests/test_queue_handler.py new file mode 100644 index 0000000..89752a8 --- /dev/null +++ b/tests/test_queue_handler.py @@ -0,0 +1,124 @@ +"""Tests für den Queue-Handler.""" + +from unittest.mock import patch + +import pytest + +from src.models.submission import Submission + + +@pytest.fixture +def queue_handler(): + """Erstellt einen Queue-Handler mit gemocktem Redis.""" + import fakeredis + + fake_redis = fakeredis.FakeRedis(decode_responses=True) + + with patch("src.queue.handler.redis.Redis", return_value=fake_redis): + from src.queue.handler import QueueHandler + + handler = QueueHandler() + + # Override stream keys für Tests + handler.stream_key = "test:submissions" + handler.processed_key = "test:processed" + handler.consumer_group = "test-workers" + + # Consumer-Group initialisieren + try: + handler.redis.xgroup_create(handler.stream_key, handler.consumer_group, id="0") + except: + pass + + handler._fake_redis = fake_redis # Für queue_size Referenz + + return handler + + +def test_enqueue_neue_submission_gibt_stream_id(queue_handler): + """Testet dass enqueue einer neuen Submission eine stream_id zurückgibt.""" + submission = Submission( + message_id="msg_001", + plattform="whatsapp", + absender="+491****7890", + text="Testnachricht", + ) + + stream_id = queue_handler.enqueue(submission) + + assert stream_id is not None + assert isinstance(stream_id, str) + + +def test_enqueue_duplikat_gibt_none(queue_handler): + """Testet dass enqueue eines Duplikats None zurückgibt.""" + submission = Submission( + message_id="msg_002", + plattform="telegram", + absender="@user123", + text="Duplikat-Test", + ) + + # Erstmal enqueueen + queue_handler.enqueue(submission) + + # Zweites Mal mit gleicher message_id → sollte None zurückgeben + stream_id = queue_handler.enqueue(submission) + + assert stream_id is None + + +def test_dequeue_holt_submission_zurueck(queue_handler): + """Testet dass dequeue eingereihte Submissions zurückgibt.""" + submission = Submission( + message_id="msg_003", + plattform="whatsapp", + absender="+499****3210", + text="Dequeue-Test", + ) + + queue_handler.enqueue(submission) + + submissions = queue_handler.dequeue(count=5) + + assert len(submissions) == 1 + assert submissions[0].message_id == "msg_003" + assert submissions[0].text == "Dequeue-Test" + + +def test_mark_done_entfernt_aus_queue(queue_handler): + """Testet dass mark_done die Nachricht ackt.""" + submission = Submission( + message_id="msg_004", + plattform="telegram", + absender="@user456", + text="Mark-done-Test", + ) + + queue_handler.enqueue(submission) + queue_handler.dequeue(count=1) # Nachricht holen + queue_handler.mark_done(submission) # Als erledigt marken + + # XAck sollte erfolgreich gewesen sein - keine Exception + # Wir prüfen indem wir erneut dequeue versuchen + submissions = queue_handler.dequeue(count=5) + assert len(submissions) == 0 # Sollte leer sein + + +def test_queue_size_gibt_korrekte_anzahl(queue_handler): + """Testet dass queue_size die korrekte Anzahl zurückgibt.""" + # Queue sollte anfangs leer sein + assert queue_handler.queue_size() == 0 + + # 3 Submissions enqueueen + for i in range(3): + submission = Submission( + message_id=f"msg_size_{i}", + plattform="whatsapp", + absender=f"+491****{i:04d}", + text=f"Nachricht {i}", + ) + queue_handler.enqueue(submission) + + size = queue_handler.queue_size() + assert size == 3 diff --git a/tests/test_submission.py b/tests/test_submission.py new file mode 100644 index 0000000..4640191 --- /dev/null +++ b/tests/test_submission.py @@ -0,0 +1,61 @@ +"""Tests für das Submission-Modell.""" + +from datetime import datetime + +import pytest +from pydantic import ValidationError + +from src.models.submission import Submission + + +def test_gueltige_submission_erstellen(): + """Testet das Erstellen einer gültigen Submission mit allen Feldern.""" + submission = Submission( + message_id="msg_123", + plattform="whatsapp", + absender="+491234567890", + text="Hallo Welt", + ) + assert submission.message_id == "msg_123" + assert submission.plattform == "whatsapp" + assert submission.absender == "+491234567890" + assert submission.text == "Hallo Welt" + assert submission.status == "neu" + assert submission.zeitstempel is not None + + +def test_leerer_text_wirft_error(): + """Testet dass ein leeres Textfeld eine ValidationError wirft.""" + with pytest.raises(ValidationError): + Submission( + message_id="msg_123", + plattform="whatsapp", + absender="+491234567890", + text="", + ) + + +def test_text_ueber_5000_zeichen_wirft_error(): + """Testet dass Text > 5000 Zeichen eine ValidationError wirft.""" + with pytest.raises(ValidationError): + Submission( + message_id="msg_123", + plattform="telegram", + absender="@user123", + text="x" * 5001, + ) + + +def test_zeitstempel_wird_automatisch_gesetzt(): + """Testet dass zeitstempel automatisch gesetzt wird und nicht None ist.""" + vor_erstellung = datetime.now() + submission = Submission( + message_id="msg_456", + plattform="whatsapp", + absender="+499876543210", + text="Testnachricht", + ) + nach_erstellung = datetime.now() + + assert submission.zeitstempel is not None + assert vor_erstellung <= submission.zeitstempel <= nach_erstellung diff --git a/tests/test_validators.py b/tests/test_validators.py new file mode 100644 index 0000000..df2bd10 --- /dev/null +++ b/tests/test_validators.py @@ -0,0 +1,53 @@ +"""Tests für die Input-Validatoren.""" + +import pytest + +from src.utils.validators import ( + validiere_spiel_name, + validiere_spieleranzahl, + validiere_telefonnummer, +) + + +class TestValidiereTelefonnummer: + """Tests für validiere_telefonnummer.""" + + def test_gueltige_telefonnummer_bereinigt(self): + """Testet dass eine gültige Telefonnummer bereinigt zurückgegeben wird.""" + ergebnis = validiere_telefonnummer("+49 123 4567890") + assert ergebnis == "+491234567890" + + def test_ungueltige_telefonnummer_wirft_error(self): + """Testet dass eine ungültige Telefonnummer ValueError wirft.""" + with pytest.raises(ValueError): + validiere_telefonnummer("12345") # Zu wenige Ziffern + + +class TestValidiereSpielName: + """Tests für validiere_spiel_name.""" + + def test_gueltiger_spielname_bereinigt(self): + """Testet dass ein gültiger Spielname bereinigt zurückgegeben wird.""" + ergebnis = validiere_spiel_name(" Monopoly ") + assert ergebnis == "Monopoly" + + def test_html_tags_im_spielnamen_entfernt(self): + """Testet dass HTML-Tags aus dem Spielnamen entfernt werden.""" + ergebnis = validiere_spiel_name("Catan") + assert ergebnis == "Catan" + assert "" not in ergebnis + assert "" not in ergebnis + + +class TestValidiereSpieleranzahl: + """Tests für validiere_spieleranzahl.""" + + def test_gueltige_spieleranzahl_unveraendert(self): + """Testet dass eine gültige Spieleranzahl unverändert zurückgegeben wird.""" + ergebnis = validiere_spieleranzahl(5) + assert ergebnis == 5 + + def test_ungueltige_spieleranzahl_wirft_error(self): + """Testet dass eine ungültige Spieleranzahl ValueError wirft.""" + with pytest.raises(ValueError): + validiere_spieleranzahl(25) # Zu viele Spieler