Server-scripts/audiobook_metadata/metadata_audiobooks_library.py

#!/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 <name>",
    )
    parser.add_argument(
        "-cmfp",
        "--create-m4b-filter-path",
        type=str,
        default="",
        help="Only create m4b from multiple files matching <path>",
    )
    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()