compare playlists m3u und navidrome

audiobook_metadata/combine_audiobookmedia_to_m4b.py

@@ -8,7 +8,6 @@ import subprocess
 import tempfile
 from pathlib import Path
 import time
-from typing import Any
 
 RED = "\033[91m"
 GREEN = "\033[92m"
@@ -145,36 +144,6 @@ def create_m4b(root_path: Path, book_dir: Path) -> None:
         else:
             log_message(f"{RED}[ERROR]{RESET} - {book_dir.name} - Code: {result.returncode}, Stderr: {result.stderr}")
 
-def find_book_dirs(root: Path) -> tuple[list[Any], int]:
-    """
-    A book directory is any directory containing more than one
-    audiobook file (.mp3, .m4a, .m4b)
-    """
-
-    media_files = (
-        list(root.rglob("*.mp3"))
-        + list(root.rglob("*.m4a"))
-        + list(root.rglob("*.m4b"))
-        + list(root.rglob("*.opus"))
-    )
-
-    counts: dict[Path, int] = {}
-
-    for file in media_files:
-        if counts.get(file.parent, 0) > 1:
-            continue
-        counts[file.parent] = counts.get(file.parent, 0) + 1
-
-    directories = []
-    size = 0
-    for directory, count in counts.items():
-        if count > 1:
-            directories.append(directory)
-            size += sum(file.stat().st_size for pattern in ("*.mp3", "*.m4a", "*.m4b", "*.opus") for file in directory.glob(pattern))
-
-
-    return sorted(directories), size
-
 def process_book(args):
     target, book_dir = args
     try:

audiobook_metadata/metadata_audiobooks.py

@@ -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}")

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

audiobook_metadata/metadata_read.sh

@@ -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
-

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

media handling/compare_playlists.py

@@ -0,0 +1,170 @@
+#!/usr/bin/env python3
+
+import re
+import sys
+from pathlib import Path
+from difflib import SequenceMatcher
+
+def normalize(name):
+    name = Path(name).stem.lower()
+
+    # remove brackets
+    name = re.sub(r"\(.*?\)|\[.*?\]", "", name)
+
+    # replace punctuation with spaces
+    name = re.sub(r"[^a-z0-9]", " ", name)
+
+    # collapse whitespace
+    name = " ".join(name.split())
+
+    return name
+
+
+def similarity(a, b):
+    words_a = set(normalize(a).split())
+    words_b = set(normalize(b).split())
+
+    if not words_a or not words_b:
+        return 0.0
+
+    # Jaccard similarity
+    intersection = words_a & words_b
+    union = words_a | words_b
+
+    return len(intersection) / len(union)
+
+
+def load_playlist(path):
+    entries = []
+
+    with open(path, encoding="utf-8", errors="ignore") as f:
+        for line in f:
+            line = line.strip()
+
+            if not line or line.startswith("#"):
+                continue
+
+            entries.append(Path(line).name)
+
+    return entries
+
+
+def compare_playlists(orig_file, new_file, threshold=0.85):
+    orig_files = load_playlist(orig_file)
+    new_files = load_playlist(new_file)
+
+    print(f"\n=== {orig_file.parent} ===")
+    print(f"Comparing:")
+    print(f"  {orig_file.name}")
+    print(f"  {new_file.name}")
+
+    missing = []
+
+    for orig in orig_files:
+        best_match = None
+        best_score = 0
+
+        for candidate in new_files:
+            score = similarity(orig, candidate)
+
+            if score > best_score:
+                best_score = score
+                best_match = candidate
+
+        if best_score >= threshold:
+            pass
+           # print(
+           #     f"MATCH   {best_score:.1%}  "
+           #     f"{orig} -> {best_match}"
+           # )
+        else:
+            missing.append((orig, best_score, best_match))
+
+    if missing:
+        print("\nMISSING:")
+        for orig, score, best_match in sorted(
+            missing,
+            key=lambda x: x[1],
+            reverse=True,
+        ):
+            print(
+                f"  {orig}"
+                f" (best={score:.1%}"
+                f"{', candidate=' + best_match if best_match else ''})"
+            )
+
+    return len(missing)
+
+
+def compare_directory(root, threshold=0.85):
+    total_missing = 0
+    playlist_pairs = 0
+
+    for orig_file in root.rglob("*_orig.m3u"):
+        new_file = orig_file.with_name(
+            orig_file.name.replace("_orig.m3u", ".m3u")
+        )
+
+        if not new_file.exists():
+            print(
+                f"\nWARNING: Missing playlist for "
+                f"{orig_file}"
+            )
+            continue
+
+        playlist_pairs += 1
+        total_missing += compare_playlists(
+            orig_file,
+            new_file,
+            threshold,
+        )
+
+    print("\n" + "=" * 80)
+    print(f"Playlists compared : {playlist_pairs}")
+    print(f"Total missing      : {total_missing}")
+
+
+def main():
+    args = sys.argv[1:]
+
+    threshold = 0.85
+
+    if len(args) == 3 and args[0] == "--threshold":
+        threshold = float(args[1])
+        args = [args[2]]
+
+    elif len(args) == 4 and args[0] == "--threshold":
+        threshold = float(args[1])
+        args = [args[2], args[3]]
+
+    if len(args) == 1:
+        path = Path(args[0])
+
+        if not path.is_dir():
+            print("Single argument must be a directory")
+            sys.exit(1)
+
+        compare_directory(path, threshold)
+
+    elif len(args) == 2:
+        compare_playlists(
+            Path(args[0]),
+            Path(args[1]),
+            threshold,
+        )
+
+    else:
+        print(
+            f"Usage:\n"
+            f"  {Path(sys.argv[0]).name} "
+            f"<playlist_orig.m3u> <playlist.m3u>\n\n"
+            f"  {Path(sys.argv[0]).name} "
+            f"<directory>\n\n"
+            f"Optional:\n"
+            f"  --threshold 0.90"
+        )
+        sys.exit(1)
+
+
+if __name__ == "__main__":
+    main()