mirror of
https://github.com/maxdorninger/MediaManager.git
synced 2026-04-18 04:53:59 +02:00
270 lines
8.9 KiB
Python
270 lines
8.9 KiB
Python
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
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|