#868: disambiguate same-name artists by owned-catalog overlap during enrichment
Enrichment matched artists by NAME ONLY (0.85 gate), so for a common name
('Rone' has ~5 artists) it stored whichever the source ranked first — often the
wrong one, which then drove a wrong/sparse library 'Standard' discography while
'Enhanced' (the real owned albums) showed the full set.
Fix — use the decisive signal the library already has (the albums you OWN):
- worker_utils: pick_artist_by_catalog() + catalog_overlap_score() +
owned_album_titles()/release_titles(). When 2+ candidates clear the name gate,
fetch each one's catalog and choose the one overlapping the owned albums; falls
back to the current best-by-name pick when there's nothing to disambiguate or
no overlap (so the common single-candidate path makes no extra API calls).
- Wired into Spotify (covers Spotify-Free, same client), iTunes, Deezer (now
multi-candidate search_artists + get_artist_info store), and MusicBrainz
(match_artist gains owned_titles; release-groups as the catalog).
Re-match path (#868):
- build_reset_query now also clears the stored source-ID column for artist/album
item resets — previously a 're-match' only nulled match_status, so the worker's
existing-id short-circuit re-confirmed the WRONG id and never re-resolved. Tracks
excluded (ids live in tags, not a column).
- MusicBrainz also self-corrects its 90-day name->mbid cache: match_artist bypasses
a cached mbid whose catalog has ZERO overlap with the owned albums, so a re-match
isn't blocked by a stale wrong cache entry.
Tests: shared selector (9), per-worker disambiguation for all 4 sources + MB
backward-compat + MB cache-revalidation (8), reset-clears-id (2). 99 worker/
enrichment tests green.
This commit is contained in:
BoulderBadgeDad
parent
030d9bf9ff
commit
177a4d8d05
10 changed files with 549 additions and 46 deletions
|
|
@ -8,7 +8,15 @@ from datetime import datetime, timedelta
|
|||
from utils.logging_config import get_logger
|
||||
from database.music_database import MusicDatabase
|
||||
from core.deezer_client import DeezerClient
|
||||
from core.worker_utils import accept_artist_match, interruptible_sleep, set_album_api_track_count
|
||||
from core.worker_utils import (
|
||||
accept_artist_match,
|
||||
artist_name_matches,
|
||||
interruptible_sleep,
|
||||
owned_album_titles,
|
||||
pick_artist_by_catalog,
|
||||
release_titles,
|
||||
set_album_api_track_count,
|
||||
)
|
||||
from core.enrichment.manual_match_honoring import honor_stored_match
|
||||
|
||||
logger = get_logger("deezer_worker")
|
||||
|
|
@ -401,7 +409,20 @@ class DeezerWorker:
|
|||
logger.debug(f"Preserving existing Deezer ID for artist '{artist_name}': {existing_id}")
|
||||
return
|
||||
|
||||
result = self.client.search_artist(artist_name)
|
||||
# Multi-candidate search (was single search_artist) so same-name artists
|
||||
# can be disambiguated: gate by name, then pick the one whose catalog
|
||||
# overlaps the albums this library owns.
|
||||
results = self.client.search_artists(artist_name, limit=5)
|
||||
gated = [a for a in (results or []) if artist_name_matches(artist_name, getattr(a, 'name', ''))]
|
||||
chosen, _overlap = pick_artist_by_catalog(
|
||||
gated,
|
||||
owned_album_titles(self.db, artist_id),
|
||||
lambda a: release_titles(self.client.get_artist_albums_list(a.id)),
|
||||
)
|
||||
|
||||
# search_artists returns lean Artist objects; fetch the full dict (same
|
||||
# shape the old search_artist returned) for storage.
|
||||
result = self.client.get_artist_info(chosen.id) if chosen else None
|
||||
if result:
|
||||
result_name = result.get('name', '')
|
||||
ok, reason = accept_artist_match(
|
||||
|
|
|
|||
|
|
@ -216,7 +216,22 @@ def build_reset_query(
|
|||
table = meta['table']
|
||||
ms = match_status_column(service)
|
||||
la = last_attempted_column(service)
|
||||
set_clause = f"SET {ms} = NULL, {la} = NULL"
|
||||
set_parts = [f"{ms} = NULL", f"{la} = NULL"]
|
||||
|
||||
# Also forget the stored source ID so re-matching actually RE-RESOLVES the
|
||||
# entity. Without this, the worker hits its existing-id short-circuit, sees
|
||||
# the old (possibly WRONG) id and just re-confirms it — which is why "click
|
||||
# to rematch" never fixed a mis-matched same-name artist (#868). Tracks keep
|
||||
# their ids in file tags rather than a column, so only artist/album clear one.
|
||||
if entity_type in ('artist', 'album'):
|
||||
try:
|
||||
from core.source_ids import id_column
|
||||
id_col = id_column(service, entity_type)
|
||||
except Exception:
|
||||
id_col = None
|
||||
if id_col:
|
||||
set_parts.append(f"{id_col} = NULL")
|
||||
set_clause = "SET " + ", ".join(set_parts)
|
||||
|
||||
if scope == 'item':
|
||||
if not entity_id:
|
||||
|
|
|
|||
|
|
@ -9,7 +9,15 @@ from datetime import datetime, timedelta
|
|||
from utils.logging_config import get_logger
|
||||
from database.music_database import MusicDatabase
|
||||
from core.itunes_client import iTunesClient
|
||||
from core.worker_utils import accept_artist_match, interruptible_sleep, set_album_api_track_count
|
||||
from core.worker_utils import (
|
||||
accept_artist_match,
|
||||
artist_name_matches,
|
||||
interruptible_sleep,
|
||||
owned_album_titles,
|
||||
pick_artist_by_catalog,
|
||||
release_titles,
|
||||
set_album_api_track_count,
|
||||
)
|
||||
from core.enrichment.manual_match_honoring import honor_stored_match
|
||||
|
||||
logger = get_logger("itunes_worker")
|
||||
|
|
@ -392,20 +400,30 @@ class iTunesWorker:
|
|||
logger.debug(f"No iTunes results for artist '{artist_name}'")
|
||||
return
|
||||
|
||||
for artist_obj in results:
|
||||
# Candidates clearing the name gate (results are source-ranked, so [0] is
|
||||
# the legacy "first passing" pick), then disambiguate same-name artists by
|
||||
# which one's catalog overlaps the albums this library owns.
|
||||
gated = [a for a in results if artist_name_matches(artist_name, a.name)]
|
||||
chosen, _overlap = pick_artist_by_catalog(
|
||||
gated,
|
||||
owned_album_titles(self.db, artist_id),
|
||||
lambda a: release_titles(self.client.get_artist_albums(a.id)),
|
||||
)
|
||||
|
||||
if chosen:
|
||||
ok, reason = accept_artist_match(
|
||||
self.db, 'itunes_artist_id', artist_obj.id, artist_id,
|
||||
artist_name, artist_obj.name,
|
||||
self.db, 'itunes_artist_id', chosen.id, artist_id,
|
||||
artist_name, chosen.name,
|
||||
)
|
||||
if ok:
|
||||
if not self._is_itunes_id(artist_obj.id):
|
||||
logger.warning(f"Rejecting non-iTunes ID '{artist_obj.id}' for artist '{artist_name}'")
|
||||
if not self._is_itunes_id(chosen.id):
|
||||
logger.warning(f"Rejecting non-iTunes ID '{chosen.id}' for artist '{artist_name}'")
|
||||
self._mark_status('artist', artist_id, 'error')
|
||||
self.stats['errors'] += 1
|
||||
return
|
||||
self._update_artist(artist_id, artist_obj)
|
||||
self._update_artist(artist_id, chosen)
|
||||
self.stats['matched'] += 1
|
||||
logger.info(f"Matched artist '{artist_name}' -> iTunes ID: {artist_obj.id}")
|
||||
logger.info(f"Matched artist '{artist_name}' -> iTunes ID: {chosen.id}")
|
||||
return
|
||||
|
||||
self._mark_status('artist', artist_id, 'not_found')
|
||||
|
|
|
|||
|
|
@ -5,6 +5,7 @@ from datetime import datetime, timedelta
|
|||
from difflib import SequenceMatcher
|
||||
from utils.logging_config import get_logger
|
||||
from core.musicbrainz_client import MusicBrainzClient
|
||||
from core.worker_utils import catalog_overlap_score, pick_artist_by_catalog
|
||||
from database.music_database import MusicDatabase
|
||||
|
||||
logger = get_logger("musicbrainz_service")
|
||||
|
|
@ -117,23 +118,51 @@ class MusicBrainzService:
|
|||
if conn:
|
||||
conn.close()
|
||||
|
||||
def match_artist(self, artist_name: str) -> Optional[Dict[str, Any]]:
|
||||
def _candidate_release_titles(self, mbid: str) -> list:
|
||||
"""Release-group titles for a candidate MBID — the catalog side of
|
||||
same-name artist disambiguation."""
|
||||
if not mbid:
|
||||
return []
|
||||
try:
|
||||
data = self.mb_client.get_artist(mbid, includes=['release-groups'])
|
||||
except Exception:
|
||||
return []
|
||||
groups = (data or {}).get('release-groups') or []
|
||||
return [g.get('title') for g in groups if isinstance(g, dict) and g.get('title')]
|
||||
|
||||
def match_artist(self, artist_name: str, owned_titles: Optional[list] = None) -> Optional[Dict[str, Any]]:
|
||||
"""
|
||||
Match an artist by name to MusicBrainz
|
||||
|
||||
Match an artist by name to MusicBrainz.
|
||||
|
||||
``owned_titles`` — the library artist's owned album titles. When given and
|
||||
more than one strong same-name candidate exists, the one whose release
|
||||
groups overlap those owned titles is chosen (disambiguates the ~5 "Rone"s);
|
||||
omitted → falls back to the highest-confidence candidate as before.
|
||||
|
||||
Returns:
|
||||
Dict with 'mbid', 'name', 'confidence' or None if no good match
|
||||
"""
|
||||
# Check cache first
|
||||
cached = self._check_cache('artist', artist_name)
|
||||
if cached:
|
||||
logger.debug(f"Cache hit for artist '{artist_name}'")
|
||||
return {
|
||||
'mbid': cached['musicbrainz_id'],
|
||||
'name': artist_name,
|
||||
'confidence': cached['confidence'],
|
||||
'cached': True
|
||||
}
|
||||
cached_mbid = cached.get('musicbrainz_id')
|
||||
# Don't trust a cached mbid whose catalog has ZERO overlap with the
|
||||
# albums this library owns — that's the wrong same-name artist (and a
|
||||
# re-match would otherwise be blocked for up to the 90-day cache TTL,
|
||||
# #868). Fall through to a fresh, disambiguated resolve in that case.
|
||||
stale_wrong_match = bool(
|
||||
cached_mbid and owned_titles
|
||||
and catalog_overlap_score(owned_titles, self._candidate_release_titles(cached_mbid)) == 0
|
||||
)
|
||||
if not stale_wrong_match:
|
||||
logger.debug(f"Cache hit for artist '{artist_name}'")
|
||||
return {
|
||||
'mbid': cached_mbid,
|
||||
'name': artist_name,
|
||||
'confidence': cached['confidence'],
|
||||
'cached': True
|
||||
}
|
||||
logger.debug(f"Cached MB match for '{artist_name}' has no owned-catalog overlap — re-resolving")
|
||||
|
||||
# Search MusicBrainz
|
||||
try:
|
||||
|
|
@ -144,25 +173,30 @@ class MusicBrainzService:
|
|||
self._save_to_cache('artist', artist_name, None, None, None, 0)
|
||||
return None
|
||||
|
||||
# Find best match
|
||||
best_match = None
|
||||
best_confidence = 0
|
||||
|
||||
# Score every candidate (name similarity 60% + MB's own relevance 40%).
|
||||
scored = []
|
||||
for result in results:
|
||||
mb_name = result.get('name', '')
|
||||
mb_score = result.get('score', 0) # MusicBrainz search score
|
||||
|
||||
# Calculate our own similarity
|
||||
similarity = self._calculate_similarity(artist_name, mb_name)
|
||||
|
||||
# Combine MusicBrainz score with our similarity (weighted)
|
||||
# Cap at 100 to prevent edge cases where MB score > 100
|
||||
confidence = min(100, int((similarity * 60) + (mb_score / 100 * 40)))
|
||||
|
||||
if confidence > best_confidence:
|
||||
best_confidence = confidence
|
||||
best_match = result
|
||||
|
||||
scored.append((confidence, result))
|
||||
scored.sort(key=lambda s: s[0], reverse=True)
|
||||
|
||||
# Among the strong (>=70) candidates, disambiguate same-name artists by
|
||||
# which one's release groups overlap the albums this library owns.
|
||||
gated = [r for conf, r in scored if conf >= 70]
|
||||
best_match = None
|
||||
best_confidence = scored[0][0] if scored else 0
|
||||
if gated:
|
||||
chosen, _overlap = pick_artist_by_catalog(
|
||||
gated, owned_titles or [],
|
||||
lambda r: self._candidate_release_titles(r.get('id')),
|
||||
)
|
||||
best_match = chosen
|
||||
best_confidence = next(conf for conf, r in scored if r is chosen)
|
||||
|
||||
# Only return matches with confidence >= 70%
|
||||
if best_match and best_confidence >= 70:
|
||||
mbid = best_match.get('id')
|
||||
|
|
|
|||
|
|
@ -5,7 +5,7 @@ from datetime import datetime, timedelta
|
|||
from utils.logging_config import get_logger
|
||||
from database.music_database import MusicDatabase
|
||||
from core.musicbrainz_service import MusicBrainzService
|
||||
from core.worker_utils import interruptible_sleep, source_id_conflict
|
||||
from core.worker_utils import interruptible_sleep, owned_album_titles, source_id_conflict
|
||||
|
||||
logger = get_logger("musicbrainz_worker")
|
||||
|
||||
|
|
@ -366,7 +366,8 @@ class MusicBrainzWorker:
|
|||
return
|
||||
|
||||
if item_type == 'artist':
|
||||
result = self.mb_service.match_artist(item_name)
|
||||
result = self.mb_service.match_artist(
|
||||
item_name, owned_titles=owned_album_titles(self.db, item_id))
|
||||
mbid = result.get('mbid') if result else None
|
||||
# MB's combined score can match a weak name ("Grant" -> "Amy
|
||||
# Grant") when its own relevance rank is high. Guard against
|
||||
|
|
|
|||
|
|
@ -12,6 +12,9 @@ from core.spotify_client import SpotifyClient, SpotifyRateLimitError
|
|||
from core.worker_utils import (
|
||||
ARTIST_NAME_MATCH_THRESHOLD,
|
||||
interruptible_sleep,
|
||||
owned_album_titles,
|
||||
pick_artist_by_catalog,
|
||||
release_titles,
|
||||
set_album_api_track_count,
|
||||
source_id_conflict,
|
||||
)
|
||||
|
|
@ -563,16 +566,23 @@ class SpotifyWorker:
|
|||
logger.debug(f"No Spotify results for artist '{artist_name}'")
|
||||
return
|
||||
|
||||
# Find best fuzzy match — score all candidates, pick highest above the
|
||||
# (stricter, artist-specific) threshold so short-name false positives
|
||||
# like "ODESZA"/"odessa" don't slip through.
|
||||
best_obj = None
|
||||
best_score = 0
|
||||
for artist_obj in results:
|
||||
score = self._name_similarity(artist_name, artist_obj.name)
|
||||
if score >= ARTIST_NAME_MATCH_THRESHOLD and score > best_score:
|
||||
best_obj = artist_obj
|
||||
best_score = score
|
||||
# Candidates clearing the (stricter, artist-specific) name gate, best
|
||||
# name-score first so [0] is the legacy "first/highest" pick.
|
||||
scored = [
|
||||
(self._name_similarity(artist_name, a.name), a)
|
||||
for a in results
|
||||
]
|
||||
gated = [a for score, a in sorted(scored, key=lambda s: s[0], reverse=True)
|
||||
if score >= ARTIST_NAME_MATCH_THRESHOLD]
|
||||
|
||||
# Same-name disambiguation: when more than one "Rone" clears the gate,
|
||||
# pick the one whose catalog overlaps the albums this library owns.
|
||||
best_obj, _overlap = pick_artist_by_catalog(
|
||||
gated,
|
||||
owned_album_titles(self.db, artist_id),
|
||||
lambda a: release_titles(self.client.get_artist_albums(a.id)),
|
||||
)
|
||||
best_score = self._name_similarity(artist_name, best_obj.name) if best_obj else 0
|
||||
|
||||
if best_obj:
|
||||
if not self._is_spotify_id(best_obj.id):
|
||||
|
|
|
|||
|
|
@ -101,6 +101,108 @@ def accept_artist_match(database, id_column: str, source_id, artist_id,
|
|||
return True, ""
|
||||
|
||||
|
||||
# --- Same-name artist disambiguation by owned-catalog overlap -------------
|
||||
# The name gate (above) can't separate two artists who share a name ("Rone" has
|
||||
# ~5). The decisive signal is the library itself: the user owns albums by the
|
||||
# RIGHT one. So when several candidates clear the name gate, fetch each one's
|
||||
# catalog and pick the one whose releases overlap the albums actually owned.
|
||||
|
||||
def normalize_release_title(title: str) -> str:
|
||||
"""Collapse an album/release title for tolerant comparison — drop edition
|
||||
suffixes ('(Deluxe)', ' - Remastered'), punctuation, and case."""
|
||||
t = (title or '').lower().strip()
|
||||
t = re.sub(r'\s*\(.*?\)\s*', ' ', t)
|
||||
t = re.sub(r'\s*\[.*?\]\s*', ' ', t)
|
||||
t = re.sub(r'\s+[-–—]\s+.*$', '', t)
|
||||
t = re.sub(r'[^\w\s]', '', t)
|
||||
t = re.sub(r'\s+', ' ', t).strip()
|
||||
return t
|
||||
|
||||
|
||||
def catalog_overlap_score(owned_titles, candidate_titles, threshold: float = 0.85) -> int:
|
||||
"""How many OWNED album titles appear in the candidate's catalog (fuzzy,
|
||||
edition-insensitive). The disambiguation signal — higher = better match."""
|
||||
owned = {normalize_release_title(t) for t in (owned_titles or []) if t}
|
||||
owned.discard('')
|
||||
cand = {normalize_release_title(t) for t in (candidate_titles or []) if t}
|
||||
cand.discard('')
|
||||
if not owned or not cand:
|
||||
return 0
|
||||
score = 0
|
||||
for o in owned:
|
||||
if o in cand or any(SequenceMatcher(None, o, c).ratio() >= threshold for c in cand):
|
||||
score += 1
|
||||
return score
|
||||
|
||||
|
||||
def pick_artist_by_catalog(candidates, owned_titles, fetch_titles) -> tuple:
|
||||
"""Choose, among same-name candidates, the one whose catalog best overlaps the
|
||||
OWNED albums. Returns ``(chosen, overlap_score)``.
|
||||
|
||||
``candidates`` — artist objects already past the name gate (order = the
|
||||
worker's existing best-by-name order; candidates[0] is the
|
||||
current behavior's pick).
|
||||
``owned_titles`` — the library artist's owned album titles.
|
||||
``fetch_titles(candidate) -> list[str]`` — that candidate's album titles;
|
||||
called ONLY when disambiguation is actually needed (2+
|
||||
candidates and we have owned albums), so the common
|
||||
single-candidate path costs no extra API calls.
|
||||
|
||||
Falls back to candidates[0] (unchanged behavior) when there's nothing to
|
||||
disambiguate or no candidate overlaps the owned catalog.
|
||||
"""
|
||||
candidates = list(candidates or [])
|
||||
if not candidates:
|
||||
return None, 0
|
||||
if len(candidates) == 1:
|
||||
return candidates[0], 0
|
||||
owned = [t for t in (owned_titles or []) if t]
|
||||
if not owned:
|
||||
return candidates[0], 0
|
||||
|
||||
best, best_score = None, 0
|
||||
for cand in candidates:
|
||||
try:
|
||||
titles = fetch_titles(cand) or []
|
||||
except Exception as exc:
|
||||
logger.debug("catalog disambiguation: fetch_titles failed: %s", exc)
|
||||
titles = []
|
||||
score = catalog_overlap_score(owned, titles)
|
||||
if score > best_score:
|
||||
best, best_score = cand, score
|
||||
if best is not None and best_score > 0:
|
||||
return best, best_score
|
||||
return candidates[0], 0 # no overlap signal → keep the best-by-name pick
|
||||
|
||||
|
||||
def owned_album_titles(database, artist_id) -> list:
|
||||
"""The album titles the library actually has for this artist — the ground
|
||||
truth used to disambiguate same-name source artists."""
|
||||
try:
|
||||
with database._get_connection() as conn:
|
||||
rows = conn.execute(
|
||||
"SELECT title FROM albums WHERE artist_id = ?", (artist_id,)
|
||||
).fetchall()
|
||||
return [r[0] for r in rows if r and r[0]]
|
||||
except Exception as exc:
|
||||
logger.debug("owned_album_titles(%s) failed: %s", artist_id, exc)
|
||||
return []
|
||||
|
||||
|
||||
def release_titles(albums) -> list:
|
||||
"""Extract titles from a list of album objects/dicts (handles ``.title``/
|
||||
``.name`` / dict keys) — the candidate side of catalog disambiguation."""
|
||||
out = []
|
||||
for al in albums or []:
|
||||
if isinstance(al, dict):
|
||||
t = al.get('title') or al.get('name')
|
||||
else:
|
||||
t = getattr(al, 'title', None) or getattr(al, 'name', None)
|
||||
if t:
|
||||
out.append(t)
|
||||
return out
|
||||
|
||||
|
||||
def interruptible_sleep(stop_event: threading.Event, seconds: float, step: float = 0.5) -> bool:
|
||||
"""Sleep in chunks so shutdown can interrupt long waits."""
|
||||
if seconds <= 0:
|
||||
|
|
|
|||
|
|
@ -187,6 +187,24 @@ def test_reset_builder_nulls_status_not_just_attempted():
|
|||
assert "WHERE spotify_match_status = 'not_found'" in sql
|
||||
|
||||
|
||||
def test_reset_builder_also_clears_artist_source_id():
|
||||
# #868: a re-match must forget the stored id so the worker actually
|
||||
# re-resolves (otherwise its existing-id short-circuit re-confirms the wrong
|
||||
# same-name artist).
|
||||
for service, col in [('spotify', 'spotify_artist_id'), ('itunes', 'itunes_artist_id'),
|
||||
('deezer', 'deezer_id'), ('musicbrainz', 'musicbrainz_id')]:
|
||||
sql, _ = build_reset_query(service, 'artist', 'item', entity_id=5)
|
||||
assert f'{col} = NULL' in sql, f'{service}: expected {col} cleared'
|
||||
assert f'{service}_match_status = NULL' in sql
|
||||
|
||||
|
||||
def test_reset_builder_does_not_clear_track_id():
|
||||
# Tracks have no source-id column (ids live in tags) — must not emit one.
|
||||
sql, _ = build_reset_query('spotify', 'track', 'item', entity_id=5)
|
||||
assert 'spotify_match_status = NULL' in sql
|
||||
assert 'spotify_track_id = NULL' not in sql
|
||||
|
||||
|
||||
def test_reset_item_requeues_to_pending(db):
|
||||
n = db.reset_enrichment('spotify', 'artist', 'item', entity_id='a2') # was not_found
|
||||
assert n == 1
|
||||
|
|
|
|||
98
tests/test_artist_catalog_disambiguation.py
Normal file
98
tests/test_artist_catalog_disambiguation.py
Normal file
|
|
@ -0,0 +1,98 @@
|
|||
"""Same-name artist disambiguation by owned-catalog overlap (#868).
|
||||
|
||||
Enrichment matched artists by NAME ONLY, so for a common name ("Rone" has ~5
|
||||
artists) it grabbed whichever the source ranked first — often the wrong one,
|
||||
which then drove a wrong/sparse library discography. The fix: when several
|
||||
candidates clear the name gate, pick the one whose catalog overlaps the albums
|
||||
the user actually OWNS. These pin the source-agnostic selector.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from core.worker_utils import (
|
||||
catalog_overlap_score,
|
||||
normalize_release_title,
|
||||
pick_artist_by_catalog,
|
||||
)
|
||||
|
||||
|
||||
# --- normalization ---------------------------------------------------------
|
||||
|
||||
def test_normalize_strips_editions_and_punctuation():
|
||||
assert normalize_release_title('Tohu Bohu (Deluxe Edition)') == 'tohu bohu'
|
||||
assert normalize_release_title('Mirapolis - Remastered') == 'mirapolis'
|
||||
assert normalize_release_title('Room with a View [2020]') == 'room with a view'
|
||||
assert normalize_release_title('') == ''
|
||||
|
||||
|
||||
# --- overlap scoring -------------------------------------------------------
|
||||
|
||||
def test_overlap_counts_matching_owned_titles():
|
||||
owned = ['Tohu Bohu', 'Creatures', 'Mirapolis']
|
||||
cand = ['Tohu Bohu (Deluxe)', 'Creatures', 'Spanish Breakfast', 'Motion']
|
||||
assert catalog_overlap_score(owned, cand) == 2 # Tohu Bohu + Creatures
|
||||
|
||||
|
||||
def test_overlap_zero_for_a_different_artists_catalog():
|
||||
owned = ['Tohu Bohu', 'Creatures', 'Mirapolis']
|
||||
cand = ['Some Other Record', 'Unrelated Album']
|
||||
assert catalog_overlap_score(owned, cand) == 0
|
||||
|
||||
|
||||
def test_overlap_zero_when_either_side_empty():
|
||||
assert catalog_overlap_score([], ['A']) == 0
|
||||
assert catalog_overlap_score(['A'], []) == 0
|
||||
|
||||
|
||||
# --- the selector ----------------------------------------------------------
|
||||
|
||||
def _cand(cid, titles):
|
||||
return {'id': cid, '_titles': titles}
|
||||
|
||||
|
||||
def _fetch(cand):
|
||||
return cand['_titles']
|
||||
|
||||
|
||||
def test_single_candidate_returns_without_fetching():
|
||||
calls = []
|
||||
chosen, score = pick_artist_by_catalog(
|
||||
[_cand('only', ['X'])], ['Tohu Bohu'],
|
||||
lambda c: calls.append(c) or c['_titles'])
|
||||
assert chosen['id'] == 'only'
|
||||
assert calls == [] # never fetched — nothing to disambiguate
|
||||
|
||||
|
||||
def test_no_owned_albums_keeps_name_order():
|
||||
calls = []
|
||||
chosen, score = pick_artist_by_catalog(
|
||||
[_cand('first', ['A']), _cand('second', ['B'])], [],
|
||||
lambda c: calls.append(c) or c['_titles'])
|
||||
assert chosen['id'] == 'first' # candidates[0] — current behavior
|
||||
assert calls == []
|
||||
|
||||
|
||||
def test_picks_the_candidate_overlapping_owned_catalog():
|
||||
# The WRONG Rone is ranked first; the right one overlaps the owned albums.
|
||||
wrong = _cand('wrong', ['Rap Mixtape Vol 1', 'Some Single'])
|
||||
right = _cand('right', ['Tohu Bohu', 'Creatures', 'Mirapolis'])
|
||||
chosen, score = pick_artist_by_catalog(
|
||||
[wrong, right], ['Tohu Bohu', 'Creatures', 'Spanish Breakfast'], _fetch)
|
||||
assert chosen['id'] == 'right'
|
||||
assert score == 2
|
||||
|
||||
|
||||
def test_no_overlap_anywhere_falls_back_to_first():
|
||||
a = _cand('a', ['Nope']); b = _cand('b', ['Also Nope'])
|
||||
chosen, score = pick_artist_by_catalog([a, b], ['Tohu Bohu'], _fetch)
|
||||
assert chosen['id'] == 'a'
|
||||
assert score == 0
|
||||
|
||||
|
||||
def test_fetch_failure_is_tolerated():
|
||||
def _boom(_c):
|
||||
raise RuntimeError('api down')
|
||||
chosen, score = pick_artist_by_catalog(
|
||||
[_cand('a', []), _cand('b', [])], ['Tohu Bohu'], _boom)
|
||||
assert chosen['id'] == 'a' # both fail → fall back to first
|
||||
assert score == 0
|
||||
186
tests/test_worker_artist_disambiguation.py
Normal file
186
tests/test_worker_artist_disambiguation.py
Normal file
|
|
@ -0,0 +1,186 @@
|
|||
"""Per-worker same-name artist disambiguation (#868).
|
||||
|
||||
Each enrichment worker, when several same-name candidates clear the name gate,
|
||||
must pick the one whose catalog overlaps the albums the library actually owns —
|
||||
not whichever the source ranked first. Covers Spotify (also the Spotify-Free
|
||||
path, same client surface), iTunes, Deezer, and MusicBrainz.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import types
|
||||
|
||||
|
||||
# A query-aware fake DB: owned-albums query → owned titles; the source_id_conflict
|
||||
# query (SELECT name FROM artists ...) → no conflict.
|
||||
class _Cur:
|
||||
def __init__(self, rows):
|
||||
self._rows = rows
|
||||
|
||||
def fetchall(self):
|
||||
return self._rows
|
||||
|
||||
|
||||
class _Conn:
|
||||
def __init__(self, owned):
|
||||
self._owned = owned
|
||||
|
||||
def __enter__(self):
|
||||
return self
|
||||
|
||||
def __exit__(self, *a):
|
||||
return False
|
||||
|
||||
def execute(self, sql, params=()):
|
||||
if 'FROM albums' in sql:
|
||||
return _Cur([(t,) for t in self._owned])
|
||||
return _Cur([]) # conflict check → none
|
||||
|
||||
def cursor(self):
|
||||
return self
|
||||
|
||||
def commit(self):
|
||||
pass
|
||||
|
||||
def close(self):
|
||||
pass
|
||||
|
||||
|
||||
class _DB:
|
||||
def __init__(self, owned):
|
||||
self._owned = owned
|
||||
|
||||
def _get_connection(self):
|
||||
return _Conn(self._owned)
|
||||
|
||||
|
||||
# The owned library: the RIGHT "Rone" made these.
|
||||
OWNED = ['Tohu Bohu', 'Creatures', 'Mirapolis']
|
||||
WRONG = types.SimpleNamespace(id='wrong_rone', name='Rone')
|
||||
RIGHT = types.SimpleNamespace(id='right_rone', name='Rone')
|
||||
ALBUMS = {
|
||||
'wrong_rone': [{'title': 'Mixtape Vol 1'}, {'title': 'Random Single'}],
|
||||
'right_rone': [{'title': 'Tohu Bohu (Deluxe)'}, {'title': 'Creatures'}],
|
||||
}
|
||||
|
||||
|
||||
def test_spotify_picks_artist_overlapping_owned_catalog():
|
||||
from core.spotify_worker import SpotifyWorker
|
||||
w = object.__new__(SpotifyWorker)
|
||||
w.db = _DB(OWNED)
|
||||
w.stats = {'matched': 0, 'not_found': 0, 'errors': 0}
|
||||
w.client = types.SimpleNamespace(
|
||||
search_artists=lambda name, limit=5: [WRONG, RIGHT], # wrong ranked first
|
||||
get_artist_albums=lambda aid: ALBUMS.get(aid, []),
|
||||
)
|
||||
captured = {}
|
||||
w._get_existing_id = lambda *a: None
|
||||
w._mark_status = lambda *a: None
|
||||
w._name_similarity = lambda a, b: 1.0 # both "Rone" clear the gate
|
||||
w._is_spotify_id = lambda i: True
|
||||
w._update_artist = lambda artist_id, obj: captured.update(id=obj.id)
|
||||
|
||||
w._process_artist({'id': 5, 'name': 'Rone'})
|
||||
assert captured.get('id') == 'right_rone'
|
||||
assert w.stats['matched'] == 1
|
||||
|
||||
|
||||
def test_itunes_picks_artist_overlapping_owned_catalog():
|
||||
from core.itunes_worker import iTunesWorker
|
||||
w = object.__new__(iTunesWorker)
|
||||
w.db = _DB(OWNED)
|
||||
w.stats = {'matched': 0, 'not_found': 0, 'errors': 0}
|
||||
w.client = types.SimpleNamespace(
|
||||
search_artists=lambda name, limit=5: [WRONG, RIGHT],
|
||||
get_artist_albums=lambda aid: ALBUMS.get(aid, []),
|
||||
)
|
||||
captured = {}
|
||||
w._get_existing_id = lambda *a: None
|
||||
w._mark_status = lambda *a: None
|
||||
w._is_itunes_id = lambda i: True
|
||||
w._update_artist = lambda artist_id, obj: captured.update(id=obj.id)
|
||||
|
||||
w._process_artist({'id': 5, 'name': 'Rone'})
|
||||
assert captured.get('id') == 'right_rone'
|
||||
assert w.stats['matched'] == 1
|
||||
|
||||
|
||||
def test_deezer_picks_artist_overlapping_owned_catalog():
|
||||
from core.deezer_worker import DeezerWorker
|
||||
w = object.__new__(DeezerWorker)
|
||||
w.db = _DB(OWNED)
|
||||
w.stats = {'matched': 0, 'not_found': 0, 'errors': 0}
|
||||
w.client = types.SimpleNamespace(
|
||||
search_artists=lambda name, limit=5: [WRONG, RIGHT],
|
||||
get_artist_albums_list=lambda aid: ALBUMS.get(aid, []),
|
||||
get_artist_info=lambda aid: {'id': aid, 'name': 'Rone'},
|
||||
)
|
||||
captured = {}
|
||||
w._get_existing_id = lambda *a: None
|
||||
w._mark_status = lambda *a: None
|
||||
w._update_artist = lambda artist_id, data: captured.update(id=data.get('id'))
|
||||
|
||||
w._process_artist(5, 'Rone')
|
||||
assert captured.get('id') == 'right_rone'
|
||||
assert w.stats['matched'] == 1
|
||||
|
||||
|
||||
def _mb_service(owned_release_groups):
|
||||
from core.musicbrainz_service import MusicBrainzService
|
||||
s = object.__new__(MusicBrainzService)
|
||||
s._check_cache = lambda *a, **k: None
|
||||
s._save_to_cache = lambda *a, **k: None
|
||||
s._calculate_similarity = lambda a, b: 1.0
|
||||
s.mb_client = types.SimpleNamespace(
|
||||
search_artist=lambda name, limit=5: [
|
||||
{'id': 'wrong_mbid', 'name': 'Rone', 'score': 100}, # ranked first
|
||||
{'id': 'right_mbid', 'name': 'Rone', 'score': 90},
|
||||
],
|
||||
get_artist=lambda mbid, includes=None: owned_release_groups.get(mbid),
|
||||
)
|
||||
return s
|
||||
|
||||
|
||||
def test_musicbrainz_picks_artist_overlapping_owned_catalog():
|
||||
rg = {
|
||||
'wrong_mbid': {'release-groups': [{'title': 'Other Stuff'}]},
|
||||
'right_mbid': {'release-groups': [{'title': 'Tohu Bohu'}, {'title': 'Creatures'}]},
|
||||
}
|
||||
result = _mb_service(rg).match_artist('Rone', owned_titles=OWNED)
|
||||
assert result is not None
|
||||
assert result['mbid'] == 'right_mbid'
|
||||
|
||||
|
||||
def test_musicbrainz_without_owned_titles_keeps_legacy_behavior():
|
||||
"""No owned titles → highest-confidence (name-order first) candidate, unchanged."""
|
||||
rg = {'wrong_mbid': {'release-groups': []}, 'right_mbid': {'release-groups': []}}
|
||||
result = _mb_service(rg).match_artist('Rone') # no owned_titles
|
||||
assert result['mbid'] == 'wrong_mbid' # the top-confidence pick
|
||||
|
||||
|
||||
def test_musicbrainz_cache_hit_used_when_catalog_overlaps():
|
||||
"""A cached mbid whose catalog overlaps the owned albums is trusted (no re-resolve)."""
|
||||
rg = {'cached_mbid': {'release-groups': [{'title': 'Tohu Bohu'}, {'title': 'Creatures'}]}}
|
||||
s = _mb_service(rg)
|
||||
s._check_cache = lambda *a, **k: {'musicbrainz_id': 'cached_mbid', 'confidence': 95}
|
||||
s.mb_client.search_artist = lambda *a, **k: (_ for _ in ()).throw(
|
||||
AssertionError("must not re-search when the cached match fits the library"))
|
||||
result = s.match_artist('Rone', owned_titles=OWNED)
|
||||
assert result['mbid'] == 'cached_mbid'
|
||||
assert result['cached'] is True
|
||||
|
||||
|
||||
def test_musicbrainz_stale_cache_is_bypassed_and_re_resolved():
|
||||
"""A cached mbid with ZERO owned-catalog overlap (wrong same-name artist) is
|
||||
bypassed → fresh disambiguated resolve. This is what makes a re-match work for
|
||||
MB despite the 90-day cache TTL (#868)."""
|
||||
rg = {
|
||||
'cached_wrong': {'release-groups': [{'title': 'Unrelated'}]}, # the stale cache
|
||||
'wrong_mbid': {'release-groups': [{'title': 'Other Stuff'}]},
|
||||
'right_mbid': {'release-groups': [{'title': 'Tohu Bohu'}, {'title': 'Creatures'}]},
|
||||
}
|
||||
s = _mb_service(rg)
|
||||
s._check_cache = lambda *a, **k: {'musicbrainz_id': 'cached_wrong', 'confidence': 95}
|
||||
result = s.match_artist('Rone', owned_titles=OWNED)
|
||||
assert result['mbid'] == 'right_mbid' # re-resolved + disambiguated
|
||||
assert result.get('cached') is not True
|
||||
Loading…
Reference in a new issue