#!/usr/bin/env python3 import subprocess import tempfile import argparse import logging import re import json from pathlib import Path from typing import Any from collections import defaultdict from concurrent.futures import ThreadPoolExecutor, as_completed RED = "\033[91m" GREEN = "\033[92m" YELLOW = "\033[93m" BLUE = "\033[94m" RESET = "\033[0m" TRACE = 5 class CustomLogger(logging.Logger): def trace(self, msg: str, *args, **kwargs) -> None: if self.isEnabledFor(TRACE): self._log(TRACE, msg, args, **kwargs) logging.setLoggerClass(CustomLogger) logger: CustomLogger = logging.getLogger(__name__) # type: ignore[assignment] LOG_LEVELS = { "TRACE": TRACE, "DEBUG": logging.DEBUG, "INFO": logging.INFO, "WARNING": logging.WARNING, "ERROR": logging.ERROR, "CRITICAL": logging.CRITICAL, } audio_extensions = { "*.m4b", "*.m4a", "*.mp3", "*.flac", "*.ogg", "*.opus" } def ffmetadata_escape(value: str) -> str: return ( value .replace("\\", "\\\\") .replace("=", "\\=") .replace(";", "\\;") .replace("#", "\\#") .replace("\n", "\\\n") ) def get_chapters(media: Path) -> list[dict]: result = subprocess.run( [ "ffprobe", "-v", "error", "-show_chapters", "-of", "json", str(media), ], capture_output=True, text=True, check=True, ) data = json.loads(result.stdout) return data.get("chapters", []) def get_metadata(file, fields): metadata=dict.fromkeys(fields, "") for field in fields: cmd= (["kid3-cli", "-c", f"get {field}", str(file)]) try: result = subprocess.run( cmd, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, text=True, check=True, ) metadata[field] = result.stdout.strip() except (OSError, subprocess.SubprocessError) as error: print(f"Error getting metadata for {file}: {error}") return metadata return metadata def get_duration_ms(media: Path) -> int: duration = subprocess.check_output( [ "ffprobe", "-v", "error", "-show_entries", "format=duration", "-of", "default=noprint_wrappers=1:nokey=1", str(media), ], text=True, ).strip() return int(float(duration) * 1000) class Audiobook(): def __init__(self, book_path: Path) -> None: self.metadata = None self.artist = None self.genre = None self.composer = None self.name = None self.comment = None self.bookPath = book_path self.files = [file for pattern in audio_extensions for file in book_path.glob(pattern)] self.firstFile = self.get_first_file() self.size = sum(file.stat().st_size for pattern in audio_extensions for file in book_path.glob(pattern)) self.fileConunt = len(self.files) self.singleFile = self.fileConunt == 1 # Metadata self.load_metadata() logger.info(f"Found {self.bookPath.name}, {self.fileConunt} files") self.changed = False def __repr__(self): return ( f"Audiobook(" f"name={self.name!r}, " f"artist={self.artist!r}, " f"composer={self.composer!r}, " f"path={str(self.bookPath)!r})," f"size={self.size!r}, " f"fileConunt={self.fileConunt!r}, " ) def get_first_file(self): return self.files[0] def load_metadata(self): fields = ["artist", "album", "composer", "comment", "genre"] metadata = get_metadata(self.get_first_file(), fields ) self.artist = metadata["artist"] self.name = metadata["album"] self.composer = metadata["composer"] self.comment = metadata["comment"] self.genre = metadata["genre"] self.metadata= metadata logger.trace(f"Load metadata {self.artist}, {self.name}, {self.composer}, {self.comment}, {self.genre}") def write_metadata(self, file, fields, values): for field,value in fields,values: cmd= (["kid3-cli", "-c" , "tag 2","-c", f"set {field} \"{value}\"", "-c" "save", str(file)]) try: result = subprocess.run( cmd, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, text=True, check=True, ) except (OSError, subprocess.SubprocessError) as error: print(f"Error writing metadata for {file}: {error}") return return def fix_composer(self): m = re.search( r"(?:read\s+by|spoken\s+by|narrated\s+by|gesprochen\s+von|gelesen\s+von)\s+([^\n.,;]+)", self.comment, re.IGNORECASE, ) if not m: logger.debug(f"No composer found in comment for {self.bookPath}") return self.changed=True self.comment = (self.comment[:m.start()] + self.comment[m.end():] ).strip(" .,;\n") self.comment = re.sub(r"\s+"," ",self.comment) if not self.composer: logger.info(f"[Updated] composer found for {self.bookPath}") self.composer = m.group(1).strip() def fix_name(self): if not self.name: self.name = self.bookPath.name logger.info(f"[Updated] name found for {self.bookPath}") def clean_genre(self,genre): if not self.genre: return genre = self.genre.lower() genre = re.sub(r'(?i)^audio\s*books?$', '', genre) genre = re.sub(r'(?i)^audiobooks?;?$', '', genre) genre = re.sub(r'(?i)\b(hörb(uch|ücher)|hoerb(uch|uecher))\b', '', genre) genre = re.sub(r'(?i)^(hörspiele?|hoerspiele?)$', '', genre) genre = re.sub(r'(?i)^spoken\s*word$', '', genre) genre = re.sub(r'(?i)gesprochene\s*inhalte$', '', genre) genre = re.sub(r'(?i)^speech$', '', genre) genre = re.sub(r'Fanasie', 'Fantasie', genre) genre = re.sub(r'Mit Videospiel-?', 'Videospielbezug', genre) genre = re.sub(r'(?i)nonfiction', 'NonFiction', genre) genre = re.sub(r'(?i)^speech$', '', genre) genre = re.sub(r'(?i)^vocal$', '', genre) genre = re.sub(r'(?i)^retail$', '', genre) genre = re.sub(r'(?i)^other$', '', genre) genre = re.sub(r'(?i)^misc$', '', genre) genre = re.sub(r'(?i)^(sci[- ]?fi|sf)$', 'Science Fiction', genre) genre = re.sub(r'(?i)^\(?science fiction\)?$', 'Science Fiction', genre) genre = re.sub(r'Thrillers', 'Thriller', genre) genre = re.sub(r"(?i)\bchildren'?s audiobooks?\b", "", genre) genre = re.sub(r'^-\s*|\s*-$', '', genre) genre = re.sub(r'(?i)Engelske\s*\sLydbånd', '', genre) genre = re.sub(r'^FIC028000.*$', '', genre) genre = re.sub(r'^800.*$', '', genre) genre = re.sub(r'^Fantastic Beasts.*$', '', genre) genre = re.sub(r'^and thei.*$', '', genre) genre = re.sub(r'^those you.*$', '', genre) # Normalize separators genre = re.sub(r'\s*[,&:/]\s*',';',genre) genre = re.sub(r'\s*;\s*','; ',genre) # Trim punctuation / whitespace genre = re.sub(r'^[\s,;:\-]+','',genre) genre = re.sub(r'[\s,;\-]+$','',genre) genre = genre.strip() self.genre = genre.lstrip() return def create_m4b_from_book(self, library_root_path, creat_m4b_no_delete: bool=False) -> None: output_file_final = self.bookPath / f"{self.bookPath.name}.m4b" if output_file_final.exists(): logger.info(f"{RED}[SKIP]{RESET} {output_file_final}") logger.info("Book already present") return logger.info(f"[CREATE] {output_file_final}") with tempfile.TemporaryDirectory(dir=library_root_path.parent) as tmp: tmpdir = Path(tmp) output_file_tmp = tmpdir / f"{self.bookPath.name}.m4b" concat_file = tmpdir / "concat.txt" metadata_file = tmpdir / "chapters.txt" # Inspect all inputs before deciding how to create chapters. file_info = [] has_existing_chapters = False for media in self.files: chapters = get_chapters(media) duration_ms = get_duration_ms(media) file_info.append({ "media": media, "chapters": chapters, "duration_ms": duration_ms, }) if chapters: has_existing_chapters = True if has_existing_chapters: logger.info( "Existing chapter tracks found; preserving chapter metadata" ) else: logger.info( "No existing chapters found; generating chapters from filenames" ) current_ms = 0 with ( concat_file.open("w", encoding="utf-8") as cf, metadata_file.open("w", encoding="utf-8") as mf, ): mf.write(";FFMETADATA1\n") for key, value in self.metadata.items(): mf.write( f"{ffmetadata_escape(key)}=" f"{ffmetadata_escape(value)}\n" ) for info in file_info: media = info["media"] chapters = info["chapters"] duration_ms = info["duration_ms"] cf.write(f"file '{ffmpeg_escape(media)}'\n") if has_existing_chapters: # Preserve actual chapter boundaries and titles. for chapter in chapters: start_ms = round( float(chapter["start_time"]) * 1000 ) end_ms = round( float(chapter["end_time"]) * 1000 ) tags = chapter.get("tags", {}) title = tags.get("title", media.stem) mf.write("[CHAPTER]\n") mf.write("TIMEBASE=1/1000\n") mf.write(f"START={current_ms + start_ms}\n") mf.write(f"END={current_ms + end_ms}\n") mf.write( f"title={ffmetadata_escape(title)}\n" ) else: # No input contains chapters: one chapter per file. mf.write("[CHAPTER]\n") mf.write("TIMEBASE=1/1000\n") mf.write(f"START={current_ms}\n") mf.write(f"END={current_ms + duration_ms}\n") mf.write( f"title={ffmetadata_escape(media.stem)}\n" ) current_ms += duration_ms cmd_backup = [ "ffmpeg", "-y", "-f", "concat", "-safe", "0", "-i", str(concat_file), "-i", str(metadata_file), "-i", str(self.firstFile), # audio from concat "-map", "0:a", # cover image from first media "-map", "2:v?", # Metadata and chapters from our generated FFMETADATA file "-map_metadata", "1", "-map_chapters", "1", "-c:a", "aac", "-b:a", "128k", "-c:v", "copy", "-disposition:v:0", "attached_pic", str(output_file_tmp), ] cmd_default = [ "ffmpeg", "-y", "-f", "concat", "-safe", "0", "-i", str(concat_file), "-i", str(metadata_file), "-i", str(self.firstFile), "-map", "0:a", "-map", "2:v?", # Metadata and chapters from our generated FFMETADATA file "-map_metadata", "1", "-map_chapters", "1", "-c:a", "copy", "-c:v", "copy", "-disposition:v:0", "attached_pic", str(output_file_tmp), ] result = subprocess.run( cmd_default, capture_output=True, text=True, ) logger.info(result.stdout) if result.returncode == 234: logger.warning( f"{YELLOW}[WARNING]{RESET} - {self.bookPath.name} - {result.returncode} retrying with aac encoder") result = subprocess.run( cmd_backup, capture_output=True, text=True, ) logger.info(result.stdout) if result.returncode == 0: output_file_tmp.rename(output_file_final) for media in self.files: if creat_m4b_no_delete: continue media.unlink() self.files = Path(output_file_final) self.firstFile = Path(output_file_final) self.fileConunt = 1 self.singleFile = True self.changed = False self.size = Path(output_file_final).stat().st_size logger.info(f"{GREEN}[SUCCESS]{RESET} - {self.bookPath.name} - delete media files") else: logger.error( f"{RED}[ERROR]{RESET} - {self.bookPath.name} - Code: {result.returncode}, Stderr: {result.stderr}") class AudiobookLibrary: def __init__(self, libraryPath,dry_run,logfilename: str="audiobook_library.log"): self.dry_run = dry_run logging.setLoggerClass(CustomLogger) logger: CustomLogger = logging.getLogger(__name__) # type: ignore[assignment] logging.basicConfig( level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s", handlers=[ logging.FileHandler(logfilename), logging.StreamHandler() ] ) logger.info('Started') self.libraryPath = Path(libraryPath) self.audiobooks = [] self._by_path = {} self._by_name = defaultdict(list) self._by_artist = defaultdict(list) self._by_composer = defaultdict(list) self.load_audiobooks_from_dir() self.size = sum( audiobook.size for audiobook in self.audiobooks ) # size in bytes def add(self, book: Audiobook): self.audiobooks.append(book) self._by_path[book.bookPath] = book self._by_name[book.name].append(book) if book.artist: self._by_artist[book.artist].append(book) if book.composer: self._by_composer[book.composer].append(book) def print_name(self, books): if not isinstance(books, list): books = [books] for book in books: if book.name: logger.info(book.name) else: logger.info(f"No name in {book.bookPath.name}") def get_by_path(self, path): return self._by_path.get(Path(path)) def get_by_name(self, name): return self._by_name.get(name, []) def get_by_artist(self, artist): return self._by_artist.get(artist, []) def get_by_composer(self, composer): return self._by_artist.get(composer, []) def find_path_contains(self, path) -> list[Audiobook]: return [ book for book in self.audiobooks if book.bookPath == path or book.bookPath.is_relative_to(path) ] def find_name_contains(self, text): text = text.lower() return [ book for book in self.audiobooks if text in book.name.lower() ] def find_artist_contains(self, text): text = text.lower() return [ book for book in self.audiobooks if book.artist and text in book.artist.lower() ] def load_audiobooks_from_dir(self): """ A book directory is any directory containing more than one audiobook file (.mp3, .m4a, .m4b) """ media_files = [] for ext in audio_extensions: media_files.extend(list(self.libraryPath.rglob( ext)) ) dirs = {file.parent for file in media_files} for directory in dirs: self.add(Audiobook(directory)) def load_metadata(self): for audiobook in self.audiobooks: audiobook.load_metadata() def write_metadata(self): for audiobook in self.audiobooks: if audiobook.changed: logger.trace(f"Writing metadata: {audiobook.name}") if self.dry_run: continue audiobook.write_metadata() def fix_composers(self): for book in self.audiobooks: book.fix_composer() def fix_names(self): for book in self.audiobooks: book.fix_name() def fix_comments(self): for book in self.audiobooks: if book.comment and len(book.comment) < 40: logger.info(f"Remove short comment: {book.comment} of {book.name}") book.comment = "" book.changed=True def create_m4b_from_dir(self,path: Path="", name: str="",create_m4b_no_delete: bool=False) -> None: audiobooks = [] if path: audiobooks = self.find_path_contains(path) if name: audiobooks.extend(self.find_name_contains(name)) if not name and not path: audiobooks = self.audiobooks jobs = [ audiobook for audiobook in audiobooks if not audiobook.singleFile ] for audiobook in jobs: logger.debug(f"Create m4b file for {audiobook.name}") if not self.dry_run: with ThreadPoolExecutor(max_workers=8) as executor: futures = { executor.submit( audiobook.create_m4b_from_book, self.libraryPath, create_m4b_no_delete, ): audiobook for audiobook in jobs } for future in as_completed(futures): audiobook = futures[future] try: future.result() except Exception: logger.exception(f"Failed to create m4b for {audiobook.name}") def ffmpeg_escape(path: Path) -> str: return str(path.resolve()).replace("'", r"'\''") def main(): parser = argparse.ArgumentParser( description="Audiobook Directory Processing" ) parser.add_argument( "library", type=Path, help="Library path" ) parser.add_argument( "--create-m4b-no-delete", action="store_true", help="create m4b without deleting" ) parser.add_argument( "-ll", "--log-level", type=str.upper, default="INFO", choices=["TRACE", "DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"], help="Set log level", ) parser.add_argument( "-cmfn", "--create-m4b-filter-name", type=str, default="", help="Only create m4b from multiple files matching ", ) parser.add_argument( "-cmfp", "--create-m4b-filter-path", type=str, default="", help="Only create m4b from multiple files matching ", ) parser.add_argument( "--dry-run", action="store_true", help="Dry run only", ) args = parser.parse_args() logging.basicConfig( level=LOG_LEVELS[args.log_level.upper()], format="%(asctime)s %(levelname)s %(message)s", ) library_dir = args.library.resolve() library = AudiobookLibrary(library_dir,args.dry_run) library.fix_composers() library.fix_names() library.fix_comments() library.create_m4b_from_dir(name=args.create_m4b_filter_name, path=args.create_m4b_filter_path, create_m4b_no_delete=args.create_m4b_no_delete) library.write_metadata() #library.printName(library.get_by_artist("Tolkien")) #library.printName(library.find_artist_contains("Tolkien")) #library.printName(library.find_name_contains("")) #library.printName(library.get_by_path("/home/lucas/Nextcloud/Syncfolder/")) print("Fin") if __name__ == "__main__": main()