metadata audiobook library
delete old scripts load metadata from file clean genre fix artist/comment/composer/series read books from database write books to database
This commit is contained in:
Lucas Vivo committed
1 parent
c52c83f7ff
commit
7fbda6b133
4 files changed
+634
-636
No files matched your search
@@ -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()
|
||||
File diff suppressed because it is too large.
Load diff
@@ -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()
|
||||
@@ -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",
|
||||
|
||||
Reference in new issue
Block a user