diff --git a/media_manager/notification/utils.py b/media_manager/notification/utils.py index 6f81a2b..683d02c 100644 --- a/media_manager/notification/utils.py +++ b/media_manager/notification/utils.py @@ -1,7 +1,9 @@ import logging +import re import smtplib from email.mime.multipart import MIMEMultipart from email.mime.text import MIMEText +from pathlib import Path from media_manager.config import AllEncompassingConfig @@ -23,3 +25,26 @@ def send_email(subject: str, html: str, addressee: str) -> None: server.sendmail(email_conf.from_email, addressee, message.as_string()) log.info(f"Successfully sent email to {addressee} with subject: {subject}") + + +def detect_unknown_media(self, path: Path) -> list[Path]: + libraries = [] + libraries.extend(AllEncompassingConfig().misc.movie_libraries) + libraries.extend(AllEncompassingConfig().misc.tv_libraries) + + show_dirs = path.glob("*") + log.debug(f"Using Directory {path}") + unknown_tv_shows = [] + for media_dir in show_dirs: + # check if directory is one created by MediaManager (contins [tmdbd/tvdbid-0000) or if it is a library + if ( + re.search(r"\[(?:tmdbid|tvdbid)-\d+]", media_dir.name, re.IGNORECASE) + or media_dir.absolute() + in [Path(library.path).absolute() for library in libraries] + or media_dir.name.startswith(".") + ): + log.debug(f"MediaManager directory detected: {media_dir.name}") + else: + log.info(f"Detected unknown media directory: {media_dir.name}") + unknown_tv_shows.append(media_dir) + return unknown_tv_shows diff --git a/media_manager/schemas.py b/media_manager/schemas.py new file mode 100644 index 0000000..658cbf3 --- /dev/null +++ b/media_manager/schemas.py @@ -0,0 +1,10 @@ +from pathlib import Path + +from pydantic import BaseModel + +from media_manager.metadataProvider.schemas import MetaDataProviderSearchResult + + +class MediaImportSuggestion(BaseModel): + directory: Path + candidates: list[MetaDataProviderSearchResult] diff --git a/media_manager/tv/router.py b/media_manager/tv/router.py index ab8ab33..db92b75 100644 --- a/media_manager/tv/router.py +++ b/media_manager/tv/router.py @@ -12,6 +12,7 @@ from media_manager.indexer.schemas import ( IndexerQueryResult, ) from media_manager.metadataProvider.schemas import MetaDataProviderSearchResult +from media_manager.notification.utils import detect_unknown_media from media_manager.torrent.schemas import Torrent from media_manager.tv import log from media_manager.exceptions import MediaAlreadyExists @@ -27,8 +28,9 @@ from media_manager.tv.schemas import ( UpdateSeasonRequest, RichSeasonRequest, Season, - TvShowImportSuggestion, ) +from media_manager.schemas import MediaImportSuggestion + from media_manager.tv.dependencies import ( season_dep, show_dep, @@ -107,7 +109,7 @@ def get_all_shows(tv_service: tv_service_dep): "/importable", status_code=status.HTTP_200_OK, dependencies=[Depends(current_superuser)], - response_model=list[TvShowImportSuggestion], + response_model=list[MediaImportSuggestion], ) def get_all_importable_shows( tv_service: tv_service_dep, metadata_provider: metadata_provider_dep @@ -115,7 +117,7 @@ def get_all_importable_shows( """ get a list of unknown shows that were detected in the tv directory and are importable """ - directories = tv_service.detect_unknown_tv_shows() + directories = detect_unknown_media(AllEncompassingConfig().misc.tv_directory) shows = [] for directory in directories: shows.append( @@ -136,7 +138,9 @@ def import_detected_show(tv_service: tv_service_dep, tv_show: show_dep, director get a list of unknown shows that were detected in the tv directory and are importable """ source_directory = Path(directory) - if source_directory not in tv_service.detect_unknown_tv_shows(): + if source_directory not in detect_unknown_media( + AllEncompassingConfig().misc.tv_directory + ): raise HTTPException(status.HTTP_400_BAD_REQUEST, "No such directory") tv_service.import_existing_tv_show( tv_show=tv_show, source_directory=source_directory diff --git a/media_manager/tv/schemas.py b/media_manager/tv/schemas.py index 83fafc8..cacb963 100644 --- a/media_manager/tv/schemas.py +++ b/media_manager/tv/schemas.py @@ -1,12 +1,10 @@ import typing import uuid -from pathlib import Path from uuid import UUID from pydantic import BaseModel, Field, ConfigDict, model_validator from media_manager.auth.schemas import UserRead -from media_manager.metadataProvider.schemas import MetaDataProviderSearchResult from media_manager.torrent.models import Quality from media_manager.torrent.schemas import TorrentId, TorrentStatus @@ -168,8 +166,3 @@ class PublicShow(BaseModel): library: str seasons: list[PublicSeason] - - -class TvShowImportSuggestion(BaseModel): - directory: Path - candidates: list[MetaDataProviderSearchResult] diff --git a/media_manager/tv/service.py b/media_manager/tv/service.py index 0c1a31e..7083831 100644 --- a/media_manager/tv/service.py +++ b/media_manager/tv/service.py @@ -30,7 +30,6 @@ from media_manager.tv.schemas import ( RichSeasonRequest, EpisodeId, Episode as EpisodeSchema, - TvShowImportSuggestion, ) from media_manager.torrent.schemas import QualityStrings from media_manager.tv.repository import TvRepository @@ -50,6 +49,7 @@ from media_manager.metadataProvider.abstractMetaDataProvider import ( ) from media_manager.metadataProvider.tmdb import TmdbMetadataProvider from media_manager.metadataProvider.tvdb import TvdbMetadataProvider +from media_manager.schemas import MediaImportSuggestion class TvService: @@ -831,33 +831,13 @@ class TvService: show_id=show_id, continuous_download=continuous_download ) - def detect_unknown_tv_shows(self) -> list[Path]: - tv_libraries = AllEncompassingConfig().misc.tv_libraries - tv_directory = AllEncompassingConfig().misc.tv_directory - show_dirs = tv_directory.glob("*") - log.debug(f"Using Directory {tv_directory}") - unknown_tv_shows = [] - for show_dir in show_dirs: - # check if directory is one created by MediaManager (contins [tmdbd/tvdbid-0000) or if it is a library - if ( - re.search(r"\[(?:tmdbid|tvdbid)-\d+]", show_dir.name, re.IGNORECASE) - or show_dir.absolute() - in [Path(library.path).absolute() for library in tv_libraries] - or show_dir.name.startswith(".") - ): - log.debug(f"MediaManager directory detected: {show_dir.name}") - else: - log.info(f"Detected unknown tv show directory: {show_dir.name}") - unknown_tv_shows.append(show_dir) - return unknown_tv_shows - def get_import_candidates( self, tv_show: Path, metadata_provider: AbstractMetadataProvider - ) -> TvShowImportSuggestion: + ) -> MediaImportSuggestion: search_result = self.search_for_show( strip_trailing_year(tv_show.name), metadata_provider ) - import_candidates = TvShowImportSuggestion( + import_candidates = MediaImportSuggestion( directory=tv_show, candidates=search_result ) log.debug(