first version of general audiobook library.py

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()