398 lines
14 KiB
Python
398 lines
14 KiB
Python
"""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 unicodedata
|
|
|
|
import httpx
|
|
|
|
from .config import LANGUAGE, TMDB_BASE, TMDB_TOKEN
|
|
from .tmdb import (
|
|
DETAIL_PARAMS,
|
|
pick_backdrops,
|
|
pick_overview,
|
|
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.
|
|
"""
|
|
|
|
async def hole(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
|
|
|
|
|
|
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:
|
|
resp = await http.get("/discover/movie", params=params)
|
|
resp.raise_for_status()
|
|
daten = resp.json()
|
|
|
|
# Bei "Zufällig" nochmal von einer beliebigen Seite holen — sonst
|
|
# stehen bei jedem Aufruf dieselben zwanzig Filme oben
|
|
if filter_.get("sort") == "random":
|
|
seiten = min(daten.get("total_pages") or 1, MAX_SEITE)
|
|
if seiten > 1:
|
|
params["page"] = random.randint(1, seiten)
|
|
resp = await http.get("/discover/movie", params=params)
|
|
resp.raise_for_status()
|
|
daten = resp.json()
|
|
|
|
results = daten.get("results", [])
|
|
if filter_.get("sort") == "random":
|
|
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):
|
|
try:
|
|
resp = await http.get(
|
|
"/search/movie",
|
|
params={
|
|
"query": begriff,
|
|
"language": LANGUAGE,
|
|
"include_adult": "false",
|
|
},
|
|
)
|
|
resp.raise_for_status()
|
|
return resp.json().get("results", [])
|
|
except httpx.HTTPError:
|
|
return []
|
|
|
|
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
|
|
|
|
|
|
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"),
|
|
"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"),
|
|
}
|