diff --git a/README.md b/README.md index ca641a8..f951561 100644 --- a/README.md +++ b/README.md @@ -65,6 +65,12 @@ Beim ersten Start wird automatisch ein Admin-Konto angelegt: | `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_NEUHEITEN_KI_AKTIV` | `0` | Optionale LLM-Hilfsstufe des Web-Quellen-Crawlers an (`1`) oder aus (`0`) | +| `SPIELE_NEUHEITEN_KI_BASIS_URL` | *(leer)* | Basis-URL einer OpenAI-kompatiblen Chat-Completions-API, z. B. `https://api.openai.com/v1` | +| `SPIELE_NEUHEITEN_KI_API_KEY` | *(leer)* | API-Key, wird als `Authorization: Bearer …` gesendet | +| `SPIELE_NEUHEITEN_KI_MODELL` | *(leer)* | Modellname, z. B. `gpt-4o-mini` | +| `SPIELE_NEUHEITEN_KI_KONFIDENZ_MIN` | `0.7` | Mindest-Konfidenz; darunter wird das KI-Ergebnis verworfen und der klassische Parser-Befund behalten | +| `SPIELE_NEUHEITEN_KI_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`) | @@ -331,6 +337,44 @@ Autor, Erscheinungsjahr, Status, aktualisiert am): - Jeder Titel verlinkt direkt auf den BGG-Eintrag. - Sync-Steuerung nur für Rolle Admin/Redakteur. +### Web-Quellen-Crawler mit optionaler KI-Hilfsstufe + +Neben BGG crawlt das Plugin zusätzlich redaktionell gepflegte Web-Quellen +(spielbox.de, brettspielbox.de, spiel-essen.de, cliquenabend.de) über je einen +Adapter in `plugins/neuheiten/quellen/` — mit Rate-Limit (1 Request/s), +Retry/Backoff, konditionalen Requests (ETag/304), Fehler-Isolation pro Quelle +und Duplikat-Gate gegen die BGG-Liste. Der Crawler arbeitet standardmäßig rein +regelbasiert. + +Optional kann eine **LLM-Hilfsstufe** (`plugins/neuheiten/quellen/ki_hilfe.py`) +zugeschaltet werden — konsistent zur Dedup-KI über die +`SPIELE_NEUHEITEN_KI_*`-Variablen (siehe Tabelle oben). Das LLM wird nur in +zwei Situationen genutzt: + +- **Struktur-Erkennung:** Liefert der Regel-Parser keine oder verdächtig wenige + Einträge (0 Treffer; 1–2 Treffer auf einer sehr linkreichen Seite), darf das + LLM aus dem gelieferten HTML Pagination-/Filter-Folge-URLs vorschlagen. + Der Vorschlag wird im Code hart gefiltert: nur URLs derselben Domain wie die + geparste Seite, maximal 20 je Seite, bereits besuchte/geplante URLs werden + übersprungen, nicht erreichbare Vorschläge brechen den Lauf nicht ab — + Schutz vor Rate-Limit-Überlastung und Domain-Ausbruch. +- **Feld-Extraktion:** Einzelne Treffer sind unsicher (fehlender Verlag oder + verdächtiger/generischer Titel). Nur solche Treffer gehen ans LLM, das + höchstens `titel`, `verlag` und `autor` korrigiert — die URL ist strukturell + nicht änderbar. + +Unter der Mindest-Konfidenz (`SPIELE_NEUHEITEN_KI_KONFIDENZ_MIN`, +Standard 0.7) wird jedes KI-Ergebnis verworfen; der klassische Parser-Befund +bleibt dann unverändert bestehen. + +- **Fallback-Pflicht:** Standardmäßig aus (`SPIELE_NEUHEITEN_KI_AKTIV=0`). + Ohne vollständige Konfiguration oder bei jedem Fehler (Timeout, + Server-Störung nach genau einem Retry, kaputtes JSON) läuft EXAKT der + klassische Crawler — ein KI-Fehler kann den Sync nie blockieren. +- **Audit:** KI-Eingriffe eines Sync-Laufs landen gebündelt im audit-log + (`korrigiert` / `neuheiten_ki_crawler`) mit Quelle, Modell, Konfidenz und + den jeweils korrigierten Feldern sowie den verfolgten KI-Folge-URLs. + ## Plugin „dedup“ (implementiert) Dedup-Prüfungen bei Import und händigem Eintrag. Die Prüf-Logik liegt diff --git a/plugins/neuheiten/__init__.py b/plugins/neuheiten/__init__.py index 8ec412d..6240071 100644 --- a/plugins/neuheiten/__init__.py +++ b/plugins/neuheiten/__init__.py @@ -340,12 +340,16 @@ class NeuheitenPlugin(BasePlugin): def _quellen_sync_ausfuehren(self) -> list[QuellenLaufZeile]: """Ein Lauf über alle aktiven Web-Quellen (Fehler pro Quelle isoliert).""" - dedup = ( - self.context.registry.get("dedup") + registry = ( + self.context.registry if self.context is not None and self.context.registry is not None else None ) - service = WebQuellenSyncService(self.context.session_factory, dedup=dedup) + dedup = registry.get("dedup") if registry is not None else None + audit = registry.get("audit-log") if registry is not None else None + service = WebQuellenSyncService( + self.context.session_factory, dedup=dedup, audit=audit + ) return service.laufe() def _quellen_sync_job(self) -> None: diff --git a/plugins/neuheiten/quellen/basis.py b/plugins/neuheiten/quellen/basis.py index 4df3e92..ab28a4d 100644 --- a/plugins/neuheiten/quellen/basis.py +++ b/plugins/neuheiten/quellen/basis.py @@ -21,6 +21,7 @@ die Testsuite führt keine echten Netzwerkaufrufe durch. """ from __future__ import annotations +import logging import re import time import xml.etree.ElementTree # noqa: F401 (dokumentiert: keine XML-Nutzung hier) @@ -31,6 +32,10 @@ from typing import ClassVar import httpx from rapidfuzz import fuzz +from .ki_hilfe import KiHilfe + +_logger = logging.getLogger(__name__) + #: Freundlicher User-Agent für alle Redaktions-Crawler (Courtesy-Regeln). USER_AGENT = "SpieleRedaktionBot/1.0 (Redaktions-Tool; Kontakt siehe Repo)" @@ -76,6 +81,12 @@ class SammelErgebnis: #: True = mindestens eine Seite kam per 304 als unverändert zurück; #: der Lauf wurde dann abgekürzt (Quelle übersprungen, kein Fehler). unveraendert: bool = False + #: KI-Feldkorrekturen dieses Laufs (Protokoll fürs Audit-Log); leer bei + #: rein regelbasiertem Lauf. + ki_eingriffe: list[dict] = field(default_factory=list) + #: Von der KI vorgeschlagene und tatsächlich verfolgte Folge-URLs + #: (Struktur-Erkennung); leer bei rein regelbasiertem Lauf. + ki_folge_urls: list[str] = field(default_factory=list) def jahr_aus_datumsangabe(angabe: str | None) -> int | None: @@ -236,18 +247,40 @@ class QuellenAdapter: def __init__(self, client: WebQuellenClient) -> None: self.client = client + #: Optionale LLM-Hilfsstufe (`ki_hilfe.KiHilfe` oder gleiches Protokoll). + #: None (Standard) = rein regelbasiert; KI-Fehler blockieren den Lauf nie. + self.ki_hilfe: KiHilfe | None = None def sammle(self) -> SammelErgebnis: - """Crawlt die Quelle: Startseiten + gefundene Folge-Links (BFS).""" + """Crawlt die Quelle: Startseiten + gefundene Folge-Links (BFS). + + Mit gesetzter `ki_hilfe`: Bei verdächtig leer geparsten Seiten darf + das LLM Pagination-/Filter-Folge-URLs vorschlagen (nur gleiche + Domain, max. 20), unsichere Treffer werden per LLM nachgebessert + (nur titel/verlag/autor, nie die URL). Jeder KI-Fehler fällt auf den + klassischen Befund zurück. + """ ergebnis = SammelErgebnis() warteschlange = list(self.start_urls) besucht: set[str] = set() + ki_vorschlaege: set[str] = set() while warteschlange and len(besucht) < self.max_seiten: url = warteschlange.pop(0) if url in besucht: continue besucht.add(url) - antwort = self.client.hole(url) + try: + antwort = self.client.hole(url) + except Exception: + if url in ki_vorschlaege: + # Eine nicht erreichbare KI-Suggestion darf den klassischen + # Lauf nie abbrechen — überspringen statt Fehler. + _logger.warning( + "KI-vorgeschlagene URL nicht erreichbar (%s) — übersprungen.", + url, + ) + continue + raise if antwort.status_code == 304: # Unverändert → Quelle überspringen (kein Fehler). ergebnis.unveraendert = True @@ -258,8 +291,48 @@ class QuellenAdapter: for link in folge_links: if link not in besucht: warteschlange.append(link) + if self.ki_hilfe is not None: + for link in self._ki_folge_urls(treffer, antwort.content, url, besucht): + if link not in besucht and link not in warteschlange: + warteschlange.append(link) + ki_vorschlaege.add(link) + ergebnis.ki_folge_urls.append(link) + if self.ki_hilfe is not None: + self._ki_nachbessern(ergebnis) return ergebnis + def _ki_folge_urls( + self, + treffer: list[QuellenTreffer], + inhalt: bytes, + url: str, + besucht: set[str], + ) -> list[str]: + """Struktur-Erkennung der Hilfsstufe (best effort — wirft nicht).""" + try: + return self.ki_hilfe.folge_urls_bei_bedarf( + len(treffer), inhalt, url, besucht + ) + except Exception: + _logger.exception( + "KI-Struktur-Erkennung fehlgeschlagen (%s) — klassischer Lauf bleibt.", + url, + ) + return [] + + def _ki_nachbessern(self, ergebnis: SammelErgebnis) -> None: + """Feld-Korrektur unsicherer Treffer (best effort — wirft nicht).""" + try: + korrigiert, eingriffe = self.ki_hilfe.treffer_nachbessern(ergebnis.treffer) + except Exception: + _logger.exception( + "KI-Feldkorrektur fehlgeschlagen — klassischer Parserbefund bleibt." + ) + return + if korrigiert is not None: + ergebnis.treffer = korrigiert + ergebnis.ki_eingriffe = eingriffe or [] + def seite_verarbeiten( self, inhalt: bytes, url: str ) -> tuple[list[QuellenTreffer], list[str]]: diff --git a/plugins/neuheiten/quellen/ki_hilfe.py b/plugins/neuheiten/quellen/ki_hilfe.py new file mode 100644 index 0000000..09e96cb --- /dev/null +++ b/plugins/neuheiten/quellen/ki_hilfe.py @@ -0,0 +1,575 @@ +"""Optionale LLM-Hilfsstufe für den Web-Quellen-Crawler des Neuheiten-Plugins. + +Der Crawler (`basis.py` + Adapter) bleibt maßgeblich und arbeitet rein +regelbasiert. Das LLM wird ausschließlich in zwei Situationen eingeschaltet: + +1. **Struktur-Erkennung:** Liefert der Regel-Parser keine oder verdächtig + wenige Einträge (0 Treffer, oder wenige Treffer auf einer sehr link- + reichen Seite), darf das LLM Pagination-/Filter-Folge-URLs aus dem + gelieferten HTML vorschlagen. Der Vorschlag wird im Code hart gefiltert: + nur URLs derselben Domain wie die geparste Seite, maximal 20 je Seite — + Schutz vor Rate-Limit-Überlastung und Domain-Ausbruch. +2. **Feld-Extraktion:** Ein einzelner Treffer ist *unsicher* (fehlender + Verlag oder verdächtiger Titel). Dann darf das LLM höchstens `titel`, + `verlag` und `autor` korrigieren — niemals die URL. + +Konfiguration konsistent zur dedup-KI über Umgebungsvariablen: + +| Variable | Bedeutung | +|----------|-----------| +| `SPIELE_NEUHEITEN_KI_AKTIV` | `1` = Hilfsstufe an (Standard `0`) | +| `SPIELE_NEUHEITEN_KI_BASIS_URL` | Basis-URL, z. B. `https://api.openai.com/v1` | +| `SPIELE_NEUHEITEN_KI_API_KEY` | API-Key (wird als Bearer gesendet) | +| `SPIELE_NEUHEITEN_KI_MODELL` | Modellname, z. B. `gpt-4o-mini` | +| `SPIELE_NEUHEITEN_KI_KONFIDENZ_MIN` | Mindest-Konfidenz (Standard `0.7`) | +| `SPIELE_NEUHEITEN_KI_TIMEOUT_SEKUNDEN` | Timeout je Aufruf (Standard `20`) | + +**Fallback-Pflicht:** Ohne Konfiguration oder bei jedem Fehler läuft EXAKT +der klassische Crawler weiter — ein KI-Fehler blockiert den Sync nie. +Ergebnisse unterhalb der Mindest-Konfidenz werden verworfen; der klassische +Parser-Befund bleibt dann unverändert bestehen. + +Schnittstelle: OpenAI-kompatible Chat-Completions-API via httpx; +Netzwerk-/Server-Fehler werden genau einmal wiederholt (insgesamt zwei +Versuche), Antwort-Parsing-Fehler nicht. +""" +from __future__ import annotations + +import dataclasses +import json +import logging +import os +import re +from dataclasses import dataclass +from typing import Any, Mapping + +from urllib.parse import urljoin, urlsplit + +import httpx + +_logger = logging.getLogger("plugins.neuheiten") + +#: 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 + +#: Maximalzahl der KI-vorgeschlagenen Folge-URLs je Seite (Rate-Limit-Schutz). +MAX_FOLGE_URLS = 20 + +#: Obergrenze des HTML-Ausschnitts, der pro Struktur-Anfrage ans LLM geht. +HTML_MAX_ZEICHEN = 12000 + +# ---------------- Verdachts-Heuristiken (wann hilft das LLM?) ---------------- + +#: Weniger Treffer als das gilt als „verdächtig wenig“ … +STRUKTUR_MIN_TREFFER = 3 +#: … aber nur, wenn die Seite zugleich sehr linkreich ist (Indiz für eine +#: nicht erkannte Liste/Pagination statt einer wirklich leeren Seite). +STRUKTUR_LINK_SCHWELLE = 30 + +_LINK_MUSTER = re.compile(r" str: + if isinstance(inhalt, bytes): + return inhalt.decode("utf-8", errors="replace") + return inhalt + + +def titel_ist_verdaechtig(titel: str | None) -> bool: + """True bei fehlendem, zu kurzem oder generischem Navigationstext.""" + if not titel or not titel.strip(): + return True + gekuerzt = titel.strip() + if len(gekuerzt) < 3: + return True + return gekuerzt.rstrip(".!? ").lower() in GENERISCHE_TITEL + + +def treffer_ist_unsicher(treffer: Any) -> bool: + """Ein Treffer ist unsicher, wenn der Verlag fehlt oder der Titel verdächtig ist. + + Nur solche Treffer gehen in die KI-Feld-Extraktion; sichere Treffer + kosten kein Token und bleiben garantiert unangetastet. + """ + verlag = getattr(treffer, "verlag", None) + if not verlag or not str(verlag).strip(): + return True + return titel_ist_verdaechtig(getattr(treffer, "titel", None)) + + +def struktur_verdaechtig(anzahl_treffer: int, inhalt: bytes | str) -> bool: + """True, wenn der Parser keine oder verdächtig wenige Einträge lieferte. + + 0 Treffer sind immer verdächtig. 1–2 Treffer nur dann, wenn die Seite + sehr viele Links enthält — sonst ist sie schlicht klein/leer. + """ + if anzahl_treffer <= 0: + return True + if anzahl_treffer >= STRUKTUR_MIN_TREFFER: + return False + links = len(_LINK_MUSTER.findall(_als_text(inhalt))) + return links > STRUKTUR_LINK_SCHWELLE + + +# ---------------- Konfiguration ---------------- + +@dataclass(frozen=True) +class KiKonfiguration: + """Effektive Konfiguration der KI-Hilfsstufe (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 führen nicht zum Fehler, sondern zum Standardwert — + die Hilfsstufe darf den Crawler niemals blockieren. + """ + env = os.environ if quelle is None else quelle + aktiv = env.get("SPIELE_NEUHEITEN_KI_AKTIV", "0").strip() == "1" + + try: + konfidenz_min = float(env.get("SPIELE_NEUHEITEN_KI_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_NEUHEITEN_KI_TIMEOUT_SEKUNDEN", "")) + except ValueError: + timeout = STANDARD_TIMEOUT + timeout = max(1.0, timeout) + + return KiKonfiguration( + aktiv=aktiv, + basis_url=env.get("SPIELE_NEUHEITEN_KI_BASIS_URL", "").strip().rstrip("/"), + api_key=env.get("SPIELE_NEUHEITEN_KI_API_KEY", "").strip(), + modell=env.get("SPIELE_NEUHEITEN_KI_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 und begleitenden Text; None, wenn nichts + Sinnvolles übrig bleibt. + """ + if not isinstance(text, str): + return None + text = text.strip() + if not text: + return None + + versuche = [text] + versuche.extend(m.group(1).strip() for m in _FENCE_MUSTER.finditer(text)) + 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 _normalisiere_konfidenz(daten: Mapping[str, Any]) -> float | None: + """Konfidenz aus der Antwort; Prozent-Skalen werden toleriert.""" + try: + konfidenz = float(daten.get("konfidenz")) + except (TypeError, ValueError): + return None + if 10.0 <= konfidenz <= 100.0: # 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 + return round(konfidenz, 4) + + +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_url_vorschlaege(daten: Any, seiten_url: str) -> dict | None: + """Prüft die Struktur-Antwort und filtert hart auf gleiche Domain + max 20. + + Erwartet `{folge_urls: [...], konfidenz: float}`. Relative URLs werden + gegen die Seiten-URL aufgelöst; fremde Domains, Nicht-HTTP-Schemata und + Duplikate (inkl. Fragment-Unterschiede) werden verworfen — auch wenn das + LLM sie vorschlägt. + """ + if not isinstance(daten, dict): + return None + roh = daten.get("folge_urls") + if not isinstance(roh, list): + return None + konfidenz = _normalisiere_konfidenz(daten) + if konfidenz is None: + return None + + basis_domain = urlsplit(seiten_url).netloc.lower() + if not basis_domain: + return None + gefiltert: list[str] = [] + gesehen: set[str] = set() + for eintrag in roh: + if not isinstance(eintrag, str) or not eintrag.strip(): + continue + absolut = urljoin(seiten_url, eintrag.strip()) + teile = urlsplit(absolut) + if teile.scheme not in ("http", "https"): + continue + if teile.netloc.lower() != basis_domain: + continue + ohne_fragment = absolut.split("#", 1)[0] + if ohne_fragment in gesehen: + continue + gesehen.add(ohne_fragment) + gefiltert.append(ohne_fragment) + if len(gefiltert) >= MAX_FOLGE_URLS: + break + return {"folge_urls": gefiltert, "konfidenz": konfidenz} + + +def validiere_feld_antwort(daten: Any) -> dict | None: + """Prüft die Feld-Antwort gegen das Schema (titel/verlag/autor/konfidenz). + + Die URL ist bewusst kein Teil des Schemas — eine Korrektur kann sie + strukturell nie ändern. + """ + if not isinstance(daten, dict): + return None + titel = daten.get("titel") + if not isinstance(titel, str) or not titel.strip(): + return None + konfidenz = _normalisiere_konfidenz(daten) + if konfidenz is None: + return None + begruendung = daten.get("begruendung") + if not isinstance(begruendung, str): + begruendung = "" + return { + "titel": titel.strip(), + "verlag": _optional_text(daten.get("verlag")), + "autor": _optional_text(daten.get("autor")), + "konfidenz": konfidenz, + "begruendung": begruendung.strip(), + } + + +# ---------------- Prompt ---------------- + +_SYSTEM_STRUKTUR = ( + "Du hilfst einem Crawler einer Spielemagazin-Redaktion, Brettspiel-" + "Neuheitenlisten im Web zu finden. Der Regel-Parser hat die folgende " + "Seite fast leer geparst — vermutlich nutzt die Liste Pagination, " + "Archiv- oder Filter-Links. Identifiziere aus dem HTML die URLs, unter " + "denen weitere Neuheiten-Einträge zu erwarten sind. Antworte " + "AUSSCHLIESSLICH mit einem JSON-Objekt nach exakt diesem Schema:\n" + '{"folge_urls": ["", …], ' + '"konfidenz": }\n' + f"Höchstens {MAX_FOLGE_URLS} URLs und nur Links derselben Domain wie die " + "Seiten-URL. Wenn du nichts Sinnvolles findest, liefere eine leere Liste " + "mit niedriger Konfidenz. Kein weiterer Text, keine Code-Fences." +) + +_SYSTEM_FELDER = ( + "Du assistierst einer Spielemagazin-Redaktion beim Parsen von Brettspiel-" + "Neuheiten. Der folgende Treffer eines regelbasierten Crawlers ist " + "unsicher. Bereinige die Felder: trenne „Verlag: Titel“-Muster, entferne " + "Navigationstext und ergänze den Verlag, falls er sicher erkennbar ist. " + "Korrigiere ausschließlich titel, verlag und autor — ändere niemals URLs " + "und erfinde nichts. Ein Feld, das du nicht sicher bestimmen kannst, " + "bleibt null. Antworte AUSSCHLIESSLICH mit einem JSON-Objekt nach exakt " + "diesem Schema:\n" + '{"titel": "", "verlag": "", ' + '"autor": "", "konfidenz": , ' + '"begruendung": ""}\n' + "Kein weiterer Text, keine Code-Fences." +) + + +def baue_struktur_prompt(seiten_url: str, html_text: str) -> str: + """User-Nachricht für die Struktur-Erkennung (gekürzter HTML-Ausschnitt).""" + ausschnitt = html_text[:HTML_MAX_ZEICHEN] + return ( + f"Seiten-URL: {seiten_url}\n\n" + "HTML-Ausschnitt der geparsten Seite:\n" + f"{ausschnitt}" + ) + + +def baue_feld_prompt(titel: str, verlag: str | None, autor: str | None) -> str: + """User-Nachricht mit dem unsicheren Treffer (ohne URL — bleibt fix).""" + zeilen = [f"Unsicherer Treffer:", f"titel: {titel}"] + zeilen.append(f"verlag: {verlag if verlag else '(leer)'}") + zeilen.append(f"autor: {autor if autor else '(leer)'}") + zeilen.append( + "Gib die bereinigten Felder zurück; die quellen_url selbst bleibt " + "nicht veränderbar und ist nicht Teil der Aufgabe." + ) + return "\n".join(zeilen) + + +# ---------------- Client ---------------- + +class KiHilfe: + """OpenAI-kompatibler Chat-Completions-Client für die Crawler-Hilfsstufe. + + Bewusst genauso defensiv wie der dedup-KI-Client: Alle öffentlichen + Methoden werfen nicht (Ausnahmen werden intern geschluckt) — ohne + Konfiguration oder bei jedem Problem greift exakt der klassische + Crawler-Befund. Transport 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() + + # ---------- Struktur-Erkennung ---------- + + def folge_urls_bei_bedarf( + self, + anzahl_treffer: int, + inhalt: bytes | str, + seiten_url: str, + bereits_geprueft: set[str] | None = None, + ) -> list[str]: + """Pagination-/Filter-URLs bei verdächtig leerem Parser-Ergebnis. + + Prüft selbst die Verdachts-Heuristik (sonst []), fragt dann das LLM + und filtert dessen Vorschläge hart: gleiche Domain, max. 20, bereits + geplante/besuchte URLs raus, Konfidenz unter der Schwelle → alles + verwerfen. Fehler → [] (klassischer Lauf bleibt). + """ + try: + if not struktur_verdaechtig(anzahl_treffer, inhalt): + return [] + if not self._konfiguration.vollstaendig: + return [] + + nutzernachricht = baue_struktur_prompt(seiten_url, _als_text(inhalt)) + daten = self._chat(_SYSTEM_STRUKTUR, nutzernachricht) + ergebnis = validiere_url_vorschlaege(daten, seiten_url) + if ergebnis is None: + return [] + if ergebnis["konfidenz"] < self._konfiguration.konfidenz_min: + _logger.info( + "neuheiten/KI: URL-Vorschläge verworfen (Konfidenz %.2f < %.2f).", + ergebnis["konfidenz"], self._konfiguration.konfidenz_min, + ) + return [] + bekannt = bereits_geprueft or set() + return [ + url for url in ergebnis["folge_urls"] if url not in bekannt + ] + except Exception: + _logger.exception( + "neuheiten/KI: Struktur-Erkennung fehlgeschlagen — klassischer Lauf bleibt." + ) + return [] + + # ---------- Feld-Extraktion ---------- + + def treffer_nachbessern( + self, treffer_liste: list + ) -> tuple[list, list[dict]]: + """Korrigiert ausschließlich unsichere Treffer (titel/verlag/autor). + + Sichere Treffer werden nicht angefasst und kosten kein Token. Die + quellen_url bleibt immer unangetastet (kein Teil des Antwort-Schemas). + Rückgabe: (neue Trefferliste, Eingriffs-Protokoll fürs Audit-Log mit + Modell, Konfidenz und den jeweils korrigierten Feldern). + """ + if not treffer_liste: + return [], [] + if not self._konfiguration.vollstaendig: + return list(treffer_liste), [] + + neue_liste: list = [] + eingriffe: list[dict] = [] + for einzel in treffer_liste: + if not treffer_ist_unsicher(einzel): + neue_liste.append(einzel) + continue + ergebnis = self._einzeln_korrigieren(einzel) + if ergebnis is None: + neue_liste.append(einzel) # klassischer Befund bleibt + continue + korrigiert, antwort = ergebnis + if korrigiert is einzel: # nichts tatsächlich geändert + neue_liste.append(einzel) + continue + neue_liste.append(korrigiert) + eingriffe.append(self._eingriff_protokoll(einzel, korrigiert, antwort)) + return neue_liste, eingriffe + + def _einzeln_korrigieren(self, treffer: Any) -> tuple[Any, dict] | None: + """Eine Feld-Korrektur; Rückgabe (neuer Treffer, validierte Antwort). + + None bzw. der unveränderte Treffer bedeutet: klassischer Befund bleibt. + """ + + try: + nutzernachricht = baue_feld_prompt( + getattr(treffer, "titel", "") or "", + getattr(treffer, "verlag", None), + getattr(treffer, "autor", None), + ) + daten = self._chat(_SYSTEM_FELDER, nutzernachricht) + antwort = validiere_feld_antwort(daten) + if antwort is None: + return None + if antwort["konfidenz"] < self._konfiguration.konfidenz_min: + _logger.info( + "neuheiten/KI: Feldkorrektur verworfen (Konfidenz %.2f < %.2f).", + antwort["konfidenz"], self._konfiguration.konfidenz_min, + ) + return None + + felder: dict[str, str | None] = {} + for feld in ("titel", "verlag", "autor"): + alter_wert = getattr(treffer, feld, None) + alter_norm = (alter_wert or "").strip() or None + neu_norm = (antwort[feld] or "").strip() or None + if neu_norm != alter_norm: + felder[feld] = neu_norm + if not felder: + return treffer, antwort # nichts tatsächlich korrigiert + try: + return dataclasses.replace(treffer, **felder), antwort + except Exception: + return None + except Exception: + _logger.exception( + "neuheiten/KI: Feldkorrektur fehlgeschlagen — klassischer Befund bleibt." + ) + return None + + def _eingriff_protokoll(self, alt: Any, neu: Any, antwort: dict) -> dict: + """Audit-Zeile: was wurde korrigiert, mit welchem Modell/welcher Konfidenz.""" + korrigiert: dict[str, str | None] = {} + for feld in ("titel", "verlag", "autor"): + vorher = getattr(alt, feld, None) + nachher = getattr(neu, feld, None) + if vorher != nachher: + korrigiert[feld] = nachher + return { + "quellen_url": getattr(alt, "quellen_url", ""), + "modell": self.modell, + "konfidenz": antwort["konfidenz"], + "begruendung": antwort.get("begruendung", ""), + "korrigiert": korrigiert, + "vorher": {f: getattr(alt, f, None) for f in korrigiert}, + } + + # ---------- Gemeinsamer Chat-Completions-Aufruf ---------- + + def _chat(self, system: str, nutzer: str) -> dict | None: + """POST /chat/completions mit genau einem Retry; None bei jedem Problem. + + Netzwerk-/Timeout-Fehler und zeitweilige Server-Störungen werden + genau einmal wiederholt (insgesamt zwei Versuche); andere 4xx und + kaputte Antworten nicht — dort hilft kein Retry. + """ + payload = { + "model": self._konfiguration.modell, + "temperature": 0, + "messages": [ + {"role": "system", "content": system}, + {"role": "user", "content": nutzer}, + ], + } + for versuch in range(1, VERSUCHE + 1): + try: + antwort = self._client.post("/chat/completions", json=payload) + except httpx.RequestError as exc: + _logger.warning( + "neuheiten/KI: Versuch %d/%d fehlgeschlagen (%s)", + versuch, VERSUCHE, exc, + ) + continue + if antwort.status_code in RETRY_STATUS: + _logger.warning( + "neuheiten/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("neuheiten/KI: Anfrage abgelehnt (%s).", exc) + return None + try: + inhalte = antwort.json()["choices"][0]["message"]["content"] + except Exception as exc: # kaputtes Antwort-Layout + _logger.warning("neuheiten/KI: unbrauchbare Antwort (%s)", exc) + return None + return extrahiere_json(inhalte) + _logger.warning( + "neuheiten/KI: alle %d Versuche fehlgeschlagen — klassischer Befund.", VERSUCHE + ) + return None diff --git a/plugins/neuheiten/quellen_sync.py b/plugins/neuheiten/quellen_sync.py index edbdada..1f818db 100644 --- a/plugins/neuheiten/quellen_sync.py +++ b/plugins/neuheiten/quellen_sync.py @@ -32,6 +32,7 @@ from typing import Callable from sqlalchemy import select from sqlalchemy.orm import sessionmaker +from .quellen.ki_hilfe import KiHilfe, lade_konfiguration as lade_ki_konfiguration from .models import STATUS_NEUHEIT, Neuheit, QuellenStatus from .quellen import ( ADAPTER_KLASSEN, @@ -108,20 +109,27 @@ class WebQuellenSyncService: dedup=None, client_fabrik: Callable[[dict], WebQuellenClient] | None = None, umgebung=None, + ki_fabrik: Callable[[], KiHilfe] | None = None, + audit=None, ) -> None: self.session_factory = session_factory self.adapter_klassen = adapter_klassen self.dedup = dedup + self.audit = audit self.umgebung = umgebung if umgebung is not None else os.environ self._client_fabrik = client_fabrik or self._standard_client_fabrik + # KI-Hilfsstufe: Fabrik für Tests; ohne vollständige Konfiguration wird + # nie ein Client gebaut (und damit nie ein LLM-Aufruf getätigt). + self._ki_fabrik = ki_fabrik # ---------- Gesamtdurchlauf ---------- def laufe(self) -> list[QuellenLaufZeile]: zeilen: list[QuellenLaufZeile] = [] + ki = self._ki_hilfe() for klasse in self.adapter_klassen: try: - zeile = self._eine_quelle(klasse) + zeile = self._eine_quelle(klasse, ki) except Exception as exc: # letzte Verteidigungslinie pro Quelle _logger.exception("Quellen-Sync fehlgeschlagen für %s", klasse.name) zeile = QuellenLaufZeile( @@ -141,7 +149,28 @@ class WebQuellenSyncService: # ---------- Eine Quelle ---------- - def _eine_quelle(self, klasse: type[QuellenAdapter]) -> QuellenLaufZeile: + def _ki_hilfe(self) -> KiHilfe | None: + """KI-Hilfsstufe nur bei vollständiger Konfiguration (sonst None). + + Die Aktivierungsprüfung gilt vor jeder injizierten Fabrik — ist die + Hilfsstufe aus oder unvollständig konfiguriert, wird nie ein Client + gebaut und damit nie ein LLM-Aufruf getätigt. + """ + try: + konfiguration = lade_ki_konfiguration(self.umgebung) + if not konfiguration.vollstaendig: + return None + if self._ki_fabrik is not None: + return self._ki_fabrik() + return KiHilfe(konfiguration) + except Exception: + # Auch ein kaputtes KI-Setup darf den Sync nie blockieren. + _logger.exception("neuheiten/KI: Hilfsstufe nicht verfügbar — klassischer Lauf.") + return None + + def _eine_quelle( + self, klasse: type[QuellenAdapter], ki: KiHilfe | None = None + ) -> QuellenLaufZeile: zeile = QuellenLaufZeile(klasse.name, klasse.anzeigename) zeile.url = klasse.start_urls[0] if klasse.start_urls else None if not quelle_aktiv(klasse.name, self.umgebung): @@ -152,6 +181,7 @@ class WebQuellenSyncService: client = self._client_fabrik(self._validatoren_laden(klasse.name)) try: adapter = klasse(client) + adapter.ki_hilfe = ki # None = rein regelbasiert (Fallback-Pflicht) try: ergebnis = adapter.sammle() except Exception as exc: @@ -160,6 +190,7 @@ class WebQuellenSyncService: zeile.fehlermeldung = str(exc)[:300] return zeile zeile.seiten = ergebnis.seiten + self._ki_auditieren(klasse.name, ergebnis) if ergebnis.unveraendert: zeile.unveraendert = True else: @@ -263,6 +294,37 @@ class WebQuellenSyncService: return True return not ergebnis.hat_konflikte + # ---------- KI-Audit ---------- + + def _ki_auditieren(self, quelle: str, ergebnis) -> None: + """KI-Eingriffe des Laufs ins audit-log schreiben (best effort). + + Ein Eintrag je Quelle und Lauf mit Modell, Konfidenz, korrigierten + Feldern und den verfolgten KI-Folge-URLs. Ohne audit-log (Plugin + nicht geladen) oder bei Fehlern wird nur protokolliert — der Sync + wird nie blockiert. + """ + eingriffe = list(getattr(ergebnis, "ki_eingriffe", []) or []) + folge_urls = list(getattr(ergebnis, "ki_folge_urls", []) or []) + if not eingriffe and not folge_urls: + return + if self.audit is None: + _logger.info( + "Quelle %s: %d KI-Korrekturen, %d KI-Folge-URLs (kein audit-log geladen).", + quelle, len(eingriffe), len(folge_urls), + ) + return + details = { + "quelle": quelle, + "modell": eingriffe[0].get("modell", "") if eingriffe else "", + "eingriffe": eingriffe, + "folge_urls": folge_urls, + } + try: + self.audit.log_sync(None, "korrigiert", "neuheiten_ki_crawler", quelle, details) + except Exception: + _logger.exception("neuheiten/KI: Audit-Eintrag konnte nicht geschrieben werden.") + # ---------- Statusprotokoll ---------- def _status_speichern(self, zeile: QuellenLaufZeile) -> None: diff --git a/tests/_neuheiten.py b/tests/_neuheiten.py index 94a1ded..c2c2165 100644 --- a/tests/_neuheiten.py +++ b/tests/_neuheiten.py @@ -59,6 +59,18 @@ jahr_aus_datumsangabe = quellen_basis.jahr_aus_datumsangabe titel_aehnlichkeit = quellen_basis.titel_aehnlichkeit titel_normalisieren = quellen_basis.titel_normalisieren +ki_hilfe_modul = _neuheiten.quellen.ki_hilfe +KiHilfe = ki_hilfe_modul.KiHilfe +KiKonfiguration = ki_hilfe_modul.KiKonfiguration +lade_ki_konfiguration = ki_hilfe_modul.lade_konfiguration +extrahiere_ki_json = ki_hilfe_modul.extrahiere_json +struktur_verdaechtig = ki_hilfe_modul.struktur_verdaechtig +treffer_ist_unsicher = ki_hilfe_modul.treffer_ist_unsicher +titel_ist_verdaechtig = ki_hilfe_modul.titel_ist_verdaechtig +validiere_url_vorschlaege = ki_hilfe_modul.validiere_url_vorschlaege +validiere_feld_antwort = ki_hilfe_modul.validiere_feld_antwort +MAX_FOLGE_URLS = ki_hilfe_modul.MAX_FOLGE_URLS + SpielboxQuelle = _neuheiten.quellen.spielbox.SpielboxQuelle BrettspielboxQuelle = _neuheiten.quellen.brettspielbox.BrettspielboxQuelle SpielEssenQuelle = _neuheiten.quellen.spielessen.SpielEssenQuelle diff --git a/tests/test_neuheiten_ki_crawler.py b/tests/test_neuheiten_ki_crawler.py new file mode 100644 index 0000000..d233ba9 --- /dev/null +++ b/tests/test_neuheiten_ki_crawler.py @@ -0,0 +1,544 @@ +"""Tests: Optionale LLM-Hilfsstufe des Web-Quellen-Crawlers (neuheiten). + +Alle HTTP- und LLM-Antworten sind gemockt (httpx.MockTransport bzw. +Fake-Transports) — es gibt niemals echte Netzwerk- oder LLM-Aufrufe. +Abgedeckt: Fallback-Fälle (aktiv=0, fehlender Key, Timeout mit genau einem +Retry, kaputtes JSON), Konfidenz-Schwelle, Struktur-Erkennung nur mit +gleiche-Domain-URLs (max. 20), Feld-Korrektur ausschließlich unsicherer +Treffer (URL bleibt fix) und die Audit-Protokollierung der KI-Eingriffe. +""" +from __future__ import annotations + +import json + +import httpx +import pytest + +from tests._neuheiten import ( + MAX_FOLGE_URLS, + KiHilfe, + KiKonfiguration, + Neuheit, + QuellenFehler, + QuellenTreffer, + SpielboxQuelle, + SammelErgebnis, + WebQuellenClient, + WebQuellenSyncService, + extrahiere_ki_json, + lade_ki_konfiguration, + struktur_verdaechtig, + titel_ist_verdaechtig, + treffer_ist_unsicher, +) + +BASIS_URL = "https://www.spielbox.de/neuheiten-neue-spiele" +KONFIG = dict( + aktiv=True, + basis_url="https://llm.example.test/v1", + api_key="test-key", + modell="test-modell", +) + + +class FakeUhr: + def __init__(self) -> None: + self.zeit = 0.0 + + def __call__(self) -> float: + return self.zeit + + +# ---------------- Gemockte Seiten ---------------- + +SEITE_LEER_VIELE_LINKS = ( + "

Keine Neuheiten erkannt.

" +) + +SEITE_MIT_TREFFERN = """ +
+

MICROMACRO Kids

+

02.03.2026 - kurz.

+
+""" + + +def _antwort(inhalt: bytes | str) -> httpx.Response: + return httpx.Response( + 200, + content=inhalt.encode("utf-8") if isinstance(inhalt, str) else inhalt, + request=httpx.Request("GET", BASIS_URL), + ) + + +def _crawler_client(seiten: dict[str, bytes | str]) -> tuple[WebQuellenClient, KartenTransport, FakeUhr]: + uhr = FakeUhr() + transport = KartenTransport({u: _antwort(i) for u, i in seiten.items()}, uhr) + client = WebQuellenClient( + transport=transport, schlaf=lambda s: None, uhr=uhr + ) + return client, transport, uhr + + +class KartenTransport(httpx.BaseTransport): + """Antworten je URL; zeichnet angefragte URLs auf.""" + + def __init__(self, antworten: dict[str, httpx.Response], uhr: FakeUhr): + self.antworten = antworten + self.anfragen: list[str] = [] + + def handle_request(self, request: httpx.Request) -> httpx.Response: + url = str(request.url) + self.anfragen.append(url) + antwort = self.antworten.get(url) + if antwort is None and "?" in url: + antwort = self.antworten.get(url.split("?")[0]) + if antwort is None: + raise QuellenFehler(f"Keine gemockte Antwort für {url}") + return antwort + + +# ---------------- LLM-Mocks ---------------- + +def llm_antwort(objekt: dict, status: int = 200) -> httpx.Response: + body = {"choices": [{"message": {"role": "assistant", "content": json.dumps(objekt)}}]} + return httpx.Response(status, content=json.dumps(body).encode()) + + +class LlmTransport(httpx.BaseTransport): + """Gestaffelte Antworten für Chat-Completions; zählt Aufrufe. + + Elemente: dict → JSON-Inhalt, int → HTTP-Status, Exception → wird geworfen. + """ + + def __init__(self, antworten: list): + self.antworten = list(antworten) + self.aufrufe = 0 + self.prompts: list[dict] = [] + + def handle_request(self, request: httpx.Request) -> httpx.Response: + index = min(self.aufrufe, len(self.antworten) - 1) + spezifikation = self.antworten[index] + self.aufrufe += 1 + self.prompts.append(json.loads(request.content.decode())) + if isinstance(spezifikation, Exception): + raise spezifikation + if isinstance(spezifikation, int): + return httpx.Response(spezifikation, json={"error": "kaputt"}) + return llm_antwort(spezifikation) + + +def ki_hilfe(transport: httpx.BaseTransport, konfidenz_min: float = 0.7) -> KiHilfe: + return KiHilfe( + KiKonfiguration(konfidenz_min=konfidenz_min, **KONFIG), + transport=transport, + ) + + +TREFFER_UNSICHER = lambda: QuellenTreffer( # noqa: E731 + titel="Pegasus Spiele: Next Station Berlin", + quellen_url="https://www.spielbox.de/x/1", +) +TREFFER_SICHER = lambda: QuellenTreffer( # noqa: E731 + titel="Galactic Cruise", verlag="Dranda Games", + quellen_url="https://www.spielbox.de/x/2", +) + +GUTE_FELDANTWORT = { + "titel": "Next Station Berlin", + "verlag": "Pegasus Spiele", + "autor": None, + "konfidenz": 0.9, + "begruendung": "Verlag stand als Präfix im Titel.", +} + + +# ---------------- Heuristiken ---------------- + +def test_struktur_verdaechtig_ohne_treffer_immmer(): + assert struktur_verdaechtig(0, "

kurz

") is True + + +def test_struktur_verdaechtig_wenige_treffer_nur_bei_linkreicher_seite(): + viele_links = "".join('x' for _ in range(40)) + assert struktur_verdaechtig(2, viele_links) is True + assert struktur_verdaechtig(2, "x") is False + assert struktur_verdaechtig(5, viele_links) is False + + +def test_unsichere_treffer_fehlender_verlag_oder_generischer_titel(): + assert treffer_ist_unsicher(QuellenTreffer(titel="X", quellen_url="u")) is True + assert treffer_ist_unsicher(TREFFER_UNSICHER()) is True # kein Verlag + assert treffer_ist_unsicher(TREFFER_SICHER()) is False + assert titel_ist_verdaechtig("Mehr") is True + assert titel_ist_verdaechtig("ab") is True + assert titel_ist_verdaechtig("Fadenfroh") is False + + +def test_extrahiere_json_toleriert_code_fences(): + text = "Antwort:\n```json\n{\"folge_urls\": []}\n```\nEnde." + assert extrahiere_ki_json(text) == {"folge_urls": []} + assert extrahiere_ki_json("völlig kaputt") is None + + +# ---------------- Konfiguration & Fallback ---------------- + +def test_konfiguration_standard_aus(): + k = lade_ki_konfiguration({}) + assert not k.aktiv and not k.vollstaendig + assert k.konfidenz_min == 0.7 + assert k.timeout >= 1.0 + + +def test_service_ohne_konfiguration_baut_keinen_ki_client(session_factory, monkeypatch): + """Fallback aktiv=0: kein KI-Client, exakt der klassische Crawler.""" + monkeypatch.setenv("SPIELE_NEUHEITEN_KI_AKTIV", "0") + + class Streng: + def __call__(self): # pragma: no cover — darf nie aufgerufen werden + raise AssertionError("KI-Client ohne Aktivierung gebaut") + def schliessen(self): pass + + quelle = StatischeQuelle("spielbox", [TREFFER_SICHER()]) + service = WebQuellenSyncService( + session_factory, adapter_klassen=(quelle,), + client_fabrik=lambda v: _ohne_netz_client(), + ki_fabrik=Streng(), + ) + + zeilen = service.laufe() + + assert zeilen[0].neu == 1 and zeilen[0].fehlermeldung is None + with session_factory() as db: + assert db.query(Neuheit).count() == 1 + + +def test_service_mit_aktiv_ohne_key_laeuft_klassisch(session_factory, monkeypatch): + """Fallback: AKTIV=1, aber Key fehlt → vollständige Konfiguration fehlt.""" + monkeypatch.setenv("SPIELE_NEUHEITEN_KI_AKTIV", "1") + monkeypatch.setenv("SPIELE_NEUHEITEN_KI_BASIS_URL", "https://llm.example.test/v1") + monkeypatch.delenv("SPIELE_NEUHEITEN_KI_API_KEY", raising=False) + monkeypatch.setenv("SPIELE_NEUHEITEN_KI_MODELL", "test-modell") + + class Streng: + def __call__(self): # pragma: no cover + raise AssertionError("KI-Client ohne Key gebaut") + def schliessen(self): pass + + quelle = StatischeQuelle("spielbox", [TREFFER_SICHER()]) + service = WebQuellenSyncService( + session_factory, adapter_klassen=(quelle,), + client_fabrik=lambda v: _ohne_netz_client(), + ki_fabrik=Streng(), + ) + + zeilen = service.laufe() + + assert zeilen[0].neu == 1 + + +def test_timeout_genau_ein_retry_dann_klassischer_befund(): + """Timeout: insgesamt zwei Versuche, danach klassischer Treffer unverändert.""" + transport = LlmTransport([httpx.ReadTimeout("zu lang"), httpx.ReadTimeout("nochmal")]) + hilfe = ki_hilfe(transport) + + treffer_liste, eingriffe = hilfe.treffer_nachbessern([TREFFER_UNSICHER()]) + + assert transport.aufrufe == 2 # erster Versuch + genau ein Retry + assert eingriffe == [] + original = treffer_liste[0] + assert original.verlag is None # klassischer Befund bleibt + assert original.quellen_url == "https://www.spielbox.de/x/1" + + +def test_kaputtes_json_kein_retry_kein_eingriff(): + transport = LlmTransport(["das ist {kaputtes JSON ohne Sinn"]) + hilfe = ki_hilfe(transport) + + treffer_liste, eingriffe = hilfe.treffer_nachbessern([TREFFER_UNSICHER()]) + + assert transport.aufrufe == 1 # Parsing-Fehler werden nicht wiederholt + assert eingriffe == [] + assert treffer_liste[0].verlag is None + + +def test_serverfehler_nach_retry_faellt_zurueck(): + transport = LlmTransport([503, 503]) + hilfe = ki_hilfe(transport) + + _, eingriffe = hilfe.treffer_nachbessern([TREFFER_UNSICHER()]) + folge_urls = hilfe.folge_urls_bei_bedarf(0, SEITE_LEER_VIELE_LINKS, BASIS_URL, set()) + + assert transport.aufrufe == 4 # je Aufrufart zwei Versuche + assert eingriffe == [] and folge_urls == [] + + +# ---------------- Konfidenz-Schwelle ---------------- + +def test_feldkorrektur_unter_konfidenz_min_wird_verworfen(): + schwach = dict(GUTE_FELDANTWORT, konfidenz=0.5) + transport = LlmTransport([schwach]) + hilfe = ki_hilfe(transport, konfidenz_min=0.7) + + treffer_liste, eingriffe = hilfe.treffer_nachbessern([TREFFER_UNSICHER()]) + + assert transport.aufrufe == 1 + assert eingriffe == [] + assert treffer_liste[0].titel == "Pegasus Spiele: Next Station Berlin" + assert treffer_liste[0].verlag is None + + +def test_url_vorschlaege_unter_konfidenz_min_werden_verworfen(): + vorschlaege = { + "folge_urls": ["/neuheiten-neue-spiele?start=6"], + "konfidenz": 0.3, + } + transport = LlmTransport([vorschlaege]) + hilfe = ki_hilfe(transport) + + urls = hilfe.folge_urls_bei_bedarf(0, SEITE_LEER_VIELE_LINKS, BASIS_URL, set()) + + assert urls == [] + + +# ---------------- Struktur-Erkennung ---------------- + +def test_struktur_erkennung_nur_gleiche_domain_urls_werden_verfolgt(): + """Fremde Domains und Nicht-HTTP-Schemata werden hart herausgefiltert.""" + gemischt = { + "folge_urls": [ + "/neuheiten-neue-spiele?start=6", # gleiche Domain ✓ + "https://www.spielbox.de/neuheiten?page=2", # gleiche Domain ✓ + "https://boese.example.org/neuheiten", # fremde Domain ✗ + "ftp://www.spielbox.de/liste", # Schema ✗ + "/neuheiten-neue-spiele?start=6#oben", # Duplikat ✗ + "javascript:zeigeListe()", # Schema ✗ + ], + "konfidenz": 0.9, + } + crawler_client, transport, _ = _crawler_client({BASIS_URL: SEITE_LEER_VIELE_LINKS}) + transport.antworten["https://www.spielbox.de/neuheiten-neue-spiele?start=6"] = _antwort(SEITE_MIT_TREFFERN) + transport.antworten["https://www.spielbox.de/neuheiten?page=2"] = _antwort(SEITE_MIT_TREFFERN) + adapter = SpielboxQuelle(crawler_client) + adapter.ki_hilfe = ki_hilfe(LlmTransport([gemischt])) + + ergebnis = adapter.sammle() + + gefolgt = [u for u in transport.anfragen if u != BASIS_URL] + assert "https://www.spielbox.de/neuheiten-neue-spiele?start=6" in gefolgt + assert all(u.startswith("https://www.spielbox.de") for u in gefolgt) + assert not any("boese" in u or u.startswith("ftp") for u in gefolgt) + assert len(ergebnis.treffer) == 2 # aus den beiden nachgeschobenen Seiten + assert ergebnis.ki_folge_urls == gefolgt + + +def test_struktur_erkennung_maximal_20_folge_urls(): + """Rate-Limit-Schutz: auch bei 30 Vorschlägen werden höchstens 20 verfolgt.""" + vorschlaege = { + "folge_urls": [f"/liste?seite={i}" for i in range(30)], + "konfidenz": 0.9, + } + crawler_client, transport, _ = _crawler_client({BASIS_URL: SEITE_LEER_VIELE_LINKS}) + for i in range(30): + transport.antworten[f"https://www.spielbox.de/liste?seite={i}"] = _antwort(SEITE_MIT_TREFFERN) + adapter = SpielboxQuelle(crawler_client) + adapter.ki_hilfe = ki_hilfe(LlmTransport([vorschlaege])) + + adapter.sammle() + + gefolgt = [u for u in transport.anfragen if u != BASIS_URL] + assert len(gefolgt) <= MAX_FOLGE_URLS + assert len(gefolgt) == 20 + assert "https://www.spielbox.de/liste?seite=25" not in gefolgt + + +def test_struktur_erkennung_greift_nicht_bei_normalen_ergebnissen(): + """Drei sichere Treffer → keine Verdachts-Heuristik, kein LLM-Aufruf.""" + seite = SEITE_MIT_TREFFERN * 3 + crawler_client, transport, _ = _crawler_client({BASIS_URL: seite}) + adapter = SpielboxQuelle(crawler_client) + adapter.ki_hilfe = ki_hilfe(LlmTransport([])) # würde bei Aufruf zählen + + ergebnis = adapter.sammle() + + assert len(ergebnis.treffer) == 3 + assert len(transport.anfragen) == 1 # nur die Startseite + + +# ---------------- Feld-Korrektur ---------------- + +def test_feldkorrektur_nur_bei_unsicheren_treffern_url_bleibt_fix(): + """Sichere Treffer kosten kein Token; korrigiert wird nie die URL.""" + transport = LlmTransport([GUTE_FELDANTWORT]) + hilfe = ki_hilfe(transport) + sicher = TREFFER_SICHER() + unsicher = TREFFER_UNSICHER() + + treffer_liste, eingriffe = hilfe.treffer_nachbessern([sicher, unsicher]) + + assert transport.aufrufe == 1 # nur der unsichere Treffer ging ans LLM + assert treffer_liste[0] is sicher # unangetastet + korrigiert = treffer_liste[1] + assert korrigiert.titel == "Next Station Berlin" + assert korrigiert.verlag == "Pegasus Spiele" + assert korrigiert.quellen_url == unsicher.quellen_url # URL bleibt fix + assert len(eingriffe) == 1 + protokoll = eingriffe[0] + assert protokoll["modell"] == "test-modell" + assert protokoll["konfidenz"] == 0.9 + assert protokoll["korrigiert"] == { + "titel": "Next Station Berlin", "verlag": "Pegasus Spiele", + } + assert protokoll["vorher"]["titel"] == "Pegasus Spiele: Next Station Berlin" + + +def test_feldkorrektur_ohne_aenderung_erzeugt_keinen_eingriff(): + identisch = { + "titel": "Pegasus Spiele: Next Station Berlin", + "verlag": None, "autor": None, "konfidenz": 0.9, "begruendung": "passt", + } + transport = LlmTransport([identisch]) + hilfe = ki_hilfe(transport) + + treffer_liste, eingriffe = hilfe.treffer_nachbessern([TREFFER_UNSICHER()]) + + assert eingriffe == [] + assert treffer_liste[0].quellen_url == "https://www.spielbox.de/x/1" + + +# ---------------- Sync-Integration + Audit ---------------- + +class StatischeQuelle: + """Adapter-Ersatz mit fester Trefferliste (Service-Tests ohne Netz).""" + + def __init__(self, name: str = "spielbox", treffer: list[QuellenTreffer] | None = None): + self.name = name + self.anzeigename = name.title() + self.start_urls = (f"https://{name}.example.org/liste",) + self._treffer = treffer if treffer is not None else [] + + def __call__(self, client): + return self + + def sammle(self) -> SammelErgebnis: + return SammelErgebnis(treffer=list(self._treffer), seiten=1) + + +def _ohne_netz_client() -> WebQuellenClient: + return WebQuellenClient(transport=None, schlaf=lambda s: None, uhr=FakeUhr()) + + +class FakeAudit: + def __init__(self): + self.eintraege: list[tuple] = [] + + def log_sync(self, actor, action, objekt_typ, objekt_id, details, **kwargs): + self.eintraege.append((actor, action, objekt_typ, objekt_id, details)) + + +def test_sync_auditert_ki_eingriffe_mit_modell_und_konfidenz(): + """End-to-End: Crawler-Seite mit unsicherem Treffer → KI-Korrektur → Audit. + + „Ringträger“ ohne Doppelpunkt: der Parser findet keinen Verlag → Treffer + ist unsicher → das LLM darf titel/verlag korrigieren. + """ + seite = ( + '

' + 'Ringträger

' + "

15.08.2026 - Text.

" + ) + crawler_client, _, _ = _crawler_client({BASIS_URL: seite}) + audit = FakeAudit() + fabrik = _session_factory() + service = WebQuellenSyncService( + fabrik, + adapter_klassen=(SpielboxQuelle,), + client_fabrik=lambda validatoren: crawler_client, + ki_fabrik=lambda: ki_hilfe(LlmTransport([GUTE_FELDANTWORT])), + umgebung={ + "SPIELE_NEUHEITEN_KI_AKTIV": "1", + "SPIELE_NEUHEITEN_KI_BASIS_URL": "https://llm.example.test/v1", + "SPIELE_NEUHEITEN_KI_API_KEY": "k", + "SPIELE_NEUHEITEN_KI_MODELL": "m", + }, + audit=audit, + ) + + zeilen = service.laufe() + + assert zeilen[0].neu == 1 + with fabrik() as db: + eintrag = db.query(Neuheit).one() + assert eintrag.titel == "Next Station Berlin" + assert eintrag.verlag == "Pegasus Spiele" + assert eintrag.erscheinungsjahr == 2026 + assert len(audit.eintraege) == 1 + actor, action, objekt_typ, objekt_id, details = audit.eintraege[0] + assert action == "korrigiert" and objekt_typ == "neuheiten_ki_crawler" + assert objekt_id == "spielbox" + assert details["modell"] == "test-modell" + assert details["eingriffe"][0]["konfidenz"] == 0.9 + assert details["eingriffe"][0]["korrigiert"]["verlag"] == "Pegasus Spiele" + + +def _session_factory(): + from sqlalchemy import create_engine + from sqlalchemy.orm import sessionmaker + + from redaktionskern.db import Base + + engine = create_engine("sqlite://", connect_args={"check_same_thread": False}) + Base.metadata.create_all(engine) + fabrik = sessionmaker(bind=engine, expire_on_commit=False, autoflush=False) + fabrik.engine_zum_aufraeumen = engine # type: ignore[attr-defined] + return fabrik + + +def test_sync_ohne_ki_eingriff_schreibt_kein_audit(): + audit = FakeAudit() + service = WebQuellenSyncService( + _session_factory(), + adapter_klassen=(StatischeQuelle("spielbox", [TREFFER_SICHER()]),), + client_fabrik=lambda v: _ohne_netz_client(), + ki_fabrik=None, + audit=audit, + ) + + zeilen = service.laufe() + + assert zeilen[0].neu == 1 + assert audit.eintraege == [] + + +def test_ki_ausfall_blockiert_den_sync_nie(): + """KI-Fabrik crasht → Sync läuft trotzdem komplett klassisch.""" + def kaputte_fabrik(): + raise RuntimeError("LLM nicht erreichbar") + + service = WebQuellenSyncService( + _session_factory(), + adapter_klassen=(StatischeQuelle("spielbox", [TREFFER_SICHER()]),), + client_fabrik=lambda v: _ohne_netz_client(), + ki_fabrik=kaputte_fabrik, + umgebung={ + "SPIELE_NEUHEITEN_KI_AKTIV": "1", + "SPIELE_NEUHEITEN_KI_BASIS_URL": "u", + "SPIELE_NEUHEITEN_KI_API_KEY": "k", + "SPIELE_NEUHEITEN_KI_MODELL": "m", + }, + ) + + zeilen = service.laufe() + + assert zeilen[0].neu == 1 and zeilen[0].fehlermeldung is None + + +@pytest.fixture +def session_factory(): + fabrik = _session_factory() + yield fabrik + fabrik.engine_zum_aufraeumen.dispose() # type: ignore[attr-defined]