import logging import tmdbsimple from pydantic_settings import BaseSettings from tmdbsimple import TV, TV_Seasons import media_manager.metadataProvider.utils from media_manager.exceptions import InvalidConfigError from media_manager.metadataProvider.abstractMetaDataProvider import ( AbstractMetadataProvider, ) from media_manager.metadataProvider.schemas import MetaDataProviderSearchResult from media_manager.tv.schemas import Episode, Season, Show, SeasonNumber, EpisodeNumber from media_manager.movies.schemas import Movie class TmdbConfig(BaseSettings): TMDB_API_KEY: str | None = None ENDED_STATUS = {"Ended", "Canceled"} log = logging.getLogger(__name__) class TmdbMetadataProvider(AbstractMetadataProvider): name = "tmdb" def __init__(self, api_key: str = None): config = TmdbConfig() if config.TMDB_API_KEY is None: raise InvalidConfigError("TMDB_API_KEY is not set") tmdbsimple.API_KEY = config.TMDB_API_KEY def download_show_poster_image(self, show: Show) -> bool: show_metadata = TV(show.external_id).info() # downloading the poster # all pictures from TMDB should already be jpeg, so no need to convert if show_metadata["poster_path"] is not None: poster_url = ( "https://image.tmdb.org/t/p/original" + show_metadata["poster_path"] ) if media_manager.metadataProvider.utils.download_poster_image( storage_path=self.storage_path, poster_url=poster_url, id=show.id ): log.info("Successfully downloaded poster image for show " + show.name) else: log.warning(f"download for image of show {show.name} failed") return False else: log.warning(f"image for show {show.name} could not be downloaded") return False return True def get_show_metadata(self, id: int = None) -> Show: """ :param id: the external id of the show :type id: int :return: returns a ShowMetadata object :rtype: ShowMetadata """ show_metadata = TV(id).info() season_list = [] # inserting all the metadata into the objects for season in show_metadata["seasons"]: season_metadata = TV_Seasons( tv_id=show_metadata["id"], season_number=season["season_number"] ).info() episode_list = [] for episode in season_metadata["episodes"]: episode_list.append( Episode( external_id=int(episode["id"]), title=episode["name"], number=EpisodeNumber(episode["episode_number"]), ) ) season_list.append( Season( external_id=int(season_metadata["id"]), name=season_metadata["name"], overview=season_metadata["overview"], number=SeasonNumber(season_metadata["season_number"]), episodes=episode_list, ) ) year = media_manager.metadataProvider.utils.get_year_from_date( show_metadata["first_air_date"] ) show = Show( external_id=id, name=show_metadata["name"], overview=show_metadata["overview"], year=year, seasons=season_list, metadata_provider=self.name, ended=show_metadata["status"] in ENDED_STATUS, ) return show def search_show( self, query: str | None = None, max_pages: int = 5 ) -> list[MetaDataProviderSearchResult]: """ Search for shows using TMDB API. If no query is provided, it will return the most popular shows. """ if query is None: result_factory = lambda page: tmdbsimple.Trending(media_type="tv").info() # noqa: E731 else: result_factory = lambda page: tmdbsimple.Search().tv( # noqa: E731 page=page, query=query, include_adult=True ) results = [] for i in range(1, max_pages + 1): result_page = result_factory(i) if not result_page["results"]: break else: results.extend(result_page["results"]) formatted_results = [] for result in results: try: if result["poster_path"] is not None: poster_url = ( "https://image.tmdb.org/t/p/original" + result["poster_path"] ) else: poster_url = None formatted_results.append( MetaDataProviderSearchResult( poster_path=poster_url, overview=result["overview"], name=result["name"], external_id=result["id"], year=media_manager.metadataProvider.utils.get_year_from_date( result["first_air_date"] ), metadata_provider=self.name, added=False, vote_average=result["vote_average"], ) ) except Exception as e: log.warning(f"Error processing search result {result}: {e}") return formatted_results def get_movie_metadata(self, id: int = None) -> Movie: """ :param id: the external id of the show :type id: int :return: returns a ShowMetadata object :rtype: ShowMetadata """ movie_metadata = tmdbsimple.Movies(id).info() year = media_manager.metadataProvider.utils.get_year_from_date( movie_metadata["release_date"] ) movie = Movie( external_id=id, name=movie_metadata["title"], overview=movie_metadata["overview"], year=year, metadata_provider=self.name, ) return movie def search_movie( self, query: str | None = None, max_pages: int = 5 ) -> list[MetaDataProviderSearchResult]: """ Search for movies using TMDB API. If no query is provided, it will return the most popular movies. """ if query is None: result_factory = lambda page: tmdbsimple.Trending(media_type="movie").info() # noqa: E731 else: result_factory = lambda page: tmdbsimple.Search().movie( # noqa: E731 page=page, query=query, include_adult=True ) results = [] for i in range(1, max_pages + 1): result_page = result_factory(i) if not result_page["results"]: break else: results.extend(result_page["results"]) formatted_results = [] for result in results: try: if result["poster_path"] is not None: poster_url = ( "https://image.tmdb.org/t/p/original" + result["poster_path"] ) else: poster_url = None formatted_results.append( MetaDataProviderSearchResult( poster_path=poster_url, overview=result["overview"], name=result["title"], external_id=result["id"], year=media_manager.metadataProvider.utils.get_year_from_date( result["release_date"] ), metadata_provider=self.name, added=False, vote_average=result["vote_average"], ) ) except Exception as e: log.warning(f"Error processing search result {result}: {e}") return formatted_results def download_movie_poster_image(self, movie: Movie) -> bool: movie_metadata = tmdbsimple.Movies(movie.external_id).info() # downloading the poster # all pictures from TMDB should already be jpeg, so no need to convert if movie_metadata["poster_path"] is not None: poster_url = ( "https://image.tmdb.org/t/p/original" + movie_metadata["poster_path"] ) if media_manager.metadataProvider.utils.download_poster_image( storage_path=self.storage_path, poster_url=poster_url, id=movie.id ): log.info("Successfully downloaded poster image for show " + movie.name) else: log.warning(f"download for image of show {movie.name} failed") return False else: log.warning(f"image for show {movie.name} could not be downloaded") return False return True