From c4e181b05bfef488d74bd9f019f62fb2d2e432fb Mon Sep 17 00:00:00 2001 From: Steppenstreuner Date: Fri, 28 Aug 2026 14:18:09 +0200 Subject: [PATCH] add(aduio): tagging files --- pyproject.toml | 1 + tests/test_import_flow.py | 34 ++++++++++++++++++ wordarr/importer/mover.py | 5 +++ wordarr/importer/tagger.py | 73 ++++++++++++++++++++++++++++++++++++++ 4 files changed, 113 insertions(+) create mode 100644 wordarr/importer/tagger.py diff --git a/pyproject.toml b/pyproject.toml index c29ae99..31d912f 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -10,6 +10,7 @@ dependencies = [ "httpx>=0.27", "rapidfuzz>=3.6", "pydantic>=2.6", + "mutagen>=1.47", ] [project.optional-dependencies] diff --git a/tests/test_import_flow.py b/tests/test_import_flow.py index e07b010..833db6b 100644 --- a/tests/test_import_flow.py +++ b/tests/test_import_flow.py @@ -134,3 +134,37 @@ def test_bulk_requests_and_split_scan(client): by_name = {i["name"]: i["suggested_request_id"] for i in items} expected = {f"Die drei Fragezeichen - Folge {n:03d}": created[n - 1]["id"] for n in (1, 2, 3)} assert by_name == expected + + +def test_audio_tags_written_on_import(client): + from mutagen.id3 import ID3 + + root = client.tmp_path / "library" / "kids_audio" + lib_id = client.post("/api/libraries", json={ + "name": "KidsAudio", "media_type": "audiobook", "root_path": str(root), + }).json()["id"] + req_id = client.post("/api/requests", json={ + "library_id": lib_id, "title": "Die drei ??? Kids Folge 085", + "authors": "Ulf Blanck", "series": "Die drei ??? Kids", "volume": 85, + }).json()["id"] + + book_dir = client.downloads / "Kids 85 Schnueffler" + book_dir.mkdir() + for i in range(2): + (book_dir / f"track{i}.mp3").write_bytes(b"") + + items = client.get("/api/import/scan").json()["items"] + resp = client.post("/api/import", json={"items": [{ + "path": items[0]["path"], "is_dir": True, + "files": items[0]["files"], "request_id": req_id, + }]}) + assert resp.json()["results"][0]["ok"] + + dest = root / "Ulf Blanck" / "Die drei Kids Folge 085" # "?" is stripped from paths + files = sorted(dest.iterdir()) + tags = ID3(files[0]) + assert str(tags["TALB"]) == "Die drei ??? Kids Folge 085" + assert str(tags["TPE1"]) == "Ulf Blanck" + assert str(tags["TXXX:SERIES"]) == "Die drei ??? Kids" + assert str(tags["TXXX:SERIES-PART"]) == "85" + assert str(tags["TRCK"]) == "1/2" diff --git a/wordarr/importer/mover.py b/wordarr/importer/mover.py index a1f760f..854c87f 100644 --- a/wordarr/importer/mover.py +++ b/wordarr/importer/mover.py @@ -2,6 +2,7 @@ import shutil from pathlib import Path from ..naming import render_template, sanitize +from . import tagger def import_item(item_path: str, files: list[str], is_dir: bool, request, library) -> str: @@ -18,6 +19,8 @@ def import_item(item_path: str, files: list[str], is_dir: bool, request, library for i, src in enumerate(paths, 1): dest = folder / sanitize(f"{base} - Part {i:0{width}d}{src.suffix.lower()}") shutil.move(str(src), dest) + if request.media_type == "audiobook": + tagger.tag_audio(dest, request, track=i, total=len(paths)) # remove the source dir if nothing meaningful is left src_dir = Path(item_path) leftovers = [p for p in src_dir.rglob("*") if p.is_file()] @@ -30,4 +33,6 @@ def import_item(item_path: str, files: list[str], is_dir: bool, request, library if dest.exists(): raise FileExistsError(f"Destination already exists: {dest}") shutil.move(str(src), dest) + if request.media_type == "audiobook": + tagger.tag_audio(dest, request) return str(dest) diff --git a/wordarr/importer/tagger.py b/wordarr/importer/tagger.py new file mode 100644 index 0000000..f84adc8 --- /dev/null +++ b/wordarr/importer/tagger.py @@ -0,0 +1,73 @@ +"""Write metadata tags into imported audio files so Audiobookshelf & Co. can +build series (album/artist/track plus SERIES / SERIES-PART).""" + +import logging +from pathlib import Path + +from mutagen.flac import FLAC +from mutagen.id3 import ID3, TALB, TIT2, TPE1, TRCK, TXXX +from mutagen.mp4 import MP4 +from mutagen.oggopus import OggOpus +from mutagen.oggvorbis import OggVorbis + +log = logging.getLogger(__name__) + + +def tag_audio(path: Path, request, track: int | None = None, total: int | None = None): + """Best effort — a tagging failure must never break the import.""" + try: + _tag_audio(path, request, track, total) + except Exception as exc: + log.warning("tagging failed for %s: %s", path, exc) + + +def _tag_audio(path: Path, request, track, total): + author = (request.authors or "").split(",")[0].strip() + title = request.title or "" + series = request.series or "" + part = str(request.volume) if request.volume is not None else "" + track_title = f"{title} - Part {track:02d}" if track else title + ext = path.suffix.lower() + + if ext == ".mp3": + try: + tags = ID3(path) + except Exception: + tags = ID3() + tags.setall("TALB", [TALB(encoding=3, text=title)]) + tags.setall("TIT2", [TIT2(encoding=3, text=track_title)]) + if author: + tags.setall("TPE1", [TPE1(encoding=3, text=author)]) + if track: + tags.setall("TRCK", [TRCK(encoding=3, text=f"{track}/{total}" if total else str(track))]) + if series: + tags.setall("TXXX:SERIES", [TXXX(encoding=3, desc="SERIES", text=series)]) + if part: + tags.setall("TXXX:SERIES-PART", [TXXX(encoding=3, desc="SERIES-PART", text=part)]) + tags.save(path) + elif ext in (".m4b", ".m4a"): + mp4 = MP4(path) + mp4["\xa9alb"] = [title] + mp4["\xa9nam"] = [track_title] + if author: + mp4["\xa9ART"] = [author] + if track: + mp4["trkn"] = [(track, total or 0)] + if series: + mp4["----:com.apple.iTunes:SERIES"] = [series.encode()] + if part: + mp4["----:com.apple.iTunes:SERIES-PART"] = [part.encode()] + mp4.save() + elif ext in (".flac", ".ogg", ".opus"): + audio = {".flac": FLAC, ".ogg": OggVorbis, ".opus": OggOpus}[ext](path) + audio["ALBUM"] = title + audio["TITLE"] = track_title + if author: + audio["ARTIST"] = author + if track: + audio["TRACKNUMBER"] = str(track) + if series: + audio["SERIES"] = series + if part: + audio["SERIES-PART"] = part + audio.save()