"""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 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. " "Mische Bekanntes mit Geheimtipps. " "Antworte nur mit JSON in dieser Form, ohne Text davor oder danach: " '{"filme": [{"titel": "Originaltitel", "jahr": 1999, ' '"grund": "ein Satz auf Deutsch, warum der Film passt"}]}' ) 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}") 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": str(f.get("grund") or "").strip(), } ) 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_)}, ], # Wie beim Teaser: sonst denkt M3 das Token-Limit voll "reasoning_effort": "low", "max_tokens": 4000, "temperature": 0.7, } async with httpx.AsyncClient( headers={"Authorization": f"Bearer {MINIMAX_KEY}"}, timeout=120.0 ) as http: resp = await http.post(MINIMAX_URL, json=payload) resp.raise_for_status() return parsen(resp.json()["choices"][0]["message"]["content"]) 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 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_)]