Plugin neuheiten: optionale LLM-Hilfsstufe für den Web-Quellen-Crawler
Neues Modul quellen/ki_hilfe.py (konsistent zum dedup-LLM-Muster): - Struktur-Erkennung: bei verdächtig leerem Parser-Ergebnis schlägt das LLM Pagination-/Filter-Folge-URLs vor; hart gefiltert auf gleiche Domain, max. 20 je Seite, nicht erreichbare Vorschläge brechen den Lauf nicht ab. - Feld-Extraktion: nur unsichere Treffer (fehlender Verlag, verdächtiger Titel); korrigiert ausschließlich titel/verlag/autor, niemals die URL. - Env-Konfiguration SPIELE_NEUHEITEN_KI_* (AKTIV default 0, KONFIDENZ_MIN default 0.7), OpenAI-kompatible Chat-Completions via httpx mit Timeout und genau einem Retry. - Fallback-Pflicht: ohne Konfiguration oder bei jedem Fehler läuft exakt der klassische Crawler; KI-Fehler blockieren den Sync nie. - Audit: KI-Eingriffe je Quelle und Lauf ins audit-log (Modell, Konfidenz, korrigierte Felder, verfolgte Folge-URLs). 20 neue Tests (gemocktes HTTP/LLM): Fallback-Fälle, Konfidenz-Schwelle, Domain-Filter, Max-20-Grenze, URL-Unveränderbarkeit, Audit.
This commit is contained in:
Flo Hartmann
@@ -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
|
||||
|
||||
544
tests/test_neuheiten_ki_crawler.py
Normal file
544
tests/test_neuheiten_ki_crawler.py
Normal file
@@ -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 = (
|
||||
"<html><body><nav>"
|
||||
+ "".join(
|
||||
f'<a href="/rubrik/{i}">Rubrik {i}</a> ' for i in range(40)
|
||||
)
|
||||
+ "</nav><p>Keine Neuheiten erkannt.</p></body></html>"
|
||||
)
|
||||
|
||||
SEITE_MIT_TREFFERN = """
|
||||
<div class="item">
|
||||
<div class="page-header"><h2><a href="/x/1">MICROMACRO Kids</a></h2></div>
|
||||
<p><strong>02.03.2026</strong> - kurz.</p>
|
||||
</div>
|
||||
"""
|
||||
|
||||
|
||||
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, "<p>kurz</p>") is True
|
||||
|
||||
|
||||
def test_struktur_verdaechtig_wenige_treffer_nur_bei_linkreicher_seite():
|
||||
viele_links = "".join('<a href="/x">x</a>' for _ in range(40))
|
||||
assert struktur_verdaechtig(2, viele_links) is True
|
||||
assert struktur_verdaechtig(2, "<a href='/x'>x</a>") 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 = (
|
||||
'<div class="item"><div class="page-header"><h2>'
|
||||
'<a href="/x/1">Ringträger</a></h2></div>'
|
||||
"<p><strong>15.08.2026</strong> - Text.</p></div>"
|
||||
)
|
||||
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]
|
||||
Reference in New Issue
Block a user