Refine artist-detail discography flow

- move artist-detail discography resolution onto the shared source-priority metadata service
- keep the variant dedup helper in the UI-facing adapter
- pass the chosen source through completion checks
- add coverage for the new adapter and dedup behavior

core/metadata_service.py

@@ -343,6 +343,92 @@ def _sort_discography_releases(releases: List[Dict[str, Any]]) -> List[Dict[str,
     return sorted(releases, key=get_release_year, reverse=True)
 
 
+def _dedup_variant_releases(releases: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
+    """Collapse obvious edition variants into a single canonical release card.
+
+    This keeps a clean UI while still preserving distinct releases when the
+    cleaned titles diverge enough that they are likely not variants.
+    """
+    if not releases:
+        return []
+
+    import re
+    from difflib import SequenceMatcher
+
+    variant_patterns = [
+        r'\s*[\(\[]\s*(explicit|clean|deluxe|deluxe edition|standard edition|clean version|explicit version|remastered|bonus track version)\s*[\)\]]',
+        r'\s*-\s*(explicit|clean|deluxe edition|single)\s*$',
+    ]
+
+    def _clean_title(title: Any) -> str:
+        cleaned = str(title or '').strip().lower()
+        for pattern in variant_patterns:
+            cleaned = re.sub(pattern, '', cleaned, flags=re.IGNORECASE)
+        cleaned = re.sub(r'\s+', ' ', cleaned).strip()
+        return cleaned
+
+    def _is_compilation(release: Dict[str, Any]) -> bool:
+        title = str(_extract_lookup_value(release, 'name', 'title', default='') or '').lower()
+        album_type = str(_extract_lookup_value(release, 'album_type', default='') or '').lower()
+        return (
+            album_type == 'compilation'
+            or 'best of' in title
+            or 'greatest hits' in title
+            or 'collection' in title
+            or 'anthology' in title
+            or 'essential' in title
+        )
+
+    def _variant_score(release: Dict[str, Any]) -> tuple:
+        title = str(_extract_lookup_value(release, 'name', 'title', default='') or '').lower()
+        has_explicit = 'explicit' in title
+        has_clean = 'clean' in title and not has_explicit
+        track_count = int(_extract_lookup_value(release, 'track_count', 'total_tracks', default=0) or 0)
+        release_date = str(_extract_lookup_value(release, 'release_date', default='') or '')
+
+        # Higher is better.
+        return (
+            1 if not _is_compilation(release) else 0,
+            2 if has_explicit else (1 if not has_clean else 0),
+            track_count,
+            release_date,
+        )
+
+    grouped: Dict[tuple, Dict[str, Any]] = {}
+    ordered_keys: List[tuple] = []
+
+    for release in releases:
+        title = _extract_lookup_value(release, 'name', 'title', default='') or ''
+        release_date = _extract_lookup_value(release, 'release_date')
+        year = _extract_lookup_value(release, 'year')
+        if not year and release_date:
+            year = str(release_date)[:4]
+        year = str(year) if year is not None else ''
+
+        cleaned_title = _clean_title(title) or str(title).strip().lower()
+        key = (cleaned_title, year)
+
+        existing = grouped.get(key)
+        if existing is None:
+            grouped[key] = release
+            ordered_keys.append(key)
+            continue
+
+        # If the cleaned titles are still materially different, keep both.
+        existing_clean = _clean_title(_extract_lookup_value(existing, 'name', 'title', default='') or '')
+        if SequenceMatcher(None, cleaned_title, existing_clean).ratio() < 0.85:
+            alt_key = (str(title).strip().lower(), year)
+            if alt_key not in grouped:
+                grouped[alt_key] = release
+                ordered_keys.append(alt_key)
+            continue
+
+        if _variant_score(release) > _variant_score(existing):
+            grouped[key] = release
+
+    return [grouped[key] for key in ordered_keys]
+
+
 def get_artist_discography(
     artist_id: str,
     artist_name: str = '',
@@ -423,6 +509,100 @@ def get_artist_discography(
     }
 
 
+def _build_artist_detail_release_card(release: Dict[str, Any]) -> Optional[Dict[str, Any]]:
+    release_id = _extract_lookup_value(release, 'id', 'album_id', 'release_id')
+    if not release_id:
+        return None
+
+    album_type = (_extract_lookup_value(release, 'album_type', default='album') or 'album').lower()
+    release_date = _extract_lookup_value(release, 'release_date')
+    release_year = None
+    if release_date:
+        try:
+            release_year = str(release_date)[:4]
+        except Exception:
+            release_year = None
+    if not release_year:
+        release_year = _extract_lookup_value(release, 'year')
+        if release_year is not None:
+            release_year = str(release_year)
+
+    card = {
+        'id': release_id,
+        'name': _extract_lookup_value(release, 'name', 'title', default=release_id),
+        'title': _extract_lookup_value(release, 'name', 'title', default=release_id),
+        'spotify_id': release_id,
+        'album_type': album_type,
+        'image_url': _extract_lookup_value(release, 'image_url', 'thumb_url', 'cover_image'),
+        'year': release_year,
+        'track_count': _extract_lookup_value(release, 'track_count', 'total_tracks', default=0) or 0,
+        'owned': None,
+        'track_completion': 'checking',
+    }
+
+    if release_date:
+        card['release_date'] = release_date
+    elif release_year:
+        card['release_date'] = f"{release_year}-01-01"
+
+    return card
+
+
+def get_artist_detail_discography(
+    artist_id: str,
+    artist_name: str = '',
+    options: Optional[MetadataLookupOptions] = None,
+) -> Dict[str, Any]:
+    """Get artist-detail-ready discography cards from the source-priority lookup flow."""
+    source_discography = get_artist_discography(
+        artist_id,
+        artist_name=artist_name,
+        options=options,
+    )
+
+    albums: List[Dict[str, Any]] = []
+    eps: List[Dict[str, Any]] = []
+    singles: List[Dict[str, Any]] = []
+    seen_ids = set()
+
+    for release in list(source_discography.get('albums', []) or []) + list(source_discography.get('singles', []) or []):
+        card = _build_artist_detail_release_card(release)
+        if not card:
+            continue
+
+        release_id = card['id']
+        if release_id in seen_ids:
+            continue
+        seen_ids.add(release_id)
+
+        album_type = (card.get('album_type') or 'album').lower()
+        if album_type == 'ep':
+            eps.append(card)
+        elif album_type == 'single':
+            singles.append(card)
+        else:
+            albums.append(card)
+
+    albums = _dedup_variant_releases(albums)
+    eps = _dedup_variant_releases(eps)
+    singles = _dedup_variant_releases(singles)
+
+    albums = _sort_discography_releases(albums)
+    eps = _sort_discography_releases(eps)
+    singles = _sort_discography_releases(singles)
+
+    has_releases = bool(albums or eps or singles)
+    return {
+        'success': has_releases,
+        'albums': albums,
+        'eps': eps,
+        'singles': singles,
+        'source': source_discography.get('source', 'unknown'),
+        'source_priority': source_discography.get('source_priority', []),
+        'error': None if has_releases else f'No releases found for artist "{artist_name or artist_id}"',
+    }
+
+
 def _get_completion_source_chain(source_override: Optional[str] = None) -> List[str]:
     primary_source = get_primary_source()
     source_chain = list(get_source_priority(primary_source))

tests/test_metadata_service_discography.py

@@ -383,3 +383,92 @@ def test_iter_artist_discography_completion_uses_release_artist_metadata(monkeyp
     assert events[1]["name"] == "Album Three"
     assert db.album_calls[0]["artist"] == "Explicit Artist"
     assert source.track_search_calls == []
+
+
+def test_get_artist_detail_discography_classifies_release_types(monkeypatch):
+    monkeypatch.setattr(
+        metadata_service,
+        "get_artist_discography",
+        lambda artist_id, artist_name='', options=None: {
+            "albums": [
+                {
+                    "id": "album-1",
+                    "name": "Album One",
+                    "album_type": "album",
+                    "image_url": "https://img.example/album-1.jpg",
+                    "release_date": "2024-01-05",
+                    "total_tracks": 10,
+                }
+            ],
+            "singles": [
+                {
+                    "id": "ep-1",
+                    "name": "EP One",
+                    "album_type": "ep",
+                    "image_url": "https://img.example/ep-1.jpg",
+                    "release_date": "2023-06-01",
+                    "total_tracks": 6,
+                },
+                {
+                    "id": "single-1",
+                    "name": "Single One",
+                    "album_type": "single",
+                    "image_url": "https://img.example/single-1.jpg",
+                    "release_date": "2022-03-10",
+                    "total_tracks": 1,
+                },
+            ],
+            "source": "deezer",
+            "source_priority": ["deezer", "spotify"],
+        },
+    )
+
+    result = metadata_service.get_artist_detail_discography("artist-1", "Artist One", MetadataLookupOptions())
+
+    assert result["success"] is True
+    assert result["source"] == "deezer"
+    assert result["source_priority"] == ["deezer", "spotify"]
+    assert [album["id"] for album in result["albums"]] == ["album-1"]
+    assert [ep["id"] for ep in result["eps"]] == ["ep-1"]
+    assert [single["id"] for single in result["singles"]] == ["single-1"]
+    assert result["albums"][0]["title"] == "Album One"
+    assert result["albums"][0]["spotify_id"] == "album-1"
+    assert result["albums"][0]["owned"] is None
+    assert result["albums"][0]["track_completion"] == "checking"
+
+
+def test_get_artist_detail_discography_dedups_variant_releases(monkeypatch):
+    monkeypatch.setattr(
+        metadata_service,
+        "get_artist_discography",
+        lambda artist_id, artist_name='', options=None: {
+            "albums": [
+                {
+                    "id": "album-standard",
+                    "name": "Variant Album",
+                    "album_type": "album",
+                    "image_url": "https://img.example/standard.jpg",
+                    "release_date": "2024-01-05",
+                    "total_tracks": 10,
+                },
+                {
+                    "id": "album-deluxe",
+                    "name": "Variant Album (Deluxe Edition)",
+                    "album_type": "album",
+                    "image_url": "https://img.example/deluxe.jpg",
+                    "release_date": "2024-01-05",
+                    "total_tracks": 14,
+                },
+            ],
+            "singles": [],
+            "source": "deezer",
+            "source_priority": ["deezer", "spotify"],
+        },
+    )
+
+    result = metadata_service.get_artist_detail_discography("artist-1", "Artist One", MetadataLookupOptions())
+
+    assert result["success"] is True
+    assert [album["id"] for album in result["albums"]] == ["album-deluxe"]
+    assert result["albums"][0]["title"] == "Variant Album (Deluxe Edition)"
+    assert result["albums"][0]["track_count"] == 14

web_server.py

@@ -10713,32 +10713,45 @@ def get_artist_detail(artist_id):
             if single.get('image_url'):
                 single['image_url'] = fix_artist_image_url(single['image_url'])
 
-        # Get Spotify discography for proper categorization and missing releases
+        # Get source-priority discography for proper categorization and missing releases
-        spotify_artist_data = None
+        artist_detail_discography = None
         try:
-            spotify_discography = get_spotify_artist_discography(artist_info['name'])
+            from core.metadata_service import MetadataLookupOptions, get_artist_detail_discography as _get_artist_detail_discography
 
-            if spotify_discography['success']:
+            artist_detail_discography = _get_artist_detail_discography(
-                print(f"Spotify discography found - Albums: {len(spotify_discography['albums'])}, EPs: {len(spotify_discography['eps'])}, Singles: {len(spotify_discography['singles'])}")
+                artist_id,
+                artist_name=artist_info['name'],
+                options=MetadataLookupOptions(
+                    allow_fallback=True,
+                    skip_cache=False,
+                    max_pages=0,
+                    limit=50,
+                ),
+            )
 
-                # Store Spotify artist data for the response
+            if artist_detail_discography['success']:
-                spotify_artist_data = {
+                print(
-                    'spotify_artist_id': spotify_discography.get('spotify_artist_id'),
+                    "Source-priority discography found - "
-                    'spotify_artist_name': spotify_discography.get('spotify_artist_name'),
+                    f"Albums: {len(artist_detail_discography['albums'])}, "
-                    'artist_image': spotify_discography.get('artist_image')
+                    f"EPs: {len(artist_detail_discography['eps'])}, "
-                }
+                    f"Singles: {len(artist_detail_discography['singles'])}"
-
+                )
-                # Merge owned and Spotify data for complete picture
+                merged_discography = artist_detail_discography
-                merged_discography = merge_discography_data(owned_releases, spotify_discography, db=database, artist_name=artist_info['name'])
             else:
-                print(f"Spotify discography not found: {spotify_discography.get('error', 'Unknown error')}")
+                print(f"Source-priority discography not found: {artist_detail_discography.get('error', 'Unknown error')}")
-                # Fall back to our database categorization
                 merged_discography = owned_releases
-        except Exception as spotify_error:
+        except Exception as detail_error:
-            print(f"Error fetching Spotify data: {spotify_error}")
+            print(f"Error fetching source-priority discography: {detail_error}")
-            # Fall back to our database categorization
             merged_discography = owned_releases
 
+        spotify_artist_data = None
+        if artist_info.get('spotify_artist_id'):
+            spotify_artist_data = {
+                'spotify_artist_id': artist_info.get('spotify_artist_id'),
+                'spotify_artist_name': artist_info.get('name'),
+                'artist_image': artist_info.get('image_url')
+            }
+
         # Compute per-artist track enrichment coverage
         enrichment_coverage = {}
         try:
@@ -11779,6 +11792,7 @@ def library_completion_stream():
         try:
             from core.metadata_service import check_album_completion, check_single_completion
             db = get_database()
+            source_override = (data.get('source') or '').strip().lower() or None
 
             categories = ['albums', 'eps', 'singles']
             all_items = []
@@ -11799,9 +11813,9 @@ def library_completion_stream():
                     }
 
                     if category == 'singles':
-                        result = check_single_completion(db, mapped, artist_name)
+                        result = check_single_completion(db, mapped, artist_name, source_override=source_override)
                     else:
-                        result = check_album_completion(db, mapped, artist_name)
+                        result = check_album_completion(db, mapped, artist_name, source_override=source_override)
 
                     result['spotify_id'] = item.get('spotify_id', '')
                     result['category'] = category
@@ -50726,232 +50740,6 @@ def start_oauth_callback_servers():
     
     print("OAuth callback servers started")
 
-# ===============================================
-# Artist Detail Spotify Integration Functions
-# ===============================================
-
-def get_spotify_artist_discography(artist_name):
-    """Get complete artist discography from Spotify using proper matching"""
-    try:
-        from core.matching_engine import MusicMatchingEngine
-
-        print(f"Searching Spotify for artist: {artist_name}")
-
-        # Reuse cached profile-aware Spotify client
-        spotify_client = get_spotify_client_for_profile()
-        matching_engine = MusicMatchingEngine()
-
-        if not spotify_client:
-            return {
-                'success': False,
-                'error': 'Spotify client unavailable'
-            }
-
-        # Search for multiple potential matches (not just 1)
-        artists = spotify_client.search_artists(artist_name, limit=5)
-
-        if not artists:
-            return {
-                'success': False,
-                'error': f'Artist "{artist_name}" not found on Spotify'
-            }
-
-        # Since database names are exact Spotify names, try exact match first
-        best_match = None
-        highest_score = 0.0
-
-        # Step 1: Try exact case-insensitive match
-        for spotify_artist in artists:
-            if artist_name.lower().strip() == spotify_artist.name.lower().strip():
-                print(f"Exact match found: '{spotify_artist.name}'")
-                best_match = spotify_artist
-                highest_score = 1.0
-                break
-
-        # Step 2: If no exact match, use matching engine with higher threshold
-        if not best_match:
-            db_artist_normalized = matching_engine.normalize_string(artist_name)
-
-            for spotify_artist in artists:
-                spotify_artist_normalized = matching_engine.normalize_string(spotify_artist.name)
-                score = matching_engine.similarity_score(db_artist_normalized, spotify_artist_normalized)
-
-                print(f"Fuzzy match candidate: '{spotify_artist.name}' (score: {score:.3f})")
-
-                if score > highest_score:
-                    highest_score = score
-                    best_match = spotify_artist
-
-        # Require high confidence threshold since DB should have exact names
-        if not best_match or highest_score < 0.95:
-            return {
-                'success': False,
-                'error': f'No confident artist match found for "{artist_name}" (best: "{getattr(best_match, "name", "N/A")}", score: {highest_score:.3f})'
-            }
-
-        artist = best_match
-        spotify_artist_id = artist.id
-
-        print(f"Found Spotify artist: {artist.name} (ID: {spotify_artist_id}, confidence: {highest_score:.3f})")
-
-        # Get all albums (albums, singles, and compilations)
-        all_albums = spotify_client.get_artist_albums(spotify_artist_id, album_type='album,single,compilation')
-
-        if not all_albums:
-            return {
-                'success': False,
-                'error': f'No albums found for artist "{artist_name}"'
-            }
-
-        print(f"Found {len(all_albums)} releases on Spotify")
-
-        # Categorize releases
-        albums = []
-        eps = []
-        singles = []
-
-        for album in all_albums:
-            # Skip albums where this artist isn't the primary (first-listed) artist
-            if getattr(album, 'artist_ids', None) and album.artist_ids[0] != spotify_artist_id:
-                continue
-
-            # Use the Album object properties
-            track_count = album.total_tracks
-
-            release_data = {
-                'title': album.name,
-                'year': album.release_date[:4] if album.release_date else None,
-                'release_date': album.release_date if album.release_date else None,
-                'image_url': album.image_url,
-                'spotify_id': album.id,
-                'owned': False,  # Will be updated when merging with owned data
-                'track_count': track_count,
-                'album_type': album.album_type.lower()
-            }
-
-            # Categorize based on album_type from source (Spotify/iTunes/Deezer)
-            album_type = album.album_type.lower() if album.album_type else 'album'
-
-            if album_type == 'single':
-                singles.append(release_data)
-            elif album_type == 'ep':
-                eps.append(release_data)
-            elif album_type == 'compilation':
-                albums.append(release_data)
-            else:
-                albums.append(release_data)
-
-        # Deduplicate variant releases (Explicit/Clean, Standard/Deluxe)
-        def _dedup_releases(releases):
-            import re
-            _VARIANT_RE = re.compile(
-                r'\s*[\(\[](explicit|clean|deluxe|deluxe edition|standard edition|'
-                r'clean version|explicit version|remastered|bonus track version)[\)\]]'
-                r'|\s*-\s*(explicit|clean|deluxe edition|single)\s*$',
-                re.IGNORECASE
-            )
-            groups = {}
-            for r in releases:
-                norm = _VARIANT_RE.sub('', r['title']).strip().lower()
-                key = (norm, r.get('year'))
-                if key not in groups:
-                    groups[key] = r
-                else:
-                    existing = groups[key]
-                    # Only dedup if titles are genuinely similar (not just same year)
-                    from difflib import SequenceMatcher
-                    title_sim = SequenceMatcher(None, norm, _VARIANT_RE.sub('', existing['title']).strip().lower()).ratio()
-                    if title_sim < 0.85:
-                        # Titles are too different — not variants, keep both
-                        # Use full title as key to avoid collision
-                        groups[(r['title'].lower(), r.get('year'))] = r
-                        continue
-                    # Prefer: studio over compilation, explicit over clean, more tracks as tiebreaker
-                    r_is_compilation = r.get('album_type', '').lower() == 'compilation' or 'best of' in r['title'].lower() or 'greatest hits' in r['title'].lower()
-                    e_is_compilation = existing.get('album_type', '').lower() == 'compilation' or 'best of' in existing['title'].lower() or 'greatest hits' in existing['title'].lower()
-                    if e_is_compilation and not r_is_compilation:
-                        groups[key] = r  # Studio album wins over compilation
-                    elif r_is_compilation and not e_is_compilation:
-                        pass  # Keep existing studio album
-                    elif (r.get('track_count', 0) > existing.get('track_count', 0) or
-                          '(explicit' in r['title'].lower()):
-                        groups[key] = r
-            return list(groups.values())
-
-        albums = _dedup_releases(albums)
-        eps = _dedup_releases(eps)
-        singles = _dedup_releases(singles)
-
-        print(f"Categorized Spotify releases - Albums: {len(albums)}, EPs: {len(eps)}, Singles: {len(singles)}")
-
-        return {
-            'success': True,
-            'albums': albums,
-            'eps': eps,
-            'singles': singles,
-            'artist_image': artist.image_url if hasattr(artist, 'image_url') else None,
-            'spotify_artist_id': spotify_artist_id,
-            'spotify_artist_name': artist.name
-        }
-
-    except Exception as e:
-        print(f"Error getting Spotify discography for {artist_name}: {e}")
-        return {
-            'success': False,
-            'error': str(e)
-        }
-
-def merge_discography_data(owned_releases, spotify_discography, db=None, artist_name=None):
-    """Build discography from Spotify data with 'checking' state - ownership is resolved via SSE stream"""
-    try:
-        print("Building discography cards (fast path - no DB matching)...")
-
-        def build_category(spotify_category, category_name):
-            """Build cards for a category with checking state"""
-            cards = []
-            for spotify_release in spotify_category:
-                card = {
-                    'title': spotify_release['title'],
-                    'spotify_id': spotify_release.get('spotify_id'),
-                    'album_type': spotify_release.get('album_type', 'album'),
-                    'image_url': spotify_release.get('image_url'),
-                    'year': spotify_release.get('year'),
-                    'track_count': spotify_release.get('track_count', 0),
-                    'owned': None,  # null = checking (resolved by completion stream)
-                    'track_completion': 'checking',
-                }
-                if spotify_release.get('release_date'):
-                    card['release_date'] = spotify_release['release_date']
-                elif spotify_release.get('year'):
-                    card['release_date'] = f"{spotify_release['year']}-01-01"
-                cards.append(card)
-            return cards
-
-        albums = build_category(spotify_discography['albums'], 'Albums')
-        eps = build_category(spotify_discography['eps'], 'EPs')
-        singles = build_category(spotify_discography['singles'], 'Singles')
-
-        print(f"Built discography cards - Albums: {len(albums)}, EPs: {len(eps)}, Singles: {len(singles)}")
-
-        return {
-            'success': True,
-            'albums': albums,
-            'eps': eps,
-            'singles': singles
-        }
-
-    except Exception as e:
-        print(f"Error building discography: {e}")
-        import traceback
-        traceback.print_exc()
-        return {
-            'success': False,
-            'error': str(e),
-            'albums': [],
-            'eps': [],
-            'singles': []
-        }
-
 # ================================================================================================
 # MUSICBRAINZ ENRICHMENT - PHASE 5 WEB UI INTEGRATION
 # ================================================================================================

webui/static/script.js

@@ -45204,7 +45204,8 @@ async function checkLibraryCompletion(artistName, discography) {
         artist_name: artistName,
         albums: discography.albums || [],
         eps: discography.eps || [],
-        singles: discography.singles || []
+        singles: discography.singles || [],
+        source: discography?.source || null
     };
 
     try {