f27940db8e
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.
576 lines
22 KiB
Python
576 lines
22 KiB
Python
"""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"<a\s", re.IGNORECASE)
|
||
|
||
#: Generische Navigations-/Platzhaltertexte, die als Spieltitel verdächtig sind.
|
||
GENERISCHE_TITEL = frozenset(
|
||
{
|
||
"hier", "mehr", "mehr infos", "details", "weiterlesen", "weiter",
|
||
"klick", "klick hier", "link", "produkt", "neuheit", "unbekannt",
|
||
"titel", "n/a", "-", "–",
|
||
}
|
||
)
|
||
|
||
|
||
def _als_text(inhalt: bytes | str) -> 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": ["<absolute oder relative URL>", …], '
|
||
'"konfidenz": <float 0.0-1.0>}\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": "<bereinigter Titel>", "verlag": "<Verlagsname oder null>", '
|
||
'"autor": "<Autor oder null>", "konfidenz": <float 0.0-1.0>, '
|
||
'"begruendung": "<kurzer deutscher Satz>"}\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
|