mirror of
https://github.com/maxdorninger/MediaManager.git
synced 2026-04-17 15:13:24 +02:00
ruff: enable RUF lint
This commit is contained in:
@@ -1,6 +1,6 @@
|
||||
import sys
|
||||
|
||||
sys.path = ["", ".."] + sys.path[1:]
|
||||
sys.path = ["", "..", *sys.path[1:]]
|
||||
|
||||
|
||||
from logging.config import fileConfig # noqa: E402
|
||||
@@ -46,19 +46,19 @@ target_metadata = Base.metadata
|
||||
# this is to keep pycharm from complaining about/optimizing unused imports
|
||||
# noinspection PyStatementEffect
|
||||
__all__ = [
|
||||
"User",
|
||||
"OAuthAccount",
|
||||
"IndexerQueryResult",
|
||||
"Torrent",
|
||||
"Show",
|
||||
"Season",
|
||||
"Episode",
|
||||
"SeasonFile",
|
||||
"SeasonRequest",
|
||||
"IndexerQueryResult",
|
||||
"Movie",
|
||||
"MovieFile",
|
||||
"MovieRequest",
|
||||
"Notification",
|
||||
"OAuthAccount",
|
||||
"Season",
|
||||
"SeasonFile",
|
||||
"SeasonRequest",
|
||||
"Show",
|
||||
"Torrent",
|
||||
"User",
|
||||
]
|
||||
|
||||
|
||||
|
||||
@@ -51,7 +51,7 @@ class UserManager(UUIDIDMixin, BaseUserManager[User, uuid.UUID]):
|
||||
request: Optional[Request] = None,
|
||||
) -> None:
|
||||
log.info(f"User {user.id} has been updated.")
|
||||
if "is_superuser" in update_dict and update_dict["is_superuser"]:
|
||||
if update_dict.get("is_superuser"):
|
||||
log.info(f"User {user.id} has been granted superuser privileges.")
|
||||
if "email" in update_dict:
|
||||
updated_user = UserUpdate(is_verified=True)
|
||||
|
||||
@@ -8,12 +8,8 @@ from media_manager.tv.schemas import Show
|
||||
class GenericIndexer(ABC):
|
||||
name: str
|
||||
|
||||
def __init__(self, name: str = None):
|
||||
if name:
|
||||
self.name = name
|
||||
else:
|
||||
msg = "indexer name must not be None"
|
||||
raise ValueError(msg)
|
||||
def __init__(self, name: str):
|
||||
self.name = name
|
||||
|
||||
@abstractmethod
|
||||
def search(self, query: str, is_tv: bool) -> list[IndexerQueryResult]:
|
||||
|
||||
@@ -38,7 +38,7 @@ class Prowlarr(GenericIndexer, TorznabMixin):
|
||||
super().__init__(name="prowlarr")
|
||||
self.config = MediaManagerConfig().indexers.prowlarr
|
||||
|
||||
def _call_prowlarr_api(self, path: str, parameters: dict = None):
|
||||
def _call_prowlarr_api(self, path: str, parameters: dict | None = None):
|
||||
url = f"{self.config.url}/api/v1{path}"
|
||||
headers = {"X-Api-Key": self.config.api_key}
|
||||
with Session() as session:
|
||||
@@ -50,7 +50,7 @@ class Prowlarr(GenericIndexer, TorznabMixin):
|
||||
)
|
||||
|
||||
def _newznab_search(
|
||||
self, indexer: IndexerInfo, parameters: dict = None
|
||||
self, indexer: IndexerInfo, parameters: dict | None = None
|
||||
) -> list[IndexerQueryResult]:
|
||||
if parameters is None:
|
||||
parameters = {}
|
||||
|
||||
@@ -18,11 +18,15 @@ class AbstractMetadataProvider(ABC):
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def get_show_metadata(self, id: int = None, language: str | None = None) -> Show:
|
||||
def get_show_metadata(
|
||||
self, id: int | None = None, language: str | None = None
|
||||
) -> Show:
|
||||
raise NotImplementedError()
|
||||
|
||||
@abstractmethod
|
||||
def get_movie_metadata(self, id: int = None, language: str | None = None) -> Movie:
|
||||
def get_movie_metadata(
|
||||
self, id: int | None = None, language: str | None = None
|
||||
) -> Movie:
|
||||
raise NotImplementedError()
|
||||
|
||||
@abstractmethod
|
||||
|
||||
@@ -52,7 +52,7 @@ class TmdbMetadataProvider(AbstractMetadataProvider):
|
||||
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: {str(e)}",
|
||||
message=f"Failed to fetch show metadata for ID {id} from TMDB. Error: {e}",
|
||||
)
|
||||
raise
|
||||
|
||||
@@ -66,7 +66,7 @@ class TmdbMetadataProvider(AbstractMetadataProvider):
|
||||
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: {str(e)}",
|
||||
message=f"Failed to fetch show external IDs for ID {id} from TMDB. Error: {e}",
|
||||
)
|
||||
raise
|
||||
|
||||
@@ -89,7 +89,7 @@ class TmdbMetadataProvider(AbstractMetadataProvider):
|
||||
if notification_manager.is_configured():
|
||||
notification_manager.send_notification(
|
||||
title="TMDB API Error",
|
||||
message=f"Failed to fetch season {season_number} metadata for show ID {show_id} from TMDB. Error: {str(e)}",
|
||||
message=f"Failed to fetch season {season_number} metadata for show ID {show_id} from TMDB. Error: {e}",
|
||||
)
|
||||
raise
|
||||
|
||||
@@ -109,7 +109,7 @@ class TmdbMetadataProvider(AbstractMetadataProvider):
|
||||
if notification_manager.is_configured():
|
||||
notification_manager.send_notification(
|
||||
title="TMDB API Error",
|
||||
message=f"Failed to search TV shows with query '{query}' on TMDB. Error: {str(e)}",
|
||||
message=f"Failed to search TV shows with query '{query}' on TMDB. Error: {e}",
|
||||
)
|
||||
raise
|
||||
|
||||
@@ -126,7 +126,7 @@ class TmdbMetadataProvider(AbstractMetadataProvider):
|
||||
if notification_manager.is_configured():
|
||||
notification_manager.send_notification(
|
||||
title="TMDB API Error",
|
||||
message=f"Failed to fetch trending TV shows from TMDB. Error: {str(e)}",
|
||||
message=f"Failed to fetch trending TV shows from TMDB. Error: {e}",
|
||||
)
|
||||
raise
|
||||
|
||||
@@ -144,7 +144,7 @@ class TmdbMetadataProvider(AbstractMetadataProvider):
|
||||
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: {str(e)}",
|
||||
message=f"Failed to fetch movie metadata for ID {id} from TMDB. Error: {e}",
|
||||
)
|
||||
raise
|
||||
|
||||
@@ -158,7 +158,7 @@ class TmdbMetadataProvider(AbstractMetadataProvider):
|
||||
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: {str(e)}",
|
||||
message=f"Failed to fetch movie external IDs for ID {id} from TMDB. Error: {e}",
|
||||
)
|
||||
raise
|
||||
|
||||
@@ -178,7 +178,7 @@ class TmdbMetadataProvider(AbstractMetadataProvider):
|
||||
if notification_manager.is_configured():
|
||||
notification_manager.send_notification(
|
||||
title="TMDB API Error",
|
||||
message=f"Failed to search movies with query '{query}' on TMDB. Error: {str(e)}",
|
||||
message=f"Failed to search movies with query '{query}' on TMDB. Error: {e}",
|
||||
)
|
||||
raise
|
||||
|
||||
@@ -195,7 +195,7 @@ class TmdbMetadataProvider(AbstractMetadataProvider):
|
||||
if notification_manager.is_configured():
|
||||
notification_manager.send_notification(
|
||||
title="TMDB API Error",
|
||||
message=f"Failed to fetch trending movies from TMDB. Error: {str(e)}",
|
||||
message=f"Failed to fetch trending movies from TMDB. Error: {e}",
|
||||
)
|
||||
raise
|
||||
|
||||
@@ -224,7 +224,9 @@ class TmdbMetadataProvider(AbstractMetadataProvider):
|
||||
return False
|
||||
return True
|
||||
|
||||
def get_show_metadata(self, id: int = None, language: str | None = None) -> Show:
|
||||
def get_show_metadata(
|
||||
self, id: int | None = None, language: str | None = None
|
||||
) -> Show:
|
||||
"""
|
||||
|
||||
:param id: the external id of the show
|
||||
@@ -355,7 +357,9 @@ class TmdbMetadataProvider(AbstractMetadataProvider):
|
||||
log.warning(f"Error processing search result: {e}")
|
||||
return formatted_results
|
||||
|
||||
def get_movie_metadata(self, id: int = None, language: str | None = None) -> Movie:
|
||||
def get_movie_metadata(
|
||||
self, id: int | None = None, language: str | None = None
|
||||
) -> Movie:
|
||||
"""
|
||||
Get movie metadata with language-aware fetching.
|
||||
|
||||
|
||||
@@ -63,7 +63,9 @@ class TvdbMetadataProvider(AbstractMetadataProvider):
|
||||
log.warning(f"image for show {show.name} could not be downloaded")
|
||||
return False
|
||||
|
||||
def get_show_metadata(self, id: int = None, language: str | None = None) -> Show:
|
||||
def get_show_metadata(
|
||||
self, id: int | None = None, language: str | None = None
|
||||
) -> Show:
|
||||
"""
|
||||
|
||||
:param id: the external id of the show
|
||||
@@ -272,7 +274,9 @@ class TvdbMetadataProvider(AbstractMetadataProvider):
|
||||
log.warning(f"image for show {movie.name} could not be downloaded")
|
||||
return False
|
||||
|
||||
def get_movie_metadata(self, id: int = None, language: str | None = None) -> Movie:
|
||||
def get_movie_metadata(
|
||||
self, id: int | None = None, language: str | None = None
|
||||
) -> Movie:
|
||||
"""
|
||||
|
||||
:param id: the external id of the movie
|
||||
|
||||
@@ -251,7 +251,7 @@ class MovieService:
|
||||
raise ValueError(msg)
|
||||
|
||||
def get_all_available_torrents_for_movie(
|
||||
self, movie: Movie, search_query_override: str = None
|
||||
self, movie: Movie, search_query_override: str | None = None
|
||||
) -> list[IndexerQueryResult]:
|
||||
"""
|
||||
Get all available torrents for a given movie.
|
||||
@@ -612,7 +612,7 @@ class MovieService:
|
||||
:param movie: The Movie object
|
||||
"""
|
||||
|
||||
video_files, subtitle_files, all_files = get_files_for_import(torrent=torrent)
|
||||
video_files, subtitle_files, _all_files = get_files_for_import(torrent=torrent)
|
||||
success: list[bool] = []
|
||||
|
||||
if len(video_files) != 1:
|
||||
@@ -695,7 +695,7 @@ class MovieService:
|
||||
msg = "Failed to rename directory"
|
||||
raise Exception(msg) from e
|
||||
|
||||
video_files, subtitle_files, all_files = get_files_for_import(
|
||||
video_files, subtitle_files, _all_files = get_files_for_import(
|
||||
directory=new_source_path
|
||||
)
|
||||
|
||||
@@ -886,7 +886,7 @@ def update_all_movies_metadata() -> None:
|
||||
continue
|
||||
except InvalidConfigError as e:
|
||||
log.error(
|
||||
f"Error initializing metadata provider {movie.metadata_provider} for movie {movie.name}: {str(e)}"
|
||||
f"Error initializing metadata provider {movie.metadata_provider} for movie {movie.name}: {e}"
|
||||
)
|
||||
continue
|
||||
movie_service.update_movie_metadata(
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
import logging
|
||||
from types import MappingProxyType
|
||||
|
||||
import transmission_rpc
|
||||
|
||||
@@ -17,15 +18,17 @@ class TransmissionDownloadClient(AbstractDownloadClient):
|
||||
name = "transmission"
|
||||
|
||||
# Transmission status mappings
|
||||
STATUS_MAPPING = {
|
||||
"stopped": TorrentStatus.unknown,
|
||||
"check pending": TorrentStatus.downloading,
|
||||
"checking": TorrentStatus.downloading,
|
||||
"download pending": TorrentStatus.downloading,
|
||||
"downloading": TorrentStatus.downloading,
|
||||
"seed pending": TorrentStatus.finished,
|
||||
"seeding": TorrentStatus.finished,
|
||||
}
|
||||
STATUS_MAPPING = MappingProxyType(
|
||||
{
|
||||
"stopped": TorrentStatus.unknown,
|
||||
"check pending": TorrentStatus.downloading,
|
||||
"checking": TorrentStatus.downloading,
|
||||
"download pending": TorrentStatus.downloading,
|
||||
"downloading": TorrentStatus.downloading,
|
||||
"seed pending": TorrentStatus.finished,
|
||||
"seeding": TorrentStatus.finished,
|
||||
}
|
||||
)
|
||||
|
||||
def __init__(self):
|
||||
self.config = MediaManagerConfig().torrents.transmission
|
||||
|
||||
@@ -422,7 +422,7 @@ def get_torrents_for_a_season(
|
||||
tv_service: tv_service_dep,
|
||||
show_id: ShowId,
|
||||
season_number: int = 1,
|
||||
search_query_override: str = None,
|
||||
search_query_override: str | None = None,
|
||||
) -> list[IndexerQueryResult]:
|
||||
"""
|
||||
Search for torrents for a specific season of a show.
|
||||
|
||||
@@ -230,18 +230,21 @@ class TvService:
|
||||
return True
|
||||
except NotFoundError:
|
||||
return False
|
||||
elif show_id:
|
||||
elif show_id is not None:
|
||||
try:
|
||||
self.tv_repository.get_show_by_id(show_id=show_id)
|
||||
return True
|
||||
except NotFoundError:
|
||||
return False
|
||||
else:
|
||||
msg = "External ID and metadata provider or Show ID must be provided"
|
||||
msg = "Use one of the provided overloads for this function!"
|
||||
raise ValueError(msg)
|
||||
|
||||
def get_all_available_torrents_for_a_season(
|
||||
self, season_number: int, show_id: ShowId, search_query_override: str = None
|
||||
self,
|
||||
season_number: int,
|
||||
show_id: ShowId,
|
||||
search_query_override: str | None = None,
|
||||
) -> list[IndexerQueryResult]:
|
||||
"""
|
||||
Get all available torrents for a given season.
|
||||
@@ -251,7 +254,6 @@ class TvService:
|
||||
:param search_query_override: Optional override for the search query.
|
||||
:return: A list of indexer query results.
|
||||
"""
|
||||
show = self.tv_repository.get_show_by_id(show_id=show_id)
|
||||
|
||||
if search_query_override:
|
||||
torrents = self.indexer_service.search(
|
||||
@@ -259,6 +261,8 @@ class TvService:
|
||||
)
|
||||
return torrents
|
||||
else:
|
||||
show = self.tv_repository.get_show_by_id(show_id=show_id)
|
||||
|
||||
torrents = self.indexer_service.search_season(
|
||||
show=show, season_number=season_number
|
||||
)
|
||||
@@ -701,7 +705,7 @@ class TvService:
|
||||
:param show: The Show object
|
||||
"""
|
||||
|
||||
video_files, subtitle_files, all_files = get_files_for_import(torrent=torrent)
|
||||
video_files, subtitle_files, _all_files = get_files_for_import(torrent=torrent)
|
||||
|
||||
success: list[bool] = []
|
||||
|
||||
@@ -716,7 +720,7 @@ class TvService:
|
||||
|
||||
for season_file in season_files:
|
||||
season = self.get_season(season_id=season_file.season_id)
|
||||
season_import_success, imported_episodes_count = self.import_season(
|
||||
season_import_success, _imported_episodes_count = self.import_season(
|
||||
show=show,
|
||||
season=season,
|
||||
video_files=video_files,
|
||||
@@ -911,7 +915,7 @@ class TvService:
|
||||
msg = "Failed to rename source directory"
|
||||
raise Exception(msg) from e
|
||||
|
||||
video_files, subtitle_files, all_files = get_files_for_import(
|
||||
video_files, subtitle_files, _all_files = get_files_for_import(
|
||||
directory=new_source_path
|
||||
)
|
||||
for season in tv_show.seasons:
|
||||
@@ -1065,7 +1069,7 @@ def update_all_non_ended_shows_metadata() -> None:
|
||||
continue
|
||||
except InvalidConfigError as e:
|
||||
log.error(
|
||||
f"Error initializing metadata provider {show.metadata_provider} for show {show.name}: {str(e)}"
|
||||
f"Error initializing metadata provider {show.metadata_provider} for show {show.name}: {e}"
|
||||
)
|
||||
continue
|
||||
updated_show = tv_service.update_show_metadata(
|
||||
|
||||
Reference in New Issue
Block a user