diff --git a/audiobook_metadata/combine_audiobookmedia_to_m4b.py b/audiobook_metadata/combine_audiobookmedia_to_m4b.py deleted file mode 100755 index 1b6cbd5..0000000 --- a/audiobook_metadata/combine_audiobookmedia_to_m4b.py +++ /dev/null @@ -1,200 +0,0 @@ -#!/usr/bin/env python3 - -from __future__ import annotations -from concurrent.futures import ProcessPoolExecutor, as_completed - -import argparse -import subprocess -import tempfile -from pathlib import Path -import time - -RED = "\033[91m" -GREEN = "\033[92m" -YELLOW = "\033[93m" -BLUE = "\033[94m" -RESET = "\033[0m" -logfile=Path(f"media_{time.strftime("%Y%m%d_%H%M%S")}.log") - -def log_message(message): - with open(logfile, "a") as log: - log.write(f"{message}\n") - print(message) - -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) - -def ffmpeg_escape(path: Path) -> str: - return str(path.resolve()).replace("'", r"'\''") - -def create_m4b(root_path: Path, book_dir: Path) -> None: - media_files = sorted(list(book_dir.glob("*.mp3")) - + list(book_dir.glob("*.m4a")) - + list(book_dir.glob("*.m4b")) - + list(book_dir.glob("*.opus"))) - if not media_files: - return - - output_file_final = book_dir / f"{book_dir.name}.m4b" - - if output_file_final.exists(): - log_message(f"{RED}[SKIP]{RESET} {output_file_final}") - log_message("Book already present") - return - - log_message(f"[CREATE] {output_file_final}") - - with (tempfile.TemporaryDirectory(dir=root_path.parent) as tmp): - tmpdir = Path(tmp) - - output_file_tmp = tmpdir / f"{book_dir.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 media_files: - cf.write(f"file '{ffmpeg_escape(media)}'\n") - - duration_ms = 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(media_files[0]), - # 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(media_files[0]), - # 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), - ] - with logfile.open("a", encoding="utf-8") as log: - result = subprocess.run( - cmd_default, - stdout=log, - stderr=subprocess.STDOUT, - text=True, - ) - if result.returncode == 234: - log_message(f"{YELLOW}[WARNING]{RESET} - {book_dir.name} - {result.returncode} retrying with aac encoder") - result = subprocess.run( - cmd_backup, - stdout=log, - stderr=subprocess.STDOUT, - text=True, - ) - if result.returncode == 0: - output_file_tmp.rename(output_file_final) - for media in media_files: - media.unlink() - log_message(f"{GREEN}[SUCCESS]{RESET} - {book_dir.name} - delete media files") - else: - log_message(f"{RED}[ERROR]{RESET} - {book_dir.name} - Code: {result.returncode}, Stderr: {result.stderr}") - -def process_book(args): - target, book_dir = args - try: - create_m4b(target, book_dir) - return book_dir, None - except Exception as exc: - return book_dir, str(exc) - -def main() -> None: - parser = argparse.ArgumentParser() - parser.add_argument( - "path", - type=Path, - help="Book folder or root folder", - ) - - args = parser.parse_args() - - target = args.path.resolve() - - if not target.exists(): - raise SystemExit(f"Path does not exist: {target}") - - # Direct book folder - if any(list(target.glob("*.mp3")) - + list(target.glob("*.m4a")) - + list(target.glob("*.m4b")) - + list(target.glob("*.opus"))): - create_m4b(target.parent,target) - return - - book_dir_set, library_size = find_book_dirs(target) - - for book_dir in book_dir_set: - log_message(f"Found Media: {book_dir}") - - print("Proceed? y/n") - name = input() - if name != "y": - log_message(f"Aborted") - exit(0) - - # Root folder containing many books - with ProcessPoolExecutor(max_workers=8) as executor: - for book_dir, error in executor.map( - process_book, - ((target, book_dir) for book_dir in book_dir_set), - ): - if error: - log_message(f"{RED}[ERROR]{RESET} {book_dir}: {error}") - - -if __name__ == "__main__": - main() diff --git a/audiobook_metadata/metadata_audiobooks_library.py b/audiobook_metadata/metadata_audiobooks_library.py index 1d14050..4e302fa 100755 --- a/audiobook_metadata/metadata_audiobooks_library.py +++ b/audiobook_metadata/metadata_audiobooks_library.py @@ -1,15 +1,16 @@ #!/usr/bin/env python3 -import subprocess -import tempfile import argparse +import json import logging import re -import json -from pathlib import Path -from typing import Any +import subprocess +import sys +import tempfile +import sqlite3 from collections import defaultdict from concurrent.futures import ThreadPoolExecutor, as_completed +from pathlib import Path RED = "\033[91m" GREEN = "\033[92m" @@ -17,11 +18,14 @@ 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 = { @@ -33,32 +37,77 @@ LOG_LEVELS = { "CRITICAL": logging.CRITICAL, } -audio_extensions = { - "*.m4b", - "*.m4a", - "*.mp3", - "*.flac", - "*.ogg", - "*.opus" -} +metadata_fields = ["artist", "album", "composer", "comment", "genre", "grouping", "album artist", "description"] +audio_extensions = {"*.m4b", "*.m4a", "*.mp3", "*.flac", "*.ogg", "*.opus"} + +def ffmpeg_escape(path: Path) -> str: + return str(path.resolve()).replace("'", r"'\''") + +def clean_genre(genre): + if not genre: + return "" + genre_old = genre + #genre = 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() + genre = genre.lstrip() + if genre != genre_old: + logger.debug(f"Update genre from {genre_old} --->>> {genre}") + return genre + def ffmetadata_escape(value: str) -> str: return ( - value - .replace("\\", "\\\\") + value.replace("\\", "\\\\") .replace("=", "\\=") .replace(";", "\\;") .replace("#", "\\#") .replace("\n", "\\\n") ) - def get_chapters(media: Path) -> list[dict]: result = subprocess.run( [ "ffprobe", - "-v", "error", + "-v", + "error", "-show_chapters", - "-of", "json", + "-of", + "json", str(media), ], capture_output=True, @@ -69,13 +118,11 @@ def get_chapters(media: Path) -> list[dict]: data = json.loads(result.stdout) return data.get("chapters", []) - -def get_metadata(file, fields): - metadata=dict.fromkeys(fields, "") +def get_metadata(file, fields) -> dict: + metadata = dict.fromkeys(fields, "") for field in fields: - cmd= (["kid3-cli", "-c", f"get {field}", str(file)]) + cmd = ["kid3-cli", "-c", f"get {field}", str(file)] try: - result = subprocess.run( cmd, stdout=subprocess.PIPE, @@ -89,7 +136,6 @@ def get_metadata(file, fields): return metadata return metadata - def get_duration_ms(media: Path) -> int: duration = subprocess.check_output( [ @@ -107,21 +153,74 @@ def get_duration_ms(media: Path) -> int: return int(float(duration) * 1000) +def write_metadata(file, fields, values,dry_run=False): + cmd_base = [ + "kid3-cli", + "-c", + "tag 2" + ] + for field, value in zip(fields, values): + cmd_base.extend( ["-c", f'set {field} "{value}"']) + cmd_base.extend(["-c", "save", str(file)]) + try: + logger.trace(f"Writing metadata for {file}") + logger.trace(f"Cmd: {" ".join(cmd_base)}") + if dry_run: + return + result = subprocess.run( + cmd_base, + stdout=subprocess.PIPE, + stderr=subprocess.DEVNULL, + text=True, + check=True, + ) + logger.trace(f"KID3: {result.stdout}") + except (OSError, subprocess.SubprocessError) as error: + print(f"Error writing metadata for {file}: {error}") + return + return -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 +def cleanup(text: str) -> str: + # Collapse repeated punctuation groups like ", ,", ". .", "! !", "? ?" + while True: + new_text = re.sub(r'([.,!?;:])(?:\s+\1)+', r'\1', text) + if new_text == text: + break + text = new_text + + # Optional: normalize multiple spaces + text = re.sub(r'\s{2,}', ' ', text) + + return text.strip() + +class Audiobook: + def __init__(self, book_path: Path, library_root_path : Path) -> None: + self.series = "" + self.singleFile = False + self.fileConunt = 0 + self.size = 0 + self.firstFile = "" + self.changed = None + self.metadata = {} + self.artist = "" + self.genre = "" + self.composer = "" + self.name = "" + self.comment = "" + self.files: list[Path] = [] self.bookPath = book_path - self.files = [file for pattern in audio_extensions for file in - book_path.glob(pattern)] + self.library_root_path = library_root_path + + def init_book(self) -> None: + self.files = [ + file for pattern in audio_extensions for file in self.bookPath.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.size = sum( + file.stat().st_size + for pattern in audio_extensions + for file in self.bookPath.glob(pattern) + ) self.fileConunt = len(self.files) self.singleFile = self.fileConunt == 1 # Metadata @@ -139,107 +238,192 @@ class Audiobook(): f"size={self.size!r}, " f"fileConunt={self.fileConunt!r}, " ) + def get_first_file(self): return self.files[0] + def write_fields_from_metadata(self) -> None: + self.name = self.metadata["album"] + self.composer = self.metadata["composer"] + self.comment = self.metadata["comment"] + self.genre = self.metadata["genre"] + self.series = self.metadata["grouping"] + artist = self.metadata.get("artist", "").strip() + self.artist = artist or self.metadata.get("album artist") + self.description = self.metadata["description"] + + def write_metadata_from_fields(self): + self.metadata["album"] = self.name + self.metadata["composer"] = self.composer + self.metadata["genre"] = self.genre + self.metadata["grouping"] = self.series + self.metadata["comment"] = self.comment + self.metadata["description"] = self.description + self.metadata["artist"] = self.artist + self.metadata["album artist"] = self.artist + 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}") + self.metadata = get_metadata(self.get_first_file(), metadata_fields) + self.write_fields_from_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): + def fix_artist_from_comment(self): m = re.search( - r"(?:read\s+by|spoken\s+by|narrated\s+by|gesprochen\s+von|gelesen\s+von)\s+([^\n.,;]+)", + r"(?:" + r"author(?:s)?:?|" + r"writer(?:s)?:?|" + r"co[- ]?author(?:s)?:?|" + r"co[- ]?writer(?:s)?:?|" + r"creator(?:s)?:?|" + r"contributor(?:s)?:?|" + r"composer(?:s)?:?|" + r"lyricist(?:s)?:?|" + r"songwriter(?:s)?:?|" + r"written\s+by:?|" + r"story\s+by:?|" + r"autorin(?:nen)?:?|" + r"autor\(en\):?|" + r"autor(?:en)?:?|" + r"text(?:e)?\s+von:?|" + r"geschrieben\s+von:?|" + r"verfasst\s+von:?|" + r"auteur(?:s)?:?|" + r"autore(?:i)?:?|" + r"autor(?:es)?:?|" + r"escritor(?:es|as)?:?" + r")\s*:?\s*([^\n.,;]+)", self.comment, re.IGNORECASE, ) if not m: - logger.debug(f"No composer found in comment for {self.bookPath}") + logger.debug(f"No artist 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) + self.changed = True + artist = re.sub(r"\s*\([^)]*\)", "", m.group(1).strip()).strip() + logger.trace(f"Remove {artist} from comment: {self.comment}") + self.comment = (self.comment[: m.start()] + self.comment[m.end() :]).strip( + " .,;\n" + ) + self.comment = cleanup(self.comment) + if not self.artist: + self.artist = artist + elif self.artist != artist: + logger.warning( + f"[Action Required] artist {artist} found not matching {self.artist} for {self.name}" + ) + else: + logger.debug(f"Artist {artist} matches file metadata for {self.name}") + + def fix_artist_from_path(self): + if not self.artist: + rel_path = self.bookPath.relative_to(self.library_root_path).parts + + if len(rel_path) < 2: + logger.trace(f"Artist not visible in folder structure: {self.bookPath}") + return + self.artist = rel_path[0] + + logger.info(f"[Updated] artist found in path for {self.bookPath}") + + def fix_composer_from_comment(self): + m = re.search( + r"(?:read\s+by:?|" + r"spoken\s+by:?|" + r"narrated\s+by:?|" + r"gesprochen\s+von:?|" + r"gelesen\s+von)\s+([^\n.,;]+)", + self.comment, + re.IGNORECASE, + ) + if not m: + logger.debug(f"No composer found in comment for {self.name}") + return + self.changed = True + self.comment = (self.comment[: m.start()] + self.comment[m.end() :]).strip( + " .,;\n" + ) + self.comment = cleanup(self.comment) if not self.composer: - logger.info(f"[Updated] composer found for {self.bookPath}") + logger.info(f"[Updated] composer found in comment for {self.name}") self.composer = m.group(1).strip() + def fix_composer_from_path(self): + if not self.composer: + rel_path = self.bookPath.relative_to(self.library_root_path) + m = re.search( + r"(?:read\s+by:?|" + r"spoken\s+by:?|" + r"narrated\s+by:?|" + r"gesprochen\s+von:?|" + r"gelesen\s+von)\s+([^\n.,;]+)", + rel_path, + re.IGNORECASE, + ) + if not m: + logger.debug(f"No composer found in path for {self.bookPath}") + return + self.changed = True + logger.info(f"[Updated] composer found in Path 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}") + logger.info(f"[Updated] name found in path for {self.bookPath}") - def clean_genre(self,genre): - if not self.genre: + def fix_series(self): + if self.series: + logger.trace(f"{self.name} already has series: {self.series}") + rel_path = self.bookPath.relative_to(self.library_root_path).parts + patterns = [ + r'^(\d+)\s*-\s*(.+)$', # 02 - Bookname + r'^(.+?)\s*-\s*(\d+)$', # Bookname - 02 + r'^(.+?)\s*-\s*Book\s+(\d+)$', # Bookname - Book 2 + ] + number="" + if len(rel_path) <= 2: + logger.trace(f"No series found for {self.name}") 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() - + for pattern in patterns: + match = re.match(pattern, rel_path[2], re.IGNORECASE) + if match: + number = next(g for g in match.groups() if g.isdigit()) + break + if not number: + logger.trace(f"No series number found for {self.name}") + return + self.series = rel_path[1] + "#" + number + logger.info(f"[Updated] name found for {self.name}") return - def create_m4b_from_book(self, library_root_path, creat_m4b_no_delete: bool=False) -> None: + def fix_genre(self): + self.genre = clean_genre(self.genre) + + def move(self, desiredPath: Path, desired_name: str, dry_run=False): + logger.trace(f"Moving {self.bookPath} to {desiredPath}") + if dry_run: + return + desiredPath.mkdir(parents=True, exist_ok=True) + + for file in self.files: + file.rename(desiredPath / file.name) + + def generate_dir_from_metadata(self): + composer ="" + name = self.name + if self.composer: + composer = f" (by {self.composer})" + name = self.name + composer + desired_path = self.library_root_path / self.artist /name + if self.series: + desired_path = self.library_root_path / self.artist / self.series / name + desired_name = self.artist + " - " + self.name + composer + self.bookPath.ext + return desired_path, desired_name + + def create_m4b_from_book(self, creat_m4b_no_delete: bool = False) -> None: output_file_final = self.bookPath / f"{self.bookPath.name}.m4b" if output_file_final.exists(): @@ -249,7 +433,7 @@ class Audiobook(): logger.info(f"[CREATE] {output_file_final}") - with tempfile.TemporaryDirectory(dir=library_root_path.parent) as tmp: + with tempfile.TemporaryDirectory(dir=self.library_root_path.parent) as tmp: tmpdir = Path(tmp) output_file_tmp = tmpdir / f"{self.bookPath.name}.m4b" @@ -264,11 +448,13 @@ class Audiobook(): chapters = get_chapters(media) duration_ms = get_duration_ms(media) - file_info.append({ - "media": media, - "chapters": chapters, - "duration_ms": duration_ms, - }) + file_info.append( + { + "media": media, + "chapters": chapters, + "duration_ms": duration_ms, + } + ) if chapters: has_existing_chapters = True @@ -289,12 +475,9 @@ class Audiobook(): 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" - ) + mf.write(f"{ffmetadata_escape(key)}={ffmetadata_escape(value)}\n") for info in file_info: media = info["media"] chapters = info["chapters"] @@ -305,12 +488,8 @@ class Audiobook(): 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 - ) + 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) @@ -319,62 +498,80 @@ class Audiobook(): 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" - ) + 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" - ) + 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), + "ffmpeg", + "-y", + "-f", + "concat", + "-safe", + "0", + "-i", + str(concat_file), + "-i", + str(metadata_file), + "-i", + str(self.firstFile), # audio from concat - "-map", "0:a", + "-map", + "0:a", # cover image from first media - "-map", "2:v?", + "-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", - + "-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?", + "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", - + "-map_metadata", + "1", + "-map_chapters", + "1", + "-c:a", + "copy", + "-c:v", + "copy", + "-disposition:v:0", + "attached_pic", str(output_file_tmp), ] @@ -382,17 +579,20 @@ class Audiobook(): cmd_default, capture_output=True, text=True, + check=True, ) logger.info(result.stdout) if result.returncode == 234: logger.warning( - f"{YELLOW}[WARNING]{RESET} - {self.bookPath.name} - {result.returncode} retrying with aac encoder") + f"{YELLOW}[WARNING]{RESET} - {self.bookPath.name} - {result.returncode} retrying with aac encoder" + ) result = subprocess.run( cmd_backup, capture_output=True, text=True, + check=True, ) logger.info(result.stdout) @@ -402,43 +602,181 @@ class Audiobook(): if creat_m4b_no_delete: continue media.unlink() - self.files = Path(output_file_final) - self.firstFile = Path(output_file_final) + self.files = [output_file_final] + self.firstFile = 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") + 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}") + f"{RED}[ERROR]{RESET} - {self.bookPath.name} - Code: {result.returncode}, Stderr: {result.stderr}" + ) + def write_metadata(self, keys=None, dry_run=False): + self.write_metadata_from_fields() + if keys is None: + keys = self.metadata.keys() + values = self.metadata.values() + else: + values = [self.metadata[key] for key in keys ] + for file in self.files: + write_metadata(file, keys, values,dry_run) + self.changed = False class AudiobookLibrary: - def __init__(self, libraryPath,dry_run,logfilename: str="audiobook_library.log"): + def __init__( + self, library_path: Path, sql_database_path: Path, dry_run: bool, logfilename: str = "audiobook_library.log" + ): + self.sql_database_instance = self.SqlDatabase(sql_database_path) 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() - ] + handlers=[logging.FileHandler(logfilename), logging.StreamHandler()], ) - logger.info('Started') - self.libraryPath = Path(libraryPath) - self.audiobooks = [] - + logger.info("Started") + self.libraryPath: Path = Path(library_path) + self.audiobooks: list[Audiobook] = [] 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 + self.size = sum( + audiobook.size for audiobook in self.audiobooks + ) # size in bytes + + class SqlDatabase: + def __init__(self, sql_database_path: Path): + self.db = None + self.path = sql_database_path + self.books = [] + + def connect(self): + if not self.path.is_file() or self.path.suffix != ".db": + logger.error(f"{self.path} does not exist") + return False + self.db = sqlite3.connect(self.path) + self.db.row_factory = sqlite3.Row + return True + + def cursor(self): + return self.db.cursor() + + def commit(self): + self.db.commit() + + def close(self): + self.db.close() + + def init(self): + if not self.connect(): + return + self.load_database_data() + + def load_database_data(self): + books = [] + + for row in self.db.execute("SELECT * FROM books"): + book = dict(row) + + # auto-decode JSON columns + for key, value in book.items(): + if isinstance(value, str) and value[:1] in "[{": + try: + book[key] = json.loads(value) + except json.JSONDecodeError: + pass + + books.append(book) + + self.books = books + + def update_books(self): + for book in self.books: + self._save_book(book) + + def _save_book(self, book): + book = book.copy() + + for key, value in book.items(): + if isinstance(value, (list, dict)): + book[key] = json.dumps(value, ensure_ascii=False) + + columns = [c for c in book if c != "id"] + + sql = f""" + UPDATE books + SET {", ".join(f'"{c}" = ?' for c in columns)} + WHERE id = ? + """ + + values = [book[c] for c in columns] + values.append(book["id"]) + + self.db.execute(sql, values) + + # commit once after all updates + + def fix_sql_database(self, dry_run): + + for book in self.sql_database_instance.books: + genres = book.get("genres") + if not genres or not any(g.strip() for g in genres): + continue + new_genres = [] + for genre in genres: + genre = clean_genre(genre) + if "," in genre: + genre = genre.split(",") + elif " - " in genre: + genre = genre.split(" - ") + elif "Historical Adventure Thriller Mystery" in genre: + genre = genre.split(" ") + + if isinstance(genre, str): + if len(genre) > 50 or (5 > len(genre) > 0): + genre = "" + logger.trace(f"Too long or too short: {genre}") + genre = genre.strip() + genre = genre.lstrip() + new_genres.append(genre) + elif isinstance(genre, list): + if len(genres) > 1: + genres = [g for g in genres if g] + new_genres.extend(genre) + + new_genres = sorted(set(new_genres)) + list_difference = False + if len(new_genres) < len(genres): + list_difference = True + for old, new in zip(genres, new_genres): + if old != new: + list_difference = True + break + + if not list_difference: + logger.trace("No difference: Dont Commit to Database") + continue + + logger.trace(f"{genres} -> {new_genres}") + book["genres"] = new_genres + if dry_run: + return + + self.sql_database_instance.update_books() + self.sql_database_instance.commit() + self.sql_database_instance.close() + + def add(self, book: Audiobook): self.audiobooks.append(book) @@ -481,11 +819,7 @@ class AudiobookLibrary: def find_name_contains(self, text): text = text.lower() - return [ - book - for book in self.audiobooks - if text in book.name.lower() - ] + return [book for book in self.audiobooks if text in book.name.lower()] def find_artist_contains(self, text): text = text.lower() @@ -503,26 +837,32 @@ class AudiobookLibrary: media_files = [] for ext in audio_extensions: - media_files.extend(list(self.libraryPath.rglob( ext)) ) + media_files.extend(list(self.libraryPath.rglob(ext))) dirs = {file.parent for file in media_files} for directory in dirs: - self.add(Audiobook(directory)) + self.add(Audiobook(directory,self.libraryPath)) + for audiobook in self.audiobooks: + audiobook.init_book() def load_metadata(self): for audiobook in self.audiobooks: audiobook.load_metadata() - def write_metadata(self): + def write_metadata(self, keys=None): for audiobook in self.audiobooks: if audiobook.changed: logger.trace(f"Writing metadata: {audiobook.name}") - if self.dry_run: - continue - audiobook.write_metadata() + audiobook.write_metadata(keys,dry_run=self.dry_run) + + def fix_artists(self): + for book in self.audiobooks: + book.fix_artist_from_comment() + book.fix_artist_from_path() def fix_composers(self): for book in self.audiobooks: - book.fix_composer() + book.fix_composer_from_comment() + book.fix_composer_from_path() def fix_names(self): for book in self.audiobooks: @@ -531,11 +871,33 @@ class AudiobookLibrary: 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}") + logger.info(f"Remove short comment of {book.name}: {book.comment}") book.comment = "" - book.changed=True + book.changed = True - def create_m4b_from_dir(self,path: Path="", name: str="",create_m4b_no_delete: bool=False) -> None: + def fix_genres(self): + for book in self.audiobooks: + book.fix_genre() + + def fix_series(self): + for book in self.audiobooks: + book.fix_series() + + def move_library(self): + for audiobook in self.audiobooks: + desired_path,desired_name = audiobook.generate_dir_from_metadata() + audiobook.move(desired_path,desired_name,dry_run=self.dry_run) + + def generate_folder_structure(self): + for audiobook in self.audiobooks: + desired_path = ( + self.libraryPath / Path(audiobook.artist) / Path(audiobook.name) + ) + audiobook.move(desired_path) + + def create_m4b_from_dir( + self, path: Path = Path(""), name: str = "", keep_originals: bool = False + ) -> None: audiobooks = [] if path: audiobooks = self.find_path_contains(path) @@ -544,11 +906,7 @@ class AudiobookLibrary: if not name and not path: audiobooks = self.audiobooks - jobs = [ - audiobook - for audiobook in audiobooks - if not audiobook.singleFile - ] + jobs = [audiobook for audiobook in audiobooks if not audiobook.singleFile] for audiobook in jobs: logger.debug(f"Create m4b file for {audiobook.name}") @@ -559,7 +917,7 @@ class AudiobookLibrary: executor.submit( audiobook.create_m4b_from_book, self.libraryPath, - create_m4b_no_delete, + keep_originals, ): audiobook for audiobook in jobs } @@ -571,26 +929,12 @@ class AudiobookLibrary: 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 = argparse.ArgumentParser(description="Audiobook Directory Processing") + + parser.add_argument("library", type=Path, help="Library path") - 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", @@ -600,48 +944,93 @@ def main(): 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 ", + m4b_group = parser.add_argument_group("M4B creation") + + m4b_group.add_argument( + "--create-m4b", + action="store_true", + help="Create M4B files from multi-file audiobooks", ) - parser.add_argument( - "-cmfp", - "--create-m4b-filter-path", - type=str, - default="", - help="Only create m4b from multiple files matching ", + + m4b_group.add_argument( + "--keep-originals", + action="store_true", + help="Do not delete source files after M4B creation", ) + + m4b_group.add_argument( + "--name-filter", + metavar="NAME", + default="", + help="Process only books whose name contains NAME", + ) + + m4b_group.add_argument( + "--path-filter", + metavar="PATH", + default="", + help="Process only books whose path contains PATH", + ) + parser.add_argument( "--dry-run", action="store_true", help="Dry run only", ) + parser.add_argument( + "--sql-database", + type=Path, + default=Path("/docker/audiobookshelf/config/absdatabase.sqlite"), + help="SQL Database Path", + ) + parser.add_argument( + "--fix-db", + action="store_true", + help="Fix the Database", + ) + 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_dir = args.library.expanduser() + library_dir = library_dir.resolve() + if not library_dir.exists(): + logger.error(f"Library directory does not exist: {library_dir}") + sys.exit(1) - library = AudiobookLibrary(library_dir,args.dry_run) + database_path = args.sql_database.expanduser() + database_path = database_path.resolve() + if args.fix_db and not database_path.is_file(): + logger.error(f"Database is not a file: {database_path}") + return + library = AudiobookLibrary(library_dir, database_path, args.dry_run) + if args.fix_db: + library.sql_database_instance.init() + + library.fix_artists() library.fix_composers() library.fix_names() + library.fix_series() 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.fix_genres() + + library.create_m4b_from_dir( + name=args.name_filter, + path=args.path_filter, + keep_originals=args.keep_originals, + ) + 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/")) + if args.fix_db: + library.fix_sql_database(True) + print("Fin") + if __name__ == "__main__": - main() \ No newline at end of file + main() diff --git a/audiobook_metadata/metadata_genre_remove_from_file.py b/audiobook_metadata/metadata_genre_remove_from_file.py deleted file mode 100644 index 9ec87c6..0000000 --- a/audiobook_metadata/metadata_genre_remove_from_file.py +++ /dev/null @@ -1,192 +0,0 @@ -#!/usr/bin/env python3 - -import argparse -import re -import subprocess -from pathlib import Path -from datetime import datetime - - -def get_genre(file_path): - try: - result = subprocess.run( - ["kid3-cli", "-c", "get genre", str(file_path)], - capture_output=True, - text=True, - check=False, - ) - return result.stdout.strip() - except Exception: - return "" - - -def set_genre(file_path, genre): - subprocess.run( - [ - "kid3-cli", - "-c", - "tag 2", - "-c", - f"set genre \"{genre}\"", - "-c", - "save", - str(file_path), - ], - check=False, - ) - - -def clean_genre(genre): - - 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() - genre = genre.lstrip() - - return genre - -def find_audio_files(library): - extensions = {".mp3", ".m4a", ".m4b"} - - return [ - p for p in Path(library).rglob("*") - if p.is_file() and p.suffix.lower() in extensions - ] - - -def main(): - parser = argparse.ArgumentParser( - description="Audiobook Genre Cleanup" - ) - - parser.add_argument( - "library", - help="Library path" - ) - - parser.add_argument( - "--delete", - action="store_true", - help="Update tags instead of reporting only" - ) - - parser.add_argument( - "--report", - default="/home/lucas/audiobook_genre_cleanup.txt", - help="Report file" - ) - - args = parser.parse_args() - - files = find_audio_files(args.library) - - total = len(files) - processed = 0 - matches = 0 - modified = 0 - - with open(args.report, "w", encoding="utf-8") as report: - report.write("Audiobook Genre Cleanup Report\n") - report.write(f"Generated: {datetime.now()}\n") - report.write(f"Library: {args.library}\n") - report.write(f"Total files: {total}\n\n") - - for file in files: - processed += 1 - - if processed % 100 == 0: - print(f"Progress: {processed}/{total} files processed") - - genre = get_genre(file) - - if not genre: - continue - - cleaned = clean_genre(genre) - - if genre != cleaned: - matches += 1 - status = "MATCH" - - if args.delete: - set_genre(file, cleaned) - modified += 1 - status = "UPDATED" - - line = ( - f'[{processed}/{total}] [{status}] ' - f'{file} | OLD="{genre}" | NEW="{cleaned}"' - ) - - print(line) - - with open(args.report, "a", encoding="utf-8") as report: - report.write(line + "\n") - - with open(args.report, "a", encoding="utf-8") as report: - report.write("\n========== SUMMARY ==========\n") - report.write(f"Files scanned : {total}\n") - report.write(f"Matches found : {matches}\n") - report.write(f"Files updated : {modified}\n") - - print("\n========== SUMMARY ==========") - print(f"Files scanned : {total}") - print(f"Matches found : {matches}") - print(f"Files updated : {modified}") - print(f"Report written to: {args.report}") - - -if __name__ == "__main__": - main() \ No newline at end of file diff --git a/pyproject.toml b/pyproject.toml index 7dba19d..b6eccbc 100755 --- a/pyproject.toml +++ b/pyproject.toml @@ -40,6 +40,7 @@ dependencies = [ "packaging==26.0", "paho-mqtt==2.1.0", "pandas==3.0.2", + "pandas-stubs~=3.0.2", "pillow==12.2.0", "pillow-heif>=1.3.0", "pip==25.1.1",