From 1f50b18b9f2344ea811e95fa2bd6bb80e3d143f5 Mon Sep 17 00:00:00 2001 From: maxid <97409287+maxdorninger@users.noreply.github.com> Date: Sat, 20 Dec 2025 19:37:34 +0100 Subject: [PATCH] refactor media detection functions to improve accuracy --- media_manager/movies/router.py | 14 +++---------- media_manager/movies/service.py | 33 ++++++++++++++++++++++++++++++ media_manager/torrent/utils.py | 36 +++++++++++++++++++++------------ media_manager/tv/router.py | 14 +++---------- media_manager/tv/service.py | 34 +++++++++++++++++++++++++++++++ 5 files changed, 96 insertions(+), 35 deletions(-) diff --git a/media_manager/movies/router.py b/media_manager/movies/router.py index 7d89fe0..4bc690f 100644 --- a/media_manager/movies/router.py +++ b/media_manager/movies/router.py @@ -12,7 +12,7 @@ from media_manager.indexer.schemas import ( ) from media_manager.metadataProvider.schemas import MetaDataProviderSearchResult from media_manager.schemas import MediaImportSuggestion -from media_manager.torrent.utils import detect_unknown_media +from media_manager.torrent.utils import get_importable_media_directories from media_manager.torrent.schemas import Torrent from media_manager.movies import log from media_manager.movies.schemas import ( @@ -104,15 +104,7 @@ def get_all_importable_movies( """ get a list of unknown movies that were detected in the movie directory and are importable """ - directories = detect_unknown_media(AllEncompassingConfig().misc.movie_directory) - movies = [] - for directory in directories: - movies.append( - movie_service.get_import_candidates( - movie=directory, metadata_provider=metadata_provider - ) - ) - return movies + return movie_service.get_importable_movies(metadata_provider=metadata_provider) @router.post( @@ -127,7 +119,7 @@ def import_detected_movie( get a list of unknown movies that were detected in the movie directory and are importable """ source_directory = Path(directory) - if source_directory not in detect_unknown_media( + if source_directory not in get_importable_media_directories( AllEncompassingConfig().misc.movie_directory ): raise HTTPException(status.HTTP_400_BAD_REQUEST, "No such directory") diff --git a/media_manager/movies/service.py b/media_manager/movies/service.py index c473df4..d9d36d3 100644 --- a/media_manager/movies/service.py +++ b/media_manager/movies/service.py @@ -38,6 +38,8 @@ from media_manager.torrent.utils import ( get_files_for_import, remove_special_characters, strip_trailing_year, + get_importable_media_directories, + extract_external_id_from_string, ) from media_manager.indexer.service import IndexerService from media_manager.metadataProvider.abstractMetaDataProvider import ( @@ -716,6 +718,37 @@ class MovieService: metadata_provider.download_movie_poster_image(movie=updated_movie) return updated_movie + def get_importable_movies( + self, metadata_provider: AbstractMetadataProvider + ) -> list[MediaImportSuggestion]: + movie_root_path = AllEncompassingConfig().misc.movie_directory + importable_movies: list[MediaImportSuggestion] = [] + candidate_dirs = get_importable_media_directories(movie_root_path) + + for movie_dir in candidate_dirs: + metadata, external_id = extract_external_id_from_string(movie_dir.name) + if metadata is not None and external_id is not None: + try: + self.movie_repository.get_movie_by_external_id( + external_id=external_id, metadata_provider=metadata + ) + log.debug( + f"Movie {movie_dir.name} already exists in the database, skipping." + ) + continue + except NotFoundError: + log.debug( + f"Movie {movie_dir.name} not found in database, checking for import candidates." + ) + + import_candidates = self.get_import_candidates( + movie=movie_dir, metadata_provider=metadata_provider + ) + importable_movies.append(import_candidates) + + log.debug(f"Found {len(importable_movies)} importable movies.") + return importable_movies + def auto_download_all_approved_movie_requests() -> None: """ diff --git a/media_manager/torrent/utils.py b/media_manager/torrent/utils.py index 25b37f9..bc55f47 100644 --- a/media_manager/torrent/utils.py +++ b/media_manager/torrent/utils.py @@ -197,24 +197,34 @@ def strip_trailing_year(title: str) -> str: return re.sub(r"\s*\(\d{4}\)\s*$", "", title).strip() -def detect_unknown_media(path: Path) -> list[Path]: +def get_importable_media_directories(path: Path) -> list[Path]: libraries = [] libraries.extend(AllEncompassingConfig().misc.movie_libraries) libraries.extend(AllEncompassingConfig().misc.tv_libraries) - show_dirs = path.glob("*") + unfiltered_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 (contains [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(".") - ): + media_dirs = [] + for media_dir in unfiltered_dirs: + if 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 + media_dirs.append(media_dir) + return media_dirs + + +def extract_external_id_from_string(input_string: str) -> tuple[str | None, int | None]: + """ + Extracts an external ID (tmdb/tvdb ID) from the given string. + + :param input_string: The string to extract the ID from. + :return: The extracted Metadata Provider and ID or None if not found. + """ + match = re.search(r"(tmdb|tvdb)(?:id)?[-_]?([0-9]+)", input_string, re.IGNORECASE) + if match: + return match.group(1).lower(), int(match.group(2)) + + return None, None diff --git a/media_manager/tv/router.py b/media_manager/tv/router.py index 69d99b5..31338ec 100644 --- a/media_manager/tv/router.py +++ b/media_manager/tv/router.py @@ -12,7 +12,7 @@ from media_manager.indexer.schemas import ( IndexerQueryResult, ) from media_manager.metadataProvider.schemas import MetaDataProviderSearchResult -from media_manager.torrent.utils import detect_unknown_media +from media_manager.torrent.utils import get_importable_media_directories from media_manager.torrent.schemas import Torrent from media_manager.tv import log from media_manager.exceptions import MediaAlreadyExists @@ -125,15 +125,7 @@ def get_all_importable_shows( """ get a list of unknown shows that were detected in the tv directory and are importable """ - directories = detect_unknown_media(AllEncompassingConfig().misc.tv_directory) - shows = [] - for directory in directories: - shows.append( - tv_service.get_import_candidates( - tv_show=directory, metadata_provider=metadata_provider - ) - ) - return shows + return tv_service.get_importable_tv_shows(metadata_provider=metadata_provider) @router.post( @@ -146,7 +138,7 @@ def import_detected_show(tv_service: tv_service_dep, tv_show: show_dep, director Import a detected show from the specified directory into the library. """ source_directory = Path(directory) - if source_directory not in detect_unknown_media( + if source_directory not in get_importable_media_directories( AllEncompassingConfig().misc.tv_directory ): raise HTTPException(status.HTTP_400_BAD_REQUEST, "No such directory") diff --git a/media_manager/tv/service.py b/media_manager/tv/service.py index 386bbaf..16fa8b2 100644 --- a/media_manager/tv/service.py +++ b/media_manager/tv/service.py @@ -43,6 +43,8 @@ from media_manager.torrent.utils import ( get_files_for_import, remove_special_characters, strip_trailing_year, + get_importable_media_directories, + extract_external_id_from_string, ) from media_manager.indexer.service import IndexerService from media_manager.metadataProvider.abstractMetaDataProvider import ( @@ -912,6 +914,38 @@ class TvService: except Exception as e: log.error(f"Failed to rename {source_directory} to {new_source_path}: {e}") + def get_importable_tv_shows( + self, metadata_provider: AbstractMetadataProvider + ) -> list[MediaImportSuggestion]: + tv_directory = AllEncompassingConfig().misc.tv_directory + import_suggestions: list[MediaImportSuggestion] = [] + candidate_dirs = get_importable_media_directories(tv_directory) + + for item in candidate_dirs: + metadata, external_id = extract_external_id_from_string(item.name) + if metadata is not None and external_id is not None: + try: + self.tv_repository.get_show_by_external_id( + external_id=external_id, + metadata_provider=metadata, + ) + log.debug( + f"Show {item.name} already exists in the database, skipping import suggestion." + ) + continue + except NotFoundError: + log.debug( + f"Show {item.name} not found in database, checking for import candidates." + ) + + import_suggestion = self.get_import_candidates( + tv_show=item, metadata_provider=metadata_provider + ) + import_suggestions.append(import_suggestion) + + log.debug(f"Detected {len(import_suggestions)} importable TV shows.") + return import_suggestions + def auto_download_all_approved_season_requests() -> None: """