feat: add language aware images(poster, logo background) from tmdb

.pre-commit-config.yaml

@@ -16,7 +16,7 @@ repos:
       - id: check-docstring-first
       - id: detect-private-key
   - repo: https://github.com/asottile/pyupgrade
-    rev: v3.15.2
+    rev: v3.21.2
     hooks:
       - id: pyupgrade
         args: [--py311-plus]

app/api/endpoints/meta.py

@@ -1,6 +1,6 @@
 import asyncio
 
-from fastapi import APIRouter, HTTPException
+from fastapi import APIRouter, HTTPException, Query
 from loguru import logger
 
 from app.services.tmdb.service import get_tmdb_service
@@ -55,3 +55,39 @@ async def get_languages():
     except Exception as e:
         logger.error(f"Failed to fetch languages: {e}")
         raise HTTPException(status_code=502, detail="Failed to fetch languages from TMDB")
+
+
+@router.get("/api/meta/images")
+async def get_meta_images(
+    imdb_id: str | None = Query(None, description="IMDb ID (e.g. tt1234567)"),
+    tmdb_id: int | None = Query(None, description="TMDB ID (use with kind)"),
+    kind: str = Query("movie", description="Type: movie or series"),
+    language: str = Query("en-US", description="Language for image preference (e.g. en-US, fr-FR)"),
+):
+    """
+    Return logo, poster and background in the requested language.
+    Provide either imdb_id (and optionally kind) or tmdb_id + kind.
+    """
+    try:
+        tmdb = get_tmdb_service(language=language)
+        media_type = "tv" if kind == "series" else "movie"
+
+        if imdb_id:
+            clean_imdb = imdb_id.strip().lower()
+            if not clean_imdb.startswith("tt"):
+                clean_imdb = "tt" + clean_imdb
+            tid, found_type = await tmdb.find_by_imdb_id(clean_imdb)
+            if tid is None:
+                raise HTTPException(status_code=404, detail="Title not found on TMDB")
+            media_type = found_type
+            tmdb_id = tid
+        elif tmdb_id is None:
+            raise HTTPException(status_code=400, detail="Provide imdb_id or tmdb_id")
+
+        images = await tmdb.get_images_for_title(media_type, tmdb_id, language=language)
+        return images
+    except HTTPException:
+        raise
+    except Exception as e:
+        logger.error(f"Failed to fetch meta images: {e}")
+        raise HTTPException(status_code=502, detail="Failed to fetch images from TMDB")

app/services/recommendation/catalog_service.py

@@ -49,6 +49,7 @@ def _clean_meta(meta: dict) -> dict | None:
         "type",
         "name",
         "poster",
+        "logo",
         "background",
         "description",
         "releaseInfo",
@@ -73,8 +74,9 @@ def _clean_meta(meta: dict) -> dict | None:
     # if id does not start with tt, return None
     if not imdb_id.startswith("tt"):
         return None
-    # Add Metahub logo URL (used by Stremio)
-    cleaned["logo"] = f"https://live.metahub.space/logo/medium/{imdb_id}/img"
+    # Use Metahub logo only when no language-aware logo was set (e.g. from TMDB)
+    if not cleaned.get("logo"):
+        cleaned["logo"] = f"https://live.metahub.space/logo/medium/{imdb_id}/img"
     return cleaned
 
 

app/services/recommendation/metadata.py

@@ -28,7 +28,11 @@ def extract_year(item: dict[str, Any]) -> int | None:
 
     @classmethod
     async def format_for_stremio(
-        cls, details: dict[str, Any], media_type: str, user_settings: Any = None
+        cls,
+        details: dict[str, Any],
+        media_type: str,
+        user_settings: Any = None,
+        logo_url: str | None = None,
     ) -> dict[str, Any] | None:
         """Format TMDB details into Stremio metadata object."""
         external_ids = details.get("external_ids", {})
@@ -69,6 +73,8 @@ async def format_for_stremio(
             "_tmdb_id": tmdb_id_raw,
             "genre_ids": [g.get("id") for g in genres_full if isinstance(g, dict) and g.get("id") is not None],
         }
+        if logo_url:
+            meta_data["logo"] = logo_url
 
         # Extensions
         runtime_str = cls._extract_runtime_string(details)
@@ -152,9 +158,26 @@ async def _fetch_one(tid: int):
         tasks = [_fetch_one(it.get("id")) for it in valid_items]
         details_list = await asyncio.gather(*tasks)
 
-        format_task = [
-            cls.format_for_stremio(details, media_type, user_settings) for details in details_list if details
-        ]
+        language = getattr(user_settings, "language", None) or "en-US"
+        mt = "movie" if media_type == "movie" else "tv"
+
+        async def _images_one(d: dict[str, Any]) -> dict[str, str]:
+            async with sem:
+                try:
+                    return await tmdb_service.get_images_for_title(mt, d["id"], language=language)
+                except Exception:
+                    return {}
+
+        successful_details = [d for d in details_list if d]
+        image_tasks = [_images_one(d) for d in successful_details]
+        images_list = await asyncio.gather(*image_tasks, return_exceptions=True)
+
+        format_task = []
+        for details, imgs in zip(successful_details, images_list):
+            logo_url = None
+            if isinstance(imgs, dict):
+                logo_url = imgs.get("logo") or None
+            format_task.append(cls.format_for_stremio(details, media_type, user_settings, logo_url=logo_url))
 
         formatted_list = await asyncio.gather(*format_task, return_exceptions=True)
 

app/services/tmdb/service.py

@@ -138,6 +138,150 @@ async def get_primary_translations(self) -> list[str]:
         """Fetch supported primary translations from TMDB."""
         return await self.client.get("/configuration/primary_translations")
 
+    @alru_cache(maxsize=2000, ttl=86400)
+    async def get_images(
+        self, media_type: str, tmdb_id: int, include_image_language: str = "en,fr,null"
+    ) -> dict[str, Any]:
+        """
+        Fetch images (posters, logos, backdrops) for a movie or TV show.
+        include_image_language: comma-separated iso_639_1 codes + "null" for language-less images.
+        """
+        if media_type not in ("movie", "tv"):
+            return {}
+        path = f"/{media_type}/{tmdb_id}/images"
+        params = {"include_image_language": include_image_language}
+        return await self.client.get(path, params=params)
+
+    @staticmethod
+    def _score_image(img: dict[str, Any]) -> tuple[float, int]:
+        """Higher is better (TMDB vote fields)."""
+        va = img.get("vote_average")
+        vc = img.get("vote_count")
+        try:
+            va_f = float(va) if va is not None else 0.0
+        except (TypeError, ValueError):
+            va_f = 0.0
+        try:
+            vc_i = int(vc) if vc is not None else 0
+        except (TypeError, ValueError):
+            vc_i = 0
+        return (va_f, vc_i)
+
+    @classmethod
+    def _pick_best_in_language_bucket(
+        cls,
+        images_list: list[dict[str, Any]],
+        iso: str | None,
+    ) -> str | None:
+        """Among images with this iso_639_1 (or None for language-neutral), pick highest-rated."""
+        if iso is None:
+            candidates = [img for img in images_list if img.get("iso_639_1") in (None, "")]
+        else:
+            iso_l = iso.lower()
+            candidates = [img for img in images_list if (img.get("iso_639_1") or "").lower() == iso_l]
+        if not candidates:
+            return None
+        best = max(candidates, key=cls._score_image)
+        path = best.get("file_path")
+        return path if path else None
+
+    @classmethod
+    def _pick_logo_by_language(
+        cls,
+        logos: list[dict[str, Any]] | None,
+        primary_iso: str,
+    ) -> str | None:
+        """
+        Logo only: exact ISO 639-1 match, else language-neutral (null). No cross-language fallback.
+
+        TMDB tags logos with iso_639_1 only (no region). Falling back to another language (e.g. another
+        "fr" market's artwork or "en") often mismatches localized titles; omit logo so Metahub/default applies.
+        """
+        if not logos:
+            return None
+        p = (primary_iso or "en").lower()
+        if path := cls._pick_best_in_language_bucket(logos, p):
+            return path
+        return cls._pick_best_in_language_bucket(logos, None)
+
+    @staticmethod
+    def _pick_image_by_language(
+        images_list: list[dict[str, Any]] | None,
+        preferred_lang_codes: list[str | None],
+    ) -> str | None:
+        """
+        Pick best image from list by language preference (same logic as no-stremio-addon).
+        preferred_lang_codes: e.g. ["en", None, "fr"] -> prefer en, then no language, then fr.
+        """
+        if not images_list:
+            return None
+        for lang in preferred_lang_codes:
+            for img in images_list:
+                iso = img.get("iso_639_1")
+                if iso == lang:
+                    path = img.get("file_path")
+                    if path:
+                        return path
+        return images_list[0].get("file_path") if images_list else None
+
+    def _language_to_image_preference(self, language: str) -> tuple[list[str | None], str]:
+        """
+        Build preferred lang order and include_image_language param from language (e.g. en-US, fr-FR).
+        Returns (preferred_lang_codes, include_image_language).
+        """
+        primary = (language or "en-US").split("-")[0].lower() if language else "en"
+        fallbacks = [c for c in ("en", "fr", "null") if c != primary]
+        preferred = [primary, None, *[c for c in fallbacks if c != "null"]]
+        include = ",".join([primary] + fallbacks)
+        return preferred, include
+
+    async def get_images_for_title(
+        self,
+        media_type: str,
+        tmdb_id: int,
+        language: str | None = None,
+    ) -> dict[str, str]:
+        """
+        Get poster, logo and background URLs for a title in the requested language.
+
+        Posters/backdrops: requested language, then null, then common fallbacks (same idea as no-stremio-addon).
+
+        Logos: only the exact ISO 639-1 language or a language-neutral (null) asset — no fallback to other
+        languages, so we do not show a logo whose text targets another locale when the exact translation
+        is missing (e.g. another French market). If neither exists, no logo is returned (callers may use Metahub).
+        """
+        lang = language or self.client.language
+        preferred, include = self._language_to_image_preference(lang)
+        data = await self.get_images(media_type, tmdb_id, include_image_language=include)
+        if not data:
+            return {}
+
+        base_poster_logo = "https://image.tmdb.org/t/p/w500"
+        base_backdrop = "https://image.tmdb.org/t/p/w780"
+
+        def to_url(base: str, path: str | None) -> str:
+            if not path:
+                return ""
+            return base + (path if path.startswith("/") else "/" + path)
+
+        posters = data.get("posters") or []
+        logos = data.get("logos") or []
+        backdrops = data.get("backdrops") or []
+
+        poster_path = self._pick_image_by_language(posters, preferred)
+        primary_iso = (lang or "en-US").split("-")[0].lower() if lang else "en"
+        logo_path = self._pick_logo_by_language(logos, primary_iso)
+        backdrop_path = self._pick_image_by_language(backdrops, preferred)
+
+        result: dict[str, str] = {}
+        if poster_path:
+            result["poster"] = to_url(base_poster_logo, poster_path)
+        if logo_path:
+            result["logo"] = to_url(base_poster_logo, logo_path)
+        if backdrop_path:
+            result["background"] = to_url(base_backdrop, backdrop_path)
+        return result
+
 
 @functools.lru_cache(maxsize=128)
 def get_tmdb_service(language: str = "en-US", api_key: str | None = None) -> TMDBService: