BSN-Livesystem/docs/alicia-aufgaben-phase2.md

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

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 <spielname> — Spiel suchen\n"
                "• bewerte <spielname> — 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 <spielname>",
        }

    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

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):

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

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(...):

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