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

Merge pull request #327 from kettui/fix/artist-details-refactoring

Browse source 
Refactor artist-detail discography flow
This commit is contained in:
BoulderBadgeDad 2026-04-19 12:53:35 -07:00 committed by GitHub
commit 12087f2407
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
5 changed files with 337 additions and 249 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
.github/workflows/build-and-test.yml vendored
View file

@ -25,4 +25,6 @@ jobs:
run: python -m compileall api core database services scripts web_server.py wsgi.py beatport_unified_scraper.py
- name: Run tests
run: pytest
env:
PYTHONPATH: ${{ github.workspace }}
run: python -m pytest

200
core/metadata_service.py
View file

@ -343,6 +343,112 @@ 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_suffix_pattern = re.compile(
r'\s*[\(\[][^()\[\]]*\b(?:edition|editions|deluxe|remaster|remastered|'
r'explicit|clean|version|anniversary|collector|expanded|redux)\b[^()\[\]]*[\)\]]\s*$',
re.IGNORECASE,
)
legacy_suffix_pattern = re.compile(
r'\s*-\s*(explicit|clean|deluxe edition|single)\s*$',
re.IGNORECASE,
)
variant_keyword_pattern = re.compile(
r'\b(?:edition|editions|deluxe|remaster|remastered|explicit|clean|version|'
r'anniversary|collector|expanded|redux)\b',
re.IGNORECASE,
)
def _clean_title(title: Any) -> str:
cleaned = str(title or '').strip().lower()
while True:
new_cleaned = variant_suffix_pattern.sub('', cleaned).strip()
new_cleaned = legacy_suffix_pattern.sub('', new_cleaned).strip()
if new_cleaned == cleaned:
break
cleaned = new_cleaned
cleaned = re.sub(r'\s+', ' ', cleaned).strip()
return cleaned
def _has_variant_suffix(title: Any) -> bool:
raw = str(title or '').strip()
return bool(re.search(r'[\(\[][^\)\]]*' + variant_keyword_pattern.pattern + r'[^\)\]]*[\)\]]\s*$', raw, flags=re.IGNORECASE))
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 '')
has_variant_suffix = _has_variant_suffix(title)
# Higher is better.
return (
1 if not _is_compilation(release) else 0,
1 if not has_variant_suffix 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 +529,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))

97
tests/test_metadata_service_discography.py
View file

@ -383,3 +383,100 @@ 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-swedish",
"name": "Variant Album (Swedish Edition)",
"album_type": "album",
"image_url": "https://img.example/swedish.jpg",
"release_date": "2024-01-05",
"total_tracks": 12,
},
{
"id": "album-remaster",
"name": "Variant Album (2023 Abbey Road Remaster)",
"album_type": "album",
"image_url": "https://img.example/remaster.jpg",
"release_date": "2024-01-05",
"total_tracks": 10,
},
],
"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-standard"]
assert result["albums"][0]["title"] == "Variant Album"
assert result["albums"][0]["track_count"] == 10

282
web_server.py
View file

@ -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
spotify_artist_data = None
# Get source-priority discography for proper categorization and missing releases
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']:
print(f"Spotify discography found - Albums: {len(spotify_discography['albums'])}, EPs: {len(spotify_discography['eps'])}, Singles: {len(spotify_discography['singles'])}")
artist_detail_discography = _get_artist_detail_discography(
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
spotify_artist_data = {
'spotify_artist_id': spotify_discography.get('spotify_artist_id'),
'spotify_artist_name': spotify_discography.get('spotify_artist_name'),
'artist_image': spotify_discography.get('artist_image')
}
# Merge owned and Spotify data for complete picture
merged_discography = merge_discography_data(owned_releases, spotify_discography, db=database, artist_name=artist_info['name'])
if artist_detail_discography['success']:
print(
"Source-priority discography found - "
f"Albums: {len(artist_detail_discography['albums'])}, "
f"EPs: {len(artist_detail_discography['eps'])}, "
f"Singles: {len(artist_detail_discography['singles'])}"
)
merged_discography = artist_detail_discography
else:
print(f"Spotify discography not found: {spotify_discography.get('error', 'Unknown error')}")
# Fall back to our database categorization
print(f"Source-priority discography not found: {artist_detail_discography.get('error', 'Unknown error')}")
merged_discography = owned_releases
except Exception as spotify_error:
print(f"Error fetching Spotify data: {spotify_error}")
# Fall back to our database categorization
except Exception as detail_error:
print(f"Error fetching source-priority discography: {detail_error}")
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
@ -50727,232 +50741,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
# ================================================================================================

3
webui/static/script.js
View file

@ -45207,7 +45207,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 {