From 4caa41428e776ab8e345607329f3898d2c17f1c4 Mon Sep 17 00:00:00 2001 From: Hermes Agent Date: Mon, 22 Jun 2026 10:29:24 +0200 Subject: [PATCH] =?UTF-8?q?docs:=20Phase=202=20=E2=80=94=20Aufgaben=2011-1?= =?UTF-8?q?5=20(Intake,=20Game-Service,=20Admin,=20Structlog,=20Integratio?= =?UTF-8?q?n)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- docs/alicia-aufgaben-phase2.md | 336 +++++++++++++++++++++++++++++++++ 1 file changed, 336 insertions(+) create mode 100644 docs/alicia-aufgaben-phase2.md diff --git a/docs/alicia-aufgaben-phase2.md b/docs/alicia-aufgaben-phase2.md new file mode 100644 index 0000000..6092be6 --- /dev/null +++ b/docs/alicia-aufgaben-phase2.md @@ -0,0 +1,336 @@ +# Alicia — Aufgaben 11–15: BSN-Livesystem Phase 2 + +> **Repo:** `git@git.datenhimmel.work:danielkrause/BSN-Livesystem.git` +> **Branch:** `main` +> **Stand:** Aufgaben 1–10 sind fertig. Alle 10 erledigt. +> **Arbeitsverzeichnis:** `/tmp/BSN-Livesystem` +> **Qualitäts-Check vor JEDEM Commit:** `ruff check . && ruff format . && mypy --config-file pyproject.toml . && pytest -v` +> **Fehler?** Korrigieren, Checks wiederholen. Nicht aufgeben bis ALLE grün. + +--- + +## Aufgabe 11 — Message-Intake-Service + +Parst eingehende Nachrichten und extrahiert Kommandos. + +### Datei: `src/services/__init__.py` +Leer lassen. + +### Datei: `src/services/intake.py` + +```python +from __future__ import annotations +from enum import StrEnum + +from src.models.submission import Submission + + +class KommandoTyp(StrEnum): + """Bekannte Benutzer-Kommandos.""" + SPIEL_SUCHEN = "spiel_suchen" + SPIEL_BEWERTEN = "spiel_bewerten" + HILFE = "hilfe" + UNBEKANNT = "unbekannt" + + +def erkenne_kommando(text: str) -> KommandoTyp: + """Erkennt das Kommando aus dem Nachrichtentext. + + Args: + text: Der Nachrichtentext. + + Returns: + Das erkannte KommandoTyp-Enum. + """ + text_lower = text.lower().strip() + + if text_lower.startswith("suche") or text_lower.startswith("finde"): + return KommandoTyp.SPIEL_SUCHEN + if text_lower.startswith("bewerte") or text_lower.startswith("bewertung"): + return KommandoTyp.SPIEL_BEWERTEN + if text_lower in ("hilfe", "help", "?", "commands"): + return KommandoTyp.HILFE + + return KommandoTyp.UNBEKANNT + + +def extrahiere_suchbegriff(text: str) -> str | None: + """Extrahiert den Suchbegriff aus einer Such-Nachricht. + + Args: + text: Der Nachrichtentext (z.B. "suche Catan"). + + Returns: + Der Suchbegriff ohne Kommando-Präfix, oder None. + """ + for praefix in ("suche ", "finde "): + if text.lower().startswith(praefix): + begriff = text[len(praefix):].strip() + return begriff if begriff else None + return None + + +def verarbeite_submission(submission: Submission) -> dict[str, object]: + """Verarbeitet eine Submission und gibt eine Antwort-Struktur zurück. + + Args: + submission: Die eingegangene Submission. + + Returns: + Ein Dict mit 'antwort' und 'kommando'-Keys. + """ + kommando = erkenne_kommando(submission.text) + + if kommando == KommandoTyp.HILFE: + return { + "kommando": kommando.value, + "antwort": ( + "📋 Verfügbare Kommandos:\n" + "• suche — Spiel suchen\n" + "• bewerte — Spiel bewerten\n" + "• hilfe — Diese Hilfe anzeigen" + ), + } + + if kommando == KommandoTyp.SPIEL_SUCHEN: + suchbegriff = extrahiere_suchbegriff(submission.text) + if suchbegriff: + return { + "kommando": kommando.value, + "antwort": f"🔍 Suche nach '{suchbegriff}'...", + "suchbegriff": suchbegriff, + } + return { + "kommando": kommando.value, + "antwort": "❌ Bitte gib einen Suchbegriff an: suche ", + } + + if kommando == KommandoTyp.SPIEL_BEWERTEN: + return { + "kommando": kommando.value, + "antwort": "⭐ Bewertungsmodus aktiviert. Welches Spiel möchtest du bewerten?", + } + + return { + "kommando": KommandoTyp.UNBEKANNT.value, + "antwort": ( + "🤔 Das habe ich nicht verstanden. Schreib 'hilfe' " + "für eine Liste der Kommandos." + ), + } +``` + +### Test-Datei: `tests/test_intake.py` + +7 Tests: +1. `"suche Catan"` → KommandoTyp.SPIEL_SUCHEN +2. `"suche"` ohne Begriff → Suchbegriff None +3. `"suche Catan"` → extrahiert "Catan" +4. `"bewerte Carcassonne"` → KommandoTyp.SPIEL_BEWERTEN +5. `"hilfe"` → KommandoTyp.HILFE +6. `"blabla"` → KommandoTyp.UNBEKANNT +7. `verarbeite_submission` mit "hilfe" → Antwort enthält "Verfügbare Kommandos" + +**Commit:** `feat: Message-Intake-Service mit Kommando-Erkennung` + +--- + +## Aufgabe 12 — Game-Service (Spiele-Datenbank & CRUD) + +### Datei: `src/services/game_service.py` + +```python +from __future__ import annotations +from pydantic import BaseModel, Field + + +class Spiel(BaseModel): + """Ein Brettspiel-Datenmodell.""" + id: int | None = None + name: str = Field(..., min_length=1, max_length=200) + min_spieler: int = Field(default=1, ge=1, le=20) + max_spieler: int = Field(default=10, ge=1, le=20) + dauer_min: int = Field(default=30, ge=1) + komplexitaet: float = Field(default=2.5, ge=1.0, le=5.0) + kategorie: str = Field(default="Familienspiel") + + +class GameService: + """Verwaltet die Spiele-Datenbank (In-Memory für jetzt, später DB).""" + + def __init__(self) -> None: + self._spiele: dict[int, Spiel] = {} + self._next_id = 1 + + def hinzufuegen(self, spiel: Spiel) -> Spiel: + """Fügt ein Spiel hinzu und weist eine ID zu.""" + spiel.id = self._next_id + self._spiele[self._next_id] = spiel + self._next_id += 1 + return spiel + + def alle(self) -> list[Spiel]: + """Gibt alle Spiele zurück.""" + return list(self._spiele.values()) + + def suchen(self, name: str) -> list[Spiel]: + """Sucht Spiele nach Name (case-insensitive Teilstring).""" + name_lower = name.lower() + return [ + spiel + for spiel in self._spiele.values() + if name_lower in spiel.name.lower() + ] + + def nach_id(self, spiel_id: int) -> Spiel | None: + """Holt ein Spiel anhand seiner ID.""" + return self._spiele.get(spiel_id) +``` + +### Test-Datei: `tests/test_game_service.py` + +6 Tests: +1. Hinzufügen eines Spiels → ID wird vergeben +2. Alle Spiele → leere Liste am Anfang +3. Alle Spiele → 2 Spiele nach Hinzufügen +4. Suchen nach "cat" → findet "Catan" +5. Suchen nach "xyz" → leere Liste +6. Nach ID holen → korrektes/nicht existentes Spiel + +**Commit:** `feat: Game-Service mit In-Memory-CRUD und Suche` + +--- + +## Aufgabe 13 — Admin-Route (Queue-Status) + +### Datei: `src/routes/admin.py` + +Blueprint `"admin"` mit `/admin/queue` (GET) und `/admin/stats` (GET): + +```python +from flask import Blueprint, current_app, jsonify + +admin_bp = Blueprint("admin", __name__) + + +@admin_bp.route("/admin/queue", methods=["GET"]) +def queue_status() -> tuple: + """Gibt Queue-Status zurück.""" + qh = current_app.queue_handler + return jsonify({ + "queue_size": qh.queue_size(), + "status": "ok", + }), 200 + + +@admin_bp.route("/admin/stats", methods=["GET"]) +def stats() -> tuple: + """Gibt System-Statistiken zurück.""" + qh = current_app.queue_handler + return jsonify({ + "queue_size": qh.queue_size(), + "uptime": "placeholder", + "version": "0.1.0", + }), 200 +``` + +### Test-Datei: `tests/test_admin.py` + +4 Tests mit `app.test_client()`: +1. GET `/admin/queue` → 200 + JSON mit queue_size +2. GET `/admin/queue` → queue_size ist Integer +3. GET `/admin/stats` → 200 + version "0.1.0" +4. GET `/admin/stats` → JSON enthält alle Keys + +**Commit:** `feat: Admin-Route mit Queue-Status und Statistiken` + +--- + +## Aufgabe 14 — Structlog-Integration + +### Datei: `src/logging_config.py` + +```python +from __future__ import annotations +import structlog + + +def setup_logging() -> None: + """Konfiguriert strukturiertes Logging mit structlog.""" + structlog.configure( + processors=[ + structlog.stdlib.filter_by_level, + structlog.stdlib.add_logger_name, + structlog.stdlib.add_log_level, + structlog.stdlib.PositionalArgumentsFormatter(), + structlog.processors.TimeStamper(fmt="iso"), + structlog.processors.StackInfoRenderer(), + structlog.processors.format_exc_info, + structlog.processors.UnicodeDecoder(), + structlog.dev.ConsoleRenderer(), + ], + context_class=dict, + logger_factory=structlog.stdlib.LoggerFactory(), + wrapper_class=structlog.stdlib.BoundLogger, + cache_logger_on_first_use=True, + ) + + +def get_logger(name: str) -> structlog.stdlib.BoundLogger: + """Erstellt einen benannten Logger. + + Args: + name: Logger-Name (üblicherweise __name__). + + Returns: + Konfigurierter structlog-Logger. + """ + return structlog.get_logger(name) +``` + +### Integration in `src/app.py` + +Ergänze in `create_app()` direkt nach `app = Flask(...)`: +```python +from src.logging_config import setup_logging +setup_logging() +``` + +### Test-Datei: `tests/test_logging.py` + +3 Tests: +1. `setup_logging()` läuft ohne Fehler durch +2. `get_logger(__name__)` gibt BoundLogger zurück +3. Logger.info() produziert keinen Fehler + +**Commit:** `feat: Structlog-Integration mit Console-Renderer` + +--- + +## Aufgabe 15 — Integration-Tests (End-to-End) + +### Datei: `tests/test_integration.py` + +Testet den kompletten Flow: Webhook → Queue → Worker → Submission. + +5 Tests mit `app.test_client()` und fakeredis: +1. WhatsApp POST mit gültiger Signatur → 200, Submission in Queue +2. WhatsApp POST mit ungültiger Signatur → 401 +3. Telegram POST mit gültigem Payload → 200 +4. Vollständiger Flow: WhatsApp POST → Worker verarbeitet → Status "erledigt" +5. Vollständiger Flow: Telegram POST → Worker verarbeitet → Status "erledigt" + +Für Test 4+5: Enqueue via POST, dann `worker_loop` mit gemocktem `time.sleep` aufrufen, danach prüfen ob Queue leer. + +**Commit:** `test: Integration-Tests für WhatsApp/Telegram-Webhook-Flows` + +--- + +## ⚠️ WICHTIG + +- **Bluesprints registrieren:** Bei Aufgaben 13 und 15 musst du in `create_app()` temporär die neuen Blueprints registrieren (admin_bp, whatsapp_bp, telegram_bp), damit die Tests mit `app.test_client()` funktionieren. +- **worker_loop-Timeout:** Mocke `time.sleep` in Integration-Tests, sonst hängt die Endlos-Schleife. +- **Qualität:** `ruff check . && ruff format . && mypy --config-file pyproject.toml . && pytest -v` vor JEDEM Commit. +- **Bei Fehlern:** Code korrigieren, nicht aufgeben. +- **Nach jeder Aufgabe:** Commit + Push. +- **Täglich vor Start:** `git pull origin main`