diff --git a/README.md b/README.md index ed09586..ca641a8 100644 --- a/README.md +++ b/README.md @@ -59,6 +59,12 @@ Beim ersten Start wird automatisch ein Admin-Konto angelegt: | `SPIELE_BGG_MAX_TREFFER_PRO_SUCHE` | `25` | Obergrenze Treffer je Suchbegriff (schont das BGG-Rate-Limit) | | `SPIELE_BGG_TOKEN` | *(leer)* | API-Token für die BGG-XML-API2, wird als `Authorization: Bearer …`-Header gesendet; seit der Token-Pflicht von BGG erforderlich (siehe [BGG-Thread 3602374](https://boardgamegeek.com/thread/3602374)) — ohne Token wird der Sync übersprungen | | `SPIELE_DEDUP_BGG_AKTIV` | `1` | BGG-Zusatzdaten für die Dedup-Prüfung an (`1`) oder aus (`0`): Alternate-Names und Erweiterungs-Relationen | +| `SPIELE_DEDUP_LLM_AKTIV` | `0` | LLM-Zweitprüfung der Dedup-Grenzfälle an (`1`) oder aus (`0`) | +| `SPIELE_DEDUP_LLM_BASIS_URL` | *(leer)* | Basis-URL einer OpenAI-kompatiblen Chat-Completions-API, z. B. `https://api.openai.com/v1` | +| `SPIELE_DEDUP_LLM_API_KEY` | *(leer)* | API-Key, wird als `Authorization: Bearer …` gesendet | +| `SPIELE_DEDUP_LLM_MODELL` | *(leer)* | Modellname, z. B. `gpt-4o-mini` | +| `SPIELE_DEDUP_LLM_KONFIDENZ_MIN` | `0.7` | Mindest-Konfidenz; darunter wird das LLM-Ergebnis verworfen und nur der Regelbefund angezeigt | +| `SPIELE_DEDUP_LLM_TIMEOUT_SEKUNDEN` | `20` | Timeout je LLM-Aufruf; Netzwerk-/Server-Fehler werden genau einmal wiederholt | | `SPIELE_ARCHIV_JOB_AKTIV` | `1` | Täglicher Archivierungs-Job an (`1`) oder aus (`0`) | | `SPIELE_ARCHIV_JOB_UHRZEIT` | `03:00` | Tageszeit des täglichen Archiv-Laufs im Format `HH:MM` | | `SPIELE_ERINNERUNG_JOB_AKTIV` | `1` | Täglicher Erinnerungs-Check an (`1`) oder aus (`0`) | @@ -356,6 +362,37 @@ Teilprüfung. BGG-Störungen blockieren nie: der Hilfsclient degradiert auf die reine Heuristik und protokolliert nur. Abschaltbar über `SPIELE_DEDUP_BGG_AKTIV=0` (Standard: an, Rate-Limit 1 s). +### LLM-Zweitprüfung für Grenzfälle + +Regeln stoßen bei Editionen, Big Boxes, Übersetzungen und Verlags-Aliasen an +Grenzen („El Grande Big Box“ vs. „El Grande: 25 Jahre“, „Catan – Das schnelle +Spiel“ vs. „Catan Fast Edition“). Deshalb kann optional ein LLM als **zweite +Stufe** eingeschaltet werden — die Regelprüfung bleibt maßgeblich: + +- **Nur Grenzfälle gehen ans LLM:** Titel mit gemeinsamen Wort-Tokens, aber + Fuzzy-Score unter der Schwelle (max. 5 Paarungen je Prüfung), sowie ein + ergänzender Call zum Verlags-/Titel-Hinweis bei einem regelbasierten + Verlags-Konflikt. Klare Treffer und klare Non-Treffer kosten kein Token. +- **Kontext:** beide Titel inkl. Alternate-Names (BGG-Sync), Verlage und ggf. + BGG-Expansion-Relationen. +- **Strukturierte Antwort** (JSON): `ist_gleiches_spiel`, `konfidenz`, + `begruendung`, `empfohlener_verlag`, `deutscher_titel`. Bestätigte Treffer + laufen in die üblichen Prüfungen a)–d) zurück; ein übernommener deutscher + Titel füllt die Titel-Empfehlung, ein Verlags-Hinweis landet im Detail des + Verlags-Konflikts. Unter der Mindest-Konfidenz + (`SPIELE_DEDUP_LLM_KONFIDENZ_MIN`, Standard 0.7) wird das Ergebnis + verworfen und nur der Regelbefund angezeigt. +- **Fallback-Pflicht:** Standardmäßig aus (`SPIELE_DEDUP_LLM_AKTIV=0`). Ohne + Key oder bei jedem Fehler (Timeout, Server-Störung nach genau einem Retry, + kaputtes JSON) verhält sich die Prüfung exakt wie rein regelbasiert — das + LLM kann die Prüfung nie blockieren. +- **Audit:** Jede Prüfung mit LLM-Befunden schreibt einen Eintrag ins + audit-log (`geprüft` / `dedup_ki`) mit Modell, Konfidenz und Entscheidung + je bewertetem Grenzfall. + +Schnittstelle: OpenAI-kompatible Chat-Completions-API via httpx, konfiguriert +über die `SPIELE_DEDUP_LLM_*`-Variablen (siehe Tabelle oben). + ### Prüfprotokoll & Prüfseite Jede `check_titel`-Prüfung landet in der eigenen Migration diff --git a/plugins/dedup/__init__.py b/plugins/dedup/__init__.py index e556991..6f0fc00 100644 --- a/plugins/dedup/__init__.py +++ b/plugins/dedup/__init__.py @@ -15,6 +15,17 @@ in `pruefung.py` beschrieben. Jede Prüfung wird in der eigenen Tabelle Zusatzdaten von BoardGameGeek (Alternate-Names, Erweiterungs-Relationen) können über SPIELE_DEDUP_BGG_AKTIV=0 abgeschaltet werden (Standard: an); ohne Netzwerk degradiert die Prüfung automatisch auf die Heuristik. + +Als zweite Stufe kann ein LLM Grenzfälle bewerten — Titel mit +Wort-Überschneidung unter der Fuzzy-Schwelle (Editionen, Big Boxes, +Übersetzungen) und Verlags-Konflikte mit ähnlichem Titel (Verlags-Aliase). +Standardmäßig aus (`SPIELE_DEDUP_LLM_AKTIV=0`); konfiguriert wird eine +OpenAI-kompatible Chat-Completions-API über `SPIELE_DEDUP_LLM_*`-Variablen +(siehe ki_pruefung.py / README). Ohne Aktivierung, ohne Key oder bei jedem +Fehler verhält sich die Prüfung exakt wie rein regelbasiert; KI-Befunde +unterhalb der Mindest-Konfidenz (`SPIELE_DEDUP_LLM_KONFIDENZ_MIN`, +Standard 0.7) werden verworfen. Übernommene Befunde laufen in die üblichen +Prüfungen a)–d) zurück und werden im audit-log protokolliert. """ from __future__ import annotations @@ -30,6 +41,7 @@ from redaktionskern.auth.models import User from redaktionskern.contracts import BasePlugin, Migration, NavEntry from .bgg import BggPruefClient +from .ki_pruefung import KiPruefClient, lade_konfiguration from .models import Pruefung from .pruefung import KONFLIKT_ANZEIGE, PruefErgebnis, fuehre_pruefung_durch @@ -59,6 +71,12 @@ class DedupPlugin(BasePlugin): #: Von Tests überschreibbare Fabrik für den BGG-Hilfsclient. self.bgg_client_fabrik = None + #: Konfiguration der KI-Zweitprüfung (wird in on_load aus Env gelesen). + self.ki_konfiguration = lade_konfiguration({}) + self._ki_client: KiPruefClient | None = None + #: Von Tests überschreibbare Fabrik für den KI-Client (Mock-Transport). + self.ki_client_fabrik = None + self._routen_registrieren() # ---------- Plugin-Vertrag ---------- @@ -72,11 +90,17 @@ class DedupPlugin(BasePlugin): def on_load(self, context) -> None: super().on_load(context) self._bgg_aktiv = os.environ.get("SPIELE_DEDUP_BGG_AKTIV", "1").strip() != "0" + # KI-Zweitprüfung (Standard aus, siehe ki_pruefung.py): ohne Key oder + # mit SPIELE_DEDUP_LLM_AKTIV=0 läuft die Prüfung exakt wie bisher. + self.ki_konfiguration = lade_konfiguration() def on_unload(self) -> None: if self._bgg_client is not None: self._bgg_client.schliessen() self._bgg_client = None + if self._ki_client is not None: + self._ki_client.schliessen() + self._ki_client = None super().on_unload() # ---------- Öffentliche Prüf-API für andere Plugins ---------- @@ -102,6 +126,7 @@ class DedupPlugin(BasePlugin): id_wert = None client = self._hilfsclient() + ki_client = self._ki_hilfsclient() try: ergebnis = fuehre_pruefung_durch( self.context.session_factory, @@ -110,15 +135,57 @@ class DedupPlugin(BasePlugin): verlag=verlag, bgg_id=id_wert, user_id=user.id if user is not None else None, + ki_client=ki_client, + ki_konfidenz_min=self.ki_konfiguration.konfidenz_min, ) finally: if client is not None: client.schliessen() + if ki_client is not None: + ki_client.schliessen() self._protokolliere(ergebnis, user) + await self._auditiere_ki_befunde(ergebnis, user) return ergebnis # ---------- Internas ---------- + def _ki_hilfsclient(self) -> KiPruefClient | None: + """KI-Client nur bei aktivierter, vollständiger Konfiguration (sonst None). + + Die Aktivierungsprüfung gilt vor jeder injizierten Fabrik — ist die + Zweitprüfung deaktiviert oder unvollständig konfiguriert, wird nie + ein Client gebaut und damit nie ein LLM-Aufruf getätigt. + """ + if self.context is None or not self.ki_konfiguration.vollstaendig: + return None + if self.ki_client_fabrik is not None: + return self.ki_client_fabrik() + if self._ki_client is None: + self._ki_client = KiPruefClient(self.ki_konfiguration) + return self._ki_client + + async def _auditiere_ki_befunde(self, ergebnis: PruefErgebnis, user) -> None: + """LLM-Befunde ins audit-log schreiben (best effort, nie blockierend). + + Ein Eintrag je Prüfung mit allen KI-Entscheidungen: Modell, Konfidenz + und Entscheidung je bewertetem Grenzfall. + """ + if not ergebnis.ki_befunde: + return + registry = self.context.registry if self.context is not None else None + audit = registry.get("audit-log") if registry is not None else None + if audit is None: + return + details = { + "titel": ergebnis.titel, + "modell": ergebnis.ki_befunde[0].get("modell", ""), + "befunde": ergebnis.ki_befunde, + } + try: + await audit.log(user, "geprüft", "dedup_ki", None, details) + except Exception: + _logger.exception("dedup: KI-Befunde konnten nicht auditiert werden.") + def _hilfsclient(self) -> BggPruefClient | None: if self.bgg_client_fabrik is not None: return self.bgg_client_fabrik() diff --git a/plugins/dedup/ki_pruefung.py b/plugins/dedup/ki_pruefung.py new file mode 100644 index 0000000..ee4b9e1 --- /dev/null +++ b/plugins/dedup/ki_pruefung.py @@ -0,0 +1,345 @@ +"""KI-Zweitprüfung des Plugins „dedup“ — LLM nur für Grenzfälle. + +Die Regelprüfung (`pruefung.py`) bleibt maßgeblich: Klare Treffer und klare +Non-Treffer werden ausschließlich regelbasiert entschieden. Nur wenn die +Regeln einen Grenzfall liefern — Titel-Score über 0, aber unter der +Fuzzy-Schwelle (z. B. „El Grande Big Box“ vs. „El Grande: 25 Jahre“) oder +ein Verlags-Konflikt bei ähnlichem Titel — wird ein LLM als zweite Stufe +gefragt. + +Schnittstelle: OpenAI-kompatible Chat-Completions-API (`POST +{basis_url}/chat/completions`), konfigurierbar über Umgebungsvariablen: + +| Variable | Bedeutung | +|----------|-----------| +| `SPIELE_DEDUP_LLM_AKTIV` | `1` = Zweitprüfung an (Standard `0`) | +| `SPIELE_DEDUP_LLM_BASIS_URL` | Basis-URL, z. B. `https://api.openai.com/v1` | +| `SPIELE_DEDUP_LLM_API_KEY` | API-Key (wird als Bearer gesendet) | +| `SPIELE_DEDUP_LLM_MODELL` | Modellname, z. B. `gpt-4o-mini` | +| `SPIELE_DEDUP_LLM_KONFIDENZ_MIN` | Mindest-Konfidenz (Standard `0.7`) | +| `SPIELE_DEDUP_LLM_TIMEOUT_SEKUNDEN` | Timeout je Aufruf (Standard `20`) | + +Der Client ist bewusst genauso defensiv wie der BGG-Hilfsclient: Ohne +Aktivierung/Key oder bei jedem Fehler liefert `bewerte()` None und die +Prüfung fällt auf den reinen Regelbefund zurück — die Prüfung wird nie +blockiert. Netzwerkfehler und Server-Störungen werden genau einmal +wiederholt (insgesamt zwei Versuche), Antwort-Parsing-Fehler nicht. +""" +from __future__ import annotations + +import json +import logging +import os +import re +from dataclasses import dataclass +from typing import Any, Mapping + +import httpx + +_logger = logging.getLogger("plugins.dedup") + +#: Standard-Timeout je Chat-Completions-Aufruf in Sekunden. +STANDARD_TIMEOUT = 20.0 + +#: Gesamtzahl der Versuche (erster Versuch + genau ein Retry). +VERSUCHE = 2 + +#: HTTP-Statuscodes, die einen Retry rechtfertigen (zeitweilige Störungen). +RETRY_STATUS = frozenset({408, 429, 500, 502, 503, 504}) + +#: Mindest-Konfidenz, ab der ein KI-Befund übernommen wird. +STANDARD_KONFIDENZ_MIN = 0.7 + + +# ---------------- Konfiguration ---------------- + +@dataclass(frozen=True) +class KiKonfiguration: + """Effektive Konfiguration der KI-Zweitprüfung (aus Env gelesen).""" + + aktiv: bool = False + basis_url: str = "" + api_key: str = "" + modell: str = "" + konfidenz_min: float = STANDARD_KONFIDENZ_MIN + timeout: float = STANDARD_TIMEOUT + + @property + def vollstaendig(self) -> bool: + """True, wenn Aktiv-Schalter und Zugangsdaten vollständig sind.""" + return bool(self.aktiv and self.basis_url and self.api_key and self.modell) + + +def lade_konfiguration(quelle: Mapping[str, str] | None = None) -> KiKonfiguration: + """Liest die KI-Konfiguration aus der Umgebung (fehlertolerant). + + Kaputte Zahlenwerte (z. B. `KONFIDENZ_MIN=abc`) führen nicht zum Fehler, + sondern zum jeweiligen Standardwert — die Zweitprüfung darf die + Regelprüfung niemals blockieren. + """ + env = os.environ if quelle is None else quelle + aktiv = env.get("SPIELE_DEDUP_LLM_AKTIV", "0").strip() == "1" + + try: + konfidenz_min = float(env.get("SPIELE_DEDUP_LLM_KONFIDENZ_MIN", "")) + except ValueError: + konfidenz_min = STANDARD_KONFIDENZ_MIN + konfidenz_min = min(1.0, max(0.0, konfidenz_min)) + + try: + timeout = float(env.get("SPIELE_DEDUP_LLM_TIMEOUT_SEKUNDEN", "")) + except ValueError: + timeout = STANDARD_TIMEOUT + timeout = max(1.0, timeout) + + return KiKonfiguration( + aktiv=aktiv, + basis_url=env.get("SPIELE_DEDUP_LLM_BASIS_URL", "").strip().rstrip("/"), + api_key=env.get("SPIELE_DEDUP_LLM_API_KEY", "").strip(), + modell=env.get("SPIELE_DEDUP_LLM_MODELL", "").strip(), + konfidenz_min=konfidenz_min, + timeout=timeout, + ) + + +# ---------------- Antwort-Parsing (robust gegen Code-Fences u. Ä.) ---------------- + +_FENCE_MUSTER = re.compile(r"```(?:json|JSON)?\s*(.*?)\s*```", re.DOTALL) + + +def extrahiere_json(text: str) -> dict | None: + """Extrahiert das erste JSON-Objekt aus einer LLM-Antwort. + + Toleriert Code-Fences (```json … ```) und begleitenden Text vor/nach dem + Objekt; None, wenn nichts Sinnvolles übrig bleibt. + """ + if not isinstance(text, str): + return None + text = text.strip() + if not text: + return None + + # 1) Direkter Versuch. + versuche = [text] + # 2) Inhalt von Code-Fences (alle, der längste gewinnt meistens). + versuche.extend(m.group(1).strip() for m in _FENCE_MUSTER.finditer(text)) + # 3) Erstes „{“ bis letztes „}“ (Text drumherum weg). + erstes, letztes = text.find("{"), text.rfind("}") + if 0 <= erstes < letztes: + versuche.append(text[erstes : letztes + 1]) + + for kandidat in versuche: + try: + daten = json.loads(kandidat) + except (ValueError, TypeError): + continue + if isinstance(daten, dict): + return daten + return None + + +def _optional_text(wert: Any) -> str | None: + """Normalisiert ein nullable String-Feld der Antwort.""" + if isinstance(wert, str): + wert = wert.strip() + return wert or None + return None + + +def validiere_antwort(daten: Any) -> dict | None: + """Prüft und normalisiert die LLM-Antwort gegen das vereinbarte Schema. + + Erwartet (nach JSON-Extraktion): + `{ist_gleiches_spiel: bool, konfidenz: float, begruendung: str, + empfohlener_verlag: str|null, deutscher_titel: str|null}` + + Kleine Nachsichten: boolesche Werte dürfen als ja/nein-Text kommen, + Konfidenz darf auf 0–100-Skala geliefert werden. Alles andere → None. + """ + if not isinstance(daten, dict): + return None + + ist_gleiches_spiel = daten.get("ist_gleiches_spiel") + if isinstance(ist_gleiches_spiel, str): + ist_gleiches_spiel = { + "ja": True, "true": True, "yes": True, + "nein": False, "false": False, "no": False, + }.get(ist_gleiches_spiel.strip().lower()) + if not isinstance(ist_gleiches_spiel, bool): + return None + + try: + konfidenz = float(daten.get("konfidenz")) + except (TypeError, ValueError): + return None + if 10.0 <= konfidenz <= 100.0: # Prozent-Skala tolerieren (z. B. 92 → 0.92) + konfidenz = konfidenz / 100.0 + if not 0.0 <= konfidenz <= 1.0: # 1 < x < 10 ist auf keiner Skala plausibel + return None + + begruendung = daten.get("begruendung") + if not isinstance(begruendung, str): + begruendung = "" + + return { + "ist_gleiches_spiel": ist_gleiches_spiel, + "konfidenz": round(konfidenz, 4), + "begruendung": begruendung.strip(), + "empfohlener_verlag": _optional_text(daten.get("empfohlener_verlag")), + "deutscher_titel": _optional_text(daten.get("deutscher_titel")), + } + + +# ---------------- Prompt ---------------- + +_SYSTEM_NACHRICHT = ( + "Du assistierst einer Spielemagazin-Redaktion beim Deduplizieren von " + "Brettspieltiteln. Entscheide, ob zwei Einträge dasselbe Spiel meinen " + "(auch across Sprachen, Editionen, Big Boxes und Verlags-Aliase) oder " + "verschiedene Spiele/Editionen sind. Antworte AUSSCHLIESSLICH mit einem " + "JSON-Objekt nach exakt diesem Schema:\n" + '{"ist_gleiches_spiel": , "konfidenz": , ' + '"begruendung": "", ' + '"empfohlener_verlag": "", ' + '"deutscher_titel": ""}\n' + "Kein weiterer Text, keine Code-Fences." +) + + +def baue_prompt( + *, + titel_a: str, + verlag_a: str | None = None, + alternativen_a: list[str] | None = None, + expansionen_a: list[str] | None = None, + titel_b: str, + verlag_b: str | None = None, + alternativen_b: list[str] | None = None, + frage: str, +) -> str: + """Baut die User-Nachricht mit vollem Kontext beider Einträge.""" + def seite(name: str, titel: str, verlag, alternativen, expansionen) -> str: + zeilen = [f"{name}: „{titel}“"] + if verlag: + zeilen.append(f"{name} Verlag: {verlag}") + if alternativen: + zeilen.append(f"{name} Alternate Names: " + "; ".join(alternativen)) + if expansionen: + zeilen.append(f"{name} ergänzt (BGG-Erweiterungs-Relation): " + "; ".join(expansionen)) + return "\n".join(zeilen) + + teile = [ + seite("Eintrag A", titel_a, verlag_a, alternativen_a, expansionen_a), + seite("Eintrag B", titel_b, verlag_b, alternativen_b, None), + frage, + ] + return "\n\n".join(teile) + + +# ---------------- Client ---------------- + +class KiPruefClient: + """OpenAI-kompatibler Chat-Completions-Client für die Zweitprüfung. + + Genauso defensiv wie der BGG-Hilfsclient: `bewerte()` wirft nicht, + sondern liefert bei jedem Problem None — die Regelprüfung bleibt dann + allein maßgeblich. Transport und Timeout sind injizierbar, damit Tests + netzwerkfrei bleiben. + """ + + def __init__( + self, + konfiguration: KiKonfiguration, + *, + transport: httpx.BaseTransport | None = None, + ) -> None: + self._konfiguration = konfiguration + self._client = httpx.Client( + base_url=konfiguration.basis_url, + timeout=konfiguration.timeout, + transport=transport, + headers={ + "Authorization": f"Bearer {konfiguration.api_key}", + "Content-Type": "application/json", + }, + ) + + @property + def modell(self) -> str: + return self._konfiguration.modell + + def schliessen(self) -> None: + self._client.close() + + # ---------- Öffentliche API ---------- + + def bewerte( + self, + *, + titel_a: str, + titel_b: str, + verlag_a: str | None = None, + verlag_b: str | None = None, + alternativen_a: list[str] | None = None, + alternativen_b: list[str] | None = None, + expansionen_a: list[str] | None = None, + frage: str | None = None, + ) -> dict | None: + """Fragt das LLM, ob zwei Einträge dasselbe Spiel meinen. + + Rückgabe ist das validierte Antwort-Dict (siehe `validiere_antwort`) + oder None — ohne Key, bei Netzwerk-/Server-Problemen (nach genau + einem Retry) oder bei unbrauchbarer Antwort. + """ + if not self._konfiguration.vollstaendig: + return None + + nutzernachricht = baue_prompt( + titel_a=titel_a, + verlag_a=verlag_a, + alternativen_a=alternativen_a, + expansionen_a=expansionen_a, + titel_b=titel_b, + verlag_b=verlag_b, + alternativen_b=alternativen_b, + frage=frage or ( + "Meinen Eintrag A und Eintrag B dasselbe Spiel? Wenn ja, nenne " + "bitte auch, welcher Verlag geführt werden sollte und welchen " + "deutschen Titel es gibt (falls bekannt)." + ), + ) + payload = { + "model": self._konfiguration.modell, + "temperature": 0, + "messages": [ + {"role": "system", "content": _SYSTEM_NACHRICHT}, + {"role": "user", "content": nutzernachricht}, + ], + } + + for versuch in range(1, VERSUCHE + 1): + try: + antwort = self._client.post("/chat/completions", json=payload) + except httpx.RequestError as exc: # Netzwerk/Timeout → genau ein Retry + _logger.warning("dedup/KI: Versuch %d/%d fehlgeschlagen (%s)", versuch, VERSUCHE, exc) + continue + if antwort.status_code in RETRY_STATUS: # zeitweilige Störung → genau ein Retry + _logger.warning( + "dedup/KI: Versuch %d/%d mit Status %d — wiederholt.", + versuch, VERSUCHE, antwort.status_code, + ) + continue + try: + antwort.raise_for_status() # andere 4xx: kein Retry hilft + except httpx.HTTPStatusError as exc: + _logger.warning("dedup/KI: Anfrage abgelehnt (%s) — nur Regelbefund.", exc) + return None + try: + inhalte = antwort.json()["choices"][0]["message"]["content"] + except Exception as exc: # kaputtes Antwort-Layout + _logger.warning("dedup/KI: unbrauchbare Antwort (%s)", exc) + return None + return validiere_antwort(extrahiere_json(inhalte)) + + _logger.warning("dedup/KI: alle %d Versuche fehlgeschlagen — nur Regelbefund.", VERSUCHE) + return None diff --git a/plugins/dedup/pruefung.py b/plugins/dedup/pruefung.py index bd33369..f3dd5aa 100644 --- a/plugins/dedup/pruefung.py +++ b/plugins/dedup/pruefung.py @@ -43,6 +43,10 @@ TITEL_SCHWELLE = 85 #: Literal, damit keine Import-Abhängigkeit zwischen den Plugins entsteht). STATUS_ABGESCHLOSSEN = "abgeschlossen" +#: Obergrenze für LLM-Aufrufe je Prüfung (Token sparen): höchstens so viele +#: Grenzfall-Paarungen plus ein Verlags-/Titel-Call bei Verlags-Konflikt. +MAX_KI_PAARUNGEN = 5 + #: Funktionswörter, die stark auf einen deutschen Spieltitel hindeuten. DEUTSCHE_INDIKATOREN = frozenset({ "der", "die", "das", "den", "dem", "des", @@ -142,6 +146,67 @@ def _verlags_schluessel(verlag: str) -> str: return normalisiere(verlag) +def _gemeinsame_woerter(a: str, b: str) -> int: + """Anzahl gemeinsamer Wort-Tokens zweiter Titel nach Normalisierung. + + Als „Titel-Score > 0, aber unter Schwelle“ gilt eine Wort-Überschneidung + über 0 bei Fuzzy-Score unter TITEL_SCHWELLE — reine Zeichenähnlichkeit + ohne gemeinsames Wort („Cascadia“ vs. „Everdell“) bleibt ein klarer + Non-Treffer und kostet kein LLM-Token. + """ + wa = set(normalisiere(a).split()) + wb = set(normalisiere(b).split()) + return len(wa & wb) + + +def _titel_score(titel_a: str, titel_b: str) -> int: + if fuzz is None: # pragma: no cover + return 0 + return int(fuzz.token_set_ratio(normalisiere(titel_a), normalisiere(titel_b))) + + +def _ki_bestaetigt(antwort: dict | None, konfidenz_min: float) -> bool: + """True, wenn ein KI-Befund übernommen wird (gleiches Spiel, genug Konfidenz).""" + return bool( + antwort + and antwort.get("ist_gleiches_spiel") + and float(antwort.get("konfidenz") or 0.0) >= konfidenz_min + ) + + +def _ki_protokoll( + typ: str, + titel_a: str, + titel_b: str, + ki_client, + antwort: dict | None, + *, + konfidenz_min: float, +) -> dict: + """Fasst einen KI-Aufruf für Audit-Log/Protokoll zusammen.""" + if antwort is None: + entscheidung = "fehler_verworfen" + konfidenz = None + elif not antwort.get("ist_gleiches_spiel"): + entscheidung = "verschiedene_spiele" + konfidenz = antwort.get("konfidenz") + elif float(antwort.get("konfidenz") or 0.0) < konfidenz_min: + entscheidung = "verworfen_unter_schwelle" + konfidenz = antwort.get("konfidenz") + else: + entscheidung = "gleiches_spiel" + konfidenz = antwort.get("konfidenz") + return { + "typ": typ, + "titel_a": titel_a[:300], + "titel_b": titel_b[:300], + "modell": getattr(ki_client, "modell", "") or "", + "konfidenz": konfidenz, + "entscheidung": entscheidung, + "begruendung": (antwort or {}).get("begruendung", ""), + } + + # ---------------- Prüfergebnis ---------------- #: Anzeigenamen der Konflikt-Arten (deutsche UI). @@ -181,6 +246,9 @@ class PruefErgebnis: verlags_optionen: list[str] = field(default_factory=list) #: Empfohlener (deutschester) Titel, falls Titel-Varianten bekannt sind. titel_empfehlung: str | None = None + #: Protokoll der KI-Zweitprüfung (Grenzfälle) — je Eintrag u. a. Modell, + #: Konfidenz und Entscheidung. Leer bei rein regelbasierten Prüfungen. + ki_befunde: list[dict] = field(default_factory=list) @classmethod def ohne_befund(cls, titel: str, verlag: str | None = None, bgg_id: int | None = None): @@ -206,6 +274,8 @@ class PruefErgebnis: and normalisiere(self.titel_empfehlung) != normalisiere(self.titel) ): zeilen.append(f"• Empfohlener Titel: „{self.titel_empfehlung}“") + if self.ki_befunde: + zeilen.append(f"• KI-Zweitprüfung: {len(self.ki_befunde)} Grenzfall/Grenzfälle bewertet.") return "\n".join(zeilen) @@ -269,12 +339,18 @@ def fuehre_pruefung_durch( verlag: str | None = None, bgg_id: int | None = None, user_id: int | None = None, + ki_client=None, + ki_konfidenz_min: float = 0.7, ) -> PruefErgebnis: """Führt alle vier Prüfungen durch und liefert ein PruefErgebnis. `bgg_hilfsclient` darf None sein (BGG-Zusatzdaten deaktiviert) — dann entfallen Alternate-Names und Vorgänger-Relationen, die Heuristik und der Fuzzy-Match arbeiten weiter. + `ki_client` darf None sein (KI-Zweitprüfung deaktiviert/fehlerhaft) — + dann läuft ausschließlich die Regelprüfung; sonst werden nur Grenzfälle + (Titel-Score > 0 unter Schwelle bzw. Verlags-Konflikt) dem LLM vorgelegt + und bestätigte Treffer in die Regelprüfungen zurückgeführt. `user_id` ist der Eintragende: eigene Planungseinträge lösen keinen „schon in Planung“-Konflikt aus. """ @@ -284,6 +360,7 @@ def fuehre_pruefung_durch( ergebnis = PruefErgebnis.ohne_befund(titel, verlag, id_wert) if not titel: return ergebnis + konfidenz_min = max(0.0, min(1.0, float(ki_konfidenz_min))) with session_factory() as db: neuheiten = _lade_zeilen(db, "neuheiten", ("id", "titel", "verlag", "bgg_id")) @@ -304,12 +381,58 @@ def fuehre_pruefung_durch( except Exception as exc: _logger.warning("dedup: BGG-Vorgänger nicht verfügbar (%s)", exc) - # Kandidaten = Einträge zum selben Spiel (Neuheiten + Planungsliste). + # Kandidaten = Einträge zum selben Spiel laut Regeln (Neuheiten + Planung). + alle_zeilen = (*neuheiten, *planungen) kandidaten = [ - zeile for zeile in (*neuheiten, *planungen) + zeile for zeile in alle_zeilen if _gleiches_spiel(zeile, titel, id_wert) ] + # ---- KI-Zweitprüfung für Grenzfälle (optional, niemals blockierend) ---- + # Klare Treffer (oben) und klare Non-Treffer bleiben regelbasiert; nur + # Titel mit Wort-Überschneidung über 0, aber Fuzzy-Score unter der + # Schwelle, werden dem LLM vorgelegt. Bestätigte Treffer laufen in die + # üblichen Prüfungen a)–d) zurück, als wären sie regelbasiert gematcht. + ki_bestaetigt: dict[int, dict] = {} # id(zeile) -> KI-Antwort-Dict + ki_titel_vorschlag: str | None = None # erster übernommener deutscher Titel + if ki_client is not None and alle_zeilen: + kandidat_ids = {id(zeile) for zeile in kandidaten} + grenzfaelle = [] + for zeile in alle_zeilen: + if id(zeile) in kandidat_ids: + continue + fremd_titel = (zeile.get("titel") or "").strip() + if not fremd_titel: + continue + if _gemeinsame_woerter(titel, fremd_titel) <= 0: + continue # klarer Non-Treffer — kein LLM-Token wert + score = _titel_score(titel, fremd_titel) + if score >= TITEL_SCHWELLE: + continue # wäre ohnehin regelbasiert gematcht + grenzfaelle.append((score, fremd_titel, zeile)) + grenzfaelle.sort(key=lambda eintrag: -eintrag[0]) + for score, fremd_titel, zeile in grenzfaelle[:MAX_KI_PAARUNGEN]: + try: + antwort = ki_client.bewerte( + titel_a=titel, titel_b=fremd_titel, + verlag_a=verlag, verlag_b=zeile.get("verlag"), + alternativen_a=alternativen, + expansionen_a=[name for _, name in vorgaenger], + ) + except Exception as exc: # Zusatzdaten dürfen nie blockieren + _logger.warning("dedup/KI: Grenzfall-Bewertung fehlgeschlagen (%s)", exc) + antwort = None + ergebnis.ki_befunde.append(_ki_protokoll( + "grenzfall", titel, fremd_titel, ki_client, antwort, + konfidenz_min=konfidenz_min, + )) + if _ki_bestaetigt(antwort, konfidenz_min): + kandidaten.append(zeile) + ki_bestaetigt[id(zeile)] = antwort + if antwort.get("deutscher_titel"): + ki_titel_vorschlag = ki_titel_vorschlag or antwort["deutscher_titel"] + + # a) Verlags-Konflikt: gleiches Spiel unter anderem Verlag/Vertrieb. bekannte_verlage: dict[str, str] = {} for zeile in kandidaten: @@ -340,8 +463,9 @@ def fuehre_pruefung_durch( zeile["titel"] for zeile in kandidaten if normalisiere(zeile.get("titel") or "") != normalisiere(titel) ] - ergebnis.titel_empfehlung = bevorzuge_deutschen_titel( - [titel, *varianten, *alternativen] + ergebnis.titel_empfehlung = ( + ki_titel_vorschlag # KI-Empfehlung (bestätigter Grenzfall) hat Vorrang + or bevorzuge_deutschen_titel([titel, *varianten, *alternativen]) ) if varianten and ergebnis.titel_empfehlung \ and normalisiere(ergebnis.titel_empfehlung) != normalisiere(titel): @@ -361,12 +485,16 @@ def fuehre_pruefung_durch( # c) Spiel oder Vorgänger bereits besprochen (Planungsstatus „abgeschlossen“). besprochen = [z for z in planungen if z.get("status") == STATUS_ABGESCHLOSSEN] for zeile in besprochen: - if _gleiches_spiel(zeile, titel, id_wert): + if _gleiches_spiel(zeile, titel, id_wert) or id(zeile) in ki_bestaetigt: wer = _rezensent_name(zeile, namen) + details = {"planungseintrag": zeile["id"], "titel": zeile["titel"], "rezensent": wer} + if id(zeile) in ki_bestaetigt: + details["ki_bestaetigt"] = True + details["ki_konfidenz"] = ki_bestaetigt[id(zeile)].get("konfidenz") ergebnis.konflikte.append(TitelKonflikt( art="besprochen", beschreibung=f"„{zeile['titel']}“ wurde bereits besprochen ({wer}).", - details={"planungseintrag": zeile["id"], "titel": zeile["titel"], "rezensent": wer}, + details=details, )) for fremd_id, fremd_titel in vorgaenger: for zeile in besprochen: @@ -394,20 +522,102 @@ def fuehre_pruefung_durch( for zeile in planungen: if user_id is not None and zeile.get("rezensent_id") == user_id: continue - if _gleiches_spiel(zeile, titel, id_wert): + if _gleiches_spiel(zeile, titel, id_wert) or id(zeile) in ki_bestaetigt: wer = _rezensent_name(zeile, namen) status = zeile.get("status") or "" + details = { + "planungseintrag": zeile["id"], + "rezensent": wer, + "status": status, + } + if id(zeile) in ki_bestaetigt: + details["ki_bestaetigt"] = True + details["ki_konfidenz"] = ki_bestaetigt[id(zeile)].get("konfidenz") ergebnis.konflikte.append(TitelKonflikt( art="planung", beschreibung=( f"„{zeile['titel']}“ ist bereits in der Planungsliste " f"von {wer} (Status: {status})." ), - details={ - "planungseintrag": zeile["id"], - "rezensent": wer, - "status": status, - }, + details=details, )) + # ---- KI-Zweitprüfung Teil 2: Verlags-/Titel-Hinweis bei Verlags-Konflikt ---- + # Verlags-Aliase („Hans im Glück“ vs. „Hans im Glück Verlag“) und + # Übersetzungsfragen kann die Regelprüfung nicht entscheiden — ein + # ergänzender LLM-Call liefert Empfehlungen, ohne den Befund zu ändern. + if ki_client is not None and ergebnis.hat_verlagskonflikt: + _ki_verlags_empfehlung( + ergebnis, ki_client, konfidenz_min=konfidenz_min, + titel=titel, verlag=verlag, kandidaten=kandidaten, + alternativen=alternativen, vorgaenger=vorgaenger, + ) + return ergebnis + + +def _uebernimm_deutschen_titel(ergebnis: PruefErgebnis, deutscher_titel: str | None, eigener_titel: str) -> None: + """Übernimmt einen KI-deutschen Titel als Empfehlung (konservativ). + + Nur wenn die Heuristik selbst nichts Deutsches fand (Empfehlung leer oder + gleich dem eigenen Titel), weißt das LLM die Redaktion ggf. besser. + """ + if not deutscher_titel: + return + aktuell = ergebnis.titel_empfehlung + if aktuell and normalisiere(aktuell) != normalisiere(eigener_titel): + return # Heuristik/KI hat bereits eine abweichende deutsche Variante gewählt + if normalisiere(deutscher_titel) == normalisiere(eigener_titel): + return + ergebnis.titel_empfehlung = deutscher_titel + + +def _ki_verlags_empfehlung( + ergebnis: PruefErgebnis, + ki_client, + *, + konfidenz_min: float, + titel: str, + verlag: str | None, + kandidaten: list[dict], + alternativen: list[str], + vorgaenger: list[tuple[int, str]], +) -> None: + """Ein ergänzender KI-Call zum Verlags-Konflikt (Verlags-Aliase, Übersetzung).""" + fremd_zeile = next( + (z for z in kandidaten + if (z.get("verlag") or "").strip() and verlag + and _verlags_schluessel(z["verlag"]) != _verlags_schluessel(verlag)), + None, + ) + if fremd_zeile is None: + return + fremd_titel = (fremd_zeile.get("titel") or "").strip() + try: + antwort = ki_client.bewerte( + titel_a=titel, titel_b=fremd_titel, + verlag_a=verlag, verlag_b=fremd_zeile.get("verlag"), + alternativen_a=alternativen, + expansionen_a=[name for _, name in vorgaenger], + frage=( + "Beide Einträge meinen laut Regelprüfung dasselbe Spiel, wurden " + "aber unter verschiedenen Verlagen geführt (ggf. Alias oder " + "Lokalausgabe). Welcher Verlag sollte geführt werden " + "(empfohlener_verlag) und welchen deutschen Titel gibt es?" + ), + ) + except Exception as exc: # Zusatzdaten dürfen nie blockieren + _logger.warning("dedup/KI: Verlags-Bewertung fehlgeschlagen (%s)", exc) + antwort = None + ergebnis.ki_befunde.append(_ki_protokoll( + "verlag", titel, fremd_titel, ki_client, antwort, + konfidenz_min=konfidenz_min, + )) + if _ki_bestaetigt(antwort, konfidenz_min): + verlags_konflikt = next( + (k for k in ergebnis.konflikte if k.art == "verlag"), None + ) + if verlags_konflikt is not None and antwort.get("empfohlener_verlag"): + verlags_konflikt.details["ki_empfohlener_verlag"] = antwort["empfohlener_verlag"] + verlags_konflikt.details["ki_konfidenz"] = antwort.get("konfidenz") + _uebernimm_deutschen_titel(ergebnis, antwort.get("deutscher_titel"), titel) diff --git a/tests/conftest.py b/tests/conftest.py index 5ea5bc6..1c89f9d 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -28,6 +28,9 @@ def _hintergrundjobs_deaktiviert(monkeypatch): monkeypatch.setenv("SPIELE_ERINNERUNG_JOB_AKTIV", "0") # Auch der BGG-Hilfsclient des dedup-Plugins bleibt in Tests offline. monkeypatch.setenv("SPIELE_DEDUP_BGG_AKTIV", "0") + # Die LLM-Zweitprüfung des dedup-Plugins bleibt in Tests aus; einzelne + # Tests schalten sie explizit per SPIELE_DEDUP_LLM_AKTIV=1 ein. + monkeypatch.setenv("SPIELE_DEDUP_LLM_AKTIV", "0") @pytest.fixture diff --git a/tests/test_dedup_ki.py b/tests/test_dedup_ki.py new file mode 100644 index 0000000..0577be1 --- /dev/null +++ b/tests/test_dedup_ki.py @@ -0,0 +1,527 @@ +"""Tests: KI-Zweitprüfung des Plugins „dedup“. + +Alle LLM-Antworten werden über httpx.MockTransport bzw. Fake-Clients gestellt — +in CI läuft niemals ein echter LLM-Call. Getestet werden: Konfiguration, +JSON-Parsing-Robustheit (Code-Fences, Zusatztext), Retry-/Fallback-Verhalten, +Konfidenz-Schwelle und die Rückführung bestätigter Treffer in die Regelprüfung. +""" +from __future__ import annotations + +import asyncio +import json + +import httpx +import pytest +from fastapi.testclient import TestClient +from sqlalchemy import select + +from redaktionskern.app import create_app + +from plugins.dedup.ki_pruefung import ( + VERSUCHE, + KiKonfiguration, + KiPruefClient, + extrahiere_json, + lade_konfiguration, + validiere_antwort, +) + +from tests.conftest import lege_benutzer_an +from tests.test_dedup import benutzer_objekt, neuheit_anlegen, planung_anlegen, plugin_modul + + +# ---------------- Hilfen ---------------- + +def gueltige_llm_antwort(inhalt: str, status: int = 200) -> httpx.Response: + """Wrappt einen Assistant-Text in eine Chat-Completions-Antwort.""" + body = {"choices": [{"message": {"role": "assistant", "content": inhalt}}]} + return httpx.Response(status, content=json.dumps(body).encode()) + + +def mock_transport(antworten: list) -> tuple[httpx.MockTransport, dict]: + """Transport mit gestaffelten Antworten: str = Assistant-Inhalt, + int = HTTP-Status, Exception = wird geworfen. Zählt die Aufrufe mit.""" + zaehler = {"aufrufe": 0} + + def handler(request: httpx.Request) -> httpx.Response: + index = min(zaehler["aufrufe"], len(antworten) - 1) + zaehler["aufrufe"] += 1 + spezifikation = antworten[index] + if isinstance(spezifikation, Exception): + raise spezifikation + if isinstance(spezifikation, int): + return httpx.Response(spezifikation, json={"error": "kaputt"}) + return gueltige_llm_antwort(spezifikation) + + return httpx.MockTransport(handler), zaehler + + +class FakeKiClient: + """Ersetzt den KI-Client komplett: feste Antwort, zählt Aufrufe.""" + + modell = "fake-modell" + + def __init__(self, antwort=None): + self.antwort = antwort + self.aufrufe: list[dict] = [] + + def bewerte(self, **kwargs): + self.aufrufe.append(kwargs) + return self.antwort + + def schliessen(self): + pass + + +GLEICHES_SPIEL_HOCH = { + "ist_gleiches_spiel": True, "konfidenz": 0.9, + "begruendung": "Beides El Grande-Jubiläumsausgaben.", + "empfohlener_verlag": None, "deutscher_titel": None, +} + + +def app_mit_ki(settings, monkeypatch, *, konfidenz_min=None): + """App, deren Plugin-Umgebung die KI-Zweitprüfung aktiviert.""" + monkeypatch.setenv("SPIELE_DEDUP_LLM_AKTIV", "1") + monkeypatch.setenv("SPIELE_DEDUP_LLM_BASIS_URL", "https://llm.example.test/v1") + monkeypatch.setenv("SPIELE_DEDUP_LLM_API_KEY", "test-key") + monkeypatch.setenv("SPIELE_DEDUP_LLM_MODELL", "test-modell") + if konfidenz_min is not None: + monkeypatch.setenv("SPIELE_DEDUP_LLM_KONFIDENZ_MIN", str(konfidenz_min)) + return create_app(settings) + + +def pruefe_mit_ki(app, antworten, *args, **kwargs): + """check_titel mit per Fabrik eingehängtem Mock-LLM (Antwort-Stafette).""" + transport, zaehler = mock_transport(antworten) + dedup = app.state.registry.get("dedup") + dedup.ki_client_fabrik = lambda: KiPruefClient( + KiKonfiguration( + aktiv=True, basis_url="https://llm.example.test/v1", + api_key="test-key", modell="test-modell", + konfidenz_min=dedup.ki_konfiguration.konfidenz_min, + ), + transport=transport, + ) + try: + ergebnis = asyncio.run(dedup.check_titel(*args, **kwargs)) + finally: + dedup.ki_client_fabrik = None + return ergebnis, zaehler + + +# ---------------- Konfiguration ---------------- + +def test_konfiguration_standard_aus(): + k = lade_konfiguration({}) + assert not k.aktiv + assert not k.vollstaendig + assert k.konfidenz_min == 0.7 + assert k.timeout >= 1.0 + + +def test_konfiguration_aus_env_und_fehlertolerant(): + k = lade_konfiguration({ + "SPIELE_DEDUP_LLM_AKTIV": "1", + "SPIELE_DEDUP_LLM_BASIS_URL": "https://api.example.test/v1/", + "SPIELE_DEDUP_LLM_API_KEY": " sk-123 ", + "SPIELE_DEDUP_LLM_MODELL": " gpt-test ", + # Kaputte Zahlen fallen auf die Standards zurück: + "SPIELE_DEDUP_LLM_KONFIDENZ_MIN": "abc", + "SPIELE_DEDUP_LLM_TIMEOUT_SEKUNDEN": "-5", + }) + assert k.vollstaendig + assert k.basis_url == "https://api.example.test/v1" + assert k.api_key == "sk-123" + assert k.modell == "gpt-test" + assert k.konfidenz_min == 0.7 # Standard bei Unsinn + assert k.timeout >= 1.0 + + +def test_konfiguration_schwellenwerte_werden_geklemmt(): + k = lade_konfiguration({"SPIELE_DEDUP_LLM_KONFIDENZ_MIN": "1.5"}) + assert k.konfidenz_min == 1.0 + k = lade_konfiguration({"SPIELE_DEDUP_LLM_TIMEOUT_SEKUNDEN": "3.5"}) + assert k.timeout == 3.5 + + +# ---------------- JSON-Parsing-Robustheit ---------------- + +def test_extrahiere_json_plain(): + assert extrahiere_json('{"ist_gleiches_spiel": true}') == {"ist_gleiches_spiel": True} + + +def test_extrahiere_json_code_fence(): + text = 'Hier die Bewertung:\n\n```json\n{"ist_gleiches_spiel": false}\n```\n\nViele Grüße' + assert extrahiere_json(text) == {"ist_gleiches_spiel": False} + + +def test_extrahiere_json_text_um_json(): + text = 'Klar! {"ist_gleiches_spiel": true, "konfidenz": 0.8} — hoffe das hilft.' + assert extrahiere_json(text)["konfidenz"] == 0.8 + + +def test_extrahiere_json_unbrauchbar(): + assert extrahiere_json("Ich weiß es leider nicht.") is None + assert extrahiere_json("[1, 2, 3]") is None # kein Objekt + assert extrahiere_json("") is None + assert extrahiere_json(None) is None + + +def test_validiere_antwort_normalisiert(): + roh = { + "ist_gleiches_spiel": "ja", # Text statt Bool + "konfidenz": 92, # Prozent-Skala + "begruendung": " Gleiche Edition. ", + "empfohlener_verlag": " Kosmos ", + "deutscher_titel": "", + } + sauber = validiere_antwort(roh) + assert sauber == { + "ist_gleiches_spiel": True, + "konfidenz": 0.92, + "begruendung": "Gleiche Edition.", + "empfohlener_verlag": "Kosmos", + "deutscher_titel": None, + } + + +@pytest.mark.parametrize("roh", [ + {"ist_gleiches_spiel": True}, # konfidenz fehlt + {"ist_gleiches_spiel": "vielleicht", "konfidenz": 0.9}, + {"ist_gleiches_spiel": True, "konfidenz": "hoch"}, + {"ist_gleiches_spiel": True, "konfidenz": 7}, # weder 0–1 noch 0–100 + ["kein", "objekt"], +]) +def test_validiere_antwort_lehnt_unbrauchbares_ab(roh): + assert validiere_antwort(roh) is None + + +# ---------------- Client: Retry, Fehler, Header ---------------- + +KONFIG = KiKonfiguration( + aktiv=True, basis_url="https://llm.example.test/v1", + api_key="geheim", modell="test-modell", +) + +ANTWORT_JSON = '{"ist_gleiches_spiel": true, "konfidenz": 0.9, "begruendung": "b", ' \ + '"empfohlener_verlag": null, "deutscher_titel": null}' + + +def test_client_sendet_modell_und_key(): + gesehen = {} + + def handler(request: httpx.Request) -> httpx.Response: + gesehen["url"] = str(request.url) + gesehen["auth"] = request.headers.get("Authorization") + gesehen["body"] = json.loads(request.content.decode()) + return gueltige_llm_antwort(ANTWORT_JSON) + + client = KiPruefClient(KONFIG, transport=httpx.MockTransport(handler)) + try: + ergebnis = client.bewerte(titel_a="A", titel_b="B") + finally: + client.schliessen() + assert ergebnis["ist_gleiches_spiel"] is True + assert gesehen["url"].endswith("/v1/chat/completions") + assert gesehen["auth"] == "Bearer geheim" + assert gesehen["body"]["model"] == "test-modell" + + +def test_client_retryt_genau_einmal_bei_serverstoerung(): + transport, zaehler = mock_transport([500, ANTWORT_JSON]) + client = KiPruefClient(KONFIG, transport=transport) + try: + ergebnis = client.bewerte(titel_a="A", titel_b="B") + finally: + client.schliessen() + assert ergebnis is not None + assert zaehler["aufrufe"] == 2 # erster Versuch + genau ein Retry + + +def test_client_gibt_nach_zweitem_fehler_auf(): + transport, zaehler = mock_transport([503, 503]) + client = KiPruefClient(KONFIG, transport=transport) + try: + assert client.bewerte(titel_a="A", titel_b="B") is None + finally: + client.schliessen() + assert zaehler["aufrufe"] == VERSUCHE + + +def test_client_timeout_wird_genau_einmal_wiederholt(): + transport, zaehler = mock_transport([ + httpx.ReadTimeout("zu langsam"), httpx.ReadTimeout("weiter zu langsam"), + ]) + client = KiPruefClient(KONFIG, transport=transport) + try: + assert client.bewerte(titel_a="A", titel_b="B") is None + finally: + client.schliessen() + assert zaehler["aufrufe"] == 2 + + +def test_client_kein_retry_bei_clientfehler_oder_kaputtem_json(): + # 404 ist nicht wiederholbar: + transport, zaehler = mock_transport([404, 404]) + client = KiPruefClient(KONFIG, transport=transport) + try: + assert client.bewerte(titel_a="A", titel_b="B") is None + finally: + client.schliessen() + assert zaehler["aufrufe"] == 1 + + # Kaputtes JSON hilft kein Retry: + transport, zaehler = mock_transport(["Leider weiß ich das nicht."]) + client = KiPruefClient(KONFIG, transport=transport) + try: + assert client.bewerte(titel_a="A", titel_b="B") is None + finally: + client.schliessen() + assert zaehler["aufrufe"] == 1 + + +def test_client_ohne_vollstaendige_konfig_ruft_nichts(): + transport, zaehler = mock_transport([ANTWORT_JSON]) + leer = KiKonfiguration(aktiv=True, basis_url="", api_key="", modell="") + client = KiPruefClient(leer, transport=transport) + try: + assert client.bewerte(titel_a="A", titel_b="B") is None + finally: + client.schliessen() + assert zaehler["aufrufe"] == 0 + + +# ---------------- Integration: Fallback & Verhalten ---------------- + +def test_ohne_ki_verhaelt_sich_alles_wie_bisher(settings, monkeypatch): + """Standard (aktiv=0): Grenzfälle bleiben regelbasiert, kein LLM-Aufruf.""" + app = create_app(settings) + lege_benutzer_an(app, "rez1", "rezensent") + rez1 = benutzer_objekt(app, "rez1") + planung_anlegen(app, "El Grande Big Box", rez1.id) + + dedup = app.state.registry.get("dedup") + + def aufschrei(): # dürfte nie aufgerufen werden + raise AssertionError("KI war deaktiviert, wurde aber aufgerufen!") + + dedup.ki_client_fabrik = lambda: aufschrei() + try: + ergebnis = asyncio.run(dedup.check_titel("El Grande: 25 Jahre")) + finally: + dedup.ki_client_fabrik = None + assert not ergebnis.hat_konflikte + assert ergebnis.ki_befunde == [] + # Empfehlung bleibt beim eigenen Titel (Heuristik ohne Varianten): + assert ergebnis.titel_empfehlung == "El Grande: 25 Jahre" + + +def test_aktiv_ohne_key_faellt_auf_regeln_zurueck(settings, monkeypatch): + """AKTIV=1, aber kein API-Key → Zweitprüfung bleibt aus, kein Fehler.""" + monkeypatch.setenv("SPIELE_DEDUP_LLM_AKTIV", "1") + monkeypatch.setenv("SPIELE_DEDUP_LLM_BASIS_URL", "https://llm.example.test/v1") + # bewusst KEIN SPIELE_DEDUP_LLM_API_KEY / MODELL: + app = create_app(settings) + lege_benutzer_an(app, "rez1", "rezensent") + rez1 = benutzer_objekt(app, "rez1") + planung_anlegen(app, "El Grande Big Box", rez1.id) + + dedup = app.state.registry.get("dedup") + dedup.ki_client_fabrik = lambda: (_ for _ in ()).throw( + AssertionError("Ohne Key darf kein KI-Client gebaut werden.") + ) + try: + ergebnis = asyncio.run(dedup.check_titel("El Grande: 25 Jahre")) + finally: + dedup.ki_client_fabrik = None + assert not ergebnis.hat_konflikte + assert ergebnis.ki_befunde == [] + + +def test_llm_fehler_blockiert_nie(settings, monkeypatch): + """Netzwerk-/Server-Probleme: Ergebnis identisch zur reinen Regelprüfung.""" + app = app_mit_ki(settings, monkeypatch) + lege_benutzer_an(app, "rez1", "rezensent") + rez1 = benutzer_objekt(app, "rez1") + planung_anlegen(app, "El Grande Big Box", rez1.id, verlag="Rio Grande Games") + + ergebnis, zaehler = pruefe_mit_ki( + app, + [httpx.ConnectError("offline")] * 6, + "El Grande: 25 Jahre", "Devir", None, + ) + assert zaehler["aufrufe"] >= 1 + # Ohne funktionierende Zweitprüfung bleibt es beim Regelbefund (hier leer — + # die Grenzfall-Paarung matchen die Regeln ja gerade nicht): + assert not ergebnis.hat_konflikte + assert all(b["entscheidung"] == "fehler_verworfen" for b in ergebnis.ki_befunde) + + # Exakt dasselbe Ergebnis wie ohne KI: + dedup = app.state.registry.get("dedup") + referenz = asyncio.run(dedup.check_titel("El Grande: 25 Jahre", "Devir", None)) + assert referenz.als_liste() == ergebnis.als_liste() + assert referenz.titel_empfehlung == ergebnis.titel_empfehlung + assert referenz.verlags_optionen == ergebnis.verlags_optionen + + +# ---------------- Integration: Grenzfälle & Schwelle ---------------- + +def test_grenzfall_wird_durch_ki_bestaetigt(settings, monkeypatch): + """Big Box vs. Jubiläumsausgabe: Regeln finden nichts, das LLM schon.""" + app = app_mit_ki(settings, monkeypatch) + lege_benutzer_an(app, "rez1", "rezensent") + rez1 = benutzer_objekt(app, "rez1") + planung_anlegen(app, "El Grande Big Box", rez1.id) + + antwort = dict(GLEICHES_SPIEL_HOCH) + antwort["begruendung"] = "Beides sind El Grande-Ausgaben der Jubiläumsreihe." + ergebnis, _ = pruefe_mit_ki(app, [json.dumps(antwort)], "El Grande: 25 Jahre") + + arten = [k.art for k in ergebnis.konflikte] + assert "planung" in arten + planungs_konflikt = next(k for k in ergebnis.konflikte if k.art == "planung") + assert planungs_konflikt.details["ki_bestaetigt"] is True + assert planungs_konflikt.details["ki_konfidenz"] == 0.9 + + assert ergebnis.ki_befunde[0]["entscheidung"] == "gleiches_spiel" + assert ergebnis.ki_befunde[0]["modell"] == "test-modell" + assert ergebnis.ki_befunde[0]["konfidenz"] == 0.9 + + +def test_grenzfall_unter_konfidenzschwelle_wird_verworfen(settings, monkeypatch): + """Gleicher Befund, aber Konfidenz 0.5 < 0.7 → nur der Regelbefund zählt.""" + app = app_mit_ki(settings, monkeypatch) + lege_benutzer_an(app, "rez1", "rezensent") + rez1 = benutzer_objekt(app, "rez1") + planung_anlegen(app, "El Grande Big Box", rez1.id) + + antwort = dict(GLEICHES_SPIEL_HOCH, konfidenz=0.5) + ergebnis, _ = pruefe_mit_ki(app, [json.dumps(antwort)], "El Grande: 25 Jahre") + assert not ergebnis.hat_konflikte + assert ergebnis.ki_befunde[0]["entscheidung"] == "verworfen_unter_schwelle" + + +def test_konfidenzschwelle_laesst_sich_per_env_verschieben(settings, monkeypatch): + app = app_mit_ki(settings, monkeypatch, konfidenz_min=0.95) + lege_benutzer_an(app, "rez1", "rezensent") + rez1 = benutzer_objekt(app, "rez1") + planung_anlegen(app, "El Grande Big Box", rez1.id) + + antwort = dict(GLEICHES_SPIEL_HOCH) # 0.9 reicht jetzt nicht mehr + ergebnis, _ = pruefe_mit_ki(app, [json.dumps(antwort)], "El Grande: 25 Jahre") + assert not ergebnis.hat_konflikte + assert ergebnis.ki_befunde[0]["entscheidung"] == "verworfen_unter_schwelle" + + +def test_llm_sagt_verschiedene_spiele(settings, monkeypatch): + app = app_mit_ki(settings, monkeypatch) + lege_benutzer_an(app, "rez1", "rezensent") + rez1 = benutzer_objekt(app, "rez1") + planung_anlegen(app, "El Grande Big Box", rez1.id) + + antwort = dict(GLEICHES_SPIEL_HOCH, ist_gleiches_spiel=False, begruendung="Big Box ≠ Jubiläum.") + ergebnis, _ = pruefe_mit_ki(app, [json.dumps(antwort)], "El Grande: 25 Jahre") + assert not ergebnis.hat_konflikte + assert ergebnis.ki_befunde[0]["entscheidung"] == "verschiedene_spiele" + + +def test_uebersetzungsgrenzfall_kriegt_deutschen_titel(settings, monkeypatch): + """Catan Fast Edition vs. Catan – Das schnelle Spiel: KI liefert den deutschen Titel.""" + app = app_mit_ki(settings, monkeypatch) + neuheit_anlegen(app, "El Grande: 25 Jahre", verlag="Hans im Glück", bgg_id=111) + + antwort = dict(GLEICHES_SPIEL_HOCH, deutscher_titel="El Grande: 25 Jahre") + ergebnis, _ = pruefe_mit_ki(app, [json.dumps(antwort)], "El Grande Big Box") + + # Ohne KI wäre hier nichts gefunden worden (beide Titel heuristisch neutral); + # mit KI gibt es eine Titel-Variante samt deutscher Empfehlung: + assert "titel" in [k.art for k in ergebnis.konflikte] + assert ergebnis.titel_empfehlung == "El Grande: 25 Jahre" + + dedup = app.state.registry.get("dedup") + referenz = asyncio.run(dedup.check_titel("El Grande Big Box")) + assert not referenz.hat_konflikte + # Ohne KI bleibt die Empfehlung beim eigenen Titel: + assert referenz.titel_empfehlung == "El Grande Big Box" + + +def test_verlagskonflikt_kriegt_ki_empfehlung(settings, monkeypatch): + """Verlags-Alias: Regelprüfung findet den Konflikt, das LLM den richtigen Verlag.""" + app = app_mit_ki(settings, monkeypatch) + neuheit_anlegen(app, "El Grande", verlag="Hans im Glück Verlag", bgg_id=21) + + antwort = dict(GLEICHES_SPIEL_HOCH, empfohlener_verlag="Hans im Glück Verlag") + ergebnis, _ = pruefe_mit_ki(app, [json.dumps(antwort)], "El Grande", "Hans im Glück", 21) + + assert ergebnis.hat_verlagskonflikt + verlags_konflikt = next(k for k in ergebnis.konflikte if k.art == "verlag") + assert verlags_konflikt.details["ki_empfohlener_verlag"] == "Hans im Glück Verlag" + assert ergebnis.ki_befunde[-1]["typ"] == "verlag" + + +# ---------------- Integration: Budget & Audit ---------------- + +def test_ki_budget_begrenzt_anzahl_aufrufe(settings, monkeypatch): + from plugins.dedup.pruefung import MAX_KI_PAARUNGEN + + app = app_mit_ki(settings, monkeypatch) + for nr in range(MAX_KI_PAARUNGEN + 3): + neuheit_anlegen(app, f"Catan Ausgabe Nummer {nr}", bgg_id=200 + nr) + + fake = FakeKiClient(dict(GLEICHES_SPIEL_HOCH)) + dedup = app.state.registry.get("dedup") + dedup.ki_client_fabrik = lambda: fake + try: + ergebnis = asyncio.run(dedup.check_titel("Catan Neuauflage")) + finally: + dedup.ki_client_fabrik = None + + assert len(fake.aufrufe) <= MAX_KI_PAARUNGEN + assert len(ergebnis.ki_befunde) == len(fake.aufrufe) + + +def test_ki_befunde_landen_im_audit_log(settings, monkeypatch): + app = app_mit_ki(settings, monkeypatch) + lege_benutzer_an(app, "rez1", "rezensent") # fremder Eintrag → echter Konflikt + lege_benutzer_an(app, "anna", "redakteur") # prüfende Person + rez1 = benutzer_objekt(app, "rez1") + anna = benutzer_objekt(app, "anna") + planung_anlegen(app, "El Grande Big Box", rez1.id) + + fake = FakeKiClient(dict(GLEICHES_SPIEL_HOCH)) + dedup = app.state.registry.get("dedup") + dedup.ki_client_fabrik = lambda: fake + try: + ergebnis = asyncio.run(dedup.check_titel("El Grande: 25 Jahre", user=anna)) + finally: + dedup.ki_client_fabrik = None + assert ergebnis.hat_konflikte + + AuditEintrag = plugin_modul(app, "audit-log").AuditEintrag + with app.state.session_factory() as db: + eintraege = db.scalars( + select(AuditEintrag).where(AuditEintrag.objekt_typ == "dedup_ki") + ).all() + assert len(eintraege) == 1 + eintrag = eintraege[0] + assert eintrag.action == "geprüft" + assert eintrag.actor_name == "anna" + assert eintrag.details["modell"] == "fake-modell" + befaelle = eintrag.details["befunde"] + assert befaelle[0]["entscheidung"] == "gleiches_spiel" + assert befaelle[0]["konfidenz"] == 0.9 + assert befaelle[0]["titel_a"] == "El Grande: 25 Jahre" + + +def test_keine_ki_befunde_heissen_kein_audit_eintrag(settings, monkeypatch): + """Rein regelbasierte Prüfung (KI aus) schreibt keinen dedup_ki-Eintrag.""" + app = create_app(settings) + asyncio.run(app.state.registry.get("dedup").check_titel("Beliebiges Spiel")) + + AuditEintrag = plugin_modul(app, "audit-log").AuditEintrag + with app.state.session_factory() as db: + eintraege = db.scalars( + select(AuditEintrag).where(AuditEintrag.objekt_typ == "dedup_ki") + ).all() + assert eintraege == []