#!/usr/bin/env python3 import subprocess import tempfile import argparse import logging import re 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" } 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.loadMetadata() 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 loadMetadata(self): fields = ["artist", "album", "composer", "comment", "genre"] metadata = self.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 get_metadata(self, 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 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 fixComposer(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 fixName(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(self, root_path) -> 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=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" 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 media in self.files: cf.write(f"file '{ffmpeg_escape(media)}'\n") duration_ms = self.get_duration_ms(media) 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={media.stem}\n") current_ms += duration_ms cmd_backup = ["ffmpeg", "-y", "-f", "concat", "-safe", "0", "-i", str(concat_file), "-i", str(metadata_file), # cover from first media "-i", str(self.firstFile), # audio from concat "-map", "0:a", # cover image from first media "-map", "2:v?", # copy artist/album/composer/etc from first media file "-map_metadata", "2", # copy chapters from ffmetadata "-map_chapters", "1", "-c:a", "copy", "-c:a", "aac", "-b:a", "128k", # copy cover image without re-encoding "-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), # cover from first media "-i", str(self.firstFile), # audio from concat "-map", "0:a", # cover image from first media "-map", "2:v?", # copy artist/album/composer/etc from first media file "-map_metadata", "2", # copy chapters from ffmetadata "-map_chapters", "1", "-c:a", "copy", # copy cover image without re-encoding "-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: 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}") def get_duration_ms(self,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 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_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.loadMetadata() def fixComposers(self): for book in self.audiobooks: book.fixComposer() def fixNames(self): for book in self.audiobooks: book.fixName() 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 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" ) args = parser.parse_args() library_dir = args.library library = AudiobookLibrary(library_dir) library.fixComposers() library.fixNames() library.fixComments() library.printName(library.get_by_artist("Astrid Lindgren")) library.printName(library.find_artist_contains("Astrid Lindgren")) library.printName(library.get_by_name("Pippi Langstrumpf")) library.printName(library.find_name_contains("Pippi Langstrumpf")) library.printName(library.get_by_composer("Heike Maklatsch")) # library.printName(library.get_by_path("/home/lucas/Nextcloud/Audiobooks/Neu/Astrid Lindgren/Junker Nils von Eka (Hoerfest)")) for audiobook in library.audiobooks: first_file = audio_files[0] artist = get_metadata("artist", first_file) album = get_metadata("album", first_file) composer = get_metadata("composer", first_file) comment = get_metadata("comment", first_file) print("====≠===============================") print(f"Folder: {library_dir.name}") print(f"Album: {album}") print(f"Artist: {artist}") print(f"Narrator: {composer}") print(f"Comment: {comment}") if __name__ == "__main__": main()