8e5f65d864
Regelbasierte Dedup-Prüfung bleibt maßgeblich; nur Grenzfälle gehen an ein LLM: Titel mit Wort-Überschneidung unter der Fuzzy-Schwelle (Editionen, Big Boxes, Übersetzungen wie 'El Grande Big Box' vs. 'El Grande: 25 Jahre') und ein ergänzender Call zum Verlags-/Titel-Hinweis bei Verlags-Konflikt (Verlags-Aliase). Klare Treffer und Non-Treffer bleiben regelbasiert. - Neu: ki_pruefung.py — Env-Konfiguration (SPIELE_DEDUP_LLM_*), robuster JSON-Parser (Code-Fences, Zusatztext), OpenAI-kompatibler Client via httpx mit Timeout und genau einem Retry; liefert bei jedem Fehler None. - pruefung.py: bestätigte Grenzfälle laufen in die üblichen Prüfungen a)-d) zurück; deutscher Titel füllt die Titel-Empfehlung, Verlags-Empfehlung landet im Konflikt-Detail. Neues Feld PruefErgebnis.ki_befunde. - __init__.py: Aktivierungsprüfung vor jeder Fabrik — aus/unvollständig heißt nie ein LLM-Aufruf; KI-Befunde werden über die audit-log-API protokolliert (Modell, Konfidenz, Entscheidung), best effort. - Tests: gemockte HTTP-Antworten (MockTransport/Fakes), Fallback-Fälle (aktiv=0, kein Key, Timeout, kaputtes JSON), Konfidenz-Schwelle, Parsing-Robustheit; kein echter LLM-Call in CI. - README.md um die neuen Env-Variablen und den Zweitprüfungs-Abschnitt ergänzt.
624 lines
25 KiB
Python
624 lines
25 KiB
Python
"""Prüflogik des Plugins „dedup“ — die vier Prüfungen laut Fachanforderung:
|
||
|
||
a) **Verlags-Konflikt** — dasselbe Spiel ist bereits unter anderem
|
||
Verlag/Vertrieb geführt. Ergebnis enthält eine Verlagsauswahl.
|
||
b) **Titel-Varianten** — dasselbe Spiel mit unterschiedlichem Titel; die
|
||
deutsche Version wird bevorzugt (Heuristik über deutsche Titel-Indikatoren,
|
||
ergänzt um BGG Alternate-Names, falls ein Hilfsclient verfügbar ist).
|
||
c) **Spiel oder Vorgänger bereits besprochen** — abgeschlossene Einträge der
|
||
Planungsliste bilden das „besprochen“-Korpus; Vorgänger werden über die
|
||
BGG `boardgameexpansion`-Relation erkannt und zusätzlich fuzzy auf Titel
|
||
gematcht (rapidfuzz).
|
||
d) **Titel schon in Planungsliste eines anderen Rezensenten.**
|
||
|
||
Die Daten der anderen Plugins (neuheiten, planung) liest dieses Plugin
|
||
lesend über die gemeinsame SQLAlchemy-Metadata (`Base.metadata.tables`) —
|
||
so entstehen keine Import-Abhängigkeiten zwischen Plugin-Paketen. Fehlt eine
|
||
Tabelle (Plugin nicht geladen), entfällt die jeweilige Teilprüfung einfach.
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
import re
|
||
import unicodedata
|
||
from collections.abc import Iterable
|
||
from dataclasses import dataclass, field
|
||
|
||
from sqlalchemy import select
|
||
from sqlalchemy.exc import SQLAlchemyError
|
||
|
||
from redaktionskern.db import Base
|
||
|
||
try:
|
||
from rapidfuzz import fuzz
|
||
except ImportError: # pragma: no cover - rapidfuzz ist eine harte Abhängigkeit
|
||
fuzz = None
|
||
|
||
_logger = logging.getLogger("plugins.dedup")
|
||
|
||
#: Fuzzy-Schwelle für „gemeintes Spiel“ (rapidfuzz token_sort_ratio 0–100).
|
||
TITEL_SCHWELLE = 85
|
||
|
||
#: Statuswert des planung-Plugins für „bereits besprochen“ (bewusst als
|
||
#: Literal, damit keine Import-Abhängigkeit zwischen den Plugins entsteht).
|
||
STATUS_ABGESCHLOSSEN = "abgeschlossen"
|
||
|
||
#: Obergrenze für LLM-Aufrufe je Prüfung (Token sparen): höchstens so viele
|
||
#: Grenzfall-Paarungen plus ein Verlags-/Titel-Call bei Verlags-Konflikt.
|
||
MAX_KI_PAARUNGEN = 5
|
||
|
||
#: Funktionswörter, die stark auf einen deutschen Spieltitel hindeuten.
|
||
DEUTSCHE_INDIKATOREN = frozenset({
|
||
"der", "die", "das", "den", "dem", "des",
|
||
"ein", "eine", "einen", "einem", "einer", "eines",
|
||
"und", "für", "von", "vom", "zu", "zum", "zur",
|
||
"im", "am", "beim", "auf", "mit", "aus",
|
||
})
|
||
|
||
#: Typische Begriffe deutscher Spieltitel (+2 Punkte je Treffer).
|
||
DEUTSCHE_SPIELWOERTER = frozenset({
|
||
"abenteuer", "berg", "blitz", "burg", "dorf", "drache", "drachen",
|
||
"erde", "feuer", "gärten", "haus", "himmel", "insel", "jagd",
|
||
"kaiser", "könig", "könige", "königreich", "land", "länder",
|
||
"meister", "mond", "norden", "osten", "pfad", "reich", "ritter",
|
||
"schiff", "schloss", "see", "spiel", "spiele", "spielen",
|
||
"stadt", "stern", "süden", "turm", "wald", "weg", "welt",
|
||
"westen", "wiese", "wilder", "wüste", "zauber", "zeit",
|
||
})
|
||
|
||
#: Wörter, die auf einen englischen Titeltitel hindeuten (−1 Punkt je Treffer).
|
||
ENGLISCHE_INDIKATOREN = frozenset({
|
||
"the", "of", "and", "game", "games", "card", "cards", "dice", "board",
|
||
})
|
||
|
||
_UM_LAUTE = frozenset("äöüßÄÖÜ")
|
||
_WORT_MUSTER = re.compile(r"[^\W\d_]+", re.UNICODE)
|
||
|
||
|
||
# ---------------- Titel-Heuristik ----------------
|
||
|
||
def normalisiere(text: str) -> str:
|
||
"""Vereinheitlicht einen Titel für Vergleiche (Groß-/Satzzeichen egal)."""
|
||
text = unicodedata.normalize("NFC", text or "")
|
||
text = text.casefold()
|
||
text = re.sub(r"[^\w\s]+", " ", text, flags=re.UNICODE)
|
||
return re.sub(r"\s+", " ", text).strip()
|
||
|
||
|
||
def titel_aehnlich(a: str, b: str, schwelle: int = TITEL_SCHWELLE) -> bool:
|
||
"""True, wenn zwei Titel dasselbe Spiel meinen (exakt oder fuzzy).
|
||
|
||
Nach der Normalisierung zählt `fuzz.token_set_ratio`: Wortreihenfolge
|
||
und Füllwörter sind egal, damit „Die Siedler von Catan“ und
|
||
„Siedler von Catan, Die“ zusammenpassen, „Catan Erweiterung“ und
|
||
„Catan Grundspiel“ aber nicht.
|
||
"""
|
||
if not a or not b:
|
||
return False
|
||
na, nb = normalisiere(a), normalisiere(b)
|
||
if not na or not nb:
|
||
return False
|
||
if na == nb:
|
||
return True
|
||
if fuzz is None: # pragma: no cover
|
||
return False
|
||
return fuzz.token_set_ratio(na, nb) >= schwelle
|
||
|
||
|
||
def deutsch_score(titel: str) -> int:
|
||
"""Heuristik-Score: > 0 spricht für einen deutschen Titel.
|
||
|
||
Umlaute/ß zählen am stärksten (+3), deutsche Funktionswörter +1,
|
||
typische deutsche Spielbegriffe +2, englische Indikatoren −1.
|
||
"""
|
||
if not titel:
|
||
return 0
|
||
score = 0
|
||
if any(zeichen in _UM_LAUTE for zeichen in titel):
|
||
score += 3
|
||
woerter = {wort.casefold() for wort in _WORT_MUSTER.findall(titel)}
|
||
score += sum(1 for wort in woerter if wort in DEUTSCHE_INDIKATOREN)
|
||
score += sum(2 for wort in woerter if wort in DEUTSCHE_SPIELWOERTER)
|
||
score -= sum(1 for wort in woerter if wort in ENGLISCHE_INDIKATOREN)
|
||
return score
|
||
|
||
|
||
def ist_deutscher_titel(titel: str) -> bool:
|
||
return deutsch_score(titel) > 0
|
||
|
||
|
||
def bevorzuge_deutschen_titel(kandidaten: Iterable[str]) -> str | None:
|
||
"""Wählt aus Kandidaten den deutschesten Titel (bei Gleichstand den ersten)."""
|
||
bester: str | None = None
|
||
bester_score = 0
|
||
for kandidat in kandidaten:
|
||
kandidat = (kandidat or "").strip()
|
||
if not kandidat:
|
||
continue
|
||
score = deutsch_score(kandidat)
|
||
if bester is None or score > bester_score:
|
||
bester, bester_score = kandidat, score
|
||
return bester
|
||
|
||
|
||
def _verlags_schluessel(verlag: str) -> str:
|
||
"""Vergleichsschlüssel für Verlage („Kosmos“ == „kosmos “)."""
|
||
return normalisiere(verlag)
|
||
|
||
|
||
def _gemeinsame_woerter(a: str, b: str) -> int:
|
||
"""Anzahl gemeinsamer Wort-Tokens zweiter Titel nach Normalisierung.
|
||
|
||
Als „Titel-Score > 0, aber unter Schwelle“ gilt eine Wort-Überschneidung
|
||
über 0 bei Fuzzy-Score unter TITEL_SCHWELLE — reine Zeichenähnlichkeit
|
||
ohne gemeinsames Wort („Cascadia“ vs. „Everdell“) bleibt ein klarer
|
||
Non-Treffer und kostet kein LLM-Token.
|
||
"""
|
||
wa = set(normalisiere(a).split())
|
||
wb = set(normalisiere(b).split())
|
||
return len(wa & wb)
|
||
|
||
|
||
def _titel_score(titel_a: str, titel_b: str) -> int:
|
||
if fuzz is None: # pragma: no cover
|
||
return 0
|
||
return int(fuzz.token_set_ratio(normalisiere(titel_a), normalisiere(titel_b)))
|
||
|
||
|
||
def _ki_bestaetigt(antwort: dict | None, konfidenz_min: float) -> bool:
|
||
"""True, wenn ein KI-Befund übernommen wird (gleiches Spiel, genug Konfidenz)."""
|
||
return bool(
|
||
antwort
|
||
and antwort.get("ist_gleiches_spiel")
|
||
and float(antwort.get("konfidenz") or 0.0) >= konfidenz_min
|
||
)
|
||
|
||
|
||
def _ki_protokoll(
|
||
typ: str,
|
||
titel_a: str,
|
||
titel_b: str,
|
||
ki_client,
|
||
antwort: dict | None,
|
||
*,
|
||
konfidenz_min: float,
|
||
) -> dict:
|
||
"""Fasst einen KI-Aufruf für Audit-Log/Protokoll zusammen."""
|
||
if antwort is None:
|
||
entscheidung = "fehler_verworfen"
|
||
konfidenz = None
|
||
elif not antwort.get("ist_gleiches_spiel"):
|
||
entscheidung = "verschiedene_spiele"
|
||
konfidenz = antwort.get("konfidenz")
|
||
elif float(antwort.get("konfidenz") or 0.0) < konfidenz_min:
|
||
entscheidung = "verworfen_unter_schwelle"
|
||
konfidenz = antwort.get("konfidenz")
|
||
else:
|
||
entscheidung = "gleiches_spiel"
|
||
konfidenz = antwort.get("konfidenz")
|
||
return {
|
||
"typ": typ,
|
||
"titel_a": titel_a[:300],
|
||
"titel_b": titel_b[:300],
|
||
"modell": getattr(ki_client, "modell", "") or "",
|
||
"konfidenz": konfidenz,
|
||
"entscheidung": entscheidung,
|
||
"begruendung": (antwort or {}).get("begruendung", ""),
|
||
}
|
||
|
||
|
||
# ---------------- Prüfergebnis ----------------
|
||
|
||
#: Anzeigenamen der Konflikt-Arten (deutsche UI).
|
||
KONFLIKT_ANZEIGE: dict[str, str] = {
|
||
"verlag": "Verlags-Konflikt",
|
||
"titel": "Titel-Variante",
|
||
"besprochen": "Bereits besprochen",
|
||
"planung": "Schon in Planung",
|
||
}
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class TitelKonflikt:
|
||
"""Ein einzelner Befund der Prüfung."""
|
||
|
||
art: str # Schlüssel aus KONFLIKT_ANZEIGE
|
||
beschreibung: str
|
||
details: dict = field(default_factory=dict)
|
||
|
||
def als_dict(self) -> dict:
|
||
return {"art": self.art, "beschreibung": self.beschreibung, "details": self.details}
|
||
|
||
@property
|
||
def anzeige(self) -> str:
|
||
return KONFLIKT_ANZEIGE.get(self.art, self.art)
|
||
|
||
|
||
@dataclass
|
||
class PruefErgebnis:
|
||
"""Strukturiertes Ergebnis einer dedup-Prüfung."""
|
||
|
||
titel: str
|
||
verlag: str | None = None
|
||
bgg_id: int | None = None
|
||
konflikte: list[TitelKonflikt] = field(default_factory=list)
|
||
#: Verlage zur Auswahl bei einem Verlags-Konflikt (welcher wird geführt?).
|
||
verlags_optionen: list[str] = field(default_factory=list)
|
||
#: Empfohlener (deutschester) Titel, falls Titel-Varianten bekannt sind.
|
||
titel_empfehlung: str | None = None
|
||
#: Protokoll der KI-Zweitprüfung (Grenzfälle) — je Eintrag u. a. Modell,
|
||
#: Konfidenz und Entscheidung. Leer bei rein regelbasierten Prüfungen.
|
||
ki_befunde: list[dict] = field(default_factory=list)
|
||
|
||
@classmethod
|
||
def ohne_befund(cls, titel: str, verlag: str | None = None, bgg_id: int | None = None):
|
||
return cls(titel=titel, verlag=verlag, bgg_id=bgg_id)
|
||
|
||
@property
|
||
def hat_konflikte(self) -> bool:
|
||
return bool(self.konflikte)
|
||
|
||
@property
|
||
def hat_verlagskonflikt(self) -> bool:
|
||
return any(konflikt.art == "verlag" for konflikt in self.konflikte)
|
||
|
||
def als_liste(self) -> list[dict]:
|
||
return [konflikt.als_dict() for konflikt in self.konflikte]
|
||
|
||
def als_text(self) -> str:
|
||
if not self.konflikte:
|
||
return "Keine Treffer."
|
||
zeilen = [f"• [{k.anzeige}] {k.beschreibung}" for k in self.konflikte]
|
||
if (
|
||
self.titel_empfehlung
|
||
and normalisiere(self.titel_empfehlung) != normalisiere(self.titel)
|
||
):
|
||
zeilen.append(f"• Empfohlener Titel: „{self.titel_empfehlung}“")
|
||
if self.ki_befunde:
|
||
zeilen.append(f"• KI-Zweitprüfung: {len(self.ki_befunde)} Grenzfall/Grenzfälle bewertet.")
|
||
return "\n".join(zeilen)
|
||
|
||
|
||
# ---------------- Datenzugriff (fremde Plugin-Tabellen, nur lesend) ----------------
|
||
|
||
def _lade_zeilen(db, tabellenname: str, spalten: tuple[str, ...]) -> list[dict]:
|
||
"""Liest Zeilen einer Plugin-Tabelle über die gemeinsame Metadata.
|
||
|
||
Fehlt die Tabelle (Plugin nicht geladen bzw. Migration nicht gelaufen),
|
||
wird eine leere Liste zurückgegeben — die jeweilige Teilprüfung entfällt.
|
||
"""
|
||
tabelle = Base.metadata.tables.get(tabellenname)
|
||
if tabelle is None:
|
||
return []
|
||
try:
|
||
zeilen = db.execute(select(*(tabelle.c[name] for name in spalten))).all()
|
||
except SQLAlchemyError as exc:
|
||
_logger.warning("dedup: Tabelle '%s' nicht lesbar (%s) — Teilprüfung übersprungen.", tabellenname, exc)
|
||
return []
|
||
return [dict(zip(spalten, zeile._tuple())) for zeile in zeilen]
|
||
|
||
|
||
def _lade_benutzernamen(db) -> dict[int, str]:
|
||
benutzer = Base.metadata.tables.get("users")
|
||
namen: dict[int, str] = {}
|
||
if benutzer is None:
|
||
return namen
|
||
try:
|
||
for zeile in db.execute(
|
||
select(benutzer.c.id, benutzer.c.display_name, benutzer.c.username)
|
||
).all():
|
||
id_wert, anzeige, username = zeile._tuple()
|
||
namen[id_wert] = anzeige or username or f"Benutzer {id_wert}"
|
||
except SQLAlchemyError: # pragma: no cover - defensive
|
||
pass
|
||
return namen
|
||
|
||
|
||
def _rezensent_name(zeile: dict, namen: dict[int, str]) -> str:
|
||
id_wert = zeile.get("rezensent_id")
|
||
if id_wert is None:
|
||
return "unbekannt"
|
||
return namen.get(id_wert, f"Benutzer {id_wert}")
|
||
|
||
|
||
def _gleiches_spiel(zeile: dict, titel: str, bgg_id: int | None) -> bool:
|
||
"""Match über eindeutige BGG-ID, sonst fuzzy über den Titel."""
|
||
fremd_bgg = zeile.get("bgg_id")
|
||
if bgg_id is not None and fremd_bgg is not None and int(fremd_bgg) == int(bgg_id):
|
||
return True
|
||
return titel_aehnlich(titel, zeile.get("titel") or "")
|
||
|
||
|
||
# ---------------- Hauptprüfung ----------------
|
||
|
||
def fuehre_pruefung_durch(
|
||
session_factory,
|
||
bgg_hilfsclient,
|
||
*,
|
||
titel: str,
|
||
verlag: str | None = None,
|
||
bgg_id: int | None = None,
|
||
user_id: int | None = None,
|
||
ki_client=None,
|
||
ki_konfidenz_min: float = 0.7,
|
||
) -> PruefErgebnis:
|
||
"""Führt alle vier Prüfungen durch und liefert ein PruefErgebnis.
|
||
|
||
`bgg_hilfsclient` darf None sein (BGG-Zusatzdaten deaktiviert) — dann
|
||
entfallen Alternate-Names und Vorgänger-Relationen, die Heuristik und
|
||
der Fuzzy-Match arbeiten weiter.
|
||
`ki_client` darf None sein (KI-Zweitprüfung deaktiviert/fehlerhaft) —
|
||
dann läuft ausschließlich die Regelprüfung; sonst werden nur Grenzfälle
|
||
(Titel-Score > 0 unter Schwelle bzw. Verlags-Konflikt) dem LLM vorgelegt
|
||
und bestätigte Treffer in die Regelprüfungen zurückgeführt.
|
||
`user_id` ist der Eintragende: eigene Planungseinträge lösen keinen
|
||
„schon in Planung“-Konflikt aus.
|
||
"""
|
||
titel = (titel or "").strip()
|
||
verlag = (verlag or "").strip() or None
|
||
id_wert = int(bgg_id) if bgg_id else None
|
||
ergebnis = PruefErgebnis.ohne_befund(titel, verlag, id_wert)
|
||
if not titel:
|
||
return ergebnis
|
||
konfidenz_min = max(0.0, min(1.0, float(ki_konfidenz_min)))
|
||
|
||
with session_factory() as db:
|
||
neuheiten = _lade_zeilen(db, "neuheiten", ("id", "titel", "verlag", "bgg_id"))
|
||
planungen = _lade_zeilen(
|
||
db, "planungsliste", ("id", "titel", "verlag", "bgg_id", "status", "rezensent_id")
|
||
)
|
||
namen = _lade_benutzernamen(db)
|
||
|
||
alternativen: list[str] = []
|
||
vorgaenger: list[tuple[int, str]] = []
|
||
if id_wert is not None and bgg_hilfsclient is not None:
|
||
try:
|
||
alternativen = list(bgg_hilfsclient.alternativen(id_wert))
|
||
except Exception as exc: # Zusatzdaten dürfen die Prüfung nie blockieren
|
||
_logger.warning("dedup: BGG-Alternativen nicht verfügbar (%s)", exc)
|
||
try:
|
||
vorgaenger = list(bgg_hilfsclient.vorgaenger(id_wert))
|
||
except Exception as exc:
|
||
_logger.warning("dedup: BGG-Vorgänger nicht verfügbar (%s)", exc)
|
||
|
||
# Kandidaten = Einträge zum selben Spiel laut Regeln (Neuheiten + Planung).
|
||
alle_zeilen = (*neuheiten, *planungen)
|
||
kandidaten = [
|
||
zeile for zeile in alle_zeilen
|
||
if _gleiches_spiel(zeile, titel, id_wert)
|
||
]
|
||
|
||
# ---- KI-Zweitprüfung für Grenzfälle (optional, niemals blockierend) ----
|
||
# Klare Treffer (oben) und klare Non-Treffer bleiben regelbasiert; nur
|
||
# Titel mit Wort-Überschneidung über 0, aber Fuzzy-Score unter der
|
||
# Schwelle, werden dem LLM vorgelegt. Bestätigte Treffer laufen in die
|
||
# üblichen Prüfungen a)–d) zurück, als wären sie regelbasiert gematcht.
|
||
ki_bestaetigt: dict[int, dict] = {} # id(zeile) -> KI-Antwort-Dict
|
||
ki_titel_vorschlag: str | None = None # erster übernommener deutscher Titel
|
||
if ki_client is not None and alle_zeilen:
|
||
kandidat_ids = {id(zeile) for zeile in kandidaten}
|
||
grenzfaelle = []
|
||
for zeile in alle_zeilen:
|
||
if id(zeile) in kandidat_ids:
|
||
continue
|
||
fremd_titel = (zeile.get("titel") or "").strip()
|
||
if not fremd_titel:
|
||
continue
|
||
if _gemeinsame_woerter(titel, fremd_titel) <= 0:
|
||
continue # klarer Non-Treffer — kein LLM-Token wert
|
||
score = _titel_score(titel, fremd_titel)
|
||
if score >= TITEL_SCHWELLE:
|
||
continue # wäre ohnehin regelbasiert gematcht
|
||
grenzfaelle.append((score, fremd_titel, zeile))
|
||
grenzfaelle.sort(key=lambda eintrag: -eintrag[0])
|
||
for score, fremd_titel, zeile in grenzfaelle[:MAX_KI_PAARUNGEN]:
|
||
try:
|
||
antwort = ki_client.bewerte(
|
||
titel_a=titel, titel_b=fremd_titel,
|
||
verlag_a=verlag, verlag_b=zeile.get("verlag"),
|
||
alternativen_a=alternativen,
|
||
expansionen_a=[name for _, name in vorgaenger],
|
||
)
|
||
except Exception as exc: # Zusatzdaten dürfen nie blockieren
|
||
_logger.warning("dedup/KI: Grenzfall-Bewertung fehlgeschlagen (%s)", exc)
|
||
antwort = None
|
||
ergebnis.ki_befunde.append(_ki_protokoll(
|
||
"grenzfall", titel, fremd_titel, ki_client, antwort,
|
||
konfidenz_min=konfidenz_min,
|
||
))
|
||
if _ki_bestaetigt(antwort, konfidenz_min):
|
||
kandidaten.append(zeile)
|
||
ki_bestaetigt[id(zeile)] = antwort
|
||
if antwort.get("deutscher_titel"):
|
||
ki_titel_vorschlag = ki_titel_vorschlag or antwort["deutscher_titel"]
|
||
|
||
|
||
# a) Verlags-Konflikt: gleiches Spiel unter anderem Verlag/Vertrieb.
|
||
bekannte_verlage: dict[str, str] = {}
|
||
for zeile in kandidaten:
|
||
kandidat_verlag = (zeile.get("verlag") or "").strip()
|
||
if kandidat_verlag:
|
||
bekannte_verlage.setdefault(_verlags_schluessel(kandidat_verlag), kandidat_verlag)
|
||
abweichend = [
|
||
name for schluessel, name in bekannte_verlage.items()
|
||
if verlag and schluessel != _verlags_schluessel(verlag)
|
||
]
|
||
if abweichend or len(bekannte_verlage) > 1:
|
||
optionen = list(bekannte_verlage.values())
|
||
if verlag and _verlags_schluessel(verlag) not in bekannte_verlage:
|
||
optionen.append(verlag)
|
||
ergebnis.verlags_optionen = optionen
|
||
ergebnis.konflikte.append(TitelKonflikt(
|
||
art="verlag",
|
||
beschreibung=(
|
||
"Gleiches Spiel ist bereits unter anderem Verlag geführt: "
|
||
+ ", ".join(sorted(bekannte_verlage.values()))
|
||
+ ". Bitte wählen, welcher Verlag geführt wird."
|
||
),
|
||
details={"eigener_verlag": verlag, "bekannte_verlage": list(bekannte_verlage.values())},
|
||
))
|
||
|
||
# b) Titel-Varianten → deutsche Version bevorzugen.
|
||
varianten = [
|
||
zeile["titel"] for zeile in kandidaten
|
||
if normalisiere(zeile.get("titel") or "") != normalisiere(titel)
|
||
]
|
||
ergebnis.titel_empfehlung = (
|
||
ki_titel_vorschlag # KI-Empfehlung (bestätigter Grenzfall) hat Vorrang
|
||
or bevorzuge_deutschen_titel([titel, *varianten, *alternativen])
|
||
)
|
||
if varianten and ergebnis.titel_empfehlung \
|
||
and normalisiere(ergebnis.titel_empfehlung) != normalisiere(titel):
|
||
ergebnis.konflikte.append(TitelKonflikt(
|
||
art="titel",
|
||
beschreibung=(
|
||
f"Gleiches Spiel unter anderem Titel bekannt — empfohlen: "
|
||
f"„{ergebnis.titel_empfehlung}“."
|
||
),
|
||
details={
|
||
"empfehlung": ergebnis.titel_empfehlung,
|
||
"varianten": sorted(set(varianten)),
|
||
"bgg_alternativen": alternativen[:10],
|
||
},
|
||
))
|
||
|
||
# c) Spiel oder Vorgänger bereits besprochen (Planungsstatus „abgeschlossen“).
|
||
besprochen = [z for z in planungen if z.get("status") == STATUS_ABGESCHLOSSEN]
|
||
for zeile in besprochen:
|
||
if _gleiches_spiel(zeile, titel, id_wert) or id(zeile) in ki_bestaetigt:
|
||
wer = _rezensent_name(zeile, namen)
|
||
details = {"planungseintrag": zeile["id"], "titel": zeile["titel"], "rezensent": wer}
|
||
if id(zeile) in ki_bestaetigt:
|
||
details["ki_bestaetigt"] = True
|
||
details["ki_konfidenz"] = ki_bestaetigt[id(zeile)].get("konfidenz")
|
||
ergebnis.konflikte.append(TitelKonflikt(
|
||
art="besprochen",
|
||
beschreibung=f"„{zeile['titel']}“ wurde bereits besprochen ({wer}).",
|
||
details=details,
|
||
))
|
||
for fremd_id, fremd_titel in vorgaenger:
|
||
for zeile in besprochen:
|
||
zeilen_bgg = zeile.get("bgg_id")
|
||
passt = (
|
||
(fremd_id and zeilen_bgg is not None and int(zeilen_bgg) == fremd_id)
|
||
or titel_aehnlich(fremd_titel, zeile.get("titel") or "")
|
||
)
|
||
if passt:
|
||
wer = _rezensent_name(zeile, namen)
|
||
ergebnis.konflikte.append(TitelKonflikt(
|
||
art="besprochen",
|
||
beschreibung=(
|
||
f"Vorgänger „{fremd_titel}“ wurde bereits besprochen ({wer}) — "
|
||
"Erweiterung prüfen, ob sie noch Redaktionsstoff bietet."
|
||
),
|
||
details={
|
||
"vorgaenger_bgg_id": fremd_id,
|
||
"vorgaenger_titel": fremd_titel,
|
||
"planungseintrag": zeile["id"],
|
||
},
|
||
))
|
||
|
||
# d) Titel schon in der Planungsliste eines anderen Rezensenten.
|
||
for zeile in planungen:
|
||
if user_id is not None and zeile.get("rezensent_id") == user_id:
|
||
continue
|
||
if _gleiches_spiel(zeile, titel, id_wert) or id(zeile) in ki_bestaetigt:
|
||
wer = _rezensent_name(zeile, namen)
|
||
status = zeile.get("status") or ""
|
||
details = {
|
||
"planungseintrag": zeile["id"],
|
||
"rezensent": wer,
|
||
"status": status,
|
||
}
|
||
if id(zeile) in ki_bestaetigt:
|
||
details["ki_bestaetigt"] = True
|
||
details["ki_konfidenz"] = ki_bestaetigt[id(zeile)].get("konfidenz")
|
||
ergebnis.konflikte.append(TitelKonflikt(
|
||
art="planung",
|
||
beschreibung=(
|
||
f"„{zeile['titel']}“ ist bereits in der Planungsliste "
|
||
f"von {wer} (Status: {status})."
|
||
),
|
||
details=details,
|
||
))
|
||
|
||
# ---- KI-Zweitprüfung Teil 2: Verlags-/Titel-Hinweis bei Verlags-Konflikt ----
|
||
# Verlags-Aliase („Hans im Glück“ vs. „Hans im Glück Verlag“) und
|
||
# Übersetzungsfragen kann die Regelprüfung nicht entscheiden — ein
|
||
# ergänzender LLM-Call liefert Empfehlungen, ohne den Befund zu ändern.
|
||
if ki_client is not None and ergebnis.hat_verlagskonflikt:
|
||
_ki_verlags_empfehlung(
|
||
ergebnis, ki_client, konfidenz_min=konfidenz_min,
|
||
titel=titel, verlag=verlag, kandidaten=kandidaten,
|
||
alternativen=alternativen, vorgaenger=vorgaenger,
|
||
)
|
||
|
||
return ergebnis
|
||
|
||
|
||
def _uebernimm_deutschen_titel(ergebnis: PruefErgebnis, deutscher_titel: str | None, eigener_titel: str) -> None:
|
||
"""Übernimmt einen KI-deutschen Titel als Empfehlung (konservativ).
|
||
|
||
Nur wenn die Heuristik selbst nichts Deutsches fand (Empfehlung leer oder
|
||
gleich dem eigenen Titel), weißt das LLM die Redaktion ggf. besser.
|
||
"""
|
||
if not deutscher_titel:
|
||
return
|
||
aktuell = ergebnis.titel_empfehlung
|
||
if aktuell and normalisiere(aktuell) != normalisiere(eigener_titel):
|
||
return # Heuristik/KI hat bereits eine abweichende deutsche Variante gewählt
|
||
if normalisiere(deutscher_titel) == normalisiere(eigener_titel):
|
||
return
|
||
ergebnis.titel_empfehlung = deutscher_titel
|
||
|
||
|
||
def _ki_verlags_empfehlung(
|
||
ergebnis: PruefErgebnis,
|
||
ki_client,
|
||
*,
|
||
konfidenz_min: float,
|
||
titel: str,
|
||
verlag: str | None,
|
||
kandidaten: list[dict],
|
||
alternativen: list[str],
|
||
vorgaenger: list[tuple[int, str]],
|
||
) -> None:
|
||
"""Ein ergänzender KI-Call zum Verlags-Konflikt (Verlags-Aliase, Übersetzung)."""
|
||
fremd_zeile = next(
|
||
(z for z in kandidaten
|
||
if (z.get("verlag") or "").strip() and verlag
|
||
and _verlags_schluessel(z["verlag"]) != _verlags_schluessel(verlag)),
|
||
None,
|
||
)
|
||
if fremd_zeile is None:
|
||
return
|
||
fremd_titel = (fremd_zeile.get("titel") or "").strip()
|
||
try:
|
||
antwort = ki_client.bewerte(
|
||
titel_a=titel, titel_b=fremd_titel,
|
||
verlag_a=verlag, verlag_b=fremd_zeile.get("verlag"),
|
||
alternativen_a=alternativen,
|
||
expansionen_a=[name for _, name in vorgaenger],
|
||
frage=(
|
||
"Beide Einträge meinen laut Regelprüfung dasselbe Spiel, wurden "
|
||
"aber unter verschiedenen Verlagen geführt (ggf. Alias oder "
|
||
"Lokalausgabe). Welcher Verlag sollte geführt werden "
|
||
"(empfohlener_verlag) und welchen deutschen Titel gibt es?"
|
||
),
|
||
)
|
||
except Exception as exc: # Zusatzdaten dürfen nie blockieren
|
||
_logger.warning("dedup/KI: Verlags-Bewertung fehlgeschlagen (%s)", exc)
|
||
antwort = None
|
||
ergebnis.ki_befunde.append(_ki_protokoll(
|
||
"verlag", titel, fremd_titel, ki_client, antwort,
|
||
konfidenz_min=konfidenz_min,
|
||
))
|
||
if _ki_bestaetigt(antwort, konfidenz_min):
|
||
verlags_konflikt = next(
|
||
(k for k in ergebnis.konflikte if k.art == "verlag"), None
|
||
)
|
||
if verlags_konflikt is not None and antwort.get("empfohlener_verlag"):
|
||
verlags_konflikt.details["ki_empfohlener_verlag"] = antwort["empfohlener_verlag"]
|
||
verlags_konflikt.details["ki_konfidenz"] = antwort.get("konfidenz")
|
||
_uebernimm_deutschen_titel(ergebnis, antwort.get("deutscher_titel"), titel)
|