252 lines
9.8 KiB
Python
252 lines
9.8 KiB
Python
"""Filmvorschläge von MiniMax.
|
||
|
||
Das Modell kennt Filme nur aus seinem Training — keine Datenbank, kein
|
||
Netz. Es bekommt eine Beschreibung und auf Wunsch die eigene Sammlung,
|
||
nennt Titel mit Jahr und einem Satz Begründung. Jeder Titel wird danach
|
||
bei TMDB nachgeschlagen. Was es dort nicht gibt, fällt weg: so bleiben
|
||
erfundene oder verwechselte Filme draußen.
|
||
"""
|
||
|
||
import asyncio
|
||
import json
|
||
from difflib import SequenceMatcher
|
||
|
||
import httpx
|
||
|
||
from . import search
|
||
from .config import LANGUAGE, MINIMAX_KEY, MINIMAX_MODEL, MINIMAX_URL
|
||
from .teaser import THINK_BLOCK
|
||
|
||
ANZAHL = 12
|
||
MAX_BESCHREIBUNG = 1000
|
||
|
||
SYSTEM = (
|
||
"Du bist ein belesener Filmkenner mit Schwerpunkt Horror und Genrekino. "
|
||
"Du schlägst Filme vor, die zu dem passen, was gesucht wird. "
|
||
f"Nenne genau {ANZAHL} Filme, keine Serien. Nur Filme, die es wirklich gibt. "
|
||
"Nichts vorschlagen, was in der Sammlung steht oder schon gesehen wurde. "
|
||
"Was in der Frage genannt ist, kennt die Person schon — auch nicht vorschlagen. "
|
||
"Mische Bekanntes mit Geheimtipps. "
|
||
"Antworte nur mit JSON in dieser Form, ohne Text davor oder danach: "
|
||
'{"merkmale": ["drei Stichpunkte: was macht das Gesuchte aus"], '
|
||
'"kandidaten": ["20 Originaltitel mit Jahr, die dazu passen könnten"], '
|
||
'"filme": [{"titel": "Originaltitel", "jahr": 1999, "grund": "..."}]}\n'
|
||
# Der Umweg über Merkmale und Kandidaten ersetzt das Nachdenken des
|
||
# Modells: es sortiert erst, dann wählt es — und trifft deutlich besser.
|
||
"Erst merkmale, dann kandidaten, dann die zwölf besten davon als filme.\n"
|
||
# Die Begründung lesen Leute, die einfach einen Film für den Abend
|
||
# suchen — nicht Kritiker. Sie muss auf vier kurze Zeilen passen.
|
||
"Für grund gilt: Höchstens 10 Wörter und 90 Zeichen, Deutsch in lateinischer "
|
||
"Schrift, für Laien. "
|
||
"Sag klar, was der Film mit dem Gesuchten gemeinsam hat. "
|
||
"Keine Fachwörter wie Whodunit, Neo-Noir, Ensemble, Slasher, Body-Horror, "
|
||
"Meta, Hommage, Genre-Namen auf Englisch. Keine Inhaltsangabe, kein Titel, "
|
||
"kein Land. Ein Regisseur nur, wenn er der Grund ist. "
|
||
"Jede Begründung nennt einen anderen Punkt, keine zwei gleich. "
|
||
'Beispiel — gesucht "Ich mochte Knives Out": '
|
||
'"Wieder rätselt eine zerstrittene Familie um einen Mord, mit Witz."'
|
||
)
|
||
|
||
# Auf der Karte hat die Begründung vier Zeilen à gut zwanzig Zeichen
|
||
MAX_GRUND = 95
|
||
|
||
|
||
def kuerzen(grund):
|
||
"""Das Modell zählt ungenau. Was nicht auf die Karte passt, endet am
|
||
letzten Komma oder Wort davor — ein halber Satz ist besser als ein
|
||
abgeschnittener."""
|
||
grund = grund.strip()
|
||
if len(grund) <= MAX_GRUND:
|
||
return grund
|
||
kopf = grund[:MAX_GRUND]
|
||
schnitt = max(kopf.rfind(","), kopf.rfind(";"), kopf.rfind(" –"), kopf.rfind("."))
|
||
if schnitt < 40:
|
||
schnitt = kopf.rfind(" ")
|
||
return kopf[:schnitt].rstrip(",; –.") + " …"
|
||
|
||
|
||
def etikett(film):
|
||
"""Titel (Jahr) — so kommt ein Film in den Prompt."""
|
||
jahr = film.get("year")
|
||
return f"{film['title']} ({jahr})" if jahr else film["title"]
|
||
|
||
|
||
def _spanne(unten, oben, einheit="", ab="mindestens", bis="höchstens"):
|
||
if unten is not None and oben is not None:
|
||
return f"zwischen {unten}{einheit} und {oben}{einheit}"
|
||
if unten is not None:
|
||
return f"{ab} {unten}{einheit}"
|
||
if oben is not None:
|
||
return f"{bis} {oben}{einheit}"
|
||
return ""
|
||
|
||
|
||
def vorgaben(filter_):
|
||
"""Die Filterleiste als Sätze für den Prompt. Nur was verstellt ist,
|
||
kommt vor — die Leiste schickt auch nur das."""
|
||
f = filter_ or {}
|
||
zeilen = []
|
||
if f.get("include"):
|
||
zeilen.append("Genre: " + ", ".join(f["include"]))
|
||
if f.get("exclude"):
|
||
zeilen.append("Auf keinen Fall: " + ", ".join(f["exclude"]))
|
||
for text in (
|
||
_spanne(f.get("jahr_min"), f.get("jahr_max"), ab="ab", bis="bis"),
|
||
_spanne(f.get("note_min"), f.get("note_max"), " von 10 auf TMDB"),
|
||
_spanne(f.get("dauer_min"), f.get("dauer_max"), " Minuten"),
|
||
_spanne(f.get("stimmen_min"), f.get("stimmen_max"), " Stimmen auf TMDB"),
|
||
):
|
||
if text:
|
||
zeilen.append(text[0].upper() + text[1:])
|
||
return zeilen
|
||
|
||
|
||
def prompt(beschreibung, sammlung, gesehen, filter_=None):
|
||
teile = []
|
||
if beschreibung:
|
||
teile.append(f"Gesucht: {beschreibung}")
|
||
# In der Nutzernachricht hält sich das Modell eher daran als im
|
||
# Systemprompt — sonst steht die Fortsetzung des Gesuchten vorn
|
||
teile.append(
|
||
"Nicht vorschlagen: Filme, die oben genannt sind, und ihre Fortsetzungen."
|
||
)
|
||
regeln = vorgaben(filter_)
|
||
if regeln:
|
||
teile.append("Vorgaben, unbedingt einhalten:\n" + "\n".join(f"- {z}" for z in regeln))
|
||
if sammlung:
|
||
teile.append("Meine Sammlung, so etwas mag ich:\n" + "\n".join(f"- {t}" for t in sammlung))
|
||
if not beschreibung and sammlung:
|
||
teile.append("Schlag vor, was zu dieser Sammlung passt.")
|
||
elif not beschreibung:
|
||
teile.append("Schlag vor, was zu den Vorgaben passt.")
|
||
if gesehen:
|
||
teile.append("Schon gesehen, nicht vorschlagen:\n" + "\n".join(f"- {t}" for t in gesehen))
|
||
return "\n\n".join(teile)
|
||
|
||
|
||
def parsen(text):
|
||
"""Das JSON aus der Antwort ziehen. Das Modell hängt gern einen
|
||
Denkblock oder einen Satz davor — es zählt, was zwischen der ersten
|
||
und der letzten Klammer steht."""
|
||
text = THINK_BLOCK.sub("", text)
|
||
a, b = text.find("{"), text.rfind("}")
|
||
if a < 0 or b < a:
|
||
return []
|
||
try:
|
||
daten = json.loads(text[a : b + 1])
|
||
except ValueError:
|
||
return []
|
||
filme = daten.get("filme") if isinstance(daten, dict) else None
|
||
aus = []
|
||
for f in filme or []:
|
||
if not isinstance(f, dict):
|
||
continue
|
||
titel = str(f.get("titel") or "").strip()
|
||
if not titel:
|
||
continue
|
||
jahr = str(f.get("jahr") or "")
|
||
aus.append(
|
||
{
|
||
"titel": titel,
|
||
"jahr": int(jahr) if jahr.isdigit() else None,
|
||
"grund": kuerzen(str(f.get("grund") or "")),
|
||
}
|
||
)
|
||
return aus[:ANZAHL]
|
||
|
||
|
||
async def fragen(beschreibung, sammlung, gesehen, filter_=None):
|
||
payload = {
|
||
"model": MINIMAX_MODEL,
|
||
"messages": [
|
||
{"role": "system", "content": SYSTEM},
|
||
{"role": "user", "content": prompt(beschreibung, sammlung, gesehen, filter_)},
|
||
],
|
||
# Ohne Denkblock: M3 grübelt sonst minutenlang über zwölf Filme
|
||
# und liefert am Token-Limit gar nichts. Die Merkmale- und
|
||
# Kandidatenfelder im Prompt leisten das Sortieren stattdessen,
|
||
# in gut zehn Sekunden. Niedrige Temperatur: weniger Ausreißer.
|
||
"thinking": {"type": "disabled"},
|
||
"max_tokens": 4000,
|
||
"temperature": 0.3,
|
||
}
|
||
async with httpx.AsyncClient(
|
||
headers={"Authorization": f"Bearer {MINIMAX_KEY}"}, timeout=120.0
|
||
) as http:
|
||
# Hin und wieder kommt kein brauchbares JSON zurück. Ein zweiter
|
||
# Anlauf kostet wenige Sekunden und sitzt fast immer.
|
||
for _ in range(2):
|
||
resp = await http.post(MINIMAX_URL, json=payload)
|
||
resp.raise_for_status()
|
||
filme = parsen(resp.json()["choices"][0]["message"]["content"])
|
||
if filme:
|
||
return filme
|
||
return []
|
||
|
||
|
||
def _aehnlich(gesucht, treffer):
|
||
"""Passt der TMDB-Treffer zum genannten Titel? Die Suche liefert
|
||
auch Streifschüsse — "The Afterparty" findet ein Winx-Special. Der
|
||
deutsche oder der Originaltitel muss dem Genannten nahekommen."""
|
||
soll = search.ohne_zeichen(gesucht).lower()
|
||
return any(
|
||
SequenceMatcher(None, soll, search.ohne_zeichen(t or "").lower()).ratio() >= 0.6
|
||
for t in (treffer.get("title"), treffer.get("original_title"))
|
||
)
|
||
|
||
|
||
def _dauer_passt(karte, filter_):
|
||
dauer = karte.get("runtime")
|
||
if dauer is None:
|
||
return True
|
||
unten, oben = filter_.get("dauer_min"), filter_.get("dauer_max")
|
||
return (unten is None or dauer >= unten) and (oben is None or dauer <= oben)
|
||
|
||
|
||
async def aufloesen(vorschlaege, filter_=None, ausser=()):
|
||
"""Jeden Titel bei TMDB suchen. Der Treffer mit dem genannten Jahr
|
||
gewinnt — das Modell liegt gern ein Jahr daneben, also mit Spiel.
|
||
Sonst der beliebteste. Ohne Treffer war der Film wohl erfunden.
|
||
|
||
Die Filter der Leiste greifen hier noch einmal hart: was die KI
|
||
trotz Vorgabe danebenlegt, sieben die TMDB-Daten aus."""
|
||
filter_ = filter_ or {}
|
||
|
||
async def einer(http, v):
|
||
try:
|
||
resp = await http.get(
|
||
"/search/movie",
|
||
params={"query": v["titel"], "language": LANGUAGE, "include_adult": "false"},
|
||
)
|
||
resp.raise_for_status()
|
||
treffer = [
|
||
t for t in resp.json().get("results", [])
|
||
if t.get("poster_path")
|
||
and _aehnlich(v["titel"], t)
|
||
and search.passt_zur_leiste(t, filter_)
|
||
]
|
||
except httpx.HTTPError:
|
||
return None
|
||
if v["jahr"]:
|
||
passend = [
|
||
t for t in treffer
|
||
if (t.get("release_date") or "")[:4].isdigit()
|
||
and abs(int(t["release_date"][:4]) - v["jahr"]) <= 1
|
||
]
|
||
treffer = passend or treffer
|
||
if not treffer:
|
||
return None
|
||
bester = max(treffer, key=lambda t: t.get("popularity") or 0)
|
||
return {**search.as_result(bester), "grund": v["grund"]}
|
||
|
||
async with search.client() as http:
|
||
gefunden = await asyncio.gather(*(einer(http, v) for v in vorschlaege))
|
||
karten, dabei = [], set(ausser)
|
||
for k in gefunden:
|
||
if k and k["id"] not in dabei:
|
||
dabei.add(k["id"])
|
||
karten.append(k)
|
||
await search.details_nachladen(http, karten)
|
||
# Die Laufzeit kennt erst die Detailantwort
|
||
return [k for k in karten if _dauer_passt(k, filter_)]
|