From 09343a7523bf6bc9a472c4f12c61ae8b3621e93d Mon Sep 17 00:00:00 2001 From: Lucas Vivo Date: Mon, 12 Oct 2026 16:33:40 +0200 Subject: [PATCH] first version of general audiobook library.py --- audiobook_metadata/metadata_audiobooks.py | 60 --- .../metadata_audiobooks_library.py | 440 ++++++++++++++++++ audiobook_metadata/metadata_read.sh | 27 -- .../metadata_read_by_from_file.sh | 2 +- 4 files changed, 441 insertions(+), 88 deletions(-) delete mode 100755 audiobook_metadata/metadata_audiobooks.py create mode 100755 audiobook_metadata/metadata_audiobooks_library.py delete mode 100755 audiobook_metadata/metadata_read.sh diff --git a/audiobook_metadata/metadata_audiobooks.py b/audiobook_metadata/metadata_audiobooks.py deleted file mode 100755 index d3e9f43..0000000 --- a/audiobook_metadata/metadata_audiobooks.py +++ /dev/null @@ -1,60 +0,0 @@ -#!/usr/bin/env python3 - -import sys -import subprocess -from pathlib import Path - - -ROOT = Path(sys.argv[1]) if len(sys.argv) > 1 else Path(".") - -audio_extensions = { - ".m4b", - ".m4a", - ".mp3", - ".flac", - ".ogg", -} - - -def get_metadata(field, file): - try: - result = subprocess.run( - ["kid3-cli", "-c", f"get {field}", str(file)], - stdout=subprocess.PIPE, - stderr=subprocess.DEVNULL, - text=True, - ) - return result.stdout.strip() - - except (OSError, subprocess.SubprocessError): - return "" - - -# Include ROOT itself, matching find "$ROOT" -type d -for directory in [ROOT] + [p for p in ROOT.rglob("*") if p.is_dir()]: - audio_files = sorted( - ( - file - for file in directory.iterdir() - if file.is_file() - and file.suffix.lower() in audio_extensions - ), - key=lambda p: str(p), - ) - - if not audio_files: - continue - - 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: {directory.name}") - print(f"Album: {album}") - print(f"Artist: {artist}") - print(f"Narrator: {composer}") - print(f"Comment: {comment}") diff --git a/audiobook_metadata/metadata_audiobooks_library.py b/audiobook_metadata/metadata_audiobooks_library.py new file mode 100755 index 0000000..7f8dada --- /dev/null +++ b/audiobook_metadata/metadata_audiobooks_library.py @@ -0,0 +1,440 @@ +#!/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() \ No newline at end of file diff --git a/audiobook_metadata/metadata_read.sh b/audiobook_metadata/metadata_read.sh deleted file mode 100755 index bd45c4a..0000000 --- a/audiobook_metadata/metadata_read.sh +++ /dev/null @@ -1,27 +0,0 @@ -#!/usr/bin/env bash - -ROOT="${1:-.}" - -find "$ROOT" -type d | while read -r dir; do - first_file=$( - find "$dir" -maxdepth 1 -type f \ - \( -iname "*.m4b" -o -iname "*.m4a" -o -iname "*.mp3" \ - -o -iname "*.flac" -o -iname "*.ogg" \) | - sort | - head -n1 - ) - - [[ -z "$first_file" ]] && continue - - artist=$(kid3-cli -c "get artist" "$first_file" 2>/dev/null) - album=$(kid3-cli -c "get album" "$first_file" 2>/dev/null) - composer=$(kid3-cli -c "get composer" "$first_file" 2>/dev/null) - comment=$(kid3-cli -c "get comment" "$first_file" 2>/dev/null) - echo "====≠===============================" - echo "Folder: $(basename "$dir")" - echo "Album: $album" - echo "Artist: $artist" - echo "Narrator: $composer" - echo "Comment: $comment" -done - diff --git a/audiobook_metadata/metadata_read_by_from_file.sh b/audiobook_metadata/metadata_read_by_from_file.sh index ae936a4..ed5abd0 100755 --- a/audiobook_metadata/metadata_read_by_from_file.sh +++ b/audiobook_metadata/metadata_read_by_from_file.sh @@ -27,7 +27,7 @@ fi list="$1" while IFS= read -r file; do - +book [[ -z "$file" ]] && continue if [[ ! -e "$file" ]]; then