danvics/soulsync
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1

Quality Upgrade: tiered structured matching (ISRC -> album->track -> artist+title)

Browse source 
Replaces the blind fuzzy search with a smart hierarchy that uses the data we
already have, best identity first:

1. ISRC embedded in the file tags (enriched track) -> exact track.
2. Album -> track: use the album's stored source ID (albums.spotify_album_id /
   itunes_album_id / deezer_id / musicbrainz_release_id / audiodb_id) when the
   ALBUM is enriched (even if the track isn't); else find the album by searching
   'artist album', then locate our track in that album's tracklist by normalized
   title (track_number breaks ties). Pins the exact album context. (artist->album->track)
3. Plain artist+title search with similarity scoring. (artist->track) — loosest.

_load_tracks now returns dict rows (adds track_number + the album source-id
columns). Findings record matched_via = isrc | album | search. All clients
(spotify/deezer/itunes/discogs) expose search_albums + get_album_tracks with a
uniform {'items': [...]} shape, so the album tier is source-agnostic.

26 repair tests pass (added album-tier + _find_track_in_album coverage).
This commit is contained in:
BoulderBadgeDad 2026-06-13 13:00:16 -07:00
parent 3ea5b5181f
commit 777781db6a
2 changed files with 186 additions and 15 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

161
core/repair_jobs/quality_upgrade.py
View file

@ -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,

40
tests/repair_jobs/test_quality_upgrade.py
View file

@ -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)]