629 lines
22 KiB
Python
629 lines
22 KiB
Python
"""Music tag maintenance — the metadata sibling to :mod:`music_covers`.
|
|
|
|
Where ``music_covers`` renames folders/files and fetches sidecar art, this module
|
|
rewrites the *tags inside* audio files. Three jobs, all honouring ``dry_run``:
|
|
|
|
* **Genres** — look the album up on MusicBrainz and write a single canonical
|
|
genre to every track, so the library's genre facet stays consistent.
|
|
* **Junk** — strip comment / encoder / URL / embedded-lyrics frames that rippers
|
|
and stores leave behind.
|
|
* **Track numbers** — normalize the ``tracknumber`` / ``discnumber`` tag value
|
|
(drop the ``/total`` suffix, zero-pad to two digits).
|
|
|
|
Format-aware: MP3 (ID3), FLAC/Ogg (Vorbis comments) and M4A (MP4 atoms) each get
|
|
their own handlers, dispatched on the loaded tag object. Like ``music_covers``,
|
|
everything is synchronous (filesystem + blocking HTTP) — call from FastAPI via
|
|
``asyncio.to_thread`` or the streaming generator below.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import queue
|
|
import re
|
|
import threading
|
|
import time
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from typing import Iterator
|
|
|
|
from mutagen import File as MutagenFile, MutagenError
|
|
from mutagen.id3 import ID3
|
|
from mutagen.mp4 import MP4Tags
|
|
|
|
from . import db
|
|
from .music_covers import (
|
|
AUDIO_EXTENSIONS,
|
|
MUSIC_ROOT,
|
|
DEFAULT_RECENT_WINDOW_SECONDS,
|
|
LogCallback,
|
|
_Recorder,
|
|
_iter_album_folders,
|
|
_was_created_recently,
|
|
clean_track_number,
|
|
get_album_metadata_from_files,
|
|
)
|
|
|
|
logger = logging.getLogger("homelabtoolkit.music_metadata")
|
|
|
|
try: # Optional: only needed for the online genre lookups.
|
|
import musicbrainzngs
|
|
|
|
_HAS_MUSICBRAINZ = True
|
|
except Exception: # pragma: no cover - optional dependency
|
|
_HAS_MUSICBRAINZ = False
|
|
|
|
|
|
# ── options ───────────────────────────────────────────────────────────────────
|
|
|
|
|
|
@dataclass
|
|
class MetadataOptions:
|
|
genres: bool = False
|
|
strip_junk: bool = False
|
|
normalize_tracks: bool = False
|
|
dry_run: bool = True
|
|
recent_only: bool = False
|
|
recent_window_seconds: int = DEFAULT_RECENT_WINDOW_SECONDS
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: dict) -> "MetadataOptions":
|
|
try:
|
|
window = int(data.get("recent_window_seconds") or DEFAULT_RECENT_WINDOW_SECONDS)
|
|
except (TypeError, ValueError):
|
|
window = DEFAULT_RECENT_WINDOW_SECONDS
|
|
return cls(
|
|
genres=bool(data.get("genres", False)),
|
|
strip_junk=bool(data.get("strip_junk", False)),
|
|
normalize_tracks=bool(data.get("normalize_tracks", False)),
|
|
dry_run=bool(data.get("dry_run", True)),
|
|
recent_only=bool(data.get("recent_only", False)),
|
|
recent_window_seconds=window,
|
|
)
|
|
|
|
@property
|
|
def any_mode(self) -> bool:
|
|
return self.genres or self.strip_junk or self.normalize_tracks
|
|
|
|
|
|
# ── genre canonicalization ────────────────────────────────────────────────────
|
|
|
|
# MusicBrainz genres are lowercase; map the messy/spaced forms to a clean label.
|
|
_GENRE_CANON = {
|
|
"hip hop": "Hip-Hop",
|
|
"hip-hop": "Hip-Hop",
|
|
"rnb": "R&B",
|
|
"r and b": "R&B",
|
|
"rhythm and blues": "R&B",
|
|
"drum and bass": "Drum & Bass",
|
|
"dnb": "Drum & Bass",
|
|
"edm": "EDM",
|
|
"idm": "IDM",
|
|
"uk garage": "UK Garage",
|
|
}
|
|
_ACRONYMS = {"edm", "idm", "uk", "us", "dj"}
|
|
|
|
|
|
def canonical_genre(name: str | None) -> str | None:
|
|
if not name:
|
|
return None
|
|
key = name.strip().lower()
|
|
if not key:
|
|
return None
|
|
if key in _GENRE_CANON:
|
|
return _GENRE_CANON[key]
|
|
words = [w.upper() if w in _ACRONYMS else w.capitalize() for w in re.split(r"\s+", key)]
|
|
return " ".join(words)
|
|
|
|
|
|
def find_album_genre(artist: str, album: str) -> str | None:
|
|
"""Return a single canonical genre for an album via MusicBrainz, or None.
|
|
|
|
Prefers the release-group's curated genres (highest vote count); falls back
|
|
to its folksonomy tags, then to the artist's genres.
|
|
"""
|
|
if not _HAS_MUSICBRAINZ or not (artist and album):
|
|
return None
|
|
try:
|
|
result = musicbrainzngs.search_release_groups(artist=artist, releasegroup=album, limit=3)
|
|
groups = result.get("release-group-list", [])
|
|
if not groups:
|
|
return None
|
|
rgid = groups[0]["id"]
|
|
genre = _genre_from_release_group(rgid)
|
|
if genre:
|
|
return genre
|
|
artist_credit = groups[0].get("artist-credit") or []
|
|
for credit in artist_credit:
|
|
mbid = (credit.get("artist") or {}).get("id") if isinstance(credit, dict) else None
|
|
if mbid:
|
|
genre = _genre_from_artist(mbid)
|
|
if genre:
|
|
return genre
|
|
except Exception as exc: # network / parse / lookup failure — never fatal
|
|
logger.debug("Genre lookup failed for %s - %s: %s", artist, album, exc)
|
|
return None
|
|
return None
|
|
|
|
|
|
def find_artist_genre(artist: str) -> str | None:
|
|
"""Return a single canonical genre for an *artist* via MusicBrainz, or None.
|
|
|
|
Resolved once per artist so every album by that artist gets the same genre
|
|
(consistency over per-album accuracy — e.g. all A Perfect Circle albums land
|
|
on one genre rather than a mix of Alternative Rock / Alternative Metal).
|
|
"""
|
|
if not _HAS_MUSICBRAINZ or not artist:
|
|
return None
|
|
try:
|
|
result = musicbrainzngs.search_artists(artist=artist, limit=3)
|
|
matches = result.get("artist-list", [])
|
|
if not matches:
|
|
return None
|
|
return _genre_from_artist(matches[0]["id"])
|
|
except Exception as exc: # network / parse / lookup failure — never fatal
|
|
logger.debug("Artist genre lookup failed for %s: %s", artist, exc)
|
|
return None
|
|
|
|
|
|
# ── manual genre overrides (user-editable, persisted) ─────────────────────────
|
|
|
|
|
|
def _artist_key(artist: str) -> str:
|
|
return (artist or "").strip().lower()
|
|
|
|
|
|
def list_genre_overrides() -> list[dict]:
|
|
with db.connect() as conn:
|
|
rows = conn.execute(
|
|
"SELECT artist, genre, updated_at FROM genre_overrides ORDER BY artist COLLATE NOCASE"
|
|
).fetchall()
|
|
return [dict(r) for r in rows]
|
|
|
|
|
|
def set_genre_override(artist: str, genre: str) -> dict:
|
|
artist = (artist or "").strip()
|
|
genre = (genre or "").strip()
|
|
if not artist or not genre:
|
|
raise ValueError("Both artist and genre are required.")
|
|
with db.connect() as conn:
|
|
conn.execute(
|
|
"INSERT INTO genre_overrides(artist_key, artist, genre, updated_at) VALUES(?,?,?,?) "
|
|
"ON CONFLICT(artist_key) DO UPDATE SET artist=excluded.artist, genre=excluded.genre, "
|
|
"updated_at=excluded.updated_at",
|
|
(_artist_key(artist), artist, genre, db.now_iso()),
|
|
)
|
|
return {"artist": artist, "genre": genre}
|
|
|
|
|
|
def delete_genre_override(artist: str) -> None:
|
|
with db.connect() as conn:
|
|
conn.execute("DELETE FROM genre_overrides WHERE artist_key=?", (_artist_key(artist),))
|
|
|
|
|
|
def _overrides_map() -> dict[str, str]:
|
|
return {_artist_key(o["artist"]): o["genre"] for o in list_genre_overrides()}
|
|
|
|
|
|
def _best_genre(entries: list[dict] | None) -> str | None:
|
|
if not entries:
|
|
return None
|
|
best = max(entries, key=lambda g: int(g.get("count") or 0))
|
|
return canonical_genre(best.get("name"))
|
|
|
|
|
|
def _genre_from_release_group(rgid: str) -> str | None:
|
|
try:
|
|
detail = musicbrainzngs.get_release_group_by_id(rgid, includes=["genres"])
|
|
except Exception:
|
|
try:
|
|
detail = musicbrainzngs.get_release_group_by_id(rgid, includes=["tags"])
|
|
except Exception:
|
|
return None
|
|
rg = detail.get("release-group") or {}
|
|
return _best_genre(rg.get("genre-list")) or _best_genre(rg.get("tag-list"))
|
|
|
|
|
|
def _genre_from_artist(mbid: str) -> str | None:
|
|
try:
|
|
detail = musicbrainzngs.get_artist_by_id(mbid, includes=["genres"])
|
|
except Exception:
|
|
try:
|
|
detail = musicbrainzngs.get_artist_by_id(mbid, includes=["tags"])
|
|
except Exception:
|
|
return None
|
|
artist = detail.get("artist") or {}
|
|
return _best_genre(artist.get("genre-list")) or _best_genre(artist.get("tag-list"))
|
|
|
|
|
|
# ── junk frame classification ─────────────────────────────────────────────────
|
|
|
|
# ID3 (MP3): match by 4-char frame id prefix. TXXX/PRIV are handled specially.
|
|
_ID3_JUNK_PREFIXES = (
|
|
"COMM", # comments
|
|
"USLT", "SYLT", # embedded lyrics
|
|
"WXXX", "WCOM", "WCOP", "WOAF", "WOAR", "WOAS", "WORS", "WPAY", "WPUB", # URLs
|
|
"TENC", "TSSE", # encoded-by / encoder settings
|
|
)
|
|
# TXXX descriptions worth dropping (store/encoder junk). ReplayGain is preserved.
|
|
_ID3_TXXX_JUNK = ("itun", "cddb", "purchase", "comment", "encoder", "encoded by", "www", "url")
|
|
|
|
# Vorbis (FLAC/Ogg): lowercase comment keys.
|
|
_VORBIS_JUNK_KEYS = {
|
|
"comment", "comments", "description",
|
|
"lyrics", "unsyncedlyrics", "unsynced lyrics",
|
|
"encoder", "encodedby", "encoded_by", "encoder_options", "encoding", "tool",
|
|
}
|
|
_VORBIS_JUNK_SUBSTRINGS = ("url", "www", "purchase", "itun", "cddb")
|
|
|
|
# MP4 (M4A): atom keys.
|
|
_MP4_JUNK_KEYS = {
|
|
"\xa9cmt", # comment
|
|
"\xa9lyr", # lyrics
|
|
"\xa9too", "tool", # encoder
|
|
"purd", # purchase date
|
|
}
|
|
|
|
# Embedded artwork is NEVER stripped. These are belt-and-braces guards so no
|
|
# current or future junk rule can ever match a picture frame/atom/comment.
|
|
_ID3_PROTECTED = ("APIC", "PIC") # ID3v2.3/2.4 and ID3v2.2 attached pictures
|
|
_MP4_PROTECTED = {"covr"} # MP4 cover atom
|
|
_VORBIS_PROTECTED = {"metadata_block_picture", "coverart", "cover art"} # FLAC/Ogg art
|
|
|
|
|
|
def _clean_label(text: str) -> str:
|
|
"""A short, printable tag name for logs.
|
|
|
|
ID3 ``PRIV``/``COMM``/``TXXX`` dict keys embed the frame's raw payload (which
|
|
can be arbitrary binary — Traktor blobs, embedded art, etc.). Strip anything
|
|
non-printable and truncate so the activity log stays readable.
|
|
"""
|
|
text = "".join(ch for ch in str(text) if ch.isprintable()).strip()
|
|
return (text[:45] + "…") if len(text) > 46 else text
|
|
|
|
|
|
def _id3_frame_label(key: str, frame) -> str:
|
|
"""Human-readable id for an ID3 frame, never including its binary data."""
|
|
fid = getattr(frame, "FrameID", None) or key.split(":", 1)[0]
|
|
if fid == "PRIV":
|
|
owner = getattr(frame, "owner", "") or ""
|
|
return _clean_label(f"PRIV:{owner}" if owner else "PRIV")
|
|
if fid in ("TXXX", "COMM", "USLT", "SYLT", "WXXX"):
|
|
desc = getattr(frame, "desc", "") or ""
|
|
return _clean_label(f"{fid}:{desc}" if desc else fid)
|
|
return _clean_label(fid)
|
|
|
|
|
|
def _strip_id3_junk(tags: ID3) -> list[str]:
|
|
removed: list[str] = []
|
|
for key in list(tags.keys()):
|
|
if key.startswith(_ID3_PROTECTED):
|
|
continue # never touch embedded artwork (APIC/PIC)
|
|
drop = key.startswith(_ID3_JUNK_PREFIXES)
|
|
if not drop and key.startswith("TXXX:"):
|
|
desc = key.split(":", 1)[1].lower()
|
|
drop = any(token in desc for token in _ID3_TXXX_JUNK)
|
|
if not drop and key.startswith("PRIV"):
|
|
drop = True # Windows Media / player breadcrumbs
|
|
if drop:
|
|
removed.append(_id3_frame_label(key, tags[key]))
|
|
del tags[key]
|
|
return removed
|
|
|
|
|
|
def _strip_vorbis_junk(tags) -> list[str]:
|
|
removed: list[str] = []
|
|
for key in list(tags.keys()):
|
|
low = key.lower()
|
|
if low in _VORBIS_PROTECTED:
|
|
continue # never touch embedded artwork
|
|
if low in _VORBIS_JUNK_KEYS or any(token in low for token in _VORBIS_JUNK_SUBSTRINGS):
|
|
del tags[key]
|
|
removed.append(_clean_label(key))
|
|
return removed
|
|
|
|
|
|
def _strip_mp4_junk(tags: MP4Tags) -> list[str]:
|
|
removed: list[str] = []
|
|
for key in list(tags.keys()):
|
|
if key in _MP4_PROTECTED:
|
|
continue # never touch the cover atom
|
|
drop = key in _MP4_JUNK_KEYS
|
|
if not drop and key.startswith("----"):
|
|
low = key.lower()
|
|
drop = any(token in low for token in ("itun", "purchase", "url", "www", "comment"))
|
|
if drop:
|
|
del tags[key]
|
|
removed.append(_clean_label(key))
|
|
return removed
|
|
|
|
|
|
# ── track-number normalization ────────────────────────────────────────────────
|
|
|
|
|
|
def _normalize_id3_tracks(tags: ID3) -> list[str]:
|
|
changes: list[str] = []
|
|
for frame_id, label in (("TRCK", "track"), ("TPOS", "disc")):
|
|
frame = tags.get(frame_id)
|
|
if frame is None:
|
|
continue
|
|
current = str(frame.text[0]) if frame.text else ""
|
|
normalized = clean_track_number(current)
|
|
if normalized and normalized != current:
|
|
frame.text = [normalized]
|
|
prefix = "" if label == "track" else f"{label} "
|
|
changes.append(f"{prefix}{current} → {normalized}")
|
|
return changes
|
|
|
|
|
|
def _normalize_vorbis_tracks(tags) -> list[str]:
|
|
changes: list[str] = []
|
|
for key, label in (("tracknumber", "track"), ("discnumber", "disc")):
|
|
values = tags.get(key)
|
|
if not values:
|
|
continue
|
|
current = str(values[0])
|
|
normalized = clean_track_number(current)
|
|
if normalized and normalized != current:
|
|
tags[key] = [normalized]
|
|
prefix = "" if label == "track" else f"{label} "
|
|
changes.append(f"{prefix}{current} → {normalized}")
|
|
return changes
|
|
|
|
|
|
# ── per-file processing ───────────────────────────────────────────────────────
|
|
|
|
|
|
def _set_genre(audio, tags, genre: str) -> bool:
|
|
"""Write ``genre`` across formats. Returns True if it changed."""
|
|
if isinstance(tags, ID3):
|
|
from mutagen.id3 import TCON
|
|
|
|
existing = tags.get("TCON")
|
|
if existing is not None and existing.text == [genre]:
|
|
return False
|
|
tags.setall("TCON", [TCON(encoding=3, text=[genre])])
|
|
return True
|
|
if isinstance(tags, MP4Tags):
|
|
if tags.get("\xa9gen") == [genre]:
|
|
return False
|
|
tags["\xa9gen"] = [genre]
|
|
return True
|
|
# Vorbis comment (FLAC/Ogg)
|
|
if list(tags.get("genre", [])) == [genre]:
|
|
return False
|
|
tags["genre"] = [genre]
|
|
return True
|
|
|
|
|
|
def _process_file(
|
|
path: Path,
|
|
opts: MetadataOptions,
|
|
genre: str | None,
|
|
rec: _Recorder,
|
|
*,
|
|
group: str,
|
|
subgroup: str,
|
|
) -> None:
|
|
try:
|
|
audio = MutagenFile(path)
|
|
except (MutagenError, OSError) as exc:
|
|
rec.emit("warn", "tag", f"Could not read {path.name}: {exc}")
|
|
return
|
|
if audio is None:
|
|
return
|
|
if audio.tags is None:
|
|
try:
|
|
audio.add_tags()
|
|
except (MutagenError, Exception):
|
|
rec.emit("skip", "tag", f"No tags and none could be added: {path.name}")
|
|
return
|
|
tags = audio.tags
|
|
|
|
removed: list[str] = []
|
|
track_changes: list[str] = []
|
|
genre_set = False
|
|
|
|
if opts.strip_junk:
|
|
if isinstance(tags, ID3):
|
|
removed = _strip_id3_junk(tags)
|
|
elif isinstance(tags, MP4Tags):
|
|
removed = _strip_mp4_junk(tags)
|
|
else:
|
|
removed = _strip_vorbis_junk(tags)
|
|
|
|
if opts.normalize_tracks:
|
|
if isinstance(tags, ID3):
|
|
track_changes = _normalize_id3_tracks(tags)
|
|
elif isinstance(tags, MP4Tags):
|
|
track_changes = [] # MP4 stores track as an integer tuple; nothing to pad
|
|
else:
|
|
track_changes = _normalize_vorbis_tracks(tags)
|
|
|
|
if opts.genres and genre and _set_genre(audio, tags, genre):
|
|
genre_set = True
|
|
|
|
if not (removed or track_changes or genre_set):
|
|
return
|
|
|
|
# Human-readable summary (also used by the collapsible raw log).
|
|
parts: list[str] = []
|
|
if removed:
|
|
parts.append(f"strip {len(removed)} junk tag(s): {', '.join(removed[:6])}")
|
|
if track_changes:
|
|
parts.append("; ".join(track_changes))
|
|
if genre_set:
|
|
parts.append(f"genre → {genre}")
|
|
|
|
rec.emit(
|
|
"dry" if opts.dry_run else "ok",
|
|
"tag",
|
|
f"{path.name}: " + " | ".join(parts),
|
|
path=str(path),
|
|
file=path.name,
|
|
group=group,
|
|
subgroup=subgroup,
|
|
junk=removed,
|
|
track="; ".join(track_changes) or None,
|
|
genre=genre if genre_set else None,
|
|
)
|
|
if opts.dry_run:
|
|
return
|
|
try:
|
|
audio.save()
|
|
except (MutagenError, OSError) as exc:
|
|
rec.emit("warn", "tag", f"Could not save {path.name}: {exc}")
|
|
|
|
|
|
def _process_album(
|
|
album_folder: Path,
|
|
opts: MetadataOptions,
|
|
rec: _Recorder,
|
|
genre_cache: dict[str, str | None],
|
|
overrides: dict[str, str],
|
|
) -> None:
|
|
audio_files = [
|
|
f for f in album_folder.iterdir()
|
|
if f.is_file() and f.suffix.lower() in AUDIO_EXTENSIONS
|
|
]
|
|
if not audio_files:
|
|
return
|
|
|
|
group = album_folder.name
|
|
subgroup = album_folder.parent.name
|
|
|
|
genre = None
|
|
if opts.genres:
|
|
artist, album, _year = get_album_metadata_from_files(album_folder)
|
|
if artist and album:
|
|
group = album
|
|
subgroup = artist
|
|
if artist:
|
|
genre = _resolve_artist_genre(artist, album, genre_cache, overrides, rec)
|
|
|
|
for file in audio_files:
|
|
_process_file(file, opts, genre, rec, group=group, subgroup=subgroup)
|
|
|
|
|
|
def _resolve_artist_genre(
|
|
artist: str,
|
|
album: str | None,
|
|
cache: dict[str, str | None],
|
|
overrides: dict[str, str],
|
|
rec: _Recorder,
|
|
) -> str | None:
|
|
"""One genre per artist, resolved once and cached for the whole run.
|
|
|
|
Priority: a user override (always wins, no lookup) → the artist's MusicBrainz
|
|
genre → one album lookup as fallback. The result is cached under the artist so
|
|
every later album by the same artist reuses it and the library stays consistent.
|
|
"""
|
|
key = artist.strip().lower()
|
|
if key in cache:
|
|
return cache[key]
|
|
|
|
override = overrides.get(key)
|
|
if override:
|
|
cache[key] = override
|
|
rec.emit("info", "genre", f"{artist} → {override} (override)")
|
|
return override
|
|
|
|
genre = find_artist_genre(artist)
|
|
if not genre and album:
|
|
genre = find_album_genre(artist, album)
|
|
cache[key] = genre
|
|
|
|
if genre:
|
|
rec.emit("info", "genre", f"{artist} → {genre}")
|
|
else:
|
|
rec.emit("skip", "genre", f"No MusicBrainz genre: {artist}")
|
|
return genre
|
|
|
|
|
|
def process_library(
|
|
options: MetadataOptions,
|
|
*,
|
|
root: Path | None = None,
|
|
log: LogCallback | None = None,
|
|
album_paths: list[str] | None = None,
|
|
) -> dict:
|
|
"""Run the selected tag-maintenance modes. Honours ``options.dry_run``."""
|
|
root = root or MUSIC_ROOT
|
|
rec = _Recorder(sink=log)
|
|
|
|
if not options.any_mode:
|
|
rec.emit("warn", "start", "No metadata modes selected.")
|
|
return {"root": str(root), "dry_run": options.dry_run, "actions": rec.actions, "counts": rec.counts}
|
|
|
|
if not root.exists():
|
|
rec.emit("warn", "root", f"Music root does not exist: {root}")
|
|
return {"root": str(root), "dry_run": options.dry_run, "actions": rec.actions, "counts": rec.counts}
|
|
|
|
if album_paths:
|
|
wanted = {str(Path(p)) for p in album_paths}
|
|
folders = [Path(p) for p in album_paths] if all(Path(p).exists() for p in album_paths) else [
|
|
f for f in _iter_album_folders(root) if str(f) in wanted
|
|
]
|
|
else:
|
|
folders = list(_iter_album_folders(root))
|
|
|
|
if options.recent_only and not album_paths:
|
|
before = len(folders)
|
|
folders = [f for f in folders if _was_created_recently(f, options.recent_window_seconds)]
|
|
rec.emit(
|
|
"info",
|
|
"recent",
|
|
f"Recent-only: {len(folders)} of {before} albums modified in the last "
|
|
f"{options.recent_window_seconds // 3600}h",
|
|
)
|
|
|
|
if options.genres and not _HAS_MUSICBRAINZ:
|
|
rec.emit("warn", "genre", "MusicBrainz library not available — genre lookups skipped.")
|
|
|
|
rec.emit(
|
|
"info",
|
|
"start",
|
|
f"{'Dry run' if options.dry_run else 'Applying'} tag maintenance across {len(folders)} album(s)",
|
|
)
|
|
genre_cache: dict[str, str | None] = {} # one genre per artist, for the whole run
|
|
overrides = _overrides_map() if options.genres else {}
|
|
for album_folder in folders:
|
|
try:
|
|
_process_album(album_folder, options, rec, genre_cache, overrides)
|
|
except OSError as exc:
|
|
rec.emit("warn", "album", f"Error processing {album_folder.name}: {exc}")
|
|
|
|
rec.emit("info", "done", "Finished")
|
|
return {
|
|
"root": str(root),
|
|
"dry_run": options.dry_run,
|
|
"actions": rec.actions,
|
|
"counts": rec.counts,
|
|
}
|
|
|
|
|
|
def process_library_stream(
|
|
options: MetadataOptions,
|
|
*,
|
|
root: Path | None = None,
|
|
album_paths: list[str] | None = None,
|
|
) -> Iterator[dict]:
|
|
"""Run tag maintenance and yield each action the moment it happens."""
|
|
events: queue.Queue = queue.Queue()
|
|
sentinel = object()
|
|
|
|
def worker():
|
|
try:
|
|
process_library(options, root=root, log=events.put, album_paths=album_paths)
|
|
except Exception as exc: # pragma: no cover - defensive
|
|
events.put({"level": "warn", "action": "error", "message": str(exc)})
|
|
finally:
|
|
events.put(sentinel)
|
|
|
|
threading.Thread(target=worker, name="music_metadata_stream", daemon=True).start()
|
|
while True:
|
|
item = events.get()
|
|
if item is sentinel:
|
|
break
|
|
yield item
|