starred/MediaManager-maxdorninger-1
1
0
Fork 0
mirror of https://github.com/maxdorninger/MediaManager.git synced 2026-04-17 15:43:28 +02:00
Code Issues Packages Projects Releases Wiki Activity

ruff: enable A lint

Browse Source
This commit is contained in:
Marcel Hellwig
2026-01-04 15:01:24 +01:00
parent 7824828bea
commit 6e46b482cb
11 changed files with 77 additions and 84 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
alembic/env.py
View File

@@ -106,7 +106,7 @@ def run_migrations_online() -> None:
"""
def include_object(object, name, type_, reflected, compare_to):
def include_object(_object, name, type_, _reflected, _compare_to):
if type_ == "table" and name == "apscheduler_jobs":
return False
return True

4
media_manager/metadataProvider/abstractMetaDataProvider.py
View File

@@ -19,13 +19,13 @@ class AbstractMetadataProvider(ABC):
@abstractmethod
def get_show_metadata(
self, id: int | None = None, language: str | None = None
self, show_id: int | None = None, language: str | None = None
) -> Show:
raise NotImplementedError()
@abstractmethod
def get_movie_metadata(
self, id: int | None = None, language: str | None = None
self, movie_id: int | None = None, language: str | None = None
) -> Movie:
raise NotImplementedError()

58
media_manager/metadataProvider/tmdb.py
View File

@@ -38,40 +38,40 @@ class TmdbMetadataProvider(AbstractMetadataProvider):
return original_language
return self.default_language
def __get_show_metadata(self, id: int, language: str | None = None) -> dict:
def __get_show_metadata(self, show_id: int, language: str | None = None) -> dict:
if language is None:
language = self.default_language
try:
response = requests.get(
url=f"{self.url}/tv/shows/{id}",
url=f"{self.url}/tv/shows/{show_id}",
params={"language": language},
timeout=60,
)
response.raise_for_status()
return response.json()
except requests.RequestException as e:
log.error(f"TMDB API error getting show metadata for ID {id}: {e}")
log.error(f"TMDB API error getting show metadata for ID {show_id}: {e}")
if notification_manager.is_configured():
notification_manager.send_notification(
title="TMDB API Error",
message=f"Failed to fetch show metadata for ID {id} from TMDB. Error: {e}",
message=f"Failed to fetch show metadata for ID {show_id} from TMDB. Error: {e}",
)
raise
def __get_show_external_ids(self, id: int) -> dict:
def __get_show_external_ids(self, show_id: int) -> dict:
try:
response = requests.get(
url=f"{self.url}/tv/shows/{id}/external_ids",
url=f"{self.url}/tv/shows/{show_id}/external_ids",
timeout=60,
)
response.raise_for_status()
return response.json()
except requests.RequestException as e:
log.error(f"TMDB API error getting show external IDs for ID {id}: {e}")
log.error(f"TMDB API error getting show external IDs for ID {show_id}: {e}")
if notification_manager.is_configured():
notification_manager.send_notification(
title="TMDB API Error",
message=f"Failed to fetch show external IDs for ID {id} from TMDB. Error: {e}",
message=f"Failed to fetch show external IDs for ID {show_id} from TMDB. Error: {e}",
)
raise
@@ -138,37 +138,37 @@ class TmdbMetadataProvider(AbstractMetadataProvider):
)
raise
def __get_movie_metadata(self, id: int, language: str | None = None) -> dict:
def __get_movie_metadata(self, movie_id: int, language: str | None = None) -> dict:
if language is None:
language = self.default_language
try:
response = requests.get(
url=f"{self.url}/movies/{id}", params={"language": language}, timeout=60
url=f"{self.url}/movies/{movie_id}", params={"language": language}, timeout=60
)
response.raise_for_status()
return response.json()
except requests.RequestException as e:
log.error(f"TMDB API error getting movie metadata for ID {id}: {e}")
log.error(f"TMDB API error getting movie metadata for ID {movie_id}: {e}")
if notification_manager.is_configured():
notification_manager.send_notification(
title="TMDB API Error",
message=f"Failed to fetch movie metadata for ID {id} from TMDB. Error: {e}",
message=f"Failed to fetch movie metadata for ID {movie_id} from TMDB. Error: {e}",
)
raise
def __get_movie_external_ids(self, id: int) -> dict:
def __get_movie_external_ids(self, movie_id: int) -> dict:
try:
response = requests.get(
url=f"{self.url}/movies/{id}/external_ids", timeout=60
url=f"{self.url}/movies/{movie_id}/external_ids", timeout=60
)
response.raise_for_status()
return response.json()
except requests.RequestException as e:
log.error(f"TMDB API error getting movie external IDs for ID {id}: {e}")
log.error(f"TMDB API error getting movie external IDs for ID {movie_id}: {e}")
if notification_manager.is_configured():
notification_manager.send_notification(
title="TMDB API Error",
message=f"Failed to fetch movie external IDs for ID {id} from TMDB. Error: {e}",
message=f"Failed to fetch movie external IDs for ID {movie_id} from TMDB. Error: {e}",
)
raise
@@ -225,7 +225,7 @@ class TmdbMetadataProvider(AbstractMetadataProvider):
"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
storage_path=self.storage_path, poster_url=poster_url, uuid=show.id
):
log.info("Successfully downloaded poster image for show " + show.name)
else:
@@ -237,7 +237,7 @@ class TmdbMetadataProvider(AbstractMetadataProvider):
return True
def get_show_metadata(
self, id: int | None = None, language: str | None = None
self, show_id: int | None = None, language: str | None = None
) -> Show:
"""
@@ -250,17 +250,17 @@ class TmdbMetadataProvider(AbstractMetadataProvider):
"""
# If language not provided, fetch once to determine original language
if language is None:
show_metadata = self.__get_show_metadata(id)
show_metadata = self.__get_show_metadata(show_id)
language = show_metadata.get("original_language")
# Determine which language to use for metadata
language = self.__get_language_param(language)
# Fetch show metadata in the appropriate language
show_metadata = self.__get_show_metadata(id, language=language)
show_metadata = self.__get_show_metadata(show_id, language=language)
# get imdb id
external_ids = self.__get_show_external_ids(id=id)
external_ids = self.__get_show_external_ids(show_id=show_id)
imdb_id = external_ids.get("imdb_id")
season_list = []
@@ -297,7 +297,7 @@ class TmdbMetadataProvider(AbstractMetadataProvider):
)
show = Show(
external_id=id,
external_id=show_id,
name=show_metadata["name"],
overview=show_metadata["overview"],
year=year,
@@ -370,7 +370,7 @@ class TmdbMetadataProvider(AbstractMetadataProvider):
return formatted_results
def get_movie_metadata(
self, id: int | None = None, language: str | None = None
self, movie_id: int | None = None, language: str | None = None
) -> Movie:
"""
Get movie metadata with language-aware fetching.
@@ -384,17 +384,17 @@ class TmdbMetadataProvider(AbstractMetadataProvider):
"""
# If language not provided, fetch once to determine original language
if language is None:
movie_metadata = self.__get_movie_metadata(id=id)
movie_metadata = self.__get_movie_metadata(movie_id=movie_id)
language = movie_metadata.get("original_language")
# Determine which language to use for metadata
language = self.__get_language_param(language)
# Fetch movie metadata in the appropriate language
movie_metadata = self.__get_movie_metadata(id=id, language=language)
movie_metadata = self.__get_movie_metadata(movie_id=movie_id, language=language)
# get imdb id
external_ids = self.__get_movie_external_ids(id=id)
external_ids = self.__get_movie_external_ids(movie_id=movie_id)
imdb_id = external_ids.get("imdb_id")
year = media_manager.metadataProvider.utils.get_year_from_date(
@@ -402,7 +402,7 @@ class TmdbMetadataProvider(AbstractMetadataProvider):
)
movie = Movie(
external_id=id,
external_id=movie_id,
name=movie_metadata["title"],
overview=movie_metadata["overview"],
year=year,
@@ -478,7 +478,7 @@ class TmdbMetadataProvider(AbstractMetadataProvider):
# Fetch metadata in the appropriate language to get localized poster
movie_metadata = self.__get_movie_metadata(
id=movie.external_id, language=language
movie_id=movie.external_id, language=language
)
# downloading the poster
@@ -488,7 +488,7 @@ class TmdbMetadataProvider(AbstractMetadataProvider):
"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
storage_path=self.storage_path, poster_url=poster_url, uuid=movie.id
):
log.info("Successfully downloaded poster image for movie " + movie.name)
else:

37
media_manager/metadataProvider/tvdb.py
View File

@@ -21,11 +21,11 @@ class TvdbMetadataProvider(AbstractMetadataProvider):
config = MediaManagerConfig().metadata.tvdb
self.url = config.tvdb_relay_url
def __get_show(self, id: int) -> dict:
return requests.get(url=f"{self.url}/tv/shows/{id}", timeout=60).json()
def __get_show(self, show_id: int) -> dict:
return requests.get(url=f"{self.url}/tv/shows/{show_id}", timeout=60).json()
def __get_season(self, id: int) -> dict:
return requests.get(url=f"{self.url}/tv/seasons/{id}", timeout=60).json()
def __get_season(self, show_id: int) -> dict:
return requests.get(url=f"{self.url}/tv/seasons/{show_id}", timeout=60).json()
def __search_tv(self, query: str) -> dict:
return requests.get(
@@ -35,8 +35,8 @@ class TvdbMetadataProvider(AbstractMetadataProvider):
def __get_trending_tv(self) -> dict:
return requests.get(url=f"{self.url}/tv/trending", timeout=60).json()
def __get_movie(self, id: int) -> dict:
return requests.get(url=f"{self.url}/movies/{id}", timeout=60).json()
def __get_movie(self, movie_id: int) -> dict:
return requests.get(url=f"{self.url}/movies/{movie_id}", timeout=60).json()
def __search_movie(self, query: str) -> dict:
return requests.get(
@@ -47,13 +47,13 @@ class TvdbMetadataProvider(AbstractMetadataProvider):
return requests.get(url=f"{self.url}/movies/trending", timeout=60).json()
def download_show_poster_image(self, show: Show) -> bool:
show_metadata = self.__get_show(id=show.external_id)
show_metadata = self.__get_show(show_id=show.external_id)
if show_metadata["image"] is not None:
media_manager.metadataProvider.utils.download_poster_image(
storage_path=self.storage_path,
poster_url=show_metadata["image"],
id=show.id,
uuid=show.id,
)
log.debug("Successfully downloaded poster image for show " + show.name)
return True
@@ -62,18 +62,14 @@ class TvdbMetadataProvider(AbstractMetadataProvider):
return False
def get_show_metadata(
self, id: int | None = None, language: str | None = None
self, show_id: int | None = None, language: str | None = None
) -> Show:
"""
:param id: the external id of the show
:type id: int
:param language: does nothing, TVDB does not support multiple languages
:type language: str | None
:return: returns a ShowMetadata object
:rtype: ShowMetadata
"""
series = self.__get_show(id=id)
series = self.__get_show(show_id)
seasons = []
seasons_ids = [season["id"] for season in series["seasons"]]
@@ -86,7 +82,7 @@ class TvdbMetadataProvider(AbstractMetadataProvider):
imdb_id = remote_id.get("id")
for season in seasons_ids:
s = self.__get_season(id=season)
s = self.__get_season(show_id=season)
# the seasons need to be filtered to a certain type,
# otherwise the same season will be imported in aired and dvd order,
# which causes duplicate season number + show ids which in turn violates a unique constraint of the season table
@@ -264,7 +260,7 @@ class TvdbMetadataProvider(AbstractMetadataProvider):
media_manager.metadataProvider.utils.download_poster_image(
storage_path=self.storage_path,
poster_url=movie_metadata["image"],
id=movie.id,
uuid=movie.id,
)
log.info("Successfully downloaded poster image for show " + movie.name)
return True
@@ -273,18 +269,15 @@ class TvdbMetadataProvider(AbstractMetadataProvider):
return False
def get_movie_metadata(
self, id: int | None = None, language: str | None = None
self, movie_id: int | None = None, language: str | None = None
) -> Movie:
"""
:param id: the external id of the movie
:type id: int
:param movie_id: the external id of the movie
:param language: does nothing, TVDB does not support multiple languages
:type language: str | None
:return: returns a Movie object
:rtype: Movie
"""
movie = self.__get_movie(id)
movie = self.__get_movie(movie_id)
# get imdb id from remote ids
imdb_id = None

4
media_manager/metadataProvider/utils.py
View File

@@ -12,11 +12,11 @@ def get_year_from_date(first_air_date: str | None) -> int | None:
return None
def download_poster_image(storage_path: Path, poster_url: str, id: UUID) -> bool:
def download_poster_image(storage_path: Path, poster_url: str, uuid: UUID) -> bool:
res = requests.get(poster_url, stream=True, timeout=60)
if res.status_code == 200:
image_file_path = storage_path.joinpath(str(id)).with_suffix("jpg")
image_file_path = storage_path.joinpath(str(uuid)).with_suffix("jpg")
image_file_path.write_bytes(res.content)
original_image = Image.open(image_file_path)

4
media_manager/movies/service.py
View File

@@ -79,7 +79,7 @@ class MovieService:
:param language: Optional language code (ISO 639-1) to fetch metadata in.
"""
movie_with_metadata = metadata_provider.get_movie_metadata(
id=external_id, language=language
movie_id=external_id, language=language
)
if not movie_with_metadata:
return None
@@ -731,7 +731,7 @@ class MovieService:
# Use stored original_language preference for metadata fetching
fresh_movie_data = metadata_provider.get_movie_metadata(
id=db_movie.external_id, language=db_movie.original_language
movie_id=db_movie.external_id, language=db_movie.original_language
)
if not fresh_movie_data:
log.warning(

20
media_manager/notification/repository.py
View File

@@ -23,11 +23,11 @@ class NotificationRepository:
def __init__(self, db: Session):
self.db = db
def get_notification(self, id: NotificationId) -> NotificationSchema:
result = self.db.get(Notification, id)
def get_notification(self, nid: NotificationId) -> NotificationSchema:
result = self.db.get(Notification, nid)
if not result:
msg = f"Notification with id {id} not found."
msg = f"Notification with id {nid} not found."
raise NotFoundError(msg)
return NotificationSchema.model_validate(result)
@@ -77,21 +77,21 @@ class NotificationRepository:
raise ConflictError(msg) from None
return
def mark_notification_as_read(self, id: NotificationId) -> None:
stmt = update(Notification).where(Notification.id == id).values(read=True)
def mark_notification_as_read(self, nid: NotificationId) -> None:
stmt = update(Notification).where(Notification.id == nid).values(read=True)
self.db.execute(stmt)
return
def mark_notification_as_unread(self, id: NotificationId) -> None:
stmt = update(Notification).where(Notification.id == id).values(read=False)
def mark_notification_as_unread(self, nid: NotificationId) -> None:
stmt = update(Notification).where(Notification.id == nid).values(read=False)
self.db.execute(stmt)
return
def delete_notification(self, id: NotificationId) -> None:
stmt = delete(Notification).where(Notification.id == id)
def delete_notification(self, nid: NotificationId) -> None:
stmt = delete(Notification).where(Notification.id == nid)
result = self.db.execute(stmt)
if result.rowcount == 0:
msg = f"Notification with id {id} not found."
msg = f"Notification with id {nid} not found."
raise NotFoundError(msg)
self.db.commit()
return

8
media_manager/notification/router.py
View File

@@ -51,7 +51,7 @@ def get_notification(
"""
Get a specific notification by ID.
"""
return notification_service.get_notification(id=notification_id)
return notification_service.get_notification(nid=notification_id)
# --------------------------------
@@ -73,7 +73,7 @@ def mark_notification_as_read(
"""
Mark a notification as read.
"""
notification_service.mark_notification_as_read(id=notification_id)
notification_service.mark_notification_as_read(nid=notification_id)
@router.patch(
@@ -90,7 +90,7 @@ def mark_notification_as_unread(
"""
Mark a notification as unread.
"""
notification_service.mark_notification_as_unread(id=notification_id)
notification_service.mark_notification_as_unread(nid=notification_id)
@router.delete(
@@ -107,4 +107,4 @@ def delete_notification(
"""
Delete a notification.
"""
notification_service.delete_notification(id=notification_id)
notification_service.delete_notification(nid=notification_id)

16
media_manager/notification/service.py
View File

@@ -11,8 +11,8 @@ class NotificationService:
self.notification_repository = notification_repository
self.notification_manager = notification_manager
def get_notification(self, id: NotificationId) -> Notification:
return self.notification_repository.get_notification(id=id)
def get_notification(self, nid: NotificationId) -> Notification:
return self.notification_repository.get_notification(nid=nid)
def get_unread_notifications(self) -> list[Notification]:
return self.notification_repository.get_unread_notifications()
@@ -23,14 +23,14 @@ class NotificationService:
def save_notification(self, notification: Notification) -> None:
return self.notification_repository.save_notification(notification)
def mark_notification_as_read(self, id: NotificationId) -> None:
return self.notification_repository.mark_notification_as_read(id=id)
def mark_notification_as_read(self, nid: NotificationId) -> None:
return self.notification_repository.mark_notification_as_read(nid=nid)
def mark_notification_as_unread(self, id: NotificationId) -> None:
return self.notification_repository.mark_notification_as_unread(id=id)
def mark_notification_as_unread(self, nid: NotificationId) -> None:
return self.notification_repository.mark_notification_as_unread(nid=nid)
def delete_notification(self, id: NotificationId) -> None:
return self.notification_repository.delete_notification(id=id)
def delete_notification(self, nid: NotificationId) -> None:
return self.notification_repository.delete_notification(nid=nid)
def send_notification_to_all_providers(self, title: str, message: str) -> None:
self.notification_manager.send_notification(title, message)

4
media_manager/tv/service.py
View File

@@ -87,7 +87,7 @@ class TvService:
:param language: Optional language code (ISO 639-1) to fetch metadata in.
"""
show_with_metadata = metadata_provider.get_show_metadata(
id=external_id, language=language
show_id=external_id, language=language
)
saved_show = self.tv_repository.save_show(show=show_with_metadata)
metadata_provider.download_show_poster_image(show=saved_show)
@@ -774,7 +774,7 @@ class TvService:
# Use stored original_language preference for metadata fetching
fresh_show_data = metadata_provider.get_show_metadata(
id=db_show.external_id, language=db_show.original_language
show_id=db_show.external_id, language=db_show.original_language
)
if not fresh_show_data:
log.warning(

4
ruff.toml
View File

@@ -3,9 +3,9 @@ line-ending = "lf"
quote-style = "double"
[lint]
# to be enabled: A, ANN, ARG, BLE, C90, CPY, D, DOC, DTZ, FBT, G, INP, INT, N, PERF, PIE, PL, PTH, RET, RSE, SLF, SIM, TC, TRY, UP
# to be enabled: ANN, ARG, BLE, C90, CPY, D, DOC, DTZ, FBT, G, INP, INT, N, PERF, PIE, PL, PTH, RET, RSE, SLF, SIM, TC, TRY, UP
extend-select = [
"ASYNC",
"A", "ASYNC",
"B",
"C4", "COM",
"E", "EM", "EXE",