"""Music-library scanning, MusicBrainz enrichment, and completeness logic. Design rules (per the feature spec): * Disk is only walked by an explicit background job, never on page render. * The UI reads exclusively from the database. * Jobs write progress to ``library_scan_runs`` so the UI can poll. * Manual user decisions (ignore / mark owned / mark missing) are never overwritten by a metadata refresh. """ from __future__ import annotations import json import logging import os import threading from pathlib import Path from . import db, musicbrainz from .music_covers import AUDIO_EXTENSIONS, MUSIC_ROOT, _first_tag, _load_audio, clean_name, clean_track_number, extract_year from .text_normalize import ( MISSING, OWNED, classify_release, is_various_artists, normalize_artist, normalize_title, ) logger = logging.getLogger("homelabtoolkit.music_library") _OWNED_STATUSES = (OWNED, "probably_owned") _BATCH_SIZE = 200 # Only one job of each kind runs at a time (single process). _running: set[str] = set() _running_lock = threading.Lock() # ── job guards ──────────────────────────────────────────────────────────────── def _try_acquire(kind: str) -> bool: with _running_lock: if kind in _running: return False _running.add(kind) return True def _release(kind: str) -> None: with _running_lock: _running.discard(kind) def _is_running(kind: str) -> bool: with _running_lock: return kind in _running # ── scan run bookkeeping ────────────────────────────────────────────────────── def _create_run(kind: str) -> int: with db.connect() as conn: cur = conn.execute( "INSERT INTO library_scan_runs(kind, status, started_at) VALUES(?, 'running', ?)", (kind, db.now_iso()), ) return cur.lastrowid def _update_run(run_id: int, **fields) -> None: if not fields: return cols = ", ".join(f"{k}=?" for k in fields) with db.connect() as conn: conn.execute(f"UPDATE library_scan_runs SET {cols} WHERE id=?", (*fields.values(), run_id)) def _finish_run(run_id: int, status: str, **fields) -> None: _update_run(run_id, status=status, completed_at=db.now_iso(), **fields) # ── tag reading ─────────────────────────────────────────────────────────────── def _read_tags(path: Path) -> dict | None: audio = _load_audio(path) if audio is None: return None album_artist = _first_tag(audio, ["albumartist", "album artist"]) track_artist = _first_tag(audio, ["artist", "albumartist"]) artist = album_artist or track_artist or clean_name(path.parent.parent.name) album = _first_tag(audio, ["album"]) or clean_name(path.parent.name) title = _first_tag(audio, ["title"]) or path.stem track_no = clean_track_number(audio.get("tracknumber")) disc_no = clean_track_number(audio.get("discnumber")) return { "artist": artist, "album": album, "title": title, "track_number": int(track_no) if track_no else None, "disc_number": int(disc_no) if disc_no else None, "year": int(extract_year(_first_tag(audio, ["date", "originaldate", "year"])) or 0), "artist_mbid": _first_tag(audio, ["musicbrainz_artistid"]), "album_mbid": _first_tag(audio, ["musicbrainz_releasegroupid", "musicbrainz_albumid"]), "track_mbid": _first_tag(audio, ["musicbrainz_trackid"]), } def _iter_audio_files(root: Path): """Yield audio file paths lazily so we never hold the library in memory.""" for dirpath, _dirnames, filenames in os.walk(root): for name in filenames: if Path(name).suffix.lower() in AUDIO_EXTENSIONS: yield Path(dirpath) / name # ── upserts (in-run caches keep artist/album lookups cheap) ─────────────────── def _get_artist_id(conn, cache: dict, meta: dict) -> int: name = meta["artist"] norm = normalize_artist(name) if norm in cache: return cache[norm] now = db.now_iso() various = 1 if is_various_artists(name) else 0 row = conn.execute("SELECT id FROM library_artists WHERE name_normalized=?", (norm,)).fetchone() if row: conn.execute( "UPDATE library_artists SET is_active=1, is_various=?, updated_at=?, mbid=COALESCE(mbid, ?) WHERE id=?", (various, now, meta.get("artist_mbid"), row["id"]), ) artist_id = row["id"] else: cur = conn.execute( "INSERT INTO library_artists(name, name_normalized, mbid, is_various, is_active, created_at, updated_at) " "VALUES(?,?,?,?,1,?,?)", (name, norm, meta.get("artist_mbid"), various, now, now), ) artist_id = cur.lastrowid cache[norm] = artist_id return artist_id def _get_album_id(conn, cache: dict, artist_id: int, meta: dict) -> int: title = meta["album"] norm = normalize_title(title) year = meta.get("year") or 0 key = (artist_id, norm, year) if key in cache: return cache[key] now = db.now_iso() row = conn.execute( "SELECT id FROM library_albums WHERE artist_id=? AND title_normalized=? AND year=?", (artist_id, norm, year), ).fetchone() if row: conn.execute( "UPDATE library_albums SET is_active=1, updated_at=?, mbid=COALESCE(mbid, ?) WHERE id=?", (now, meta.get("album_mbid"), row["id"]), ) album_id = row["id"] else: cur = conn.execute( "INSERT INTO library_albums(artist_id, title, title_normalized, year, mbid, is_active, created_at, updated_at) " "VALUES(?,?,?,?,?,1,?,?)", (artist_id, title, norm, year, meta.get("album_mbid"), now, now), ) album_id = cur.lastrowid cache[key] = album_id return album_id def _upsert_track(conn, album_id: int, meta: dict, path: Path, size: int, mtime: float, run_id: int) -> None: now = db.now_iso() conn.execute( """ INSERT INTO library_tracks(album_id, title, track_number, disc_number, file_path, file_mtime, file_size, mbid, is_active, last_seen_scan_id, created_at, updated_at) VALUES(?,?,?,?,?,?,?,?,1,?,?,?) ON CONFLICT(file_path) DO UPDATE SET album_id=excluded.album_id, title=excluded.title, track_number=excluded.track_number, disc_number=excluded.disc_number, file_mtime=excluded.file_mtime, file_size=excluded.file_size, mbid=excluded.mbid, is_active=1, last_seen_scan_id=excluded.last_seen_scan_id, updated_at=excluded.updated_at """, ( album_id, meta["title"], meta["track_number"], meta["disc_number"], str(path), mtime, size, meta.get("track_mbid"), run_id, now, now, ), ) # ── scanner ─────────────────────────────────────────────────────────────────── def run_scan(root: Path | None = None) -> dict: """Walk the library and upsert artists/albums/tracks. Unchanged files (same path + size + mtime) are skipped without reading tags. Files that vanished are marked inactive, never hard-deleted.""" root = Path(root) if root else MUSIC_ROOT run_id = _create_run("scan") logger.info("Scan %d started: %s", run_id, root) if not root.exists(): _finish_run(run_id, "failed", error_message=f"Music root not found: {root}") logger.warning("Scan %d aborted: root missing", run_id) return {"run_id": run_id, "status": "failed"} files_scanned = 0 try: with db.connect() as conn: artist_cache: dict = {} album_cache: dict = {} batch = 0 for path in _iter_audio_files(root): try: stat = path.stat() except OSError: continue size, mtime = stat.st_size, stat.st_mtime existing = conn.execute( "SELECT id, file_size, file_mtime FROM library_tracks WHERE file_path=?", (str(path),), ).fetchone() if existing and existing["file_size"] == size and abs((existing["file_mtime"] or 0) - mtime) < 1: conn.execute( "UPDATE library_tracks SET is_active=1, last_seen_scan_id=? WHERE id=?", (run_id, existing["id"]), ) else: meta = _read_tags(path) if meta is not None: artist_id = _get_artist_id(conn, artist_cache, meta) album_id = _get_album_id(conn, album_cache, artist_id, meta) _upsert_track(conn, album_id, meta, path, size, mtime, run_id) files_scanned += 1 batch += 1 if batch >= _BATCH_SIZE: conn.commit() batch = 0 conn.execute( "UPDATE library_scan_runs SET files_scanned=?, progress=? WHERE id=?", (files_scanned, f"Scanned {files_scanned} files", run_id), ) conn.commit() conn.commit() # Mark vanished files inactive, then cascade activity up. conn.execute( "UPDATE library_tracks SET is_active=0 WHERE COALESCE(last_seen_scan_id, -1) != ? AND is_active=1", (run_id,), ) conn.execute( "UPDATE library_albums SET is_active = " "CASE WHEN EXISTS(SELECT 1 FROM library_tracks t WHERE t.album_id=library_albums.id AND t.is_active=1) " "THEN 1 ELSE 0 END" ) conn.execute( "UPDATE library_artists SET is_active = " "CASE WHEN EXISTS(SELECT 1 FROM library_albums al WHERE al.artist_id=library_artists.id AND al.is_active=1) " "THEN 1 ELSE 0 END" ) artists_found = conn.execute("SELECT COUNT(*) c FROM library_artists WHERE is_active=1").fetchone()["c"] albums_found = conn.execute("SELECT COUNT(*) c FROM library_albums WHERE is_active=1").fetchone()["c"] conn.commit() _finish_run( run_id, "completed", files_scanned=files_scanned, albums_found=albums_found, artists_found=artists_found, progress="Done", ) logger.info("Scan %d completed: %d files, %d artists, %d albums", run_id, files_scanned, artists_found, albums_found) return {"run_id": run_id, "status": "completed", "files_scanned": files_scanned} except Exception as exc: # pragma: no cover - defensive logger.exception("Scan %d failed", run_id) _finish_run(run_id, "failed", files_scanned=files_scanned, error_message=str(exc)) return {"run_id": run_id, "status": "failed", "error": str(exc)} # ── completeness ────────────────────────────────────────────────────────────── def _local_albums_for_artist(conn, artist_id: int) -> list[dict]: rows = conn.execute( "SELECT id, title, title_normalized, year, mbid FROM library_albums WHERE artist_id=? AND is_active=1", (artist_id,), ).fetchall() return [dict(r) for r in rows] def recompute_completeness(conn, artist_id: int) -> None: """Rebuild completeness rows for one artist from stored external releases and local albums. Manual decisions (manual_override=1) are preserved.""" local_albums = _local_albums_for_artist(conn, artist_id) releases = conn.execute( "SELECT mb_release_group_mbid, title, first_release_year, primary_type, secondary_types " "FROM external_releases WHERE artist_id=?", (artist_id,), ).fetchall() existing = { r["release_group_mbid"]: dict(r) for r in conn.execute( "SELECT release_group_mbid, manual_override FROM collection_completeness WHERE artist_id=?", (artist_id,), ).fetchall() } qualifying_mbids: list[str] = [] now = db.now_iso() for rel in releases: secondary = json.loads(rel["secondary_types"] or "[]") if not musicbrainz.is_official_album(rel["primary_type"], secondary): continue mbid = rel["mb_release_group_mbid"] qualifying_mbids.append(mbid) prior = existing.get(mbid) if prior and prior["manual_override"]: # Keep the user's decision; only refresh descriptive fields. conn.execute( "UPDATE collection_completeness SET title=?, year=?, source='musicbrainz', updated_at=? " "WHERE artist_id=? AND release_group_mbid=?", (rel["title"], rel["first_release_year"] or 0, now, artist_id, mbid), ) continue status, confidence, reason, local_id = classify_release( rel["title"], rel["first_release_year"], mbid, local_albums ) conn.execute( """ INSERT INTO collection_completeness(artist_id, release_group_mbid, local_album_id, title, year, status, confidence, reason, source, manual_override, updated_at) VALUES(?,?,?,?,?,?,?,?, 'musicbrainz', 0, ?) ON CONFLICT(artist_id, release_group_mbid) DO UPDATE SET local_album_id=excluded.local_album_id, title=excluded.title, year=excluded.year, status=excluded.status, confidence=excluded.confidence, reason=excluded.reason, source='musicbrainz', updated_at=excluded.updated_at WHERE collection_completeness.manual_override=0 """, (artist_id, mbid, local_id, rel["title"], rel["first_release_year"] or 0, status, confidence, reason, now), ) # Drop non-manual rows that are no longer qualifying (e.g. filter changes). placeholders = ",".join("?" for _ in qualifying_mbids) or "''" conn.execute( f"DELETE FROM collection_completeness WHERE artist_id=? AND manual_override=0 " f"AND release_group_mbid NOT IN ({placeholders})", (artist_id, *qualifying_mbids), ) # ── metadata refresh job (per-artist MusicBrainz lookups) ───────────────────── def run_metadata_refresh() -> dict: run_id = _create_run("metadata") logger.info("Metadata refresh %d started", run_id) processed = 0 try: with db.connect() as conn: artists = conn.execute( "SELECT id, name, mbid, is_various FROM library_artists WHERE is_active=1 ORDER BY name" ).fetchall() total = len(artists) for artist in artists: artist_id = artist["id"] if artist["is_various"]: with db.connect() as conn: conn.execute( "INSERT INTO external_artist_matches(artist_id, status, checked_at) VALUES(?, 'skipped', ?) " "ON CONFLICT(artist_id) DO UPDATE SET status='skipped', checked_at=excluded.checked_at", (artist_id, db.now_iso()), ) processed += 1 continue match = musicbrainz.search_artist(artist["name"]) with db.connect() as conn: if not match or not match.get("mbid"): conn.execute( "INSERT INTO external_artist_matches(artist_id, status, checked_at) VALUES(?, 'not_found', ?) " "ON CONFLICT(artist_id) DO UPDATE SET status='not_found', checked_at=excluded.checked_at", (artist_id, db.now_iso()), ) processed += 1 _bump_metadata_progress(run_id, processed, total) continue conn.execute( "INSERT INTO external_artist_matches(artist_id, mb_artist_mbid, mb_artist_name, confidence, status, checked_at) " "VALUES(?,?,?,?, 'matched', ?) " "ON CONFLICT(artist_id) DO UPDATE SET mb_artist_mbid=excluded.mb_artist_mbid, " "mb_artist_name=excluded.mb_artist_name, confidence=excluded.confidence, status='matched', checked_at=excluded.checked_at", (artist_id, match["mbid"], match["name"], match["confidence"], db.now_iso()), ) release_groups = musicbrainz.fetch_release_groups(match["mbid"]) with db.connect() as conn: for rg in release_groups: if not rg.get("mbid"): continue conn.execute( """ INSERT INTO external_releases(artist_id, mb_release_group_mbid, title, title_normalized, first_release_year, primary_type, secondary_types, fetched_at) VALUES(?,?,?,?,?,?,?,?) ON CONFLICT(artist_id, mb_release_group_mbid) DO UPDATE SET title=excluded.title, title_normalized=excluded.title_normalized, first_release_year=excluded.first_release_year, primary_type=excluded.primary_type, secondary_types=excluded.secondary_types, fetched_at=excluded.fetched_at """, ( artist_id, rg["mbid"], rg["title"], normalize_title(rg["title"]), rg["first_release_year"], rg["primary_type"], json.dumps(rg["secondary_types"]), db.now_iso(), ), ) recompute_completeness(conn, artist_id) processed += 1 _bump_metadata_progress(run_id, processed, total) _finish_run(run_id, "completed", artists_found=processed, progress=f"Checked {processed} artists") logger.info("Metadata refresh %d completed: %d artists", run_id, processed) return {"run_id": run_id, "status": "completed", "artists": processed} except Exception as exc: # pragma: no cover - defensive logger.exception("Metadata refresh %d failed", run_id) _finish_run(run_id, "failed", error_message=str(exc)) return {"run_id": run_id, "status": "failed", "error": str(exc)} def _bump_metadata_progress(run_id: int, processed: int, total: int) -> None: _update_run(run_id, artists_found=processed, progress=f"Checked {processed}/{total} artists") # ── job launchers ───────────────────────────────────────────────────────────── def start_scan_job(root: Path | None = None) -> dict: if not _try_acquire("scan"): return {"started": False, "reason": "A scan is already running."} def _worker(): try: run_scan(root) finally: _release("scan") threading.Thread(target=_worker, name="scan_music_collection", daemon=True).start() return {"started": True} def start_metadata_job() -> dict: if not _try_acquire("metadata"): return {"started": False, "reason": "A metadata refresh is already running."} def _worker(): try: run_metadata_refresh() finally: _release("metadata") threading.Thread(target=_worker, name="refresh_music_metadata", daemon=True).start() return {"started": True} # ── read-side queries (UI; database only) ───────────────────────────────────── def _latest_run(conn, kind: str) -> dict | None: row = conn.execute( "SELECT * FROM library_scan_runs WHERE kind=? ORDER BY id DESC LIMIT 1", (kind,) ).fetchone() return dict(row) if row else None def get_status() -> dict: with db.connect() as conn: scan = _latest_run(conn, "scan") metadata = _latest_run(conn, "metadata") return { "scan": scan, "metadata": metadata, "scan_running": _is_running("scan"), "metadata_running": _is_running("metadata"), } def get_overview() -> dict: with db.connect() as conn: scan = _latest_run(conn, "scan") metadata = _latest_run(conn, "metadata") totals = conn.execute( """ SELECT SUM(CASE WHEN status IN ('owned','probably_owned') THEN 1 ELSE 0 END) AS owned, SUM(CASE WHEN status='missing' THEN 1 ELSE 0 END) AS missing, SUM(CASE WHEN status='uncertain' THEN 1 ELSE 0 END) AS uncertain, SUM(CASE WHEN status='ignored' THEN 1 ELSE 0 END) AS ignored FROM collection_completeness c JOIN library_artists a ON a.id=c.artist_id AND a.is_active=1 """ ).fetchone() artist_count = conn.execute("SELECT COUNT(*) c FROM library_artists WHERE is_active=1").fetchone()["c"] album_count = conn.execute("SELECT COUNT(*) c FROM library_albums WHERE is_active=1").fetchone()["c"] owned = totals["owned"] or 0 missing = totals["missing"] or 0 uncertain = totals["uncertain"] or 0 ignored = totals["ignored"] or 0 denom = owned + missing + uncertain completeness = round(owned / denom * 100, 1) if denom else 0.0 return { "last_scan": scan, "last_metadata": metadata, "scan_running": _is_running("scan"), "metadata_running": _is_running("metadata"), "owned": owned, "missing": missing, "uncertain": uncertain, "ignored": ignored, "completeness": completeness, "library_artists": artist_count, "library_albums": album_count, } def get_artists_completeness(search: str = "") -> list[dict]: where = "WHERE a.is_active=1" params: list = [] if search.strip(): where += " AND a.name LIKE ?" params.append(f"%{search.strip()}%") with db.connect() as conn: rows = conn.execute( f""" SELECT a.id, a.name, SUM(CASE WHEN c.status IN ('owned','probably_owned') THEN 1 ELSE 0 END) AS owned, SUM(CASE WHEN c.status='missing' THEN 1 ELSE 0 END) AS missing, SUM(CASE WHEN c.status='uncertain' THEN 1 ELSE 0 END) AS uncertain, SUM(CASE WHEN c.status='ignored' THEN 1 ELSE 0 END) AS ignored, COUNT(c.id) AS total FROM library_artists a JOIN collection_completeness c ON c.artist_id=a.id {where} GROUP BY a.id HAVING total > 0 ORDER BY missing DESC, a.name COLLATE NOCASE """, params, ).fetchall() result = [] for r in rows: owned, missing, uncertain = r["owned"] or 0, r["missing"] or 0, r["uncertain"] or 0 denom = owned + missing + uncertain result.append( { "id": r["id"], "name": r["name"], "owned": owned, "missing": missing, "uncertain": uncertain, "ignored": r["ignored"] or 0, "completeness": round(owned / denom * 100, 1) if denom else 0.0, } ) return result def get_artist_albums(artist_id: int) -> dict: with db.connect() as conn: artist = conn.execute("SELECT id, name FROM library_artists WHERE id=?", (artist_id,)).fetchone() rows = conn.execute( "SELECT id, release_group_mbid, title, year, status, confidence, reason, source, manual_override " "FROM collection_completeness WHERE artist_id=? ORDER BY year, title COLLATE NOCASE", (artist_id,), ).fetchall() return { "artist": dict(artist) if artist else None, "albums": [dict(r) for r in rows], } def set_album_decision(completeness_id: int, action: str) -> dict: action = (action or "").lower() valid = {"ignore", "owned", "missing", "reset"} if action not in valid: raise ValueError(f"Unknown action: {action}") now = db.now_iso() with db.connect() as conn: row = conn.execute( "SELECT id, artist_id FROM collection_completeness WHERE id=?", (completeness_id,) ).fetchone() if not row: raise LookupError("Completeness row not found.") if action == "ignore": conn.execute( "UPDATE collection_completeness SET status='ignored', manual_override=1, reason='Manually ignored', updated_at=? WHERE id=?", (now, completeness_id), ) elif action == "owned": conn.execute( "UPDATE collection_completeness SET status='owned', confidence=1.0, manual_override=1, reason='Manually marked owned', updated_at=? WHERE id=?", (now, completeness_id), ) elif action == "missing": conn.execute( "UPDATE collection_completeness SET status='missing', confidence=0, manual_override=1, reason='Manually marked missing', updated_at=? WHERE id=?", (now, completeness_id), ) elif action == "reset": conn.execute( "UPDATE collection_completeness SET manual_override=0, updated_at=? WHERE id=?", (now, completeness_id), ) recompute_completeness(conn, row["artist_id"]) return {"status": "ok"}