rippy/docker/api/clients/omdb.py

"""OMDb API Client — Fallback-Quelle für die Metadaten-Erkennung.

Kommt zum Zug, wenn TMDB nichts (Sicheres) findet. Kostenloser Key von
https://www.omdbapi.com/apikey.aspx (1000 Anfragen/Tag reichen fürs Heimlab).
"""

from typing import Dict, Optional

import requests

from cache import get, set as cache_set
from config import settings

OMDB_BASE_URL = "https://www.omdbapi.com/"


def parse_runtime_minutes(runtime: str) -> int:
    """OMDb liefert '142 min' als Text → Minuten als Zahl (pure Funktion, testbar)."""
    if not runtime:
        return 0
    teil = runtime.strip().split(" ")[0]
    return int(teil) if teil.isdigit() else 0


def parse_year(year: str) -> Optional[int]:
    """OMDb-Jahr kann '1999' oder '2005–2012' (Serie) sein → erstes Jahr."""
    if not year:
        return None
    erste = year.strip()[:4]
    return int(erste) if erste.isdigit() else None


class OMDbClient:
    def __init__(self):
        # DB-Einstellung (Settings-UI/Wizard) gewinnt gegen die Env-Variable
        from db import get_settings
        self.api_key = get_settings().get("omdbApiKey") or settings.omdb_api_key
        self.session = requests.Session()

    def suche(self, title: str) -> list:
        """Roh-Kandidaten für die manuelle Korrektur (unscharfe s=-Suche)."""
        if not self.api_key:
            return []
        try:
            antwort = self.session.get(
                OMDB_BASE_URL,
                params={"apikey": self.api_key, "s": title, "r": "json"},
                timeout=10,
            ).json()
        except Exception:
            return []
        if antwort.get("Response") != "True":
            return []
        ergebnisse = []
        for e in (antwort.get("Search") or [])[:5]:
            poster = e.get("Poster", "")
            ergebnisse.append({
                "title": e.get("Title", ""),
                "year": parse_year(e.get("Year", "")),
                "poster": poster if poster != "N/A" else "",
                "overview": "",
                "type": "tv" if e.get("Type") == "series" else "movie",
                "source": "omdb",
                "id": e.get("imdbID", ""),
            })
        return ergebnisse

    def lookup(self, title: str, year: Optional[int] = None) -> Optional[Dict]:
        """Sucht per Titel (t=), liefert normalisierte Metadaten oder None."""
        if not self.api_key:
            return None

        cache_key = f"omdb:{title.lower()}:{year or ''}"
        cached = get(cache_key)
        if cached:
            return cached

        params = {"apikey": self.api_key, "t": title, "r": "json"}
        if year:
            params["y"] = str(year)

        try:
            response = self.session.get(OMDB_BASE_URL, params=params, timeout=10)
            response.raise_for_status()
            data = response.json()

            # t= ist exakt — wenn leer, unscharfe Suche (s=) und den besten
            # Treffer per imdbID nachladen
            if data.get("Response") != "True":
                suche = self.session.get(
                    OMDB_BASE_URL,
                    params={"apikey": self.api_key, "s": title, "r": "json"},
                    timeout=10,
                ).json()
                treffer = (suche.get("Search") or []) if suche.get("Response") == "True" else []
                if treffer:
                    # Bester Kandidat statt blind Treffer #1: echte Filme vor
                    # Episoden/Reviews, mit Poster vor ohne, kurzer Titel vor
                    # langem (Zusatz-Geschwafel wie "Episode 110 - …" verliert).
                    def rang(t: Dict):
                        return (
                            0 if t.get("Type") == "movie" else (1 if t.get("Type") == "series" else 2),
                            0 if t.get("Poster") not in (None, "", "N/A") else 1,
                            len(t.get("Title") or ""),
                        )

                    bester = sorted(treffer, key=rang)[0]
                    data = self.session.get(
                        OMDB_BASE_URL,
                        params={"apikey": self.api_key, "i": bester["imdbID"], "r": "json"},
                        timeout=10,
                    ).json()
        except Exception as e:
            print(f"OMDb API Error: {e}")
            return None

        if data.get("Response") != "True":
            return None

        plot = data.get("Plot", "")
        if plot == "N/A":
            plot = ""
        poster = data.get("Poster", "")
        if poster == "N/A":
            poster = ""

        # Qualitäts-Gate: Einträge ohne Poster UND ohne Beschreibung sind
        # fast immer Datenmüll (Review-Episoden etc.) — als Nicht-Treffer
        # werten, damit die Kandidaten-Kette weiterprobiert.
        if not poster and not plot:
            return None

        ergebnis = {
            "type": "tv" if data.get("Type") == "series" else "movie",
            "id": data.get("imdbID", ""),
            "title": data.get("Title", title),
            "year": parse_year(data.get("Year", "")),
            "overview": plot,
            "poster_path": poster,
            "backdrop_path": "",
            "runtime": parse_runtime_minutes(data.get("Runtime", "")),
            "genres": [g.strip() for g in data.get("Genre", "").split(",") if g.strip() and g.strip() != "N/A"],
            "source": "omdb",
        }
        cache_set(cache_key, ergebnis)
        return ergebnis