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

Merge pull request #504 from Nezreka/fix/enhance-quality-without-spotify

Browse source 
Stable-ID dispatch for Enhance Quality + Download Discography
This commit is contained in:
BoulderBadgeDad 2026-05-06 13:10:43 -07:00 committed by GitHub
commit bc5644d5c2
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
6 changed files with 1211 additions and 289 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

563
core/artists/quality.py
View file

@ -2,8 +2,8 @@
`enhance_artist_quality(artist_id, track_ids, deps)` is the route-handler
body for the `/api/library/artist/<artist_id>/enhance` endpoint. It walks
the user's selected tracks, finds the best Spotify (preferred) or iTunes
(fallback) match for each, and queues high-quality re-downloads on the
the user's selected tracks, finds the best metadata match against the
configured primary source, and queues high-quality re-downloads on the
wishlist with `source_type='enhance'`.
Per-track flow:
@ -12,13 +12,43 @@ Per-track flow:
front from `database.get_artist_full_detail`).
2. Read current quality tier from the file extension.
3. Build `matched_track_data` for the wishlist entry, in priority order:
- Direct Spotify lookup via stored `spotify_track_id` (preferred).
- Spotify search fallback using matching_engine queries.
- iTunes/fallback source search.
4. Add to wishlist via `wishlist_service.add_spotify_track_to_wishlist`
- **Direct lookup using stored source IDs** — for every source the
user has configured, if the library track has the corresponding
stored ID (`spotify_track_id` / `deezer_id` / `itunes_track_id` /
`soul_id`), call `client.get_track_details(stored_id)` and convert
the result to the wishlist payload. First success wins; the user's
configured primary source is tried first. Mirrors what Download
Discography does — stable IDs straight to the source's API, no
fuzzy text matching.
- **Multi-source parallel text search fallback** — if no stored ID
resolved, run the shared `core.metadata.multi_source_search`
against every configured source in parallel and pick the best
cross-source match (auto-accept threshold 0.7).
4. Validate the match has non-empty title, album, and artists. Reject
matches with empty fields — those propagated as
"unknown artist - unknown album - unknown track" wishlist entries
pre-fix because the wishlist payload normalizer's truthy-check
passthrough accepted dicts with empty string fields.
5. Add to wishlist via `wishlist_service.add_spotify_track_to_wishlist`
with `source_type='enhance'` and a `source_context` carrying the
original file path, format tier, bitrate, and artist name.
5. Tally `enhanced_count` / `failed_count` / per-track failure reasons.
6. Tally `enhanced_count` / `failed_count` / per-track failure reasons.
The flow originally had Spotify-only logic with an iTunes search-only
fallback. Two failure modes drove the rewrite:
- Users with neither Spotify nor Deezer connected got silent failures
("unknown artist - unknown album - unknown track" wishlist entries)
because iTunes's text search returned junk matches with empty fields
that cleared the 0.7 confidence threshold.
- Library tracks with messy tags ("Title (Live)", featured artists in
the artist field, etc.) failed fuzzy text search even when a perfect
stored ID was available — Download Discography had no such problem
because it resolves albums by stable ID.
Direct-lookup-via-stored-ID matches the Download Discography contract
for every source where we have an ID column. Text search is only the
fallback now.
Returns `(payload_dict, http_status_code)` so the route wrapper can
`jsonify()` and return.
@ -26,28 +56,286 @@ Returns `(payload_dict, http_status_code)` so the route wrapper can
from __future__ import annotations
import logging
import os
from dataclasses import dataclass
from typing import Any, Callable
from typing import Any, Callable, Optional
logger = logging.getLogger(__name__)
from utils.logging_config import get_logger
logger = get_logger('artists.quality')
@dataclass
class ArtistQualityDeps:
"""Bundle of cross-cutting deps the artist quality enhancement needs."""
spotify_client: Any
matching_engine: Any
get_database: Callable[[], Any]
get_wishlist_service: Callable[[], Any]
get_current_profile_id: Callable[[], int]
get_quality_tier_from_extension: Callable
get_metadata_fallback_client: Callable[[], Any]
# Returns ``[(source_name, client), ...]`` for every metadata source
# the user has configured. Powers both the direct-lookup fast path
# (resolves stored source IDs straight from each source's API,
# like Download Discography) and the multi-source parallel text
# search fallback (shared with Track Redownload via
# ``core.metadata.multi_source_search``).
get_metadata_search_sources: Callable[[], list]
def _has_complete_metadata(payload: Optional[dict]) -> bool:
"""Reject matches with empty / missing core fields. Pre-fix, iTunes
returned matches that cleared the 0.7 confidence threshold while
having empty artist / album / title — those propagated as junk
wishlist entries displayed as 'unknown artist - unknown album -
unknown track'."""
if not payload:
return False
if not (payload.get('name') or '').strip():
return False
artists = payload.get('artists') or []
has_artist = any(
(a.get('name') or '').strip() if isinstance(a, dict) else (a or '').strip()
for a in artists
)
if not has_artist:
return False
album = payload.get('album') or {}
if isinstance(album, dict):
if not (album.get('name') or '').strip():
return False
elif not (album or '').strip():
return False
return True
def _build_payload_from_track(track_obj) -> dict:
"""Build a Spotify-shaped wishlist payload from any metadata source's
Track-shaped object (Spotify Track, iTunes Track, Deezer Track,
Discogs Track — they all have the same .id / .name / .artists /
.album / .duration_ms / etc shape because each client mimics
Spotify's surface).
The wishlist's downstream pipeline expects Spotify shape; this helper
is the single place that knows how to produce it. Replaces the
duplicated payload construction that used to live in the Spotify
search path AND the iTunes fallback path.
Does NOT substitute defaults for missing artists / album / title —
``_has_complete_metadata`` rejects empty matches downstream so the
user sees a clear failure instead of a junk wishlist entry with
fabricated values.
"""
image_url = getattr(track_obj, 'image_url', '') or ''
album_images = (
[{'url': image_url, 'height': 600, 'width': 600}]
if image_url else []
)
artist_names = list(getattr(track_obj, 'artists', None) or [])
return {
'id': getattr(track_obj, 'id', ''),
'name': getattr(track_obj, 'name', '') or '',
'artists': [{'name': a} for a in artist_names],
'album': {
'name': getattr(track_obj, 'album', '') or '',
'artists': [{'name': a} for a in artist_names],
'album_type': getattr(track_obj, 'album_type', None) or 'album',
'images': album_images,
'release_date': getattr(track_obj, 'release_date', '') or '',
'total_tracks': 1,
},
'duration_ms': getattr(track_obj, 'duration_ms', 0) or 0,
'track_number': getattr(track_obj, 'track_number', None) or 1,
'disc_number': getattr(track_obj, 'disc_number', None) or 1,
'popularity': getattr(track_obj, 'popularity', None) or 0,
'preview_url': getattr(track_obj, 'preview_url', None),
'external_urls': getattr(track_obj, 'external_urls', None) or {},
}
# Map metadata source name → DB column on the ``tracks`` table that
# stores that source's native track ID. Used to drive the direct-lookup
# fast path: when a library track has a stored ID for source X and the
# user has source X configured, skip fuzzy text search and resolve
# straight from X's API. Mirrors what Download Discography does — stable
# IDs all the way, no fuzzy text matching.
#
# Discogs is release-based and has no per-track ID column; not listed
# here, so direct lookup never tries Discogs (search-fallback still
# runs for Discogs as one of the parallel sources).
_STORED_ID_COLUMNS = {
'spotify': 'spotify_track_id',
'deezer': 'deezer_id',
'itunes': 'itunes_track_id',
'hydrabase': 'soul_id',
}
def _enhanced_to_wishlist_payload(enhanced: dict,
fallback_title: str,
fallback_artist: str,
fallback_album: str) -> Optional[dict]:
"""Convert a ``get_track_details`` enhanced-shape dict to the
Spotify-shape wishlist payload.
Every metadata source's ``get_track_details`` returns the same
"enhanced" intermediate shape (top-level ``id``, ``name``,
``artists`` as a list of strings, ``album.artists`` as strings),
documented and pinned across spotify_client / itunes_client /
deezer_client / hydrabase_client. The wishlist downstream expects
Spotify's native shape (``artists`` as ``[{'name': ...}]``), so
this helper does the conversion in one place.
Spotify's ``raw_data`` field is already in wishlist shape (the
raw Spotify API response), so we return it as-is when detected,
preserving full ``album.images`` and ``external_urls`` that the
enhanced top-level fields drop. Other sources' ``raw_data`` is
in source-native shape and gets ignored.
"""
if not enhanced:
return None
raw = enhanced.get('raw_data')
if isinstance(raw, dict):
raw_artists = raw.get('artists')
if (isinstance(raw_artists, list) and raw_artists
and isinstance(raw_artists[0], dict)):
return raw
artists = enhanced.get('artists') or [fallback_artist]
album_data = enhanced.get('album') or {}
album_artists = album_data.get('artists') or artists
def _to_dict_artists(seq):
return [a if isinstance(a, dict) else {'name': a} for a in seq]
image_url = enhanced.get('image_url') or ''
album_images_field = album_data.get('images')
if isinstance(album_images_field, list) and album_images_field:
album_images = album_images_field
elif image_url:
album_images = [{'url': image_url, 'height': 600, 'width': 600}]
else:
album_images = []
return {
'id': str(enhanced.get('id', '')),
'name': enhanced.get('name') or fallback_title,
'artists': _to_dict_artists(artists),
'album': {
'id': str(album_data.get('id', '')),
'name': album_data.get('name') or fallback_album,
'album_type': album_data.get('album_type', 'album'),
'release_date': album_data.get('release_date', ''),
'total_tracks': album_data.get('total_tracks', 1),
'artists': _to_dict_artists(album_artists),
'images': album_images,
},
'duration_ms': enhanced.get('duration_ms', 0),
'track_number': enhanced.get('track_number', 1),
'disc_number': enhanced.get('disc_number', 1),
'popularity': enhanced.get('popularity', 0),
'preview_url': enhanced.get('preview_url'),
'external_urls': enhanced.get('external_urls', {}),
}
def _try_direct_lookup_all_sources(track: dict,
sources: list,
preferred_source: Optional[str],
title: str,
artist_name: str,
album_title: str
) -> tuple:
"""Try direct ID-based lookup on every source where the library
track has a stored ID. Returns ``(payload, source_name)`` on first
success, or ``(None, None)`` if no source has a stored ID with a
successful lookup.
Mirrors what Download Discography does — stable IDs straight to the
source's API, no fuzzy text matching. Avoids the failure mode where
library text tags don't match the source's canonical title (the
Discord report case: track tagged "Title (Live)" and source has
"Title" → fuzzy search misses, but stored ID resolves directly).
Preferred source attempted first when present in ``sources``,
typically the user's configured primary metadata source — so a
Deezer-primary user gets Deezer art / album shape on the wishlist
entry instead of whichever source happened to have a stored ID
first in iteration order.
"""
def _priority(entry):
name = entry[0]
return 0 if name == preferred_source else 1
ordered = sorted(sources, key=_priority)
for source_name, client in ordered:
column = _STORED_ID_COLUMNS.get(source_name)
if not column:
continue
stored_id = track.get(column)
if not stored_id:
continue
if not hasattr(client, 'get_track_details'):
continue
try:
enhanced = client.get_track_details(str(stored_id))
except Exception as exc:
logger.error(
f"[Enhance] {source_name} direct lookup failed for "
f"ID {stored_id}: {exc}"
)
continue
if not enhanced:
continue
payload = _enhanced_to_wishlist_payload(
enhanced, title, artist_name, album_title,
)
if _has_complete_metadata(payload):
logger.info(
f"[Enhance] Direct lookup matched: {source_name} "
f"ID {stored_id} → '{payload.get('name')}'"
)
return payload, source_name
return None, None
# Minimum match-score threshold for accepting a search-fallback match
# without user confirmation. Mirrors the legacy threshold the enhance
# flow has always used.
_AUTO_ACCEPT_SCORE_THRESHOLD = 0.7
def enhance_artist_quality(artist_id, track_ids, deps: ArtistQualityDeps):
"""Add selected tracks to wishlist for quality enhancement re-download."""
"""Add selected tracks to wishlist for quality enhancement re-download.
Per-track flow:
1. **Direct lookup using stored source IDs** (mirrors what Download
Discography does — stable IDs straight to the source's API, no
fuzzy text matching). For each source the user has configured,
if the library track has the corresponding stored ID
(``spotify_track_id`` / ``deezer_id`` / ``itunes_track_id`` /
``soul_id``), call ``client.get_track_details(stored_id)`` and
convert to wishlist payload. First success wins; preferred
source (user's configured primary) tried first.
2. **Multi-source parallel text search fallback** (via the shared
``core.metadata.multi_source_search`` module — same code path
Track Redownload uses) for tracks with no stored IDs / lookup
misses.
3. **Validation**: reject matches with empty title / album / artists
so the user sees a clear failure instead of an "unknown artist"
wishlist entry.
Pre-refactor: only Spotify had a direct-lookup fast path; everything
else went through fuzzy text search. Discogs / Hydrabase / Deezer-
primary users got far worse coverage than Download Discography
despite both flows asking the same question.
"""
from core.metadata.multi_source_search import TrackQuery, search_all_sources
from core.metadata.registry import get_primary_source
try:
if not track_ids:
return {"success": False, "error": "No track IDs provided"}, 400
@ -73,6 +361,18 @@ def enhance_artist_quality(artist_id, track_ids, deps: ArtistQualityDeps):
track['_album_id'] = album.get('id')
track_lookup[tid] = track
# Resolve every configured metadata source up front.
search_sources = deps.get_metadata_search_sources()
# User's configured primary source — direct-lookup tries this
# first so Deezer-primary users get Deezer payloads on the
# wishlist entry (correct cover art / album shape) even when
# other sources also have stored IDs for the same track.
try:
preferred_source = get_primary_source()
except Exception:
preferred_source = None
enhanced_count = 0
failed_count = 0
failed_tracks = []
@ -95,200 +395,67 @@ def enhance_artist_quality(artist_id, track_ids, deps: ArtistQualityDeps):
title = track.get('title', '') or ''
if not title.strip():
title = os.path.splitext(os.path.basename(file_path))[0]
spotify_tid = track.get('spotify_track_id')
album_title = track.get('_album_title', '')
# Build Spotify track data for wishlist
matched_track_data = None
chosen_source = None
if spotify_tid and deps.spotify_client:
# Direct lookup via stored Spotify ID — raw_data has full Spotify API format
# 1. Direct lookup via every stored source ID — like Download
# Discography. Stable IDs, no fuzzy text matching.
if search_sources:
matched_track_data, chosen_source = _try_direct_lookup_all_sources(
track, search_sources, preferred_source,
title, artist_name, album_title,
)
# 2. Multi-source parallel text search fallback — for tracks
# with no stored IDs / lookup misses.
if not matched_track_data and search_sources:
try:
track_details = deps.spotify_client.get_track_details(spotify_tid)
if track_details and track_details.get('raw_data'):
matched_track_data = track_details['raw_data']
elif track_details:
# Enhanced format — rebuild with images for wishlist compatibility
album_data = track_details.get('album', {})
album_images = []
# Try to get album art from a full album lookup
if album_data.get('id'):
try:
full_album = deps.spotify_client.get_album(album_data['id'])
if full_album and full_album.get('images'):
album_images = full_album['images']
except Exception:
pass
matched_track_data = {
'id': spotify_tid,
'name': track_details.get('name', title),
'artists': [{'name': a} for a in track_details.get('artists', [artist_name])],
'album': {
'id': album_data.get('id', ''),
'name': album_data.get('name', track.get('_album_title', '')),
'album_type': album_data.get('album_type', 'album'),
'release_date': album_data.get('release_date', ''),
'total_tracks': album_data.get('total_tracks', 1),
'artists': [{'name': a} for a in album_data.get('artists', [artist_name])],
'images': album_images,
},
'duration_ms': track_details.get('duration_ms', track.get('duration', 0)),
'track_number': track_details.get('track_number', track.get('track_number', 1)),
'disc_number': track_details.get('disc_number', 1),
'popularity': 0,
'preview_url': None,
'external_urls': {},
}
except Exception as e:
logger.error(f"[Enhance] Spotify lookup failed for {spotify_tid}: {e}")
if not matched_track_data and deps.spotify_client:
# Fallback: Spotify search matching — need full track data for wishlist
try:
temp_track = type('TempTrack', (), {
'name': title, 'artists': [artist_name],
'album': track.get('_album_title', '')
})()
search_queries = deps.matching_engine.generate_download_queries(temp_track)
best_match = None
best_match_raw = None
best_confidence = 0.0
for search_query in search_queries[:3]: # Limit queries
try:
results = deps.spotify_client.search_tracks(search_query, limit=5)
if not results:
continue
for sp_track in results:
artist_conf = max(
(deps.matching_engine.similarity_score(
deps.matching_engine.normalize_string(artist_name),
deps.matching_engine.normalize_string(a)
) for a in (sp_track.artists or [artist_name])),
default=0
)
title_conf = deps.matching_engine.similarity_score(
deps.matching_engine.normalize_string(title),
deps.matching_engine.normalize_string(sp_track.name)
)
combined = artist_conf * 0.5 + title_conf * 0.5
# Small bonus for album tracks over singles
_at = getattr(sp_track, 'album_type', None) or ''
if _at == 'album':
combined += 0.02
elif _at == 'ep':
combined += 0.01
if combined > best_confidence and combined >= 0.7:
best_confidence = combined
best_match = sp_track
if best_confidence >= 0.9:
break
except Exception:
continue
if best_match:
# Fetch full track data from Spotify for proper wishlist format
try:
full_details = deps.spotify_client.get_track_details(best_match.id)
if full_details and full_details.get('raw_data'):
matched_track_data = full_details['raw_data']
else:
raise ValueError("No raw_data from get_track_details")
except Exception:
# Build from Track dataclass with image
album_images = [{'url': best_match.image_url}] if best_match.image_url else []
matched_track_data = {
'id': best_match.id,
'name': best_match.name,
'artists': [{'name': a} for a in best_match.artists],
'album': {
'name': best_match.album,
'artists': [{'name': a} for a in best_match.artists],
'album_type': 'album',
'release_date': getattr(best_match, 'release_date', '') or '',
'images': album_images,
},
'duration_ms': best_match.duration_ms,
'popularity': best_match.popularity or 0,
'preview_url': best_match.preview_url,
'external_urls': best_match.external_urls or {},
}
except Exception as e:
logger.error(f"[Enhance] Search match failed for {title}: {e}")
# Fallback source when Spotify unavailable or no match found
if not matched_track_data:
try:
fallback_client = deps.get_metadata_fallback_client()
itunes_best = None
itunes_best_conf = 0.0
itunes_queries = deps.matching_engine.generate_download_queries(
type('TempTrack', (), {
'name': title, 'artists': [artist_name],
'album': track.get('_album_title', '')
})()
track_query = TrackQuery(
title=title,
artist=artist_name,
album=album_title,
duration_ms=track.get('duration', 0) or 0,
spotify_track_id=track.get('spotify_track_id'),
deezer_id=track.get('deezer_id'),
)
multi_result = search_all_sources(track_query, search_sources)
if multi_result.best_match and multi_result.best_match['score'] >= _AUTO_ACCEPT_SCORE_THRESHOLD:
chosen_source = multi_result.best_match['source']
best_track_obj = multi_result.best_track()
if best_track_obj:
matched_track_data = _build_payload_from_track(best_track_obj)
except Exception as exc:
logger.error(f"[Enhance] Multi-source search failed for {title}: {exc}")
for search_query in itunes_queries[:3]:
try:
itunes_results = fallback_client.search_tracks(search_query, limit=5)
if not itunes_results:
continue
for it_track in itunes_results:
artist_conf = max(
(deps.matching_engine.similarity_score(
deps.matching_engine.normalize_string(artist_name),
deps.matching_engine.normalize_string(a)
) for a in (it_track.artists or [artist_name])),
default=0
)
title_conf = deps.matching_engine.similarity_score(
deps.matching_engine.normalize_string(title),
deps.matching_engine.normalize_string(it_track.name)
)
combined = artist_conf * 0.5 + title_conf * 0.5
# Small bonus for album tracks over singles
_at = getattr(it_track, 'album_type', None) or ''
if _at == 'album':
combined += 0.02
elif _at == 'ep':
combined += 0.01
if combined > itunes_best_conf and combined >= 0.7:
itunes_best_conf = combined
itunes_best = it_track
if itunes_best_conf >= 0.9:
break
except Exception:
continue
if itunes_best:
album_images = [{'url': itunes_best.image_url, 'height': 600, 'width': 600}] if itunes_best.image_url else []
matched_track_data = {
'id': itunes_best.id,
'name': itunes_best.name,
'artists': [{'name': a} for a in itunes_best.artists],
'album': {
'name': itunes_best.album,
'artists': [{'name': a} for a in itunes_best.artists],
'album_type': 'album',
'images': album_images,
'release_date': itunes_best.release_date or '',
'total_tracks': 1,
},
'duration_ms': itunes_best.duration_ms,
'track_number': itunes_best.track_number or 1,
'disc_number': itunes_best.disc_number or 1,
'popularity': itunes_best.popularity or 0,
'preview_url': itunes_best.preview_url,
'external_urls': itunes_best.external_urls or {},
}
logger.warning(f"[Enhance] Fallback match for {title}: {itunes_best.artists[0]} - {itunes_best.name} (conf: {itunes_best_conf:.3f})")
except Exception as e:
logger.error(f"[Enhance] Fallback source failed for {title}: {e}")
# 3. Reject matches with empty / missing core fields.
if not _has_complete_metadata(matched_track_data):
if matched_track_data:
logger.warning(
f"[Enhance] {chosen_source} match for '{title}' rejected — "
f"empty title / album / artists (would render as 'unknown')"
)
matched_track_data = None
if not matched_track_data:
failed_count += 1
failed_tracks.append({'track_id': track_id, 'title': title, 'reason': 'No Spotify or fallback match'})
source_list = ', '.join(name for name, _ in (search_sources or []))
if not source_list:
reason = (
'No metadata source configured — connect Spotify / '
'iTunes / Deezer / Discogs / Hydrabase to enable enhance'
)
else:
reason = (
f'No usable match across {source_list} — '
f'try connecting an additional metadata source'
)
failed_tracks.append({
'track_id': track_id,
'title': title,
'reason': reason,
})
continue
# Add to wishlist with enhance source

241
core/metadata/multi_source_search.py Normal file
View file

@ -0,0 +1,241 @@
"""Multi-source parallel metadata search.
Both the Track Redownload modal and the Artist Enhance Quality flow
need to find the best metadata match for a known track (we have the
title + artist + duration from the user's library; we want to find
the matching entry in Spotify / iTunes / Deezer / Discogs / Hydrabase
to drive the wishlist re-download).
Pre-extraction, redownload had a fully-fledged multi-source parallel
search (parallel ThreadPoolExecutor, per-source query optimization,
"current match" flagging via stored source IDs, per-result scoring)
while enhance had a hardcoded Spotify-direct → Spotify-search →
iTunes-fallback chain that only searched ONE source. That's why
redownload "worked" for users without Spotify (it'd find matches via
iTunes / Deezer in parallel) and enhance silently failed (single
fallback returned junk).
This module owns the search logic. Both endpoints call
``search_all_sources`` and get back the same shape — same scoring,
same source-optimized queries, same "current match" semantics. UI
behavior diverges per-endpoint (redownload renders a picker, enhance
auto-picks the best across all sources) but the metadata-search
contract is shared.
"""
from __future__ import annotations
from concurrent.futures import ThreadPoolExecutor, as_completed
from dataclasses import dataclass, field
from difflib import SequenceMatcher
from typing import Any, Dict, List, Optional, Tuple
from utils.logging_config import get_logger
logger = get_logger('metadata.multi_source_search')
@dataclass
class TrackQuery:
"""Inputs needed to run a multi-source metadata search for one track."""
title: str
artist: str
album: str = ''
# Library-side duration in milliseconds. Used for the duration
# similarity component of scoring; pass 0 when unknown (scoring
# falls back to a neutral 0.5 weight).
duration_ms: int = 0
# Source-native track IDs already stored on the library track,
# used for the "is_current_match" flag in the per-result rendering
# so the UI can highlight the entry that produced the existing file.
spotify_track_id: Optional[str] = None
deezer_id: Optional[str] = None
@dataclass
class MultiSourceResult:
"""Aggregated output from ``search_all_sources``."""
# source_name → list of result dicts (already per-source sorted:
# is_current_match first, then descending match_score). Dict shape
# is JSON-serializable for direct return to the frontend (the
# redownload picker uses these as-is).
metadata_results: Dict[str, List[dict]] = field(default_factory=dict)
# source_name → list of source-native Track objects, parallel-indexed
# to ``metadata_results[source_name]``. Used by callers (Enhance
# Quality) that need to build a wishlist payload from the chosen
# match — the dict shape lacks per-source fields like full album
# data / external_urls / popularity that the wishlist needs.
raw_tracks: Dict[str, List[Any]] = field(default_factory=dict)
# Best match across all sources, or None if every source returned
# nothing. Shape: ``{'source': str, 'index': int, 'score': float}``.
best_match: Optional[dict] = None
def best_track(self) -> Optional[Any]:
"""Convenience: return the source-native Track object for the
cross-source best match, or None if no match was found."""
if not self.best_match:
return None
source = self.best_match['source']
index = self.best_match['index']
tracks = self.raw_tracks.get(source) or []
return tracks[index] if index < len(tracks) else None
def _score_match(query: TrackQuery, result: dict) -> float:
"""Score one result dict against the query.
Weights: title 0.5, artist 0.35, duration 0.15. Duration component
is neutral (0.5) when the library track has no duration on file.
These weights match the redownload pre-extraction implementation
so existing callers keep their scoring behavior identical.
"""
title_sim = SequenceMatcher(
None, query.title.lower(), (result.get('name') or '').lower()
).ratio()
artist_sim = SequenceMatcher(
None, query.artist.lower(), (result.get('artist') or '').lower()
).ratio()
if query.duration_ms:
dur_diff = abs(query.duration_ms - (result.get('duration_ms') or 0))
dur_score = max(0.0, 1.0 - dur_diff / 30000.0)
else:
dur_score = 0.5
return round((title_sim * 0.5 + artist_sim * 0.35 + dur_score * 0.15), 3)
def _build_source_query(source_name: str, query: TrackQuery, clean_title: str) -> str:
"""Build the source-optimized search query string.
Deezer's API responds best to its native field-prefixed syntax
(``artist:"X" track:"Y"``) — empirically returns better matches
than a plain query for ambiguous track names. Other sources use
the artist + clean-title concatenation.
"""
if source_name == 'deezer':
return f'artist:"{query.artist}" track:"{clean_title}"'
return f"{query.artist} {clean_title}"
def _search_one_source(source_name: str, client: Any,
query: TrackQuery, clean_title: str
) -> Tuple[str, List[dict], List[Any]]:
"""Run one source's search with three-tier query fallback.
Tier 1: source-optimized query (Deezer's structured form, others' plain).
Tier 2: plain ``artist + title`` if tier 1 returned nothing.
Tier 3: title-only as last resort.
Returns ``(source_name, results, raw_tracks)``:
- ``results`` are the JSON-serializable dicts (id / name / artist /
etc.), sorted by is_current_match first, then descending match_score
- ``raw_tracks`` are the source-native Track objects, parallel-indexed
to ``results``, for callers that need richer per-source fields
than the dict surface (album_type, external_urls, etc).
"""
try:
primary_q = _build_source_query(source_name, query, clean_title)
plain_q = f"{query.artist} {clean_title}"
title_q = clean_title
logger.info(f"[MultiSourceSearch] Searching {source_name} for: {primary_q}")
track_objs = client.search_tracks(primary_q, limit=10)
if not track_objs and primary_q != plain_q:
track_objs = client.search_tracks(plain_q, limit=10)
if not track_objs and clean_title != plain_q:
track_objs = client.search_tracks(title_q, limit=10)
logger.info(f"[MultiSourceSearch] {source_name} returned {len(track_objs)} results")
scored: List[Tuple[dict, Any]] = []
for t in track_objs:
r = {
'id': str(getattr(t, 'id', '')),
'name': getattr(t, 'name', '') or '',
'artist': ', '.join(t.artists) if getattr(t, 'artists', None) else '',
'album': getattr(t, 'album', '') or '',
'duration_ms': getattr(t, 'duration_ms', 0) or 0,
'image_url': getattr(t, 'image_url', '') or '',
'is_current_match': False,
}
# Flag the result that backs the user's existing library
# track so the UI can highlight it.
if source_name == 'spotify' and query.spotify_track_id and r['id'] == str(query.spotify_track_id):
r['is_current_match'] = True
elif source_name == 'deezer' and query.deezer_id and r['id'] == str(query.deezer_id):
r['is_current_match'] = True
r['match_score'] = _score_match(query, r)
scored.append((r, t))
# Sort dict + raw track in lockstep so raw_tracks[i] is the
# source-native object behind metadata_results[source][i].
scored.sort(key=lambda pair: (-int(pair[0]['is_current_match']), -pair[0]['match_score']))
results = [pair[0] for pair in scored]
raw_tracks = [pair[1] for pair in scored]
return source_name, results, raw_tracks
except Exception as exc:
logger.error(
f"[MultiSourceSearch] Search failed for {source_name}: {exc}",
exc_info=True,
)
return source_name, [], []
def search_all_sources(query: TrackQuery,
sources: List[Tuple[str, Any]],
clean_title: Optional[str] = None,
max_workers: int = 3) -> MultiSourceResult:
"""Run a parallel metadata search across every source in ``sources``.
Args:
query: TrackQuery describing the library track we want to match.
sources: List of ``(name, client)`` pairs. Each client must
implement ``search_tracks(query: str, limit: int) -> List[Track]``
where each Track has ``.id``, ``.name``, ``.artists`` (list),
``.album``, ``.duration_ms``, ``.image_url`` attributes.
All five primary metadata clients (Spotify / iTunes /
Deezer / Discogs / Hydrabase) satisfy this contract.
clean_title: Optional pre-cleaned track title (e.g. with
"(Remastered)" / "(Single Version)" suffixes stripped).
Defaults to ``query.title`` if not supplied.
max_workers: ThreadPoolExecutor pool size. Default 3 matches
the redownload endpoint's pre-extraction default — bumping
higher rate-limits on slower sources without speeding up
the slowest source's response.
Returns:
MultiSourceResult with per-source results + cross-source best match.
"""
if clean_title is None:
clean_title = query.title
if not sources:
return MultiSourceResult()
metadata_results: Dict[str, List[dict]] = {}
raw_tracks: Dict[str, List[Any]] = {}
with ThreadPoolExecutor(max_workers=max_workers) as pool:
futures = {
pool.submit(_search_one_source, name, client, query, clean_title): name
for name, client in sources
}
for future in as_completed(futures):
source_name, results, raws = future.result()
metadata_results[source_name] = results
raw_tracks[source_name] = raws
best_match: Optional[dict] = None
for source, results in metadata_results.items():
if results:
top = results[0]
if best_match is None or top['match_score'] > best_match['score']:
best_match = {
'source': source,
'index': 0,
'score': top['match_score'],
}
return MultiSourceResult(
metadata_results=metadata_results,
raw_tracks=raw_tracks,
best_match=best_match,
)

492
tests/artists/test_quality.py
View file

@ -91,22 +91,50 @@ def _build_deps(
artist_detail=None,
wishlist=None,
fallback_client=None,
fallback_source=None,
search_sources=None,
profile_id=1,
quality_tier=('mp3_320', 4),
):
"""Build deps for tests.
``search_sources`` is the new contract — list of ``(name, client)``
pairs the multi-source search dispatches across. For convenience,
``fallback_client`` + ``fallback_source`` are still supported and
auto-translate to a single-source list when ``search_sources``
isn't passed (preserves the older test ergonomics).
"""
if search_sources is None:
# Default: build a single-source list from fallback_client +
# fallback_source, mirroring the legacy single-source contract.
if fallback_source is None:
fallback_source = 'spotify' if spotify is not None else 'itunes'
if fallback_client is None and fallback_source == 'spotify':
fallback_client = spotify
search_sources = (
[(fallback_source, fallback_client)] if fallback_client else []
)
# Mirror web_server._resolve_search_sources: Spotify is always
# added to the source list when a Spotify client is configured,
# alongside whatever else the user has connected. Tests passing
# ``spotify=`` get Spotify in the source list automatically so
# the direct-lookup fast path can find it.
if spotify is not None and not any(name == 'spotify' for name, _ in search_sources):
search_sources = [('spotify', spotify)] + list(search_sources)
deps = aq.ArtistQualityDeps(
spotify_client=spotify,
matching_engine=matching_engine or _FakeMatchingEngine(),
get_database=lambda: _FakeDatabase(artist_detail=artist_detail),
get_wishlist_service=lambda: wishlist or _FakeWishlist(),
get_current_profile_id=lambda: profile_id,
get_quality_tier_from_extension=lambda fp: quality_tier,
get_metadata_fallback_client=lambda: fallback_client,
get_metadata_search_sources=lambda: list(search_sources),
)
return deps
def _artist_with_track(*, track_id='t1', file_path='/file.mp3', spotify_tid=None):
def _artist_with_track(*, track_id='t1', file_path='/file.mp3',
spotify_tid=None, deezer_tid=None, itunes_tid=None,
soul_id=None):
return {
'success': True,
'artist': {'name': 'Artist Name'},
@ -118,6 +146,9 @@ def _artist_with_track(*, track_id='t1', file_path='/file.mp3', spotify_tid=None
'title': 'Track One',
'file_path': file_path,
'spotify_track_id': spotify_tid,
'deezer_id': deezer_tid,
'itunes_track_id': itunes_tid,
'soul_id': soul_id,
'track_number': 1,
'duration': 180000,
'bitrate': 320,
@ -168,12 +199,17 @@ def test_spotify_direct_lookup_via_track_id_uses_raw_data():
def test_spotify_direct_lookup_enhanced_format_rebuilds_payload():
"""Track details without raw_data → rebuild payload with album images via get_album."""
enhanced = {'name': 'Track One', 'artists': ['Artist Name'],
"""Track details without raw_data → rebuild wishlist-shape payload
from the enhanced top-level fields. Spotify enhanced shape returns
``artists`` as a list of strings; the converter normalizes to
Spotify's wishlist shape (``[{'name': ...}]``). Album images stay
empty when the source doesn't surface them on the enhanced dict —
the wishlist re-download fetches art at download time."""
enhanced = {'id': 'sp-stored', 'name': 'Track One',
'artists': ['Artist Name'],
'album': {'id': 'alb-id', 'name': 'Album X'},
'duration_ms': 180000, 'track_number': 1, 'disc_number': 1}
full_album = {'images': [{'url': 'http://art'}]}
spotify = _FakeSpotify(track_details=enhanced, album=full_album)
spotify = _FakeSpotify(track_details=enhanced)
wishlist = _FakeWishlist()
deps = _build_deps(
spotify=spotify,
@ -185,7 +221,12 @@ def test_spotify_direct_lookup_enhanced_format_rebuilds_payload():
assert payload['enhanced_count'] == 1
md = wishlist.added[0]['spotify_track_data']
assert md['album']['images'] == [{'url': 'http://art'}]
assert md['id'] == 'sp-stored'
assert md['name'] == 'Track One'
# Enhanced format: artists normalized from [str] → [{'name': str}].
assert md['artists'] == [{'name': 'Artist Name'}]
assert md['album']['name'] == 'Album X'
assert md['album']['artists'] == [{'name': 'Artist Name'}]
# ---------------------------------------------------------------------------
@ -193,11 +234,15 @@ def test_spotify_direct_lookup_enhanced_format_rebuilds_payload():
# ---------------------------------------------------------------------------
def test_spotify_search_fallback_when_no_stored_id():
"""No spotify_track_id → search via matching_engine, pick best match."""
track = _SpotifyTrack(name='Track One', artists=['Artist Name'])
raw = {'id': 'sp-search', 'name': 'Track One', 'artists': [{'name': 'Artist Name'}],
'album': {'name': 'Album X'}}
spotify = _FakeSpotify(track_details={'raw_data': raw}, search_results=[track])
"""No spotify_track_id → multi-source text search runs, builds
wishlist payload from the source-native Track object via
``_build_payload_from_track`` (not ``get_track_details`` — search
results already carry enough Track-shape fields, no extra API call
needed)."""
track = _SpotifyTrack(id='sp-search', name='Track One',
artists=['Artist Name'])
track.album = 'Album X'
spotify = _FakeSpotify(search_results=[track])
wishlist = _FakeWishlist()
deps = _build_deps(
spotify=spotify,
@ -208,7 +253,11 @@ def test_spotify_search_fallback_when_no_stored_id():
payload, _ = aq.enhance_artist_quality('artist-1', ['t1'], deps)
assert payload['enhanced_count'] == 1
assert wishlist.added[0]['spotify_track_data'] == raw
md = wishlist.added[0]['spotify_track_data']
assert md['id'] == 'sp-search'
assert md['name'] == 'Track One'
assert md['artists'] == [{'name': 'Artist Name'}]
assert md['album']['name'] == 'Album X'
# ---------------------------------------------------------------------------
@ -240,6 +289,291 @@ def test_fallback_source_when_spotify_none():
assert md['album']['images'] == [{'url': 'http://it', 'height': 600, 'width': 600}]
def test_dispatches_through_primary_source_not_spotify_specific():
"""Architectural pin: when the user's primary metadata source is
Discogs (or any non-Spotify source), the enhance flow searches
THAT source's client, not Spotify. Pre-fix the flow had a
hardcoded Spotify-direct → Spotify-search → iTunes-fallback chain
that ignored the user's actual configured primary source.
"""
discogs_track = _SpotifyTrack(id='dc-1', name='Track One',
artists=['Artist Name'])
discogs_track.track_number = 1
discogs_track.disc_number = 1
discogs_track.album = 'Album X'
discogs_calls = []
class _DiscogsStub:
def search_tracks(self, q, limit=5):
discogs_calls.append(q)
return [discogs_track]
wishlist = _FakeWishlist()
deps = _build_deps(
spotify=None,
artist_detail=_artist_with_track(),
wishlist=wishlist,
fallback_client=_DiscogsStub(),
fallback_source='discogs',
)
payload, _ = aq.enhance_artist_quality('artist-1', ['t1'], deps)
# Discogs (the configured primary) was the one searched; match queued.
assert discogs_calls, "Primary source (Discogs) was not searched"
assert payload['enhanced_count'] == 1
assert wishlist.added[0]['spotify_track_data']['id'] == 'dc-1'
def test_spotify_direct_lookup_runs_as_fast_path_then_falls_back():
"""Architectural pin: Spotify direct-lookup is a fast-path
optimization, NOT a primary-source gate. When the user has Spotify
configured AND a stored spotify_track_id, direct lookup runs first
regardless of which other sources are wired. If it returns nothing,
the multi-source parallel search runs and the cross-source best
match wins (in this test, Discogs).
This pins the post-refactor behavior — pre-refactor direct lookup
was gated on Spotify being the active primary source, which broke
enhance for users without Spotify primary even when they had a
stored Spotify ID.
"""
spotify_calls = []
class _SpotifyStub:
def get_track_details(self, tid):
spotify_calls.append(('details', tid))
return None
def search_tracks(self, q, limit=10):
spotify_calls.append(('search', q))
return []
discogs_track = _SpotifyTrack(id='dc-1', name='Track One',
artists=['Artist Name'])
discogs_track.track_number = 1
discogs_track.disc_number = 1
discogs_track.album = 'Album X'
class _DiscogsStub:
def search_tracks(self, q, limit=10):
return [discogs_track]
wishlist = _FakeWishlist()
deps = _build_deps(
spotify=_SpotifyStub(),
artist_detail=_artist_with_track(spotify_tid='sp-stored'),
wishlist=wishlist,
fallback_client=_DiscogsStub(),
fallback_source='discogs',
)
payload, _ = aq.enhance_artist_quality('artist-1', ['t1'], deps)
# Fast path was attempted (Spotify configured + stored ID present).
assert ('details', 'sp-stored') in spotify_calls
# Fast path returned None → multi-source search ran → Discogs won.
assert payload['enhanced_count'] == 1
assert wishlist.added[0]['spotify_track_data']['id'] == 'dc-1'
# ---------------------------------------------------------------------------
# Direct-lookup-by-stored-ID (priority 1) — applies to every source
# with a stored ID column, not just Spotify
# ---------------------------------------------------------------------------
def test_direct_lookup_via_deezer_id_skips_text_search():
"""Library track has stored deezer_id + Deezer is configured →
enhance fast-paths via deezer_client.get_track_details(id) and skips
fuzzy text search entirely. Mirrors what Download Discography does
(stable IDs, no fuzzy matching). Pre-fix Deezer-primary users went
through text search even when the deezer_id was already on the row.
"""
deezer_calls = []
enhanced_dict = {
'id': '12345', 'name': 'Track One',
'artists': ['Artist Name'],
'album': {'id': 'alb-1', 'name': 'Album X'},
'duration_ms': 180000, 'track_number': 1, 'disc_number': 1,
}
class _DeezerStub:
def get_track_details(self, tid):
deezer_calls.append(('details', tid))
return enhanced_dict
def search_tracks(self, q, limit=10):
deezer_calls.append(('search', q))
return []
wishlist = _FakeWishlist()
deps = _build_deps(
spotify=None,
artist_detail=_artist_with_track(deezer_tid='12345'),
wishlist=wishlist,
fallback_client=_DeezerStub(),
fallback_source='deezer',
)
payload, _ = aq.enhance_artist_quality('artist-1', ['t1'], deps)
assert payload['enhanced_count'] == 1
# Direct-lookup ran, text search did NOT (fast path skipped it).
assert ('details', '12345') in deezer_calls
assert not any(call[0] == 'search' for call in deezer_calls)
md = wishlist.added[0]['spotify_track_data']
assert md['id'] == '12345'
assert md['artists'] == [{'name': 'Artist Name'}]
def test_direct_lookup_via_itunes_id_skips_text_search():
"""Stored itunes_track_id triggers iTunes direct lookup. Same
contract as Spotify / Deezer — get_track_details called, search
not."""
itunes_calls = []
enhanced_dict = {
'id': 'it-9001', 'name': 'Track One',
'artists': ['Artist Name'],
'album': {'id': 'alb-it', 'name': 'Album X'},
'duration_ms': 180000,
}
class _ItunesStub:
def get_track_details(self, tid):
itunes_calls.append(('details', tid))
return enhanced_dict
def search_tracks(self, q, limit=10):
itunes_calls.append(('search', q))
return []
wishlist = _FakeWishlist()
deps = _build_deps(
spotify=None,
artist_detail=_artist_with_track(itunes_tid='it-9001'),
wishlist=wishlist,
fallback_client=_ItunesStub(),
fallback_source='itunes',
)
payload, _ = aq.enhance_artist_quality('artist-1', ['t1'], deps)
assert payload['enhanced_count'] == 1
assert ('details', 'it-9001') in itunes_calls
assert not any(call[0] == 'search' for call in itunes_calls)
def test_direct_lookup_prefers_user_primary_source():
"""Track has stored IDs on multiple sources; user's configured
primary source is tried first. Pin: a Deezer-primary user with
both spotify_track_id and deezer_id stored gets the Deezer payload
(correct cover art / album shape for their setup), not whichever
source happened to come first in registry order.
"""
spotify_calls = []
deezer_calls = []
spotify_enhanced = {
'id': 'sp-1', 'name': 'Track One',
'artists': ['Artist Name'],
'album': {'id': 'sp-alb', 'name': 'Album X'},
}
deezer_enhanced = {
'id': 'dz-1', 'name': 'Track One',
'artists': ['Artist Name'],
'album': {'id': 'dz-alb', 'name': 'Album X'},
}
class _SpotifyStub:
def get_track_details(self, tid):
spotify_calls.append(tid)
return spotify_enhanced
class _DeezerStub:
def get_track_details(self, tid):
deezer_calls.append(tid)
return deezer_enhanced
# Patch get_primary_source to return 'deezer' so we don't depend
# on the test-runner's actual config.
import core.metadata.registry as registry_mod
original = registry_mod.get_primary_source
registry_mod.get_primary_source = lambda **kw: 'deezer'
try:
wishlist = _FakeWishlist()
deps = _build_deps(
spotify=_SpotifyStub(),
artist_detail=_artist_with_track(
spotify_tid='sp-stored', deezer_tid='dz-stored',
),
wishlist=wishlist,
fallback_client=_DeezerStub(),
fallback_source='deezer',
)
payload, _ = aq.enhance_artist_quality('artist-1', ['t1'], deps)
finally:
registry_mod.get_primary_source = original
assert payload['enhanced_count'] == 1
# Deezer (primary) tried first and won — Spotify never queried.
assert deezer_calls == ['dz-stored']
assert spotify_calls == []
md = wishlist.added[0]['spotify_track_data']
assert md['id'] == 'dz-1'
def test_direct_lookup_falls_through_to_text_search_when_no_stored_ids():
"""Track has ZERO stored source IDs → direct lookup yields nothing
→ falls through to multi-source text search. Pin the contract:
direct-lookup is best-effort, not required."""
text_search_track = _SpotifyTrack(id='dz-search', name='Track One',
artists=['Artist Name'])
text_search_track.album = 'Album X'
class _DeezerStub:
def get_track_details(self, tid):
raise AssertionError("should not be called — no stored ID")
def search_tracks(self, q, limit=10):
return [text_search_track]
wishlist = _FakeWishlist()
deps = _build_deps(
spotify=None,
artist_detail=_artist_with_track(), # no stored IDs
wishlist=wishlist,
fallback_client=_DeezerStub(),
fallback_source='deezer',
)
payload, _ = aq.enhance_artist_quality('artist-1', ['t1'], deps)
assert payload['enhanced_count'] == 1
assert wishlist.added[0]['spotify_track_data']['id'] == 'dz-search'
def test_direct_lookup_failure_falls_through_to_text_search():
"""Stored ID exists but direct lookup returns None (network blip,
catalog removal, etc.) → flow falls through to text search rather
than hard-failing. Pin: direct-lookup miss is non-fatal."""
text_search_track = _SpotifyTrack(id='dz-search', name='Track One',
artists=['Artist Name'])
text_search_track.album = 'Album X'
class _DeezerStub:
def get_track_details(self, tid):
return None # direct lookup miss
def search_tracks(self, q, limit=10):
return [text_search_track]
wishlist = _FakeWishlist()
deps = _build_deps(
spotify=None,
artist_detail=_artist_with_track(deezer_tid='9999'),
wishlist=wishlist,
fallback_client=_DeezerStub(),
fallback_source='deezer',
)
payload, _ = aq.enhance_artist_quality('artist-1', ['t1'], deps)
assert payload['enhanced_count'] == 1
# Text-search winner used.
assert wishlist.added[0]['spotify_track_data']['id'] == 'dz-search'
# ---------------------------------------------------------------------------
# Failure modes
# ---------------------------------------------------------------------------
@ -266,21 +600,139 @@ def test_track_with_no_file_path_marked_failed():
def test_no_match_anywhere_marked_failed():
"""No Spotify match AND no fallback match → failed reason 'No Spotify or fallback match'."""
"""No source returns a usable match → failed reason lists the
sources that were searched so user knows what was tried."""
spotify = _FakeSpotify(track_details=None, search_results=[])
fallback = type('FB', (), {
'search_tracks': lambda self, q, limit=5: [],
})()
deps = _build_deps(
spotify=spotify,
artist_detail=_artist_with_track(),
fallback_client=fallback,
)
payload, _ = aq.enhance_artist_quality('artist-1', ['t1'], deps)
assert payload['failed_count'] == 1
assert 'No Spotify or fallback match' in payload['failed_tracks'][0]['reason']
reason = payload['failed_tracks'][0]['reason']
assert 'No usable match across' in reason
assert 'spotify' in reason
assert 'try connecting an additional metadata source' in reason
def test_no_match_without_spotify_lists_searched_sources():
"""When Spotify isn't connected and the configured sources find
nothing, the failure reason lists every searched source. Discord
case: user with no Spotify / Deezer saw enhance silently produce
'unknown artist - unknown album - unknown track' wishlist entries
instead of a clear failure."""
fallback = type('FB', (), {
'search_tracks': lambda self, q, limit=5: [],
})()
deps = _build_deps(
spotify=None,
artist_detail=_artist_with_track(),
fallback_client=fallback,
fallback_source='itunes',
)
payload, _ = aq.enhance_artist_quality('artist-1', ['t1'], deps)
assert payload['failed_count'] == 1
reason = payload['failed_tracks'][0]['reason']
assert 'No usable match across' in reason
assert 'itunes' in reason
def test_no_match_with_no_sources_configured_prompts_setup():
"""User with literally zero metadata sources configured gets a
clear prompt to connect one — instead of a generic 'no match'."""
deps = _build_deps(
spotify=None,
artist_detail=_artist_with_track(),
search_sources=[],
)
payload, _ = aq.enhance_artist_quality('artist-1', ['t1'], deps)
assert payload['failed_count'] == 1
reason = payload['failed_tracks'][0]['reason']
assert 'No metadata source configured' in reason
def test_fallback_match_with_empty_artist_rejected():
"""Per the user-reported bug: an iTunes match that clears the 0.7
confidence threshold but has empty/missing artists is rejected
instead of producing a wishlist entry with empty artist field
(which the wishlist payload normalizer happily accepts and the
UI then displays as 'unknown artist')."""
fallback_track = _SpotifyTrack(id='it-empty', name='Track One',
artists=[], # empty artists list
image_url='')
fallback_track.track_number = 1
fallback_track.disc_number = 1
fallback_track.album = 'Album X'
fallback = type('FB', (), {
'search_tracks': lambda self, q, limit=5: [fallback_track],
})()
wishlist = _FakeWishlist()
deps = _build_deps(
spotify=None,
artist_detail=_artist_with_track(),
wishlist=wishlist,
fallback_client=fallback,
)
payload, _ = aq.enhance_artist_quality('artist-1', ['t1'], deps)
# No wishlist entry with empty fields — match was rejected.
assert payload['enhanced_count'] == 0
assert payload['failed_count'] == 1
assert wishlist.added == []
def test_fallback_match_with_empty_album_rejected():
"""Empty album field on iTunes match → reject (was producing
'unknown album' wishlist entries)."""
fallback_track = _SpotifyTrack(id='it-no-album', name='Track One',
artists=['Artist Name'])
fallback_track.track_number = 1
fallback_track.disc_number = 1
fallback_track.album = '' # empty
fallback = type('FB', (), {
'search_tracks': lambda self, q, limit=5: [fallback_track],
})()
wishlist = _FakeWishlist()
deps = _build_deps(
spotify=None,
artist_detail=_artist_with_track(),
wishlist=wishlist,
fallback_client=fallback,
)
payload, _ = aq.enhance_artist_quality('artist-1', ['t1'], deps)
assert payload['enhanced_count'] == 0
assert payload['failed_count'] == 1
assert wishlist.added == []
def test_fallback_match_with_empty_name_rejected():
"""Empty title on iTunes match → reject."""
fallback_track = _SpotifyTrack(id='it-no-name', name='',
artists=['Artist Name'])
fallback_track.track_number = 1
fallback_track.disc_number = 1
fallback_track.album = 'Album X'
fallback = type('FB', (), {
'search_tracks': lambda self, q, limit=5: [fallback_track],
})()
wishlist = _FakeWishlist()
deps = _build_deps(
spotify=None,
artist_detail=_artist_with_track(),
wishlist=wishlist,
fallback_client=fallback,
)
payload, _ = aq.enhance_artist_quality('artist-1', ['t1'], deps)
assert payload['enhanced_count'] == 0
assert payload['failed_count'] == 1
assert wishlist.added == []
# ---------------------------------------------------------------------------

176
web_server.py
View file

@ -8627,6 +8627,58 @@ def get_artist_discography(artist_id):
from core.metadata.lookup import MetadataLookupOptions
from core.metadata_service import get_artist_discography as _get_artist_discography
# Server-side per-source ID resolution. Look up the library row
# by ANY of the IDs the frontend might send: library DB id,
# spotify_artist_id, itunes_artist_id, deezer_id, or
# musicbrainz_id. Once matched, pull every stored provider ID
# and dispatch the right ID to each source via
# ``artist_source_ids``. Mirrors what the watchlist scanner
# already does.
#
# Without this, the frontend's ID choice fully decides which
# source can answer correctly:
# - sends DB id 194687 → Deezer accepts (wrong: it's a real
# Deezer ID for a different artist)
# - sends Spotify ID `1bDWGdIC...` → Deezer rejects → falls
# back to fuzzy name search → may pick wrong artist
# With server-side resolution, every source gets its OWN stored
# ID regardless of which one the URL carries.
artist_source_ids = {}
try:
_db = get_database()
_conn = _db._get_connection()
try:
_cur = _conn.cursor()
_cur.execute("""
SELECT spotify_artist_id, itunes_artist_id,
deezer_id, musicbrainz_id
FROM artists
WHERE id = ?
OR spotify_artist_id = ?
OR itunes_artist_id = ?
OR deezer_id = ?
OR musicbrainz_id = ?
LIMIT 1
""", (artist_id, artist_id, artist_id, artist_id, artist_id))
_row = _cur.fetchone()
if _row:
if _row['spotify_artist_id']:
artist_source_ids['spotify'] = str(_row['spotify_artist_id'])
if _row['itunes_artist_id']:
artist_source_ids['itunes'] = str(_row['itunes_artist_id'])
if _row['deezer_id']:
artist_source_ids['deezer'] = str(_row['deezer_id'])
if _row['musicbrainz_id']:
artist_source_ids['musicbrainz'] = str(_row['musicbrainz_id'])
logger.info(
f"Discography: resolved per-source IDs for artist_id={artist_id} → "
f"{artist_source_ids}"
)
finally:
_conn.close()
except Exception as _id_exc:
logger.debug(f"Could not resolve per-source artist IDs for {artist_id}: {_id_exc}")
discography = _get_artist_discography(
artist_id,
artist_name=artist_name,
@ -8636,6 +8688,7 @@ def get_artist_discography(artist_id):
skip_cache=False,
max_pages=0,
limit=50,
artist_source_ids=artist_source_ids or None,
),
)
@ -9403,14 +9456,44 @@ def _build_artist_quality_deps():
"""Build the ArtistQualityDeps bundle from web_server.py globals on each call."""
from core.wishlist_service import get_wishlist_service as _get_ws
def _resolve_search_sources():
"""Mirror the Track Redownload modal's source list. Every
configured metadata source contributes to the parallel multi-source
search — Spotify (only when authenticated), iTunes, Deezer, plus
Discogs / Hydrabase when configured."""
sources = []
if spotify_client and spotify_client.is_authenticated():
sources.append(('spotify', spotify_client))
try:
sources.append(('itunes', _get_itunes_client()))
except Exception:
pass
try:
sources.append(('deezer', _get_deezer_client()))
except Exception:
pass
# Discogs needs an explicit token; only include when configured.
try:
_discogs_token = config_manager.get('discogs.token', '')
if _discogs_token:
sources.append(('discogs', _get_discogs_client(_discogs_token)))
except Exception:
pass
# Hydrabase only when connected (dev-mode + active client).
try:
if hydrabase_client and hydrabase_client.is_connected():
sources.append(('hydrabase', hydrabase_client))
except Exception:
pass
return sources
return _artists_quality.ArtistQualityDeps(
spotify_client=spotify_client,
matching_engine=matching_engine,
get_database=get_database,
get_wishlist_service=_get_ws,
get_current_profile_id=get_current_profile_id,
get_quality_tier_from_extension=_get_quality_tier_from_extension,
get_metadata_fallback_client=_get_metadata_fallback_client,
get_metadata_search_sources=_resolve_search_sources,
)
@ -11557,21 +11640,16 @@ def redownload_search_metadata(track_id):
'thumb_url': thumb_url,
}
# Search all available metadata sources in parallel
from concurrent.futures import ThreadPoolExecutor, as_completed
from difflib import SequenceMatcher
# Search all available metadata sources in parallel via the
# shared module — same code path the Enhance Quality flow uses
# so both endpoints have identical scoring + per-source query
# optimization + current-match flagging.
from core.metadata.multi_source_search import (
TrackQuery,
search_all_sources,
)
def _score_metadata_match(result):
"""Score a metadata result against the current track."""
title_sim = SequenceMatcher(None, track_title.lower(), (result.get('name') or '').lower()).ratio()
artist_sim = SequenceMatcher(None, artist_name.lower(), (result.get('artist') or '').lower()).ratio()
dur_diff = abs((row['duration'] or 0) - (result.get('duration_ms') or 0))
dur_score = max(0, 1 - dur_diff / 30000) if row['duration'] else 0.5
return round((title_sim * 0.5 + artist_sim * 0.35 + dur_score * 0.15), 3)
metadata_results = {}
sources_to_search = []
if spotify_client and spotify_client.is_authenticated():
sources_to_search.append(('spotify', spotify_client))
try:
@ -11583,65 +11661,23 @@ def redownload_search_metadata(track_id):
except Exception as e:
logger.debug(f"Deezer client not available for redownload search: {e}")
# Build source-optimized queries
deezer_query = f'artist:"{artist_name}" track:"{clean_title}"'
def _search_source(source_name, client):
try:
# Deezer works best with structured artist:/track: queries
search_q = deezer_query if source_name == 'deezer' else query
logger.info(f"[Redownload] Searching {source_name} for: {search_q}")
track_objs = client.search_tracks(search_q, limit=10)
# If no results, try plain query as fallback
if not track_objs and search_q != query:
track_objs = client.search_tracks(query, limit=10)
# Last resort: title only
if not track_objs and clean_title != query:
track_objs = client.search_tracks(clean_title, limit=10)
logger.info(f"[Redownload] {source_name} returned {len(track_objs)} results")
results = []
for t in track_objs:
r = {
'id': str(t.id),
'name': t.name,
'artist': ', '.join(t.artists) if t.artists else '',
'album': t.album or '',
'duration_ms': t.duration_ms or 0,
'image_url': t.image_url or '',
'is_current_match': False,
}
# Flag current match
if source_name == 'spotify' and row['spotify_track_id'] and str(t.id) == str(row['spotify_track_id']):
r['is_current_match'] = True
elif source_name == 'deezer' and row['deezer_id'] and str(t.id) == str(row['deezer_id']):
r['is_current_match'] = True
r['match_score'] = _score_metadata_match(r)
results.append(r)
results.sort(key=lambda x: (-int(x['is_current_match']), -x['match_score']))
return source_name, results
except Exception as e:
logger.error(f"[Redownload] Metadata search failed for {source_name}: {e}", exc_info=True)
return source_name, []
with ThreadPoolExecutor(max_workers=3) as pool:
futures = {pool.submit(_search_source, name, client): name for name, client in sources_to_search}
for future in as_completed(futures):
source_name, results = future.result()
metadata_results[source_name] = results
# Find best overall match
best_match = None
for source, results in metadata_results.items():
if results:
top = results[0]
if not best_match or top['match_score'] > best_match['score']:
best_match = {'source': source, 'index': 0, 'score': top['match_score']}
track_query = TrackQuery(
title=track_title,
artist=artist_name,
album=row['album_title'] or '',
duration_ms=row['duration'] or 0,
spotify_track_id=row['spotify_track_id'],
deezer_id=row['deezer_id'],
)
result = search_all_sources(
track_query, sources_to_search, clean_title=clean_title,
)
return jsonify({
"success": True,
"current_track": current_track,
"metadata_results": metadata_results,
"best_match": best_match,
"metadata_results": result.metadata_results,
"best_match": result.best_match,
})
except Exception as e:
logger.error(f"Error in redownload metadata search: {e}", exc_info=True)

26
webui/static/helper.js
View file

@ -3432,6 +3432,8 @@ const WHATS_NEW = {
'2.4.2': [
// --- post-2.4.1 dev work — entries hidden by _getLatestWhatsNewVersion until the build version bumps ---
{ date: 'Unreleased — 2.4.2 dev cycle' },
{ title: 'Fix: Download Discography Showed Wrong Artist\'s Albums', desc: 'clicked download discography on 50 cent → modal showed young hot rod\'s albums. clicked weird al → modal showed the beatles. cause: the endpoint received whichever single artist id the frontend happened to pick (spotify or itunes or deezer or library db id) and dispatched it as-is to whichever source it queried. when the picked id didn\'t match the queried source\'s id format, lookup either returned wrong-artist results (numeric collisions — db id 194687 was a real deezer artist for someone else) or fell back to a fuzzy name search that picked a wrong artist. the per-source id dispatch mechanism (`MetadataLookupOptions.artist_source_ids`) already existed and the watchlist scanner already used it; the on-demand discography endpoint just wasn\'t wired to it. fixed: when the url artist_id matches a library row by ANY stored id (db id, spotify_artist_id, itunes_artist_id, deezer_id, musicbrainz_id), backend pulls every stored provider id and dispatches the right id to each source. each source gets its OWN stored id regardless of what the url carries. when the url id is a non-library source-native id and the row lookup misses entirely, behavior is identical to before (single-id fallback). also fixed two log-namespace bugs: enhance quality and multi-source search were writing through `getLogger(__name__)` which resolves to `core.artists.quality` / `core.metadata.multi_source_search` — neither under the soulsync handler — so every diagnostic line was silently dropped. switched both to `get_logger()` from utils.logging_config so they actually land in app.log.', page: 'library' },
{ title: 'Enhance Quality: Direct ID Lookup Like Download Discography', desc: 'two related fixes. (1) discord report: enhance quality on an artist with neither spotify nor deezer connected added tracks as "unknown artist - unknown album - unknown track". enhance was running a single-source itunes fallback chain that returned junk matches with empty fields, while track redownload had been doing parallel multi-source search the whole time. extracted that search into `core/metadata/multi_source_search.py` and pointed both flows at it. (2) followup: enhance was still using fuzzy text search even when the library track had a stored source ID (spotify_track_id / deezer_id / itunes_track_id / soul_id) on the row, which meant tracks with messy tags ("Title (Live)", featured artists in the artist field, etc.) failed to match even though a perfect ID was sitting right there. download discography never had this problem because it resolves albums by stable ID, not by name. enhance now does the same: for every source you have configured, if the track has a stored ID for that source, it calls `get_track_details(id)` directly — no fuzzy matching. preferred source (your configured primary) is tried first so a deezer-primary user gets deezer payloads on the wishlist entry. text search is only the fallback now (kicks in for tracks with no stored IDs). also fixed the modal toast that lied "matching tracks to spotify..." regardless of which sources were actually being queried.', page: 'library' },
{ title: 'Internal: Media Server Engine Cin/JohnBaumb Pass', desc: 'internal — applied the same architectural cleanups the download engine PR went through to the media server engine PR before review. (1) every server client (Plex / Jellyfin / Navidrome / SoulSync) now explicitly inherits `MediaServerClient` instead of relying on structural typing — drift in any class fails at the conformance test boundary. (2) generic accessors on the engine: `configured_clients()` (replaces per-server `if X and X.is_connected(): clients[name] = X` chains in web_server.py) and `reload_config(name=None)` (generic dispatch instead of per-client reload calls). (3) singleton factory: `get_media_server_engine()` / `set_media_server_engine()` matching the metadata + download engine shape. web_server.py boots via `set_media_server_engine(...)` so factory + global handle share state. (4) ~70 direct `plex_client.X` / `jellyfin_client.X` / `navidrome_client.X` / `soulsync_library_client.X` attribute reaches in web_server.py migrated to `media_server_engine.client(\'<name>\').X`. ~60 standalone refs (truthy checks, media_client assignments, source-name tuples) also routed through the engine. (5) the per-server `plex_client` / `jellyfin_client` / `navidrome_client` / `soulsync_library_client` globals in web_server.py are gone entirely — engine owns the client instances now, every caller reaches via `media_server_engine.client(\'<name>\')`. four multi-client consumers (`PlaylistSyncService`, `ListeningStatsWorker`, `WebScanManager`, discovery `SyncDeps`) refactored to take the engine instead of separate per-server kwargs. (6) `TrackInfo` and `PlaylistInfo` lifted out of `core/plex_client.py` / `jellyfin_client.py` / `navidrome_client.py` (each was defining a near-identical copy) into the neutral `core/media_server/types.py` module — same lift Cin caught on the download `TrackResult`/`AlbumResult`/`DownloadStatus` situation. consumers (matching engine, sync service) get one import. zero behavior change.' },
{ title: 'Internal: Media Server Engine Foundation', desc: 'internal — companion to the download engine refactor. introduces a media-server engine + plugin contract on top of the four server clients (plex / jellyfin / navidrome / soulsync standalone). pre-refactor web_server.py held four separate per-server client globals that every dispatch site reached individually. new `core/media_server/` package provides `MediaServerEngine` that owns the per-server clients + a small set of generic accessors (`client(name)`, `active_client()`, `configured_clients()`, `reload_config(name)`) so call sites use one canonical lookup pattern. plugin contract requires the four methods every server actually implements (is_connected, ensure_connection, get_all_artists, get_all_album_ids) — methods that exist on most-but-not-all servers (search_tracks, trigger_library_scan, get_library_stats, etc.) are listed in `KNOWN_PER_SERVER_METHODS` for discoverability and reached directly via `engine.client(name).<method>` since there\'s no uniform safe-default that fits every method. honest scope: the four uniform-shape `is_connected` dispatches were lifted into `engine.is_connected()` (the one cross-server wrapper kept on the engine); the ~18 server-specific chains that do genuinely different per-server work (playlist track replace, metadata sync, scan strategies) stay explicit at the call site per the "lift what\'s truly shared" standard, but reach the per-server client through the engine. 42 tests pin: per-server observable behavior (4 server pinning files, 20 tests), engine surface + accessor contracts (15 tests), structural conformance + explicit-inheritance (9 tests). zero behavior change for users.' },
{ title: 'Drop SoundCloud Preview Snippets at Search Time', desc: 'soundcloud serves a ~30s preview clip for tracks that are gated behind go+ / login (very common for major-label uploads — official content basically doesn\'t exist on soundcloud, so what shows up is bootlegs, fan uploads, type beats, and these 30s previews). yt-dlp accepts the preview as the download payload, the post-download integrity check catches the duration mismatch and quarantines the file, but the user just sees "all candidates failed" with no explanation. previews also showed up in the candidate-review modal where clicking one bypassed validation and downloaded the same broken file. now `filter_soundcloud_previews` drops these candidates at every entry point — auto-search scoring, modal-cache fallback, AND the not-found raw-results path — so previews never reach the matcher OR the user. drops candidates < 35s or below half the expected duration, gated on expected being non-trivially long (>60s) so genuine short tracks still pass. also fixed a silent regression in the hybrid-fallback path where the per-source attribute removal left `getattr(orch, \'youtube\', None)` returning None for every source — fallback never fired. now resolves through the orchestrator\'s `client(name)` accessor.', page: 'downloads' },
@ -3774,6 +3776,30 @@ const WHATS_NEW = {
// Section shape: { title, description, features: [bullet strings],
// usage_note?: 'optional hint shown at the bottom' }
const VERSION_MODAL_SECTIONS = [
{
title: "Download Discography No Longer Shows Wrong Artist",
description: "clicked download discog on 50 cent → modal showed young hot rod\'s albums. clicked weird al → modal showed beatles albums. real bug, not just data weirdness.",
features: [
"• endpoint received whichever single artist id the frontend happened to pick and dispatched it as-is to whichever source it queried — when the picked id didn\'t match the queried source\'s id format, lookup either returned wrong-artist results (numeric id collisions) or fell back to a fuzzy name search that picked a different artist",
"• fixed: backend now looks up the library row by ANY stored id (db id, spotify, itunes, deezer, musicbrainz) and dispatches the correct stored id to each source — every source gets its OWN id regardless of what the frontend chose to send",
"• mechanism already existed (`MetadataLookupOptions.artist_source_ids`) and the watchlist scanner already used it — discog endpoint just wasn\'t wired to it",
"• also fixed two log-namespace bugs: enhance quality + multi-source search were writing to a logger with no handlers, so every diagnostic line was silently dropped — now lands in app.log where you can actually see them",
],
usage_note: "no settings to change — applies on next click of download discography",
},
{
title: "Enhance Quality Now Behaves Like Download Discography",
description: "discord report — clicking enhance quality on an artist with no spotify/deezer was adding tracks as \"unknown artist - unknown album - unknown track\". root issue ran deeper than that single edge case.",
features: [
"• enhance used to fuzzy text search the configured primary source only — a single-source itunes fallback returning junk matches with empty fields, while track redownload had been doing parallel multi-source search the whole time",
"• extracted that search into a shared module; both enhance and redownload now hit every configured source in parallel and pick the cross-source best match",
"• bigger fix: enhance now uses stored source IDs (spotify_track_id / deezer_id / itunes_track_id / soul_id) the same way download discography uses album IDs — direct lookup against each source's API, no fuzzy text matching, no failures from messy tags like \"Title (Live)\" or featured artists in the artist field",
"• preferred source (your configured primary) tried first so a deezer-primary user gets deezer payloads on the wishlist entry",
"• text search is only the fallback now — kicks in only when no stored IDs exist for the track",
"• modal toast no longer lies \"matching tracks to spotify\" regardless of which sources are actually configured",
],
usage_note: "no settings to change — applies on next click of enhance quality",
},
{
title: "Watchlist No Longer Re-Downloads Compilations",
description: "compilation / soundtrack tracks were getting redownloaded on every watchlist scan because the album-name fuzzy check failed on naming drift between spotify and your media server.",

2
webui/static/stats-automations.js
View file

@ -7669,7 +7669,7 @@ async function submitEnhanceQuality() {
submitBtn.disabled = true;
submitBtn.innerHTML = '<span class="enhance-spinner"></span>Processing...';
}
if (footerInfo) footerInfo.textContent = 'Matching tracks to Spotify and adding to wishlist...';
if (footerInfo) footerInfo.textContent = 'Matching tracks across metadata sources and adding to wishlist...';
try {
const resp = await fetch(`/api/library/artist/${_enhanceArtistId}/enhance`, {