diff --git a/audiobook_metadata/metadata_audiobooks_library.py b/audiobook_metadata/metadata_audiobooks_library.py index 11fd1dd..f03b7b9 100755 --- a/audiobook_metadata/metadata_audiobooks_library.py +++ b/audiobook_metadata/metadata_audiobooks_library.py @@ -17,7 +17,6 @@ logging.setLoggerClass(CustomLogger) logger: CustomLogger = logging.getLogger(__name__) # type: ignore[assignment] metadata_fields = ["artist", "album", "composer", "comment", "genre", "grouping", "album artist", "description"] -audio_extensions = {"*.m4b", "*.m4a", "*.mp3", "*.flac", "*.ogg", "*.opus"} def ffmpeg_escape(path: Path) -> str: return str(path.resolve()).replace("'", r"'\''") @@ -192,12 +191,12 @@ class Audiobook: def init_book(self) -> None: self.files = [ - file for pattern in audio_extensions for file in self.bookPath.glob(pattern) + file for pattern in AUDIO_SUFFIXES for file in self.bookPath.glob(pattern) ] self.firstFile = self.get_first_file() self.size = sum( file.stat().st_size - for pattern in audio_extensions + for pattern in AUDIO_SUFFIXES for file in self.bookPath.glob(pattern) ) self.fileConunt = len(self.files) @@ -815,7 +814,7 @@ class AudiobookLibrary: """ media_files = [] - for ext in audio_extensions: + for ext in AUDIO_SUFFIXES: media_files.extend(list(self.libraryPath.rglob(ext))) dirs = {file.parent for file in media_files} for directory in dirs: diff --git a/audiobook_metadata/metadata_rank_from_list.py b/audiobook_metadata/metadata_rank_from_list.py index 00c026e..da65775 100755 --- a/audiobook_metadata/metadata_rank_from_list.py +++ b/audiobook_metadata/metadata_rank_from_list.py @@ -6,15 +6,6 @@ import subprocess from pathlib import Path from difflib import SequenceMatcher -AUDIO_EXTENSIONS = { - ".mp3", - ".m4b", - ".m4a", -} - -RED = "\033[31m" -GREEN = "\033[32m" -RESET = "\033[0m" USE_KID3 = True def normalize(text: str) -> str: return re.sub(r"[^a-z0-9]", "", text.lower()) diff --git a/media_editing/ffmpeg_handling.py b/media_editing/ffmpeg_handling.py index 91f1495..cf38e69 100755 --- a/media_editing/ffmpeg_handling.py +++ b/media_editing/ffmpeg_handling.py @@ -23,9 +23,6 @@ from serverlibrary import * logging.setLoggerClass(CustomLogger) logger: CustomLogger = logging.getLogger(__name__) # type: ignore[assignment] -VIDEO_EXTS = {".mp4", ".mkv", ".avi", ".mov", ".webm"} -AUDIO_EXTS = {".mp3", ".m4a", ".m4b", ".opus"} - # ---------------------------- # OUTPUT ENFORCEMENT (clean + centralized) # ---------------------------- diff --git a/serverlibrary/src/serverlibrary/__init__.py b/serverlibrary/src/serverlibrary/__init__.py index a9caf26..a6a4b18 100755 --- a/serverlibrary/src/serverlibrary/__init__.py +++ b/serverlibrary/src/serverlibrary/__init__.py @@ -1,8 +1,8 @@ from .core import ( - TRACE, LOG_LEVELS, + TRACE, CustomLogger, - setup_logging, + AUDIO_SUFFIXES, blue, combined_score, get_audio_tracks, @@ -15,18 +15,20 @@ from .core import ( score_match_fuzzy, score_match_jaccard, score_match_recall_jaccard, + setup_logging, tmdb_get, tvdb_get, + VIDEO_SUFFIXES, word_similarity, words, yellow, ) __all__ = [ - "TRACE", "LOG_LEVELS", + "TRACE", "CustomLogger", - "setup_logging", + "AUDIO_SUFFIXES", "blue", "combined_score", "get_audio_tracks", @@ -39,10 +41,11 @@ __all__ = [ "score_match_fuzzy", "score_match_jaccard", "score_match_recall_jaccard", + "setup_logging", "tmdb_get", "tvdb_get", + "VIDEO_SUFFIXES", "word_similarity", "words", "yellow", ] - diff --git a/serverlibrary/src/serverlibrary/core.py b/serverlibrary/src/serverlibrary/core.py index 7466d55..1d49435 100755 --- a/serverlibrary/src/serverlibrary/core.py +++ b/serverlibrary/src/serverlibrary/core.py @@ -26,6 +26,7 @@ reset = "\033[0m" class CustomLogger(logging.Logger): def trace(self, msg: str, *args, **kwargs) -> None: if self.isEnabledFor(TRACE): + kwargs.setdefault("stacklevel", 2) self._log(TRACE, msg, args, **kwargs) @@ -38,12 +39,13 @@ LOG_LEVELS = { "CRITICAL": logging.CRITICAL, } + def setup_logging(log_level: str = "INFO", logfile: str | None = None) -> None: logging.addLevelName(TRACE, "TRACE") file_formatter = logging.Formatter("%(asctime)s %(levelname)s %(name)s %(message)s") console_formatter = colorlog.ColoredFormatter( - "%(log_color)s%(asctime)s %(levelname)-8s%(reset)s %(name)s %(message)s", + "%(log_color)s%(asctime)s %(levelname)-8s%(reset)s %(funcName)s %(message)s", log_colors={ "TRACE": "cyan", "DEBUG": "blue", @@ -78,9 +80,19 @@ def setup_logging(log_level: str = "INFO", logfile: str | None = None) -> None: root_logger.info(f"Set logger to {log_level}, Logfile: {logfile_data}") +# ----------------------------- +# Media Files +# ----------------------------- + +AUDIO_SUFFIXES = {"*.m4b", "*.m4a", "*.mp3", "*.flac", "*.ogg", "*.opus"} +VIDEO_SUFFIXES = {".mp4", ".mkv", ".avi", ".mov", ".webm"} + + # ----------------------------- # Helpers # ----------------------------- + + def safe_delete(path: Path): if path.is_dir(): shutil.rmtree(path) @@ -89,52 +101,84 @@ def safe_delete(path: Path): def safe_rename( - src: Path, - dst: Path, - dry: bool = False, + src_filepath: Path, + dst_filename: Path, + dry_run: bool = False, confirm: bool = False, silent=False, overwrite=False, + logger=None, ) -> tuple[str, bool]: """ - Safely rename a file or directory without overwriting existing files. - Automatically appends _1, _2, ... if needed. + Safely rename or move a file or directory to a new destination. + + If the destination already exists, a unique name is automatically + generated by appending a numeric suffix (_1, _2, ...) unless + overwrite is enabled. Supports dry-run mode, optional user + confirmation, and silent operation. + + Args: + src_filepath: Source file or directory to rename. + dst_filename: Target path and filename. + dry_run: If True, only report the action without performing it. + confirm: If True, prompt for confirmation before renaming. + silent: If True, suppress informational output. + overwrite: If True, replace an existing destination. + logger: Logger handler + Returns: + A tuple containing: + - The final destination path as a string. + - True if a rename was performed successfully, otherwise False. + + Raises: + FileNotFoundError: If the source path does not exist. + OSError: If the rename operation fails. """ - src = Path(src) - dst = Path(dst) + src_filepath = Path(src_filepath).expanduser().resolve() + dst_filename = Path(dst_filename).expanduser() + if dst_filename.parent == Path("."): + dst_filename = src_filepath.parent / dst_filename + dst_filename = dst_filename.resolve() + if not src_filepath.exists(): + raise FileNotFoundError(f"File {src_filepath} not found") # nothing to do - if src.resolve() == dst.resolve(): + if src_filepath == dst_filename: return "", False - directory = dst.parent - base = dst.stem - ext = dst.suffix + # directory = dst_filename.parent + # base = dst_filename.stem + # ext = dst_filename.suffix - final_dst = dst - counter = 1 + # final_dst = dst_filename + # counter = 1 - # find free name - while final_dst.exists(): - final_dst = directory / f"{base}_{counter}{ext}" - counter += 1 - ret = f"{src} → {dst}" + # # find free name + # while final_dst.exists(): + # final_dst = directory / f"{base}_{counter}{ext}" + # counter += 1 + ret = f"{src_filepath} → {dst_filename}" if not silent: - print(ret) - - if dry: + if logger is not None: + logger.info(ret) + else: + print(ret) + if dry_run: return ret, True if confirm: input("Press Enter to confirm...") - dst.parent.mkdir(parents=True, exist_ok=True) + dst_filename.parent.mkdir(parents=True, exist_ok=True) - if dst.exists() and not overwrite: - print("File already exists...") + if dst_filename.exists() and not overwrite: + if logger is not None: + logger.warning("File already exists...") + else: + print(ret) return "", False - src.rename(dst) + src_filepath.rename(dst_filename) return ret, True @@ -177,10 +221,19 @@ def get_subtitle_tracks(source): ) if not tracks: - tracks.append({"number": 0, "description": "", "lang": "", "language": "", "primary": False}) + tracks.append( + { + "number": 0, + "description": "", + "lang": "", + "language": "", + "primary": False, + } + ) return tracks + def get_audio_tracks( source: str, title: int | None = None, handbrake: bool = False ) -> list[dict]: @@ -286,10 +339,19 @@ def get_audio_tracks( ) if not tracks: - tracks.append({"number": 0, "description": "","lang":"","language":"", "primary": False}) + tracks.append( + { + "number": 0, + "description": "", + "lang": "", + "language": "", + "primary": False, + } + ) return tracks + def words(text): return set(re.findall(r"[^\W_]+", text.lower())) @@ -360,7 +422,7 @@ def tmdb_get(endpoint, params=None): BASE_URL = "https://api.themoviedb.org/3" url = BASE_URL + endpoint - print(f"Set of Tmdb query: {url}") + # print_or_log(f"Set of Tmdb query: {url}") response = requests.get( url, @@ -378,7 +440,7 @@ def tvdb_get(endpoint, params=None): """Perform a GET request against TVDB.""" BASE_URL = "https://api4.thetvdb.com/v4" url = BASE_URL + endpoint - print(f"Set of TVDB query: {url}") + # print_or_log(f"Set of TVDB query: {url}") response = requests.get( url,