"""Freie Filmsuche über TMDB. Der Katalog deckt nur Netflix und den Kanon ab. Für den Filmabend braucht es auch Titel, die nirgends davon stehen — die kommen live von TMDB und landen nicht in der Datenbank. """ import asyncio import random import time import unicodedata import httpx from .config import LANGUAGE, TMDB_BASE, TMDB_TOKEN from .tmdb import ( DETAIL_PARAMS, pick_backdrops, pick_overview, pick_rentable, pick_title, pick_trailer, ) HEADERS = {"Authorization": f"Bearer {TMDB_TOKEN}", "accept": "application/json"} def client(): return httpx.AsyncClient(base_url=TMDB_BASE, headers=HEADERS, timeout=30.0) # Die Namen kommen von TMDB selbst und ändern sich praktisch nie. Sie hier # festzuhalten spart bei jeder Suche einen Abruf. GENRES = { 28: "Action", 12: "Abenteuer", 16: "Animation", 35: "Komödie", 80: "Krimi", 99: "Dokumentarfilm", 18: "Drama", 10751: "Familie", 14: "Fantasy", 36: "Historie", 27: "Horror", 10402: "Musik", 9648: "Mystery", 10749: "Liebesfilm", 878: "Science Fiction", 10770: "TV-Film", 53: "Thriller", 10752: "Kriegsfilm", 37: "Western", } NACH_NAME = {name: kennung for kennung, name in GENRES.items()} def as_result(item): release = item.get("release_date") or "" return { "id": item["id"], "title": item.get("title") or item.get("original_title") or "Ohne Titel", "original_title": item.get("original_title"), "year": int(release[:4]) if release[:4].isdigit() else None, "release_date": release or None, "overview": item.get("overview") or "", "poster_path": item.get("poster_path"), "vote_average": item.get("vote_average"), "vote_count": item.get("vote_count"), "genres": [GENRES[g] for g in item.get("genre_ids", []) if g in GENRES], # Trefferlisten sind knapp — details_nachladen() füllt nach "images": [], "runtime": None, } async def details_nachladen(http, treffer, je_film=4): """Bilder, Laufzeit und lesbarer Titel — alle Abrufe gleichzeitig. Trefferlisten von TMDB sind knapp: keine Bilder, keine Laufzeit, und bei indischen oder japanischen Filmen steht der Titel in Originalschrift. Das alles steckt in der Detailansicht, ein Abruf je Film reicht. """ # Nicht alle auf einmal: bei einer langen Filmografie wären das # über hundert gleichzeitige Abrufe bremse = asyncio.Semaphore(8) async def hole(film): async with bremse: await hole_einen(film) async def hole_einen(film): try: resp = await http.get( f"/movie/{film['id']}", params={ "language": LANGUAGE, "append_to_response": "images,translations", "include_image_language": "de,en,null", }, ) resp.raise_for_status() daten = resp.json() except httpx.HTTPError: return film["title"] = pick_title(daten) film["runtime"] = daten.get("runtime") backdrops = daten.get("images", {}).get("backdrops", []) # Ohne Sprachkennung steht kein Text im Bild — das sind die Szenen ohne_text = [b for b in backdrops if not b.get("iso_639_1")] film["images"] = [ b["file_path"] for b in (ohne_text or backdrops)[:je_film] ] await asyncio.gather(*(hole(f) for f in treffer)) SORTIERUNG = { "rating": "vote_average.desc", "newest": "primary_release_date.desc", "oldest": "primary_release_date.asc", "title": "title.asc", # TMDB kennt keine Zufallsordnung. Statt dessen holt discover() eine # zufällige Seite aus den Beliebtesten. "random": "popularity.desc", } # Weiter hinten liegen nur noch Filme, die kaum jemand kennt MAX_SEITE = 20 # TMDB liefert zwanzig je Seite; bis hundert sind es fünf Abrufe MAX_TREFFER = 100 JE_SEITE = 20 async def seiten(http, pfad, params, limit, start=1): """So viele Seiten, wie das Limit braucht — alle auf einmal. Was fehlschlägt, fehlt eben; die Reihenfolge der Seiten bleibt.""" anzahl = -(-limit // JE_SEITE) async def eine(nummer): try: resp = await http.get(pfad, params={**params, "page": nummer}) resp.raise_for_status() return resp.json() except httpx.HTTPError: return {} antworten = await asyncio.gather(*(eine(start + i) for i in range(anzahl))) results = [r for a in antworten for r in a.get("results", [])] return results, antworten[0] if antworten else {} async def discover(filter_, limit=40): """Stöbern ohne Suchbegriff: TMDB sucht nach den Filtern aus der Leiste. Anders als /search/movie versteht /discover/movie Genres, Jahre und Bewertungen — damit lässt sich die Leiste direkt weiterreichen. """ params = { "language": LANGUAGE, "include_adult": "false", "sort_by": SORTIERUNG.get(filter_.get("sort"), "popularity.desc"), # Ohne Untergrenze stehen Filme mit einer einzigen Stimme ganz oben "vote_count.gte": filter_.get("stimmen_min") or 50, } einbeziehen = [ NACH_NAME[g] for g in (filter_.get("include") or []) if g in NACH_NAME ] ausschliessen = [ NACH_NAME[g] for g in (filter_.get("exclude") or []) if g in NACH_NAME ] if einbeziehen: params["with_genres"] = ",".join(str(g) for g in einbeziehen) if ausschliessen: params["without_genres"] = ",".join(str(g) for g in ausschliessen) for schluessel, ziel in ( ("stimmen_max", "vote_count.lte"), ("dauer_min", "with_runtime.gte"), ("dauer_max", "with_runtime.lte"), ("note_min", "vote_average.gte"), ("note_max", "vote_average.lte"), ): wert = filter_.get(schluessel) if wert is not None: params[ziel] = wert if filter_.get("jahr_min"): params["primary_release_date.gte"] = f"{filter_['jahr_min']}-01-01" if filter_.get("jahr_max"): params["primary_release_date.lte"] = f"{filter_['jahr_max']}-12-31" async with client() as http: results, erste = await seiten(http, "/discover/movie", params, limit) # Bei "Zufällig" nochmal von einer beliebigen Stelle holen — sonst # stehen bei jedem Aufruf dieselben Filme oben if filter_.get("sort") == "random": gesamt = min(erste.get("total_pages") or 1, MAX_SEITE) noetig = -(-limit // JE_SEITE) if gesamt > noetig: start = random.randint(1, gesamt - noetig + 1) results, _ = await seiten(http, "/discover/movie", params, limit, start) random.shuffle(results) usable = [r for r in results if r.get("poster_path")] treffer = [as_result(r) for r in usable[:limit]] await details_nachladen(http, treffer) return treffer def ohne_zeichen(text): """Umlaute und Akzente auf den Grundbuchstaben zurückführen.""" zerlegt = unicodedata.normalize("NFD", text) return "".join(z for z in zerlegt if not unicodedata.combining(z)) async def search(query, limit=24): # TMDB findet angefangene Wörter nur ohne Umlaut: "Wilden Hüh" liefert # nichts, "Wilden Huh" die drei Hühner-Filme. Wer beim Tippen sucht, # steht sonst vor einer leeren Seite. Also beide Schreibweisen fragen. schlicht = ohne_zeichen(query) begriffe = [query] if schlicht == query else [query, schlicht] async def einmal(begriff): results, _ = await seiten( http, "/search/movie", {"query": begriff, "language": LANGUAGE, "include_adult": "false"}, limit, ) return results async with client() as http: results = [] gesehen = set() # Die Treffer der wörtlichen Suche stehen vorn for liste in await asyncio.gather(*(einmal(b) for b in begriffe)): for eintrag in liste: if eintrag["id"] not in gesehen: gesehen.add(eintrag["id"]) results.append(eintrag) # Ohne Poster taugt eine Kachel nichts, und Namenlose sind fast immer # Dubletten oder Making-ofs usable = [r for r in results if r.get("poster_path")] usable.sort(key=lambda r: -(r.get("popularity") or 0)) treffer = [as_result(r) for r in usable[:limit]] await details_nachladen(http, treffer) return treffer # TMDB hat je Film gut dreißig Seiten Empfehlungen. Weiter hinten wird die # Verwandtschaft dünn, und jede Seite kostet einen Abruf. MAX_AEHNLICH_SEITEN = 8 def passt_zur_leiste(item, filter_): """Die Filter der Leiste auf einen knappen Treffer anwenden. Die Laufzeit fehlt hier noch — die steht erst in der Detailantwort. Nach ihr filtert weiter die Oberfläche. """ genres = {GENRES[g] for g in item.get("genre_ids", []) if g in GENRES} einbeziehen = set(filter_.get("include") or []) ausschliessen = set(filter_.get("exclude") or []) if einbeziehen and not (genres & einbeziehen): return False if genres & ausschliessen: return False release = item.get("release_date") or "" jahr = int(release[:4]) if release[:4].isdigit() else None for wert, unten, oben in ( (item.get("vote_count"), filter_.get("stimmen_min"), filter_.get("stimmen_max")), (item.get("vote_average"), filter_.get("note_min"), filter_.get("note_max")), (jahr, filter_.get("jahr_min"), filter_.get("jahr_max")), ): if wert is None: continue if unten is not None and wert < unten: return False if oben is not None and wert > oben: return False return True def brauchbare(roh, filter_): """Dubletten und Posterlose raus, dann die Filter drüber.""" raus, gesehen = [], set() for item in roh: if not item.get("poster_path") or item["id"] in gesehen: continue gesehen.add(item["id"]) if passt_zur_leiste(item, filter_): raus.append(item) return raus # "relevanz" gehört nicht dazu: dort bleibt TMDBs eigene Ordnung stehen. # Die geht nach Verwandtschaft — dieselbe, die IMDb unter "More like this" # zeigt — und ist für Ähnliche das sinnvollste Erste. UMSORTIERBAR = ("rating", "newest", "oldest", "title", "random") def sortiert(kandidaten, sort): """Die Reihenfolge aus der Leiste auf die Kandidaten anwenden.""" if sort == "rating": return sorted(kandidaten, key=lambda r: -(r.get("vote_average") or 0)) if sort == "newest": return sorted( kandidaten, key=lambda r: r.get("release_date") or "", reverse=True ) if sort == "oldest": return sorted(kandidaten, key=lambda r: r.get("release_date") or "9999") if sort == "title": return sorted(kandidaten, key=lambda r: (r.get("title") or "").lower()) if sort == "random": gemischt = list(kandidaten) random.shuffle(gemischt) return gemischt return kandidaten async def aehnliche(movie_id, limit=12, voll=False, filter_=None): """Was Leute noch geschaut haben, die diesen Film mochten. TMDB bietet zwei Wege an. /similar geht nur nach Genre und Schlagwörtern und liefert für einen Horrorfilm hunderttausend Treffer — unbrauchbar. /recommendations beruht auf dem Verhalten der Nutzer und trifft. Filtern muss diese Funktion selbst: /recommendations kennt anders als /discover keine Genres oder Jahre. Reicht die erste Seite nicht, holt sie weitere nach — sonst bleiben bei enger Leiste drei Karten übrig. """ filter_ = filter_ or {} sort = filter_.get("sort") or "" async with client() as http: async def seite(nr): try: resp = await http.get( f"/movie/{movie_id}/recommendations", params={"language": LANGUAGE, "page": nr}, ) resp.raise_for_status() return resp.json() except httpx.HTTPError: return {} erste = await seite(1) roh = list(erste.get("results", [])) seiten = min(erste.get("total_pages") or 1, MAX_AEHNLICH_SEITEN) # Alle übrigen Seiten gleichzeitig — nacheinander wären das bei # enger Leiste sieben Wartezeiten hintereinander. # Auch beim Sortieren: die besten zwanzig von zwanzig wären keine. zu_wenig = len(brauchbare(roh, filter_)) < limit if (zu_wenig or sort in UMSORTIERBAR) and seiten > 1: for daten in await asyncio.gather( *(seite(nr) for nr in range(2, seiten + 1)) ): roh.extend(daten.get("results", [])) kandidaten = sortiert(brauchbare(roh, filter_), sort) treffer = [as_result(r) for r in kandidaten[:limit]] # Für die Kachelreihe reicht der Titel, für Karten braucht es mehr if voll: await details_nachladen(http, treffer) else: await asyncio.gather(*(titel_richten(http, f) for f in treffer)) return treffer # ── Personen: Schauspieler und Regisseure ──────────────────────────────── MAX_PERSONEN = 12 # Laufzeit und Szenenbilder kosten je Film einen Abruf — bei einer langen # Filmografie ein paar Sekunden. Einmal geholt, bleibt es einen Tag im # Speicher; so oft ändert sich eine Filmografie nicht. PERSONEN_CACHE = {} PERSONEN_FRIST = 24 * 3600 FACH = { "Acting": "Schauspiel", "Directing": "Regie", "Writing": "Drehbuch", "Production": "Produktion", } async def personen(query): """Leute suchen. Wer ein Foto hat, steht vorn — sonst die Bekannten.""" async with client() as http: try: resp = await http.get( "/search/person", params={"query": query, "language": LANGUAGE, "include_adult": "false"}, ) resp.raise_for_status() roh = resp.json().get("results", []) except httpx.HTTPError: return [] roh.sort(key=lambda p: (not p.get("profile_path"), -(p.get("popularity") or 0))) return [ { "id": p["id"], "name": p.get("name"), "fach": FACH.get(p.get("known_for_department"), p.get("known_for_department") or ""), "profile_path": p.get("profile_path"), "bekannt_fuer": [ k.get("title") or k.get("name") for k in p.get("known_for", []) if k.get("title") or k.get("name") ][:3], } for p in roh[:MAX_PERSONEN] ] def credits_aufbereiten(daten): """Aus den movie_credits zwei Listen: was jemand gespielt und wo er Regie geführt hat. Ohne Poster oder doppelt fliegt raus, die bekanntesten Filme stehen vorn. Kurzfilme und Making-ofs bleiben drin — die sortiert die Filterleiste aus, und zwar sichtbar.""" def liste(eintraege): raus, dabei = [], set() for e in sorted(eintraege, key=lambda e: -(e.get("popularity") or 0)): if not e.get("poster_path") or e["id"] in dabei: continue dabei.add(e["id"]) raus.append(as_result(e)) return raus return { "gespielt": liste(daten.get("cast", [])), "regie": liste([c for c in daten.get("crew", []) if c.get("job") == "Director"]), } # Mehr Namen passen nicht in die Filmansicht, und ab da sind es Nebenrollen MAX_CAST = 8 async def credits(movie_id): """Regie und die wichtigsten Darsteller — als Sprungbrett zur Personensuche. Fehlt etwas, bleiben die Listen leer.""" async with client() as http: try: # Ohne Sprache: auf Deutsch kämen russische Namen in Kyrillisch # und eingedeutschte Umschriften wie "Ilja Wiktorowitsch" resp = await http.get(f"/movie/{movie_id}/credits", params={"language": "en-US"}) resp.raise_for_status() daten = resp.json() except httpx.HTTPError: return {"regie": [], "cast": []} def person(p, fach): return {"id": p["id"], "name": p.get("name"), "fach": fach, "profile_path": p.get("profile_path")} regie, dabei = [], set() for c in daten.get("crew", []): if c.get("job") == "Director" and c["id"] not in dabei: dabei.add(c["id"]) regie.append(person(c, "Regie")) cast = sorted(daten.get("cast", []), key=lambda c: c.get("order", 999))[:MAX_CAST] return {"regie": regie, "cast": [person(c, "Schauspiel") for c in cast]} async def person_filme(person_id): gemerkt = PERSONEN_CACHE.get(person_id) if gemerkt and time.monotonic() - gemerkt[0] < PERSONEN_FRIST: return gemerkt[1] async with client() as http: resp = await http.get(f"/person/{person_id}/movie_credits", params={"language": LANGUAGE}) resp.raise_for_status() rollen = credits_aufbereiten(resp.json()) await details_nachladen(http, rollen["gespielt"] + rollen["regie"]) PERSONEN_CACHE[person_id] = (time.monotonic(), rollen) return rollen async def titel_richten(http, film): """Originalschrift durch den englischen Titel ersetzen.""" if not any( z.isalpha() and not ("a" <= z.lower() <= "z" or z in "äöüßàâçéèêëîïôùûÿñæœ") for z in film["title"] ): return try: resp = await http.get( f"/movie/{film['id']}", params={"language": LANGUAGE, "append_to_response": "translations"}, ) resp.raise_for_status() film["title"] = pick_title(resp.json()) except httpx.HTTPError: pass async def detail(movie_id): """Vollbild-Daten für einen Treffer, der nicht im Katalog steht.""" async with client() as http: resp = await http.get(f"/movie/{movie_id}", params=DETAIL_PARAMS) resp.raise_for_status() data = resp.json() release = data.get("release_date") or "" collection = data.get("belongs_to_collection") or {} return { "id": data["id"], "title": data.get("title") or data.get("original_title") or "Ohne Titel", "original_title": data.get("original_title"), "release_date": release or None, "year": int(release[:4]) if release[:4].isdigit() else None, "overview": pick_overview(data), "teaser": "", "poster_path": data.get("poster_path"), "backdrop_path": data.get("backdrop_path"), "vote_average": data.get("vote_average"), "vote_count": data.get("vote_count"), "runtime": data.get("runtime"), "genres": [g["name"] for g in data.get("genres", [])], "images": pick_backdrops(data)[:12], "trailer_key": pick_trailer(data), "imdb_id": data.get("imdb_id"), "rentable": pick_rentable(data), "netflix_id": None, "netflix_url": None, "on_netflix": 0, "canon_group": None, "canon_pos": None, "collection_id": collection.get("id"), "collection_name": collection.get("name"), }