extract functions and schemas from tv module

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

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]

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

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]

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(