diff --git a/.gitattributes b/.gitattributes new file mode 100644 index 00000000..64fd7a0b --- /dev/null +++ b/.gitattributes @@ -0,0 +1,25 @@ +*.py text eol=lf +*.md text eol=lf +*.toml text eol=lf +*.yml text eol=lf +*.yaml text eol=lf +*.sh text eol=lf +*.txt text eol=lf +*.ini text eol=lf +*.mako text eol=lf +*.iml text eol=lf +*.js text eol=lf +*.ts text eol=lf +*.json text eol=lf +*.css text eol=lf +*.svelte text eol=lf + +Makefile text eol=lf +Dockerfile text eol=lf + +# Binary files +*.png binary +*.jpg binary +*.jpeg binary +*.gif binary +*.ico binary \ No newline at end of file diff --git a/media_manager/indexer/schemas.py b/media_manager/indexer/schemas.py index 0236da3b..8317e250 100644 --- a/media_manager/indexer/schemas.py +++ b/media_manager/indexer/schemas.py @@ -35,7 +35,7 @@ class IndexerQueryResult(BaseModel): @computed_field @property def quality(self) -> Quality: - high_quality_pattern = r"\b(4k|2160p|uhd)\b" + high_quality_pattern = r"\b(4k|2160p|uhd|ultra[ ._-]?hd)\b" medium_quality_pattern = r"\b(1080p|full[ ._-]?hd)\b" low_quality_pattern = r"\b(720p|(? bool: directory=new_source_path ) + file_quality = extract_quality_video_file(video_files[0]) + file_suffix = f"{QualityStrings.get_label(file_quality)} - IMPORTED" + success = self.import_movie( movie=movie, video_files=video_files, subtitle_files=subtitle_files, - file_path_suffix="IMPORTED", + file_path_suffix=file_suffix, ) if success: self.movie_repository.add_movie_file( MovieFile( movie_id=movie.id, - file_path_suffix="IMPORTED", + file_path_suffix=file_suffix, torrent_id=None, - quality=Quality.unknown, + quality=file_quality, ) ) diff --git a/media_manager/torrent/schemas.py b/media_manager/torrent/schemas.py index 1ae4650f..8c6b5263 100644 --- a/media_manager/torrent/schemas.py +++ b/media_manager/torrent/schemas.py @@ -22,6 +22,10 @@ class QualityStrings(Enum): sd = "400p" unknown = "unknown" + @staticmethod + def get_label(quality: Quality) -> str: + return QualityStrings[quality.name].value + class TorrentStatus(Enum): finished = 1 diff --git a/media_manager/torrent/service.py b/media_manager/torrent/service.py index 36f9980d..278fc054 100644 --- a/media_manager/torrent/service.py +++ b/media_manager/torrent/service.py @@ -56,7 +56,7 @@ def get_torrent_status(self, torrent: Torrent) -> Torrent: self.torrent_repository.save_torrent(torrent=torrent) return torrent - def cancel_download(self, torrent: Torrent, delete_files: bool = False) -> Torrent: + def cancel_download(self, torrent: Torrent, delete_files: bool = False) -> None: """ cancels download of a torrent @@ -65,7 +65,6 @@ def cancel_download(self, torrent: Torrent, delete_files: bool = False) -> Torre """ log.info(f"Cancelling download for torrent: {torrent.title}") self.download_manager.remove_torrent(torrent, delete_data=delete_files) - return self.get_torrent_status(torrent=torrent) def pause_download(self, torrent: Torrent) -> Torrent: """ diff --git a/media_manager/torrent/utils.py b/media_manager/torrent/utils.py index c3e7b6cb..bd54a2d6 100644 --- a/media_manager/torrent/utils.py +++ b/media_manager/torrent/utils.py @@ -1,13 +1,8 @@ import hashlib import logging -import mimetypes -import re -import shutil -from pathlib import Path, UnsupportedOperation import bencoder import libtorrent -import patoolib import requests from pathvalidate import sanitize_filename from requests.exceptions import InvalidSchema @@ -15,116 +10,10 @@ from media_manager.config import MediaManagerConfig from media_manager.indexer.schemas import IndexerQueryResult from media_manager.indexer.utils import follow_redirects_to_final_torrent_url -from media_manager.torrent.schemas import Torrent log = logging.getLogger(__name__) -def list_files_recursively(path: Path = Path()) -> list[Path]: - files = list(path.glob("**/*")) - log.debug(f"Found {len(files)} entries via glob") - valid_files = [] - for x in files: - if x.is_dir(): - log.debug(f"'{x}' is a directory") - elif x.is_symlink(): - log.debug(f"'{x}' is a symlink") - else: - valid_files.append(x) - log.debug(f"Returning {len(valid_files)} files after filtering") - return valid_files - - -def extract_archives(files: list) -> None: - archive_types = { - "application/zip", - "application/x-zip-compressedapplication/x-compressed", - "application/vnd.rar", - "application/x-7z-compressed", - "application/x-freearc", - "application/x-bzip", - "application/x-bzip2", - "application/gzip", - "application/x-gzip", - "application/x-tar", - } - for file in files: - file_type = mimetypes.guess_type(file) - log.debug(f"File: {file}, Size: {file.stat().st_size} bytes, Type: {file_type}") - - if file_type[0] in archive_types: - log.info( - f"File {file} is a compressed file, extracting it into directory {file.parent}" - ) - try: - patoolib.extract_archive(str(file), outdir=str(file.parent)) - except patoolib.util.PatoolError: - log.exception(f"Failed to extract archive {file}") - - -def get_torrent_filepath(torrent: Torrent) -> Path: - return MediaManagerConfig().misc.torrent_directory / torrent.title - - -def import_file(target_file: Path, source_file: Path) -> None: - if target_file.exists(): - target_file.unlink() - - try: - target_file.hardlink_to(source_file) - except FileExistsError: - log.exception(f"File already exists at {target_file}.") - except (OSError, UnsupportedOperation, NotImplementedError): - log.exception( - f"Failed to create hardlink from {source_file} to {target_file}. Falling back to copying the file." - ) - shutil.copy(src=source_file, dst=target_file) - - -def get_files_for_import( - torrent: Torrent | None = None, directory: Path | None = None -) -> tuple[list[Path], list[Path], list[Path]]: - """ - Extracts all files from the torrent download directory, including extracting archives. - Returns a tuple containing: seperated video files, subtitle files, and all files found in the torrent directory. - """ - if torrent: - log.info(f"Importing torrent {torrent}") - search_directory = get_torrent_filepath(torrent=torrent) - elif directory: - log.info(f"Importing files from directory {directory}") - search_directory = directory - else: - msg = "Either torrent or directory must be provided." - raise ValueError(msg) - - all_files: list[Path] = list_files_recursively(path=search_directory) - log.debug(f"Found {len(all_files)} files downloaded by the torrent") - extract_archives(all_files) - all_files = list_files_recursively(path=search_directory) - - video_files: list[Path] = [] - subtitle_files: list[Path] = [] - for file in all_files: - file_type, _ = mimetypes.guess_type(str(file)) - if file_type is not None: - if file_type.startswith("video"): - video_files.append(file) - log.debug(f"File is a video, it will be imported: {file}") - elif file_type.startswith("text") and Path(file).suffix == ".srt": - subtitle_files.append(file) - log.debug(f"File is a subtitle, it will be imported: {file}") - else: - log.debug( - f"File is neither a video nor a subtitle, will not be imported: {file}" - ) - - log.info( - f"Found {len(all_files)} files ({len(video_files)} video files, {len(subtitle_files)} subtitle files) for further processing." - ) - return video_files, subtitle_files, all_files - - def get_torrent_hash(torrent: IndexerQueryResult) -> str: """ Helper method to get the torrent hash from the torrent object. @@ -179,73 +68,3 @@ def get_torrent_hash(torrent: IndexerQueryResult) -> str: return torrent_hash -def remove_special_characters(filename: str) -> str: - """ - Removes special characters from the filename to ensure it works with Jellyfin. - - :param filename: The original filename. - :return: A sanitized version of the filename. - """ - # Remove invalid characters - sanitized = re.sub(r"([<>:\"/\\|?*])", "", filename) - - # Remove leading and trailing dots or spaces - return sanitized.strip(" .") - - -def remove_special_chars_and_parentheses(title: str) -> str: - """ - Removes special characters and bracketed information from the title. - - :param title: The original title. - :return: A sanitized version of the title. - """ - - # Remove content within brackets - sanitized = re.sub(r"\[.*?\]", "", title) - - # Remove content within curly brackets - sanitized = re.sub(r"\{.*?\}", "", sanitized) - - # Remove year within parentheses - sanitized = re.sub(r"\(\d{4}\)", "", sanitized) - - # Remove special characters - sanitized = remove_special_characters(sanitized) - - # Collapse multiple whitespace characters and trim the result - return re.sub(r"\s+", " ", sanitized).strip() - - -def get_importable_media_directories(path: Path) -> list[Path]: - libraries = [ - *MediaManagerConfig().misc.movie_libraries, - *MediaManagerConfig().misc.tv_libraries, - ] - - library_paths = {Path(library.path).absolute() for library in libraries} - - unfiltered_dirs = [d for d in path.glob("*") if d.is_dir()] - - return [ - media_dir - for media_dir in unfiltered_dirs - if media_dir.absolute() not in library_paths - and not media_dir.name.startswith(".") - ] - - -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"\b(tmdb|tvdb)(?:id)?[-_]?([0-9]+)\b", 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 eaf2a7b3..0e3ebbbf 100644 --- a/media_manager/tv/router.py +++ b/media_manager/tv/router.py @@ -13,7 +13,6 @@ from media_manager.metadataProvider.schemas import MetaDataProviderSearchResult from media_manager.schemas import MediaImportSuggestion from media_manager.torrent.schemas import Torrent -from media_manager.torrent.utils import get_importable_media_directories from media_manager.tv.dependencies import ( season_dep, show_dep, @@ -27,6 +26,7 @@ Show, ShowId, ) +from media_manager.utils.file_handler import get_importable_media_directories router = APIRouter() diff --git a/media_manager/tv/schemas.py b/media_manager/tv/schemas.py index c9096684..dcb00322 100644 --- a/media_manager/tv/schemas.py +++ b/media_manager/tv/schemas.py @@ -23,6 +23,7 @@ class Episode(BaseModel): external_id: int title: str overview: str | None = None + quality: Quality = Quality.unknown class Season(BaseModel): diff --git a/media_manager/tv/service.py b/media_manager/tv/service.py index 25a21008..c578acb4 100644 --- a/media_manager/tv/service.py +++ b/media_manager/tv/service.py @@ -20,19 +20,11 @@ from media_manager.notification.service import NotificationService from media_manager.schemas import MediaImportSuggestion from media_manager.torrent.schemas import ( - Quality, + QualityStrings, Torrent, TorrentStatus, ) from media_manager.torrent.service import TorrentService -from media_manager.torrent.utils import ( - extract_external_id_from_string, - get_files_for_import, - get_importable_media_directories, - import_file, - remove_special_characters, - remove_special_chars_and_parentheses, -) from media_manager.tv import log from media_manager.tv.repository import TvRepository from media_manager.tv.schemas import ( @@ -50,6 +42,16 @@ Show, ShowId, ) +from media_manager.utils.file_handler import ( + extract_external_id_from_string, + extract_quality_video_file, + get_files_for_import, + get_importable_media_directories, + import_file, + import_subtitle, + remove_special_characters, + remove_special_chars_and_parentheses, +) class TvService: @@ -615,7 +617,6 @@ def import_episode( pattern = ( r".*[. ]S0?" + str(season.number) + r"E0?" + str(episode_number) + r"[. ].*" ) - subtitle_pattern = pattern + r"[. ]([A-Za-z]{2})[. ]srt" target_file_name = ( self.get_root_season_directory(show=show, season_number=season.number) / episode_file_name @@ -623,29 +624,21 @@ def import_episode( # import subtitle for subtitle_file in subtitle_files: - regex_result = re.search( - subtitle_pattern, subtitle_file.name, re.IGNORECASE - ) - if regex_result: - language_code = regex_result.group(1) - target_subtitle_file = target_file_name.with_suffix( - f".{language_code}.srt" - ) - import_file(target_file=target_subtitle_file, source_file=subtitle_file) - else: - log.debug( - f"Didn't find any pattern {subtitle_pattern} in subtitle file: {subtitle_file.name}" - ) + import_subtitle(subtitle_file=subtitle_file, target_file=target_file_name) # import episode videos for file in video_files: - if re.search(pattern, file.name, re.IGNORECASE): - target_video_file = target_file_name.with_suffix(file.suffix) - import_file(target_file=target_video_file, source_file=file) - return True - else: - msg = f"Could not find any video file for episode {episode_number} of show {show.name} S{season.number}" - raise Exception(msg) # noqa: TRY002 # TODO: resolve this + try: + if re.search(pattern, file.name, re.IGNORECASE): + target_video_file = target_file_name.with_suffix(file.suffix) + import_file(target_file=target_video_file, source_file=file) + return True + except FileNotFoundError: + log.error( + f"Could not find any video file for episode {episode_number} of show {show.name} S{season.number}" + ) + return False + return False def import_season( self, @@ -669,28 +662,44 @@ def import_season( for episode in season.episodes: try: + episode.quality = extract_quality_video_file(video_files[0]) + new_path_suffix = ( + f"{QualityStrings.get_label(episode.quality)} - {file_path_suffix}" + ) imported = self.import_episode( show=show, subtitle_files=subtitle_files, video_files=video_files, season=season, episode_number=episode.number, - file_path_suffix=file_path_suffix, + file_path_suffix=new_path_suffix, ) if imported: imported_episodes.append(episode) - except Exception: + except FileNotFoundError as e: + log.exception( + f"IO/FST problem while importing S{season.number:02d}E{episode.number:02d}: {e}" + ) # Send notification about missing episode file if self.notification_service: self.notification_service.send_notification_to_all_providers( title="Missing Episode File", message=f"No video file found for S{season.number:02d}E{episode.number:02d} for show {show.name}. Manual intervention may be required.", ) - success = False - log.warning( - f"S{season.number}E{episode.number} not found when trying to import episode for show {show.name}." + raise + except Exception as e: + # Send notification about unknown error + log.exception( + f"Unexpected error importing S{season.number:02d}E{episode.number:02d}: {e}" ) + if self.notification_service: + self.notification_service.send_notification_to_all_providers( + title="Unexpected Error", + message=f"An unexpected error occurred while importing S{season.number:02d}E{episode.number:02d} for {show.name}. Check server logs.", + ) + success = False + return success, imported_episodes def import_episode_files( @@ -708,7 +717,6 @@ def import_episode_files( pattern = ( r".*[. ]S0?" + str(season.number) + r"E0?" + str(episode.number) + r"[. ].*" ) - subtitle_pattern = pattern + r"[. ]([A-Za-z]{2})[. ]srt" target_file_name = ( self.get_root_season_directory(show=show, season_number=season.number) / episode_file_name @@ -716,19 +724,7 @@ def import_episode_files( # import subtitle for subtitle_file in subtitle_files: - regex_result = re.search( - subtitle_pattern, subtitle_file.name, re.IGNORECASE - ) - if regex_result: - language_code = regex_result.group(1) - target_subtitle_file = target_file_name.with_suffix( - f".{language_code}.srt" - ) - import_file(target_file=target_subtitle_file, source_file=subtitle_file) - else: - log.debug( - f"Didn't find any pattern {subtitle_pattern} in subtitle file: {subtitle_file.name}" - ) + import_subtitle(subtitle_file=subtitle_file, target_file=target_file_name) found_video = False @@ -1025,7 +1021,7 @@ def import_existing_tv_show(self, tv_show: Show, source_directory: Path) -> None for episode in imported_episodes: episode_file = EpisodeFile( episode_id=episode.id, - quality=Quality.unknown, + quality=episode.quality, file_path_suffix="IMPORTED", torrent_id=None, ) diff --git a/media_manager/utils/__init__.py b/media_manager/utils/__init__.py new file mode 100644 index 00000000..3988bf12 --- /dev/null +++ b/media_manager/utils/__init__.py @@ -0,0 +1,3 @@ +import logging + +log = logging.getLogger(__name__) diff --git a/media_manager/utils/file_handler.py b/media_manager/utils/file_handler.py new file mode 100644 index 00000000..9dc0986c --- /dev/null +++ b/media_manager/utils/file_handler.py @@ -0,0 +1,234 @@ +import mimetypes +import re +import shutil +from pathlib import Path, UnsupportedOperation + +import patoolib + +from media_manager.config import MediaManagerConfig +from media_manager.torrent.schemas import Quality, Torrent +from media_manager.utils import log + + +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"\b(tmdb|tvdb)(?:id)?[-_]?([0-9]+)\b", input_string, re.IGNORECASE + ) + if match: + return match.group(1).lower(), int(match.group(2)) + + return None, None + + +def extract_quality_video_file(file: Path) -> Quality: + """ + Extracts the quality of a video file based on its name. + + :param file: The path to the video file. + :return: The extracted quality or None if not found. + """ + quality_pattern = ( + r"\b(4k|UHD|ultra[ ._-]?hd|2160p|FHD|full[ ._-]?hd|1080p|HD|720p|SD|480p)\b" + ) + match = re.search(quality_pattern, file.stem, re.IGNORECASE) + if match: + quality = match.group(1).lower() + if quality in {"4k", "uhd", "2160p"}: + return Quality.uhd + if quality in {"fhd", "1080p"}: + return Quality.fullhd + if quality in {"hd", "720p"}: + return Quality.hd + if quality in {"sd", "480p"}: + return Quality.sd + return Quality.unknown + + +def list_files_recursively(path: Path = Path()) -> list[Path]: + files = list(path.glob("**/*")) + log.debug(f"Found {len(files)} entries via glob") + valid_files = [] + for x in files: + if x.is_dir(): + log.debug(f"'{x}' is a directory") + elif x.is_symlink(): + log.debug(f"'{x}' is a symlink") + else: + valid_files.append(x) + log.debug(f"Returning {len(valid_files)} files after filtering") + return valid_files + + +def extract_archives(files: list) -> None: + archive_types = { + "application/zip", + "application/x-zip-compressedapplication/x-compressed", + "application/vnd.rar", + "application/x-7z-compressed", + "application/x-freearc", + "application/x-bzip", + "application/x-bzip2", + "application/gzip", + "application/x-gzip", + "application/x-tar", + } + for file in files: + file_type = mimetypes.guess_type(file) + log.debug(f"File: {file}, Size: {file.stat().st_size} bytes, Type: {file_type}") + + if file_type[0] in archive_types: + log.info( + f"File {file} is a compressed file, extracting it into directory {file.parent}" + ) + try: + patoolib.extract_archive(str(file), outdir=str(file.parent)) + except patoolib.util.PatoolError: + log.exception(f"Failed to extract archive {file}") + + +def get_importable_media_directories(path: Path) -> list[Path]: + libraries = [ + *MediaManagerConfig().misc.movie_libraries, + *MediaManagerConfig().misc.tv_libraries, + ] + + library_paths = {Path(library.path).absolute() for library in libraries} + + unfiltered_dirs = [d for d in path.glob("*") if d.is_dir()] + + return [ + media_dir + for media_dir in unfiltered_dirs + if media_dir.absolute() not in library_paths + and not media_dir.name.startswith(".") + ] + + +def get_torrent_filepath(torrent: Torrent) -> Path: + return MediaManagerConfig().misc.torrent_directory / torrent.title + + +def get_files_for_import( + torrent: Torrent | None = None, directory: Path | None = None +) -> tuple[list[Path], list[Path], list[Path]]: + """ + Extracts all files from the torrent download directory, including extracting archives. + Returns a tuple containing: seperated video files, subtitle files, and all files found in the torrent directory. + """ + if torrent: + log.info(f"Importing torrent {torrent}") + search_directory = get_torrent_filepath(torrent=torrent) + elif directory: + log.info(f"Importing files from directory {directory}") + search_directory = directory + else: + msg = "Either torrent or directory must be provided." + raise ValueError(msg) + + all_files: list[Path] = list_files_recursively(path=search_directory) + log.debug(f"Found {len(all_files)} files downloaded by the torrent") + extract_archives(all_files) + all_files = list_files_recursively(path=search_directory) + + video_files: list[Path] = [] + subtitle_files: list[Path] = [] + for file in all_files: + file_type, _ = mimetypes.guess_type(str(file)) + if file_type is not None: + if file_type.startswith("video"): + video_files.append(file) + log.debug(f"File is a video, it will be imported: {file}") + elif file_type.startswith("text") and Path(file).suffix == ".srt": + subtitle_files.append(file) + log.debug(f"File is a subtitle, it will be imported: {file}") + else: + log.debug( + f"File is neither a video nor a subtitle, will not be imported: {file}" + ) + else: + log.debug(f"Unknown file type, will not be imported: {file}") + + log.info( + f"Found {len(all_files)} files ({len(video_files)} video files, {len(subtitle_files)} subtitle files) for further processing." + ) + return video_files, subtitle_files, all_files + + +def import_subtitle(subtitle_file: Path, target_file: Path) -> None: + """ + Imports a subtitle file by renaming it to match the target video file and placing it in the same directory. + + :param subtitle_file: The path to the subtitle file to be imported. + :return None + """ + subtitle_pattern = r"[. ]?([a-z]{2,3})(?:\.(\d+))?\.srt$" + regex_result = re.search(subtitle_pattern, subtitle_file.name, re.IGNORECASE) + if regex_result: + language_code = regex_result.group(1) + if regex_result.group(2): + language_code += f".{regex_result.group(2)}" + target_subtitle_file = target_file.with_suffix(f".{language_code}.srt") + import_file(target_file=target_subtitle_file, source_file=subtitle_file) + else: + log.debug( + f"Didn't find any pattern {subtitle_pattern} in subtitle file: {subtitle_file.name}" + ) + + +def import_file(target_file: Path, source_file: Path) -> None: + if target_file.exists(): + target_file.unlink() + + try: + target_file.hardlink_to(source_file) + except FileExistsError: + log.exception(f"File already exists at {target_file}.") + except (OSError, UnsupportedOperation, NotImplementedError): + log.exception( + f"Failed to create hardlink from {source_file} to {target_file}. Falling back to copying the file." + ) + shutil.copy(src=source_file, dst=target_file) + + +def remove_special_characters(filename: str) -> str: + """ + Removes special characters from the filename to ensure it works with Jellyfin. + + :param filename: The original filename. + :return: A sanitized version of the filename. + """ + # Remove invalid characters + sanitized = re.sub(r"([<>:\"/\\|?*])", "", filename) + + # Remove leading and trailing dots or spaces + return sanitized.strip(" .") + + +def remove_special_chars_and_parentheses(title: str) -> str: + """ + Removes special characters and bracketed information from the title. + + :param title: The original title. + :return: A sanitized version of the title. + """ + + # Remove content within brackets + sanitized = re.sub(r"\[.*?\]", "", title) + + # Remove content within curly brackets + sanitized = re.sub(r"\{.*?\}", "", sanitized) + + # Remove year within parentheses + sanitized = re.sub(r"\(\d{4}\)", "", sanitized) + + # Remove special characters + sanitized = remove_special_characters(sanitized) + + # Collapse multiple whitespace characters and trim the result + return re.sub(r"\s+", " ", sanitized).strip()