diff --git a/app/api/endpoints/meta.py b/app/api/endpoints/meta.py index db2367d..8e35222 100644 --- a/app/api/endpoints/meta.py +++ b/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") diff --git a/app/services/recommendation/catalog_service.py b/app/services/recommendation/catalog_service.py index 089c005..e277765 100644 --- a/app/services/recommendation/catalog_service.py +++ b/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 diff --git a/app/services/recommendation/metadata.py b/app/services/recommendation/metadata.py index 9979186..e96a120 100644 --- a/app/services/recommendation/metadata.py +++ b/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,29 @@ 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 {} + + image_tasks = [_images_one(d) for d in details_list if d] + images_list = await asyncio.gather(*image_tasks, return_exceptions=True) + + format_task = [] + for i, details in enumerate(details_list): + if not details: + continue + logo_url = None + if i < len(images_list): + imgs = images_list[i] + 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) diff --git a/app/services/tmdb/service.py b/app/services/tmdb/service.py index 3157bf6..74cd15e 100644 --- a/app/services/tmdb/service.py +++ b/app/services/tmdb/service.py @@ -138,6 +138,91 @@ 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 _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 or None) == (lang if lang else None): + 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. + Same approach as no-stremio-addon: request images with include_image_language, + then pick by preferred language (requested lang, then null, then fallbacks). + """ + 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) + logo_path = self._pick_image_by_language(logos, preferred) + 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: