#!/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 RED = "\033[91m" GREEN = "\033[92m" YELLOW = "\033[93m" BLUE = "\033[94m" RESET = "\033[0m" logger = logging.getLogger(__name__) 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, bookPath: Path) -> None: self.bookPath = bookPath self.files = [file for pattern in audio_extensions for file in bookPath.glob(pattern)] self.firstFile = self.get_first_file() self.size = sum(file.stat().st_size for pattern in audio_extensions for file in bookPath.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") 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"] def write_metadata(self, file, fields, values): metadata=dict.fromkeys(fields, "") 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): if not self.composer: 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.info(f"No composer found for {self.bookPath}") return logger.info(f"[Updated] composer found for {self.bookPath}") self.composer = m.group(1).strip() self.comment = (self.comment[:m.start()] + self.comment[m.end():] ).strip(" .,;\n") self.comment = re.sub(r"\s+"," ",self.comment) 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 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), "-map", "0:a", "-map", "2:v?", "-map_metadata", "2", "-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?", "-map_metadata", "2", "-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.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,logfilename: str="audiobook_library.log"): 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.loadAudiobooksFromDir() 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 printName(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 loadAudiobooksFromDir(self) -> tuple[list[Any], int]: """ 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 loadMetadata(self): for audiobook in self.audiobooks: audiobook.load_metadata() def fixComposers(self): for book in self.audiobooks: book.fix_composer() def fixNames(self): for book in self.audiobooks: book.fix_name() def fixComments(self): for book in self.audiobooks: if len(book.comment) < 40: logger.info(f"Remove short comment: {book.comment} of {book.name}") book.comment = "" def create_m4b_from_dir(self,path: Path="", name: str="",create_m4b_no_delete: bool=False) -> None: audiobooks = self.find_path_contains(path) audiobooks.extend(self.find_name_contains(name)) for audiobook in audiobooks: self.printName(audiobook) audiobook.create_m4b_from_book( self.libraryPath,create_m4b_no_delete) 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" ) args = parser.parse_args() library_dir = args.library library = AudiobookLibrary(library_dir) library.fixComposers() library.fixNames() library.fixComments() #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/")) library.create_m4b_from_dir(name="Fellowship", create_m4b_no_delete=args.create_m4b_no_delete) print("Fin") if __name__ == "__main__": main()