mirror of https://github.com/Nezreka/SoulSync.git
Pulls the 284-line artist quality enhancement helper out of
`web_server.py` into a new `core/artists/` package. Flask route handler
split: route + request parsing stay in web_server.py, the body lifts to
a pure function returning `(payload_dict, http_status_code)`.
What `enhance_artist_quality` does:
1. Validate request: track_ids must be non-empty, artist must exist.
2. Build a `track_lookup` from `database.get_artist_full_detail` so each
selected track resolves with its album context.
3. Per track:
- Read current quality tier from the file extension.
- Build `matched_track_data` for the wishlist entry, in priority
order:
- Spotify direct lookup via stored `spotify_track_id` (preferred).
Uses raw API data when available; otherwise rebuilds the payload
and pulls album images via a follow-up `get_album` call.
- Spotify search fallback using matching_engine queries with
artist+title similarity scoring (album-type bonus for albums,
smaller bonus for EPs). Stops at first >= 0.9 confidence match.
- iTunes/fallback source search with the same scoring shape.
- 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, original_tier, and
artist_name.
- Tally `enhanced_count` / `failed_count` / per-track failure reasons.
4. Return `{success, enhanced_count, failed_count, failed_tracks}` 200.
Dependencies injected via `ArtistQualityDeps` (7 fields) — spotify_client,
matching_engine, get_database, get_wishlist_service,
get_current_profile_id, get_quality_tier_from_extension,
get_metadata_fallback_client.
Diff vs original after `deps.X` → global X normalization is **1 line of
cosmetic drift** — the success return now uses an explicit `(payload, 200)`
tuple to keep all returns shape-consistent for the wrapper. Flask treats
`jsonify(x)` and `(jsonify(x), 200)` identically. 284 lines orig = 285
lines lifted, body otherwise byte-identical.
Tests: 10 new under tests/artists/test_quality.py covering input
validation (empty track_ids, artist not found), Spotify direct lookup
via raw_data, Spotify direct lookup with enhanced format requiring
album image rebuild, Spotify search fallback, iTunes/fallback source
match path, track-not-found and no-file-path failure modes, complete
no-match failure, and source_context payload assertions (enhance flag,
file path, format tier, bitrate, source_type).
Full suite: 1340 passing (was 1330). Ruff clean.
pull/423/head
Broque Thomas
4 weeks ago
parent
501ef1ba63
commit
91978656a5
@ -0,0 +1,329 @@
|
||||
"""Artist quality enhancement helper.
|
||||
|
||||
`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
|
||||
wishlist with `source_type='enhance'`.
|
||||
|
||||
Per-track flow:
|
||||
|
||||
1. Resolve the existing track via the artist's full detail map (built up
|
||||
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`
|
||||
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.
|
||||
|
||||
Returns `(payload_dict, http_status_code)` so the route wrapper can
|
||||
`jsonify()` and return.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import os
|
||||
from dataclasses import dataclass
|
||||
from typing import Any, Callable
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
@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]
|
||||
|
||||
|
||||
def enhance_artist_quality(artist_id, track_ids, deps: ArtistQualityDeps):
|
||||
"""Add selected tracks to wishlist for quality enhancement re-download."""
|
||||
try:
|
||||
if not track_ids:
|
||||
return {"success": False, "error": "No track IDs provided"}, 400
|
||||
|
||||
database = deps.get_database()
|
||||
wishlist_service = deps.get_wishlist_service()
|
||||
profile_id = deps.get_current_profile_id()
|
||||
|
||||
# Get artist info
|
||||
artist_result = database.get_artist_full_detail(artist_id)
|
||||
if not artist_result.get('success'):
|
||||
return {"success": False, "error": "Artist not found"}, 404
|
||||
|
||||
artist_name = artist_result.get('artist', {}).get('name', 'Unknown Artist')
|
||||
|
||||
# Build lookup of all tracks for this artist
|
||||
track_lookup = {}
|
||||
for album in artist_result.get('albums', []):
|
||||
album_title = album.get('title', '')
|
||||
for track in album.get('tracks', []):
|
||||
tid = str(track.get('id', ''))
|
||||
track['_album_title'] = album_title
|
||||
track['_album_id'] = album.get('id')
|
||||
track_lookup[tid] = track
|
||||
|
||||
enhanced_count = 0
|
||||
failed_count = 0
|
||||
failed_tracks = []
|
||||
|
||||
for track_id in track_ids:
|
||||
track_id_str = str(track_id)
|
||||
track = track_lookup.get(track_id_str)
|
||||
if not track:
|
||||
failed_count += 1
|
||||
failed_tracks.append({'track_id': track_id, 'reason': 'Track not found'})
|
||||
continue
|
||||
|
||||
file_path = track.get('file_path')
|
||||
if not file_path:
|
||||
failed_count += 1
|
||||
failed_tracks.append({'track_id': track_id, 'reason': 'No file path'})
|
||||
continue
|
||||
|
||||
tier_name, tier_num = deps.get_quality_tier_from_extension(file_path)
|
||||
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')
|
||||
|
||||
# Build Spotify track data for wishlist
|
||||
matched_track_data = None
|
||||
|
||||
if spotify_tid and deps.spotify_client:
|
||||
# Direct lookup via stored Spotify ID — raw_data has full Spotify API format
|
||||
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', '')
|
||||
})()
|
||||
)
|
||||
|
||||
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}")
|
||||
|
||||
if not matched_track_data:
|
||||
failed_count += 1
|
||||
failed_tracks.append({'track_id': track_id, 'title': title, 'reason': 'No Spotify or fallback match'})
|
||||
continue
|
||||
|
||||
# Add to wishlist with enhance source
|
||||
source_context = {
|
||||
'enhance': True,
|
||||
'original_file_path': file_path,
|
||||
'original_format': tier_name,
|
||||
'original_bitrate': track.get('bitrate'),
|
||||
'original_tier': tier_num,
|
||||
'artist_name': artist_name,
|
||||
}
|
||||
|
||||
success = wishlist_service.add_spotify_track_to_wishlist(
|
||||
spotify_track_data=matched_track_data,
|
||||
failure_reason=f"Quality enhance - upgrading from {tier_name.replace('_', ' ').title()}",
|
||||
source_type='enhance',
|
||||
source_context=source_context,
|
||||
profile_id=profile_id
|
||||
)
|
||||
|
||||
if success:
|
||||
enhanced_count += 1
|
||||
logger.info(f"[Enhance] Queued for upgrade: {artist_name} - {title} ({tier_name})")
|
||||
else:
|
||||
failed_count += 1
|
||||
failed_tracks.append({'track_id': track_id, 'title': title, 'reason': 'Wishlist add failed'})
|
||||
|
||||
return {
|
||||
'success': True,
|
||||
'enhanced_count': enhanced_count,
|
||||
'failed_count': failed_count,
|
||||
'failed_tracks': failed_tracks
|
||||
}, 200
|
||||
except Exception as e:
|
||||
logger.error(f"[Enhance] {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
return {"success": False, "error": str(e)}, 500
|
||||
@ -0,0 +1,312 @@
|
||||
"""Tests for core/artists/quality.py — artist quality enhancement helper."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
|
||||
import pytest
|
||||
|
||||
from core.artists import quality as aq
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Fakes
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@dataclass
|
||||
class _SpotifyTrack:
|
||||
id: str = 'sp-1'
|
||||
name: str = 'Found'
|
||||
artists: list = None
|
||||
album: str = 'Album'
|
||||
duration_ms: int = 200000
|
||||
image_url: str = ''
|
||||
popularity: int = 50
|
||||
preview_url: str = ''
|
||||
external_urls: dict = None
|
||||
album_type: str = 'album'
|
||||
release_date: str = '2024-01-01'
|
||||
|
||||
def __post_init__(self):
|
||||
if self.artists is None:
|
||||
self.artists = ['Artist Name']
|
||||
if self.external_urls is None:
|
||||
self.external_urls = {}
|
||||
|
||||
|
||||
class _FakeSpotify:
|
||||
def __init__(self, track_details=None, search_results=None, album=None):
|
||||
self._track_details = track_details
|
||||
self._search_results = search_results or []
|
||||
self._album = album
|
||||
self.search_calls = []
|
||||
|
||||
def get_track_details(self, track_id):
|
||||
return self._track_details
|
||||
|
||||
def get_album(self, album_id):
|
||||
return self._album
|
||||
|
||||
def search_tracks(self, query, limit=5):
|
||||
self.search_calls.append((query, limit))
|
||||
return self._search_results
|
||||
|
||||
|
||||
class _FakeMatchingEngine:
|
||||
def generate_download_queries(self, track):
|
||||
return [f"{track.artists[0]} {track.name}"]
|
||||
|
||||
def normalize_string(self, s):
|
||||
return (s or '').lower().strip()
|
||||
|
||||
def similarity_score(self, a, b):
|
||||
if a == b:
|
||||
return 1.0
|
||||
if not a or not b:
|
||||
return 0.0
|
||||
return 0.95 if a in b or b in a else 0.0
|
||||
|
||||
|
||||
class _FakeWishlist:
|
||||
def __init__(self):
|
||||
self.added = []
|
||||
|
||||
def add_spotify_track_to_wishlist(self, **kwargs):
|
||||
self.added.append(kwargs)
|
||||
return True
|
||||
|
||||
|
||||
class _FakeDatabase:
|
||||
def __init__(self, artist_detail=None):
|
||||
self._artist_detail = artist_detail or {'success': False}
|
||||
|
||||
def get_artist_full_detail(self, artist_id):
|
||||
return self._artist_detail
|
||||
|
||||
|
||||
def _build_deps(
|
||||
*,
|
||||
spotify=None,
|
||||
matching_engine=None,
|
||||
artist_detail=None,
|
||||
wishlist=None,
|
||||
fallback_client=None,
|
||||
profile_id=1,
|
||||
quality_tier=('mp3_320', 4),
|
||||
):
|
||||
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,
|
||||
)
|
||||
return deps
|
||||
|
||||
|
||||
def _artist_with_track(*, track_id='t1', file_path='/file.mp3', spotify_tid=None):
|
||||
return {
|
||||
'success': True,
|
||||
'artist': {'name': 'Artist Name'},
|
||||
'albums': [{
|
||||
'id': 'a1',
|
||||
'title': 'Album X',
|
||||
'tracks': [{
|
||||
'id': track_id,
|
||||
'title': 'Track One',
|
||||
'file_path': file_path,
|
||||
'spotify_track_id': spotify_tid,
|
||||
'track_number': 1,
|
||||
'duration': 180000,
|
||||
'bitrate': 320,
|
||||
}],
|
||||
}],
|
||||
}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Input validation
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_no_track_ids_returns_400():
|
||||
deps = _build_deps()
|
||||
payload, status = aq.enhance_artist_quality('artist-1', [], deps)
|
||||
assert status == 400
|
||||
assert payload == {"success": False, "error": "No track IDs provided"}
|
||||
|
||||
|
||||
def test_artist_not_found_returns_404():
|
||||
deps = _build_deps(artist_detail={'success': False})
|
||||
payload, status = aq.enhance_artist_quality('artist-x', ['t1'], deps)
|
||||
assert status == 404
|
||||
assert payload == {"success": False, "error": "Artist not found"}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Spotify direct lookup (priority 1)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_spotify_direct_lookup_via_track_id_uses_raw_data():
|
||||
"""Track has spotify_track_id → get_track_details, raw_data fed to wishlist."""
|
||||
raw = {'id': 'sp-stored', 'name': 'Track One', 'artists': [{'name': 'Artist Name'}],
|
||||
'album': {'name': 'Album X', 'images': [{'url': 'http://i'}]}, 'duration_ms': 180000}
|
||||
spotify = _FakeSpotify(track_details={'raw_data': raw})
|
||||
wishlist = _FakeWishlist()
|
||||
deps = _build_deps(
|
||||
spotify=spotify,
|
||||
artist_detail=_artist_with_track(spotify_tid='sp-stored'),
|
||||
wishlist=wishlist,
|
||||
)
|
||||
|
||||
payload, status = aq.enhance_artist_quality('artist-1', ['t1'], deps)
|
||||
|
||||
assert status == 200
|
||||
assert payload['enhanced_count'] == 1
|
||||
assert wishlist.added[0]['spotify_track_data'] == raw
|
||||
|
||||
|
||||
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'],
|
||||
'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)
|
||||
wishlist = _FakeWishlist()
|
||||
deps = _build_deps(
|
||||
spotify=spotify,
|
||||
artist_detail=_artist_with_track(spotify_tid='sp-stored'),
|
||||
wishlist=wishlist,
|
||||
)
|
||||
|
||||
payload, _ = aq.enhance_artist_quality('artist-1', ['t1'], deps)
|
||||
|
||||
assert payload['enhanced_count'] == 1
|
||||
md = wishlist.added[0]['spotify_track_data']
|
||||
assert md['album']['images'] == [{'url': 'http://art'}]
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Spotify search fallback (priority 2)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
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])
|
||||
wishlist = _FakeWishlist()
|
||||
deps = _build_deps(
|
||||
spotify=spotify,
|
||||
artist_detail=_artist_with_track(spotify_tid=None),
|
||||
wishlist=wishlist,
|
||||
)
|
||||
|
||||
payload, _ = aq.enhance_artist_quality('artist-1', ['t1'], deps)
|
||||
|
||||
assert payload['enhanced_count'] == 1
|
||||
assert wishlist.added[0]['spotify_track_data'] == raw
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Fallback source (iTunes/Deezer)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_fallback_source_when_spotify_none():
|
||||
"""Spotify client None → iTunes/fallback search runs."""
|
||||
fallback_track = _SpotifyTrack(id='it-1', name='Track One', artists=['Artist Name'],
|
||||
image_url='http://it')
|
||||
fallback_track.track_number = 1
|
||||
fallback_track.disc_number = 1
|
||||
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'] == 1
|
||||
md = wishlist.added[0]['spotify_track_data']
|
||||
assert md['id'] == 'it-1'
|
||||
assert md['album']['images'] == [{'url': 'http://it', 'height': 600, 'width': 600}]
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Failure modes
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_track_not_in_artist_detail_marked_failed():
|
||||
"""Track ID provided but missing from artist's albums → failed_tracks entry."""
|
||||
deps = _build_deps(artist_detail=_artist_with_track(track_id='t1'))
|
||||
payload, _ = aq.enhance_artist_quality('artist-1', ['t99'], deps)
|
||||
|
||||
assert payload['enhanced_count'] == 0
|
||||
assert payload['failed_count'] == 1
|
||||
assert payload['failed_tracks'][0]['reason'] == 'Track not found'
|
||||
|
||||
|
||||
def test_track_with_no_file_path_marked_failed():
|
||||
"""Track has no file_path → failed reason 'No file path'."""
|
||||
detail = _artist_with_track()
|
||||
detail['albums'][0]['tracks'][0]['file_path'] = None
|
||||
deps = _build_deps(artist_detail=detail)
|
||||
payload, _ = aq.enhance_artist_quality('artist-1', ['t1'], deps)
|
||||
|
||||
assert payload['failed_count'] == 1
|
||||
assert payload['failed_tracks'][0]['reason'] == 'No file path'
|
||||
|
||||
|
||||
def test_no_match_anywhere_marked_failed():
|
||||
"""No Spotify match AND no fallback match → failed reason 'No Spotify or fallback match'."""
|
||||
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']
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Wishlist source_context payload
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_wishlist_source_context_carries_quality_metadata():
|
||||
"""source_context includes original_file_path, format tier, bitrate, artist_name."""
|
||||
raw = {'id': 'sp-1', 'name': 'Track One', 'artists': [{'name': 'Artist Name'}],
|
||||
'album': {'name': 'Album X'}}
|
||||
spotify = _FakeSpotify(track_details={'raw_data': raw})
|
||||
wishlist = _FakeWishlist()
|
||||
deps = _build_deps(
|
||||
spotify=spotify,
|
||||
artist_detail=_artist_with_track(spotify_tid='sp-1'),
|
||||
wishlist=wishlist,
|
||||
quality_tier=('mp3_192', 4),
|
||||
)
|
||||
|
||||
aq.enhance_artist_quality('artist-1', ['t1'], deps)
|
||||
|
||||
ctx = wishlist.added[0]['source_context']
|
||||
assert ctx['enhance'] is True
|
||||
assert ctx['original_file_path'] == '/file.mp3'
|
||||
assert ctx['original_format'] == 'mp3_192'
|
||||
assert ctx['original_bitrate'] == 320
|
||||
assert ctx['original_tier'] == 4
|
||||
assert ctx['artist_name'] == 'Artist Name'
|
||||
assert wishlist.added[0]['source_type'] == 'enhance'
|
||||
Loading…
Reference in new issue