534 lines
18 KiB
Python
534 lines
18 KiB
Python
"""Music library maintenance, refactored from the original ``music-covers.py`` CLI.
|
|
|
|
The interactive terminal UI is gone; what remains is pure, importable logic the
|
|
web app drives. Two entry points matter:
|
|
|
|
* :func:`scan_library` — read-only analysis. Walks ``MUSIC_ROOT`` and reports, per
|
|
album, what each maintenance mode *would* do. Safe to call any time.
|
|
* :func:`process_library` — performs the work. Honours ``dry_run`` (the web UI
|
|
default) so nothing is renamed, deleted, or downloaded unless the caller opts in.
|
|
|
|
Both are synchronous (filesystem + blocking HTTP); call them from FastAPI via
|
|
``asyncio.to_thread``. Progress is reported through an optional ``log`` callback.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import re
|
|
import time
|
|
from dataclasses import dataclass, field
|
|
from pathlib import Path
|
|
from typing import Callable
|
|
|
|
import requests
|
|
|
|
try: # Optional: only needed for the "find missing year/cover" online lookups.
|
|
import musicbrainzngs
|
|
|
|
musicbrainzngs.set_useragent("HomelabToolkit", "1.0", "homelab-toolkit@example.com")
|
|
_HAS_MUSICBRAINZ = True
|
|
except Exception: # pragma: no cover - optional dependency
|
|
_HAS_MUSICBRAINZ = False
|
|
|
|
from mutagen import File as MutagenFile, MutagenError
|
|
|
|
MUSIC_ROOT = Path(os.environ.get("MUSIC_ROOT", r"\\Matt-htpc\d\Music"))
|
|
|
|
COVER_NAME_PRIORITY = ("cover.jpg", "folder.jpg", "front.jpg")
|
|
COVER_NAMES = set(COVER_NAME_PRIORITY)
|
|
COVER_MISSING_MARKER = ".cover-not-found"
|
|
LYRICS_MISSING_MARKER = ".lyrics-not-found"
|
|
LYRICS_SIDECAR_EXTENSIONS = {".lrc", ".txt"}
|
|
AUDIO_EXTENSIONS = {".mp3", ".flac", ".m4a"}
|
|
YEAR_ALBUM_FOLDER_RE = re.compile(r"^\s*(\d{4})\s*-\s*(.+?)\s*$")
|
|
|
|
LogCallback = Callable[[dict], None]
|
|
|
|
|
|
@dataclass
|
|
class ProcessOptions:
|
|
folder_cleanup: bool = False
|
|
rename: bool = False
|
|
file_cleanup: bool = False
|
|
lyrics: bool = False
|
|
covers: bool = True
|
|
dry_run: bool = True
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: dict) -> "ProcessOptions":
|
|
return cls(
|
|
folder_cleanup=bool(data.get("folder_cleanup", False)),
|
|
rename=bool(data.get("rename", False)),
|
|
file_cleanup=bool(data.get("file_cleanup", False)),
|
|
lyrics=bool(data.get("lyrics", False)),
|
|
covers=bool(data.get("covers", True)),
|
|
dry_run=bool(data.get("dry_run", True)),
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class _Recorder:
|
|
"""Collects structured action records and forwards them to an optional sink."""
|
|
|
|
sink: LogCallback | None = None
|
|
actions: list[dict] = field(default_factory=list)
|
|
counts: dict[str, int] = field(default_factory=dict)
|
|
|
|
def emit(self, level: str, action: str, message: str, **extra) -> None:
|
|
record = {"level": level, "action": action, "message": message, **extra}
|
|
self.actions.append(record)
|
|
self.counts[action] = self.counts.get(action, 0) + 1
|
|
if self.sink:
|
|
self.sink(record)
|
|
|
|
|
|
# ── pure string helpers ──────────────────────────────────────────────────────
|
|
|
|
|
|
def clean_name(text: str) -> str:
|
|
text = re.sub(r"\[(.*?)\]|\((.*?)\)", "", text)
|
|
text = text.replace("_", " ").replace("-", " ")
|
|
return " ".join(text.split()).strip()
|
|
|
|
|
|
def clean_album_folder_name(text: str) -> str:
|
|
match = YEAR_ALBUM_FOLDER_RE.match(text)
|
|
if match:
|
|
text = match.group(2)
|
|
return clean_name(text)
|
|
|
|
|
|
def get_year_from_album_folder_name(text: str) -> str | None:
|
|
match = YEAR_ALBUM_FOLDER_RE.match(text)
|
|
return match.group(1) if match else None
|
|
|
|
|
|
def safe_filename(text: str) -> str:
|
|
text = re.sub(r'[<>:"/\\|?*]', "", text)
|
|
text = text.strip().rstrip(".")
|
|
return " ".join(text.split())
|
|
|
|
|
|
def clean_track_number(value) -> str | None:
|
|
if not value:
|
|
return None
|
|
text = str(value[0] if isinstance(value, list) else value).strip()
|
|
text = text.split("/")[0].strip()
|
|
return text.zfill(2) if text.isdigit() else None
|
|
|
|
|
|
def extract_year(value: str | None) -> str | None:
|
|
if not value:
|
|
return None
|
|
match = re.search(r"\b(19\d{2}|20\d{2})\b", str(value))
|
|
return match.group(1) if match else None
|
|
|
|
|
|
# ── metadata reading ─────────────────────────────────────────────────────────
|
|
|
|
|
|
def _load_audio(path: Path):
|
|
try:
|
|
return MutagenFile(path, easy=True)
|
|
except (MutagenError, OSError):
|
|
return None
|
|
|
|
|
|
def _first_tag(audio, names):
|
|
for name in names:
|
|
value = audio.get(name)
|
|
if value:
|
|
return str(value[0]).strip()
|
|
return None
|
|
|
|
|
|
def get_album_metadata_from_files(album_folder: Path):
|
|
for file in album_folder.iterdir():
|
|
if not file.is_file() or file.suffix.lower() not in AUDIO_EXTENSIONS:
|
|
continue
|
|
audio = _load_audio(file)
|
|
if audio is None:
|
|
continue
|
|
artist = _first_tag(audio, ["albumartist", "artist"])
|
|
album = _first_tag(audio, ["album"])
|
|
year = extract_year(_first_tag(audio, ["date", "originaldate", "year"]))
|
|
if artist or album or year:
|
|
return artist, album, year
|
|
return None, None, None
|
|
|
|
|
|
def get_track_metadata(path: Path):
|
|
audio = _load_audio(path)
|
|
if audio is None:
|
|
return None, None, None, None
|
|
artist = _first_tag(audio, ["artist", "albumartist"])
|
|
album = _first_tag(audio, ["album"])
|
|
title = _first_tag(audio, ["title"])
|
|
duration = None
|
|
info = getattr(audio, "info", None)
|
|
if info and getattr(info, "length", None):
|
|
duration = round(info.length)
|
|
return artist, album, title, duration
|
|
|
|
|
|
# ── album inspection ─────────────────────────────────────────────────────────
|
|
|
|
|
|
def _cover_to_keep(album_folder: Path) -> str | None:
|
|
existing = {p.name.lower() for p in album_folder.iterdir() if p.is_file()}
|
|
for name in COVER_NAME_PRIORITY:
|
|
if name in existing:
|
|
return name
|
|
return None
|
|
|
|
|
|
def _has_cover(album_folder: Path) -> bool:
|
|
existing = {p.name.lower() for p in album_folder.iterdir() if p.is_file()}
|
|
return any(name in existing for name in COVER_NAMES)
|
|
|
|
|
|
def _should_keep_file(path: Path, cover_to_keep: str | None) -> bool:
|
|
name = path.name.lower()
|
|
suffix = path.suffix.lower()
|
|
return (
|
|
suffix in AUDIO_EXTENSIONS
|
|
or suffix in LYRICS_SIDECAR_EXTENSIONS
|
|
or name == cover_to_keep
|
|
)
|
|
|
|
|
|
def _has_lyrics(audio_file: Path) -> bool:
|
|
return any(
|
|
audio_file.with_suffix(extension).exists()
|
|
for extension in LYRICS_SIDECAR_EXTENSIONS
|
|
)
|
|
|
|
|
|
def analyze_album(album_folder: Path) -> dict:
|
|
"""Read-only summary of an album folder and the pending maintenance work."""
|
|
tag_artist, tag_album, tag_year = get_album_metadata_from_files(album_folder)
|
|
artist = tag_artist or clean_name(album_folder.parent.name)
|
|
album = tag_album or clean_album_folder_name(album_folder.name)
|
|
year = tag_year or get_year_from_album_folder_name(album_folder.name)
|
|
|
|
audio_files = [
|
|
f for f in album_folder.iterdir()
|
|
if f.is_file() and f.suffix.lower() in AUDIO_EXTENSIONS
|
|
]
|
|
cover_to_keep = _cover_to_keep(album_folder)
|
|
extra_files = [
|
|
f.name for f in album_folder.iterdir()
|
|
if f.is_file() and not _should_keep_file(f, cover_to_keep)
|
|
]
|
|
|
|
suggested_folder = None
|
|
if album and year:
|
|
candidate = f"{year} - {safe_filename(album)}"
|
|
if candidate != album_folder.name:
|
|
suggested_folder = candidate
|
|
|
|
missing_lyrics = sum(1 for f in audio_files if not _has_lyrics(f))
|
|
|
|
return {
|
|
"path": str(album_folder),
|
|
"folder_name": album_folder.name,
|
|
"artist": artist or "",
|
|
"album": album or "",
|
|
"year": year,
|
|
"track_count": len(audio_files),
|
|
"has_cover": _has_cover(album_folder),
|
|
"suggested_folder": suggested_folder,
|
|
"needs_folder_rename": suggested_folder is not None,
|
|
"extra_files": extra_files,
|
|
"extra_file_count": len(extra_files),
|
|
"missing_lyrics_count": missing_lyrics,
|
|
}
|
|
|
|
|
|
def _iter_album_folders(root: Path):
|
|
for first_level in root.iterdir():
|
|
if not first_level.is_dir():
|
|
continue
|
|
try:
|
|
has_audio = any(
|
|
p.is_file() and p.suffix.lower() in AUDIO_EXTENSIONS
|
|
for p in first_level.iterdir()
|
|
)
|
|
except OSError:
|
|
continue
|
|
if has_audio:
|
|
# Top-level folder holding audio directly is not an Artist/Album tree.
|
|
continue
|
|
for album_folder in first_level.iterdir():
|
|
if album_folder.is_dir():
|
|
yield album_folder
|
|
|
|
|
|
def scan_library(root: Path | None = None) -> dict:
|
|
root = root or MUSIC_ROOT
|
|
if not root.exists():
|
|
return {"root": str(root), "exists": False, "albums": []}
|
|
|
|
albums = []
|
|
for album_folder in _iter_album_folders(root):
|
|
try:
|
|
albums.append(analyze_album(album_folder))
|
|
except OSError:
|
|
continue
|
|
albums.sort(key=lambda a: (a["artist"].lower(), a["year"] or "", a["album"].lower()))
|
|
|
|
return {
|
|
"root": str(root),
|
|
"exists": True,
|
|
"album_count": len(albums),
|
|
"missing_cover_count": sum(1 for a in albums if not a["has_cover"]),
|
|
"needs_rename_count": sum(1 for a in albums if a["needs_folder_rename"]),
|
|
"extra_file_count": sum(a["extra_file_count"] for a in albums),
|
|
"albums": albums,
|
|
}
|
|
|
|
|
|
# ── online lookups (year / cover / lyrics) ───────────────────────────────────
|
|
|
|
|
|
def find_album_year(artist: str, album: str) -> str | None:
|
|
if not _HAS_MUSICBRAINZ:
|
|
return None
|
|
try:
|
|
result = musicbrainzngs.search_releases(artist=artist, release=album, limit=5)
|
|
for release in result.get("release-list", []):
|
|
year = extract_year(release.get("date"))
|
|
if year:
|
|
return year
|
|
except Exception:
|
|
return None
|
|
return None
|
|
|
|
|
|
def find_album_cover(artist: str, album: str) -> bytes | None:
|
|
if not _HAS_MUSICBRAINZ:
|
|
return None
|
|
try:
|
|
result = musicbrainzngs.search_releases(artist=artist, release=album, limit=3)
|
|
releases = result.get("release-list", [])
|
|
if not releases:
|
|
return None
|
|
mbid = releases[0]["id"]
|
|
url = f"https://coverartarchive.org/release/{mbid}/front-500"
|
|
response = requests.get(url, timeout=20, allow_redirects=True)
|
|
if response.status_code == 200 and response.headers.get("content-type", "").startswith("image"):
|
|
return response.content
|
|
except Exception:
|
|
return None
|
|
return None
|
|
|
|
|
|
def find_track_lyrics(artist: str, album: str, title: str, duration: int | None):
|
|
if not duration:
|
|
return None
|
|
try:
|
|
response = requests.get(
|
|
"https://lrclib.net/api/get",
|
|
params={
|
|
"artist_name": artist,
|
|
"track_name": title,
|
|
"album_name": album,
|
|
"duration": duration,
|
|
},
|
|
headers={"User-Agent": "HomelabToolkit/1.0 (local music library tool)"},
|
|
timeout=20,
|
|
)
|
|
if response.status_code != 200:
|
|
return None
|
|
data = response.json()
|
|
if data.get("syncedLyrics"):
|
|
return ".lrc", data["syncedLyrics"].strip() + "\n"
|
|
if data.get("plainLyrics"):
|
|
return ".txt", data["plainLyrics"].strip() + "\n"
|
|
except Exception:
|
|
return None
|
|
return None
|
|
|
|
|
|
# ── mutating operations (respect dry_run) ────────────────────────────────────
|
|
|
|
|
|
def _is_top_level(folder: Path, root: Path) -> bool:
|
|
try:
|
|
return folder.resolve().parent == root.resolve()
|
|
except OSError:
|
|
return folder.parent == root
|
|
|
|
|
|
def _rename_album_folder(album_folder, artist, album, year, root, opts, rec) -> Path:
|
|
if not album or not year or _is_top_level(album_folder, root):
|
|
return album_folder
|
|
new_name = f"{year} - {safe_filename(album)}"
|
|
if album_folder.name == new_name:
|
|
return album_folder
|
|
new_folder = album_folder.with_name(new_name)
|
|
if new_folder.exists():
|
|
rec.emit("skip", "folder", f"Rename target already exists: {new_name}")
|
|
return album_folder
|
|
rec.emit(
|
|
"dry" if opts.dry_run else "ok",
|
|
"folder",
|
|
f"{album_folder.name} → {new_name}",
|
|
path=str(album_folder),
|
|
)
|
|
if opts.dry_run:
|
|
return album_folder
|
|
album_folder.rename(new_folder)
|
|
return new_folder
|
|
|
|
|
|
def _rename_track(path: Path, opts, rec) -> Path:
|
|
audio = _load_audio(path)
|
|
if audio is None:
|
|
return path
|
|
artist = _first_tag(audio, ["artist", "albumartist"])
|
|
album = _first_tag(audio, ["album"])
|
|
title = _first_tag(audio, ["title"])
|
|
track = clean_track_number(audio.get("tracknumber"))
|
|
if not (artist and album and title and track):
|
|
return path
|
|
new_name = f"{track} - {safe_filename(title)}{path.suffix.lower()}"
|
|
if path.name == new_name:
|
|
return path
|
|
new_path = path.with_name(new_name)
|
|
if new_path.exists():
|
|
rec.emit("skip", "rename", f"Target exists: {new_name}")
|
|
return path
|
|
rec.emit("dry" if opts.dry_run else "ok", "rename", f"{path.name} → {new_name}")
|
|
if opts.dry_run:
|
|
return path
|
|
path.rename(new_path)
|
|
return new_path
|
|
|
|
|
|
def _clean_files(album_folder: Path, opts, rec) -> None:
|
|
cover_to_keep = _cover_to_keep(album_folder)
|
|
for file in album_folder.iterdir():
|
|
if not file.is_file() or _should_keep_file(file, cover_to_keep):
|
|
continue
|
|
rec.emit("dry" if opts.dry_run else "ok", "remove", f"Remove {file.name}", path=str(file))
|
|
if opts.dry_run:
|
|
continue
|
|
try:
|
|
file.unlink()
|
|
except OSError as exc:
|
|
rec.emit("warn", "remove", f"Could not remove {file.name}: {exc}")
|
|
|
|
|
|
def _fetch_cover(album_folder, artist, album, opts, rec) -> None:
|
|
if _has_cover(album_folder):
|
|
return
|
|
if not (artist and album):
|
|
return
|
|
rec.emit("info", "cover", f"Looking up cover: {artist} - {album}")
|
|
image = find_album_cover(artist, album)
|
|
if not image:
|
|
rec.emit("skip", "cover", f"No cover found: {artist} - {album}")
|
|
return
|
|
output = album_folder / "cover.jpg"
|
|
rec.emit("dry" if opts.dry_run else "ok", "cover", f"Save cover.jpg for {album}")
|
|
if not opts.dry_run:
|
|
output.write_bytes(image)
|
|
time.sleep(1)
|
|
|
|
|
|
def _fetch_lyrics(audio_file, album_artist, album_name, opts, rec) -> None:
|
|
if _has_lyrics(audio_file):
|
|
return
|
|
t_artist, t_album, title, duration = get_track_metadata(audio_file)
|
|
artist = t_artist or album_artist
|
|
album = t_album or album_name
|
|
if not (artist and album and title):
|
|
return
|
|
lyrics = find_track_lyrics(artist, album, title, duration)
|
|
if not lyrics:
|
|
rec.emit("skip", "lyrics", f"No lyrics: {artist} - {title}")
|
|
return
|
|
extension, text = lyrics
|
|
output = audio_file.with_suffix(extension)
|
|
rec.emit("dry" if opts.dry_run else "ok", "lyrics", f"Save lyrics for {title}")
|
|
if not opts.dry_run:
|
|
output.write_text(text, encoding="utf-8")
|
|
time.sleep(1)
|
|
|
|
|
|
def _process_album(album_folder: Path, root: Path, opts: ProcessOptions, rec: _Recorder) -> None:
|
|
tag_artist, tag_album, tag_year = get_album_metadata_from_files(album_folder)
|
|
artist = tag_artist or clean_name(album_folder.parent.name)
|
|
album = tag_album or clean_album_folder_name(album_folder.name)
|
|
year = tag_year or get_year_from_album_folder_name(album_folder.name)
|
|
|
|
if opts.folder_cleanup and not year and artist and album:
|
|
year = find_album_year(artist, album)
|
|
if opts.folder_cleanup:
|
|
album_folder = _rename_album_folder(album_folder, artist, album, year, root, opts, rec)
|
|
|
|
audio_files = [
|
|
f for f in album_folder.iterdir()
|
|
if f.is_file() and f.suffix.lower() in AUDIO_EXTENSIONS
|
|
]
|
|
if opts.rename:
|
|
audio_files = [_rename_track(f, opts, rec) for f in audio_files]
|
|
|
|
if opts.lyrics:
|
|
for file in audio_files:
|
|
_fetch_lyrics(file, artist, album, opts, rec)
|
|
|
|
if opts.covers:
|
|
_fetch_cover(album_folder, artist, album, opts, rec)
|
|
|
|
if opts.file_cleanup:
|
|
_clean_files(album_folder, opts, rec)
|
|
|
|
|
|
def process_library(
|
|
options: ProcessOptions,
|
|
*,
|
|
root: Path | None = None,
|
|
log: LogCallback | None = None,
|
|
album_paths: list[str] | None = None,
|
|
) -> dict:
|
|
"""Run the selected maintenance modes. Honours ``options.dry_run``.
|
|
|
|
``album_paths`` optionally limits the run to specific album folders.
|
|
"""
|
|
root = root or MUSIC_ROOT
|
|
rec = _Recorder(sink=log)
|
|
|
|
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))
|
|
|
|
rec.emit(
|
|
"info",
|
|
"start",
|
|
f"{'Dry run' if options.dry_run else 'Applying'} across {len(folders)} album(s)",
|
|
)
|
|
for album_folder in folders:
|
|
try:
|
|
_process_album(album_folder, root, options, rec)
|
|
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,
|
|
}
|