Hermes Agent
337 lines
9.9 KiB
Markdown
337 lines
9.9 KiB
Markdown
# 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 <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`
|
||
|
||
```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`
|