diff --git a/core/repair_jobs/quality_upgrade.py b/core/repair_jobs/quality_upgrade.py index 993bfad9..8f98f0fa 100644 --- a/core/repair_jobs/quality_upgrade.py +++ b/core/repair_jobs/quality_upgrade.py @@ -215,6 +215,109 @@ def _match_via_isrc(isrc: str, source_priority: List[str]) -> Tuple[Optional[Any return None, None +# Column order for the _load_tracks SELECT — rows come back as dicts keyed by these. +_TRACK_COLS = ( + 'id', 'title', 'file_path', 'bitrate', 'artist_name', 'album_title', 'album_id', + 'track_number', 'spotify_album_id', 'itunes_album_id', 'deezer_id', + 'musicbrainz_release_id', 'audiodb_id', +) + +# Human-readable note per match tier (search uses a confidence % instead). +_MATCH_NOTE = {'isrc': 'exact ISRC match', 'album': 'matched within album'} + +# Per-source column holding that source's album ID on the albums table. +_SOURCE_ALBUM_ID_COL = { + 'spotify': 'spotify_album_id', + 'itunes': 'itunes_album_id', + 'deezer': 'deezer_id', + 'musicbrainz': 'musicbrainz_release_id', + 'audiodb': 'audiodb_id', +} + + +def _norm_title(value: Any) -> str: + """Collapse a title to alphanumerics for tolerant comparison.""" + return ''.join(ch for ch in str(value or '').lower() if ch.isalnum()) + + +def _find_track_in_album(items: Any, title: str, track_number: Any, engine: Any) -> Optional[Any]: + """Pick the track in an album's tracklist that matches ours — exact normalized + title first (track_number breaks ties), then a high-similarity fuzzy fallback.""" + want = _norm_title(title) + exact = [] + best, best_score = None, 0.0 + for it in items or []: + it_name = _extract_lookup_value(it, 'name', 'title', default='') + if want and _norm_title(it_name) == want: + exact.append(it) + continue + if engine and it_name: + score = engine.similarity_score( + engine.normalize_string(title), engine.normalize_string(it_name)) + if score > best_score and score >= 0.85: + best, best_score = it, score + if exact: + if track_number: + for it in exact: + if _extract_lookup_value(it, 'track_number') == track_number: + return it + return exact[0] + return best + + +def _match_via_album(engine: Any, source_priority: List[str], artist: str, album_title: str, + title: str, track_number: Any, + stored_album_ids: Dict[str, str]) -> Tuple[Optional[Any], Optional[str]]: + """Structured artist → album → track match. For each source: use the album's + stored source ID if we already have it (enriched album), else find the album + by searching ``artist album``; then pull that album's tracklist and locate our + track in it. This pins the right album (exact context) without needing the + track itself to be enriched. Returns (track, source) or (None, None).""" + if not album_title: + return None, None + for source in source_priority: + client = get_client_for_source(source) + if not client or not hasattr(client, 'get_album_tracks'): + continue + + album_id = stored_album_ids.get(source) + album_name = album_title + if not album_id and hasattr(client, 'search_albums'): + try: + albums = client.search_albums(f'{artist} {album_title}'.strip(), limit=5) + except Exception: + albums = [] + best_alb, best_s = None, 0.0 + for alb in albums or []: + aname = _extract_lookup_value(alb, 'name', 'title', default='') + s = engine.similarity_score( + engine.normalize_string(album_title), engine.normalize_string(aname)) + if s > best_s and s >= 0.80: + best_alb, best_s = alb, s + if best_alb is not None: + album_id = _extract_lookup_value(best_alb, 'id') + album_name = _extract_lookup_value(best_alb, 'name', 'title', default=album_title) + if not album_id: + continue + + try: + resp = client.get_album_tracks(str(album_id)) + except Exception: + resp = None + items = resp.get('items') if isinstance(resp, dict) else None + match = _find_track_in_album(items, title, track_number, engine) + if match is None: + continue + # The album tracklist's tracks usually omit the album object — attach it so + # the wishlist add carries the correct album context. + if isinstance(match, dict): + alb = match.get('album') + if not isinstance(alb, dict) or not alb.get('name'): + match['album'] = {'name': album_name, 'images': []} + return match, source + return None, None + + def _find_best_match(engine: Any, source_priority: List[str], title: str, artist: str, album: str, min_confidence: float) -> Tuple[Optional[Any], float, Optional[str], bool]: """Search the configured metadata sources for the best replacement match. @@ -297,12 +400,14 @@ class QualityUpgradeJob(RepairJob): min_conf = 0.7 return {'scope': scope, 'min_confidence': min_conf} - def _load_tracks(self, db: Any, scope: str) -> List[tuple]: + def _load_tracks(self, db: Any, scope: str) -> List[dict]: conn = db._get_connection() try: base = ( "SELECT t.id, t.title, t.file_path, t.bitrate, a.name AS artist_name, " - "al.title AS album_title, t.album_id " + "al.title AS album_title, t.album_id, t.track_number, " + "al.spotify_album_id, al.itunes_album_id, al.deezer_id, " + "al.musicbrainz_release_id, al.audiodb_id " "FROM tracks t " "JOIN artists a ON t.artist_id = a.id " "JOIN albums al ON t.album_id = al.id " @@ -319,7 +424,7 @@ class QualityUpgradeJob(RepairJob): base + f" AND a.name IN ({placeholders})", names).fetchall() else: rows = conn.execute(base).fetchall() - return rows + return [dict(zip(_TRACK_COLS, r, strict=False)) for r in rows] finally: conn.close() @@ -365,8 +470,17 @@ class QualityUpgradeJob(RepairJob): if i % 10 == 0 and context.wait_if_paused(): return result - track_id, title, file_path, bitrate, artist_name, album_title, album_id = ( - row[0], row[1], row[2], row[3], row[4], row[5], row[6]) + track_id = row['id'] + title = row['title'] + file_path = row['file_path'] + bitrate = row['bitrate'] + artist_name = row['artist_name'] + album_title = row['album_title'] + album_id = row['album_id'] + track_number = row.get('track_number') + stored_album_ids = { + src: row[col] for src, col in _SOURCE_ALBUM_ID_COL.items() if row.get(col) + } result.scanned += 1 if meets_preferred_quality(file_path, bitrate, quality_profile): @@ -396,14 +510,31 @@ class QualityUpgradeJob(RepairJob): log_line=f'Low quality ({current_label}): {artist_name} - {title}', log_type='info') - # Fast path: enrichment embeds the ISRC (and per-source track IDs) in - # the file's tags, so an already-enriched track carries its exact - # identity. Resolve the EXACT track by ISRC — no fuzzy matching, and - # the real album comes with it. Only fall back to name/artist search - # for tracks that were never enriched / have no usable ISRC. + # Tiered match, best identity first, loosest last: + # 1. ISRC embedded in the file tags (enriched track) → EXACT track. + # 2. Album → track: use the album's stored source ID if we have it + # (enriched album), else find the album by search, then locate our + # track in its tracklist. Pins the right album even when the track + # itself isn't enriched. (artist → album → track) + # 3. Plain artist+title search with similarity scoring. (artist → track) + best, source, conf, attempted = None, None, 0.0, False + matched_via = 'isrc' best, source = _match_via_isrc(_read_track_isrc(file_path), source_priority) - conf, attempted = (1.0, True) if best else (0.0, False) + if best: + conf, attempted = 1.0, True + + if not best: + matched_via = 'album' + try: + best, source = _match_via_album( + engine, source_priority, artist_name or '', album_title or '', + title, track_number, stored_album_ids) + except Exception as e: + logger.debug("[Quality Upgrade] Album match error for %s - %s: %s", artist_name, title, e) + best = None + if best: + conf, attempted = 1.0, True if not best: matched_via = 'search' @@ -415,10 +546,10 @@ class QualityUpgradeJob(RepairJob): result.errors += 1 continue - if not attempted: - logger.warning("[Quality Upgrade] No metadata provider responded — stopping") - return result if not best: + if matched_via == 'search' and not attempted: + logger.warning("[Quality Upgrade] No metadata provider responded — stopping") + return result result.skipped += 1 continue @@ -442,7 +573,7 @@ class QualityUpgradeJob(RepairJob): description=( f'"{title}" by {artist_name} is {current_label}, below your preferred ' f'quality. Best match: "{_track_name(best)}" via {source} ' - f'({"exact ISRC match" if matched_via == "isrc" else f"confidence {conf:.0%}"}). ' + f'({_MATCH_NOTE.get(matched_via, "matched") if matched_via != "search" else f"confidence {conf:.0%}"}). ' 'Apply to add it to the wishlist.'), details={ 'track_id': track_id, diff --git a/tests/repair_jobs/test_quality_upgrade.py b/tests/repair_jobs/test_quality_upgrade.py index 0640f843..f4d8ab4c 100644 --- a/tests/repair_jobs/test_quality_upgrade.py +++ b/tests/repair_jobs/test_quality_upgrade.py @@ -153,6 +153,9 @@ def test_scan_creates_finding_for_low_quality_track(monkeypatch): ) fake_match = {'id': 'sp1', 'name': 'Song One', 'artists': ['Artist A'], 'album': {'name': 'Album X', 'images': []}} + # No ISRC / album hit → exercise the search tier. + monkeypatch.setattr(qu, '_read_track_isrc', lambda fp: '') + monkeypatch.setattr(qu, '_match_via_album', lambda *a, **k: (None, None)) monkeypatch.setattr(qu, '_find_best_match', lambda *a, **k: (fake_match, 0.95, 'spotify', True)) monkeypatch.setattr(qu, '_normalize_track_match', lambda track, src: dict(fake_match)) @@ -230,6 +233,7 @@ def test_scan_falls_back_to_search_without_isrc(monkeypatch): monkeypatch.setattr(qu, 'get_source_priority', lambda src: ['spotify']) monkeypatch.setattr('core.matching_engine.MusicMatchingEngine', lambda: types.SimpleNamespace()) monkeypatch.setattr(qu, '_read_track_isrc', lambda fp: '') # un-enriched + monkeypatch.setattr(qu, '_match_via_album', lambda *a, **k: (None, None)) # no album hit fake = {'id': 'sp1', 'name': 'Song One', 'artists': ['Artist A'], 'album': {'name': 'Album X'}} monkeypatch.setattr(qu, '_find_best_match', lambda *a, **k: (fake, 0.88, 'spotify', True)) monkeypatch.setattr(qu, '_normalize_track_match', lambda t, s: dict(fake)) @@ -241,6 +245,42 @@ def test_scan_falls_back_to_search_without_isrc(monkeypatch): assert findings[0]['details']['matched_via'] == 'search' +def test_scan_uses_album_tier_when_no_isrc(monkeypatch): + """No ISRC, but the album→track lookup resolves it → matched_via 'album', + and the fuzzy search is never reached.""" + rows = [(1, 'Song One', '/music/a.mp3', 128, 'Artist A', 'Album X', 10)] + db = _FakeDB(rows, BALANCED) + monkeypatch.setattr(qu, 'get_primary_source', lambda: 'spotify') + monkeypatch.setattr(qu, 'get_source_priority', lambda src: ['spotify']) + monkeypatch.setattr('core.matching_engine.MusicMatchingEngine', lambda: types.SimpleNamespace()) + monkeypatch.setattr(qu, '_read_track_isrc', lambda fp: '') + fake = {'id': 'sp1', 'name': 'Song One', 'artists': ['Artist A'], 'album': {'name': 'Album X'}} + monkeypatch.setattr(qu, '_match_via_album', lambda *a, **k: (fake, 'spotify')) + monkeypatch.setattr(qu, '_normalize_track_match', lambda t, s: dict(fake)) + monkeypatch.setattr(qu, '_track_name', lambda t: 'Song One') + + def _boom(*a, **k): + raise AssertionError("fuzzy search must not run when the album tier matches") + monkeypatch.setattr(qu, '_find_best_match', _boom) + + findings = [] + result = qu.QualityUpgradeJob().scan(_ctx(db, findings)) + assert result.findings_created == 1 + assert findings[0]['details']['matched_via'] == 'album' + assert findings[0]['details']['match_confidence'] == 1.0 + + +def test_find_track_in_album_exact_title_with_track_number(monkeypatch): + items = [ + {'id': 'a', 'name': 'Intro', 'track_number': 1}, + {'id': 'b', 'name': 'Karma Police', 'track_number': 6}, + {'id': 'c', 'name': 'Karma Police (Live)', 'track_number': 12}, + ] + eng = types.SimpleNamespace(similarity_score=lambda a, b: 0.0, normalize_string=lambda s: s) + got = qu._find_track_in_album(items, 'Karma Police', 6, eng) + assert got['id'] == 'b' + + def test_scan_skips_tracks_meeting_quality(monkeypatch): # A 320 kbps MP3 meets the balanced profile → no finding, no metadata calls. rows = [(2, 'Good Song', '/music/b.mp3', 320, 'Artist A', 'Album Y', 11)]