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

Discovery lift (1/N): convert_*_results_to_spotify_tracks -> shared helper

Browse source 
First cluster of the per-source playlist-discovery deduplication. The
convert_<source>_results_to_spotify_tracks functions (Tidal, Deezer, Qobuz,
Spotify-Public, YouTube, ListenBrainz) plus the already-generic
_convert_link_results_to_spotify_tracks were byte-identical apart from the
source label used in their log line.

Lift the shared body into core/discovery/endpoints.py as
convert_results_to_spotify_tracks(results, source_label); the 7 web_server
functions become 1-line delegations (names/signatures unchanged, so all
callers and behavior are identical — 1:1).

Beatport is intentionally NOT folded in: its converter coerces artist
objects to strings and emits a different track shape (source field, album
dict), so it keeps its own implementation.

Tests: tests/discovery/test_discovery_endpoints.py (12) pin both input
shapes (manual spotify_data / auto spotify_track+found), optional
track/disc numbers, falsy-0 omission, field defaults, skip-on-neither,
order preservation, if/elif precedence, empty input.

web_server.py: -209 lines. Full discovery suite: 151 passed.
This commit is contained in:
BoulderBadgeDad 2026-05-28 15:57:23 -07:00
parent abdea631a7
commit 628395eda5
3 changed files with 212 additions and 218 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

76
core/discovery/endpoints.py Normal file
View file

@ -0,0 +1,76 @@
"""Generic, source-agnostic helpers for the playlist-discovery route layer.
The discovery/sync endpoints in ``web_server.py`` were copy-pasted once per
source (Tidal, Deezer, Qobuz, Spotify-public, iTunes-link, YouTube,
ListenBrainz, Beatport). The per-source copies differ only by a source label
string and which ``<source>_discovery_states`` global they read. This module
lifts the source-agnostic pieces into importable, unit-testable helpers so the
route functions become thin wrappers — exactly preserving behavior (1:1).
Each helper is lifted verbatim from its web_server.py counterpart; any
per-source quirk that genuinely differs (e.g. Beatport's distinct result
shape) is intentionally NOT routed through here and stays in its own function.
"""
from __future__ import annotations
from typing import Any, Dict, List
from utils.logging_config import get_logger
logger = get_logger("discovery.endpoints")
def convert_results_to_spotify_tracks(
discovery_results: List[Dict[str, Any]],
source_label: str,
) -> List[Dict[str, Any]]:
"""Convert a source's discovery results into the Spotify-track dicts the
sync pipeline expects.
Lifted verbatim from the per-source ``convert_<source>_results_to_spotify_tracks``
functions (and the already-generic ``_convert_link_results_to_spotify_tracks``),
which were byte-identical apart from the ``source_label`` used in the log
line. Two input shapes are supported, matching the originals exactly:
- ``spotify_data`` (manual-fix shape): copied through, preserving optional
``track_number`` / ``disc_number``.
- ``spotify_track`` + ``status_class == 'found'`` (auto-discovery shape):
rebuilt from the flat ``spotify_*`` fields.
Any result matching neither shape is skipped, identical to the originals.
NOTE: Beatport deliberately does NOT use this — its converter coerces
artist objects to strings and emits a different track shape (``source``
field, album dict), so it keeps its own implementation.
"""
spotify_tracks: List[Dict[str, Any]] = []
for result in discovery_results:
# Support both data formats: spotify_data (manual fixes) and individual
# fields (automatic discovery).
if result.get('spotify_data'):
spotify_data = result['spotify_data']
track = {
'id': spotify_data['id'],
'name': spotify_data['name'],
'artists': spotify_data['artists'],
'album': spotify_data['album'],
'duration_ms': spotify_data.get('duration_ms', 0),
}
if spotify_data.get('track_number'):
track['track_number'] = spotify_data['track_number']
if spotify_data.get('disc_number'):
track['disc_number'] = spotify_data['disc_number']
spotify_tracks.append(track)
elif result.get('spotify_track') and result.get('status_class') == 'found':
spotify_tracks.append({
'id': result.get('spotify_id', 'unknown'),
'name': result.get('spotify_track', 'Unknown Track'),
'artists': [result.get('spotify_artist', 'Unknown Artist')] if result.get('spotify_artist') else ['Unknown Artist'],
'album': result.get('spotify_album', 'Unknown Album'),
'duration_ms': 0,
})
logger.info(f"Converted {len(spotify_tracks)} {source_label} matches to Spotify tracks for sync")
return spotify_tracks

127
tests/discovery/test_discovery_endpoints.py Normal file
View file

@ -0,0 +1,127 @@
"""Tests for the lifted, source-agnostic discovery route helpers in
``core.discovery.endpoints``.
These pin the exact behavior the per-source ``convert_<source>_results_to_spotify_tracks``
functions had in web_server.py, so the lift is provably 1:1. Each input shape
the originals handled is exercised here.
"""
from __future__ import annotations
from core.discovery.endpoints import convert_results_to_spotify_tracks
# ---------------------------------------------------------------------------
# spotify_data (manual-fix) shape
# ---------------------------------------------------------------------------
def test_spotify_data_shape_basic():
results = [{
'spotify_data': {
'id': 'sp1', 'name': 'Song', 'artists': ['A'], 'album': 'Alb',
'duration_ms': 1234,
}
}]
assert convert_results_to_spotify_tracks(results, 'Tidal') == [{
'id': 'sp1', 'name': 'Song', 'artists': ['A'], 'album': 'Alb',
'duration_ms': 1234,
}]
def test_spotify_data_duration_defaults_to_zero():
results = [{'spotify_data': {'id': 'x', 'name': 'n', 'artists': [], 'album': 'a'}}]
out = convert_results_to_spotify_tracks(results, 'Deezer')
assert out[0]['duration_ms'] == 0
def test_spotify_data_includes_track_and_disc_number_when_present():
results = [{'spotify_data': {
'id': 'x', 'name': 'n', 'artists': [], 'album': 'a',
'track_number': 5, 'disc_number': 2,
}}]
out = convert_results_to_spotify_tracks(results, 'Qobuz')
assert out[0]['track_number'] == 5
assert out[0]['disc_number'] == 2
def test_spotify_data_omits_track_disc_when_absent_or_falsy():
# track_number/disc_number of 0 are falsy -> omitted, matching original.
results = [{'spotify_data': {
'id': 'x', 'name': 'n', 'artists': [], 'album': 'a',
'track_number': 0, 'disc_number': 0,
}}]
out = convert_results_to_spotify_tracks(results, 'YouTube')
assert 'track_number' not in out[0]
assert 'disc_number' not in out[0]
# ---------------------------------------------------------------------------
# spotify_track + status_class == 'found' (auto-discovery) shape
# ---------------------------------------------------------------------------
def test_auto_discovery_shape_full():
results = [{
'spotify_track': 'Track', 'status_class': 'found',
'spotify_id': 'id9', 'spotify_artist': 'Artist', 'spotify_album': 'Album',
}]
assert convert_results_to_spotify_tracks(results, 'ListenBrainz') == [{
'id': 'id9', 'name': 'Track', 'artists': ['Artist'], 'album': 'Album',
'duration_ms': 0,
}]
def test_auto_discovery_defaults_when_fields_missing():
results = [{'spotify_track': 'T', 'status_class': 'found'}]
out = convert_results_to_spotify_tracks(results, 'Spotify Public')
assert out == [{
'id': 'unknown', 'name': 'T', 'artists': ['Unknown Artist'],
'album': 'Unknown Album', 'duration_ms': 0,
}]
def test_auto_discovery_empty_artist_yields_unknown_artist():
results = [{
'spotify_track': 'T', 'status_class': 'found', 'spotify_artist': '',
}]
out = convert_results_to_spotify_tracks(results, 'Tidal')
assert out[0]['artists'] == ['Unknown Artist']
# ---------------------------------------------------------------------------
# skip / mixed / empty
# ---------------------------------------------------------------------------
def test_auto_discovery_requires_found_status():
# spotify_track present but status_class != 'found' -> skipped.
results = [{'spotify_track': 'T', 'status_class': 'not_found'}]
assert convert_results_to_spotify_tracks(results, 'Tidal') == []
def test_result_matching_neither_shape_is_skipped():
results = [{'irrelevant': True}, {'spotify_track': 'T'}] # 2nd has no status_class
assert convert_results_to_spotify_tracks(results, 'Tidal') == []
def test_mixed_results_preserve_order():
results = [
{'spotify_data': {'id': '1', 'name': 'a', 'artists': [], 'album': ''}},
{'irrelevant': True},
{'spotify_track': 'b', 'status_class': 'found', 'spotify_id': '2'},
]
out = convert_results_to_spotify_tracks(results, 'Tidal')
assert [t['id'] for t in out] == ['1', '2']
def test_empty_input():
assert convert_results_to_spotify_tracks([], 'Tidal') == []
def test_spotify_data_takes_precedence_over_auto_fields():
# A result carrying both shapes uses spotify_data (the if-branch wins),
# matching the original if/elif ordering.
results = [{
'spotify_data': {'id': 'D', 'name': 'd', 'artists': [], 'album': ''},
'spotify_track': 'IGNORED', 'status_class': 'found', 'spotify_id': 'A',
}]
out = convert_results_to_spotify_tracks(results, 'Tidal')
assert out[0]['id'] == 'D'

227
web_server.py
View file

@ -21073,6 +21073,8 @@ from core.discovery.scoring import (
# Tidal discovery worker logic lives in core/discovery/tidal.py.
from core.discovery import tidal as _discovery_tidal
# Source-agnostic discovery route helpers (lifted from the per-source copies).
from core.discovery.endpoints import convert_results_to_spotify_tracks
def _build_tidal_discovery_deps():
@ -21103,39 +21105,7 @@ def _run_tidal_discovery_worker(playlist_id):
def convert_tidal_results_to_spotify_tracks(discovery_results):
"""Convert Tidal discovery results to Spotify tracks format for sync"""
spotify_tracks = []
for result in discovery_results:
# Support both data formats: spotify_data (manual fixes) and individual fields (automatic discovery)
if result.get('spotify_data'):
spotify_data = result['spotify_data']
# Create track object matching the expected format
track = {
'id': spotify_data['id'],
'name': spotify_data['name'],
'artists': spotify_data['artists'],
'album': spotify_data['album'],
'duration_ms': spotify_data.get('duration_ms', 0)
}
if spotify_data.get('track_number'):
track['track_number'] = spotify_data['track_number']
if spotify_data.get('disc_number'):
track['disc_number'] = spotify_data['disc_number']
spotify_tracks.append(track)
elif result.get('spotify_track') and result.get('status_class') == 'found':
# Build from individual fields (automatic discovery format)
track = {
'id': result.get('spotify_id', 'unknown'),
'name': result.get('spotify_track', 'Unknown Track'),
'artists': [result.get('spotify_artist', 'Unknown Artist')] if result.get('spotify_artist') else ['Unknown Artist'],
'album': result.get('spotify_album', 'Unknown Album'),
'duration_ms': 0
}
spotify_tracks.append(track)
logger.info(f"Converted {len(spotify_tracks)} Tidal matches to Spotify tracks for sync")
return spotify_tracks
return convert_results_to_spotify_tracks(discovery_results, "Tidal")
# ===================================================================
@ -21817,37 +21787,7 @@ def _run_deezer_discovery_worker(playlist_id):
def convert_deezer_results_to_spotify_tracks(discovery_results):
"""Convert Deezer discovery results to Spotify tracks format for sync"""
spotify_tracks = []
for result in discovery_results:
# Support both data formats: spotify_data (manual fixes) and individual fields (automatic discovery)
if result.get('spotify_data'):
spotify_data = result['spotify_data']
track = {
'id': spotify_data['id'],
'name': spotify_data['name'],
'artists': spotify_data['artists'],
'album': spotify_data['album'],
'duration_ms': spotify_data.get('duration_ms', 0)
}
if spotify_data.get('track_number'):
track['track_number'] = spotify_data['track_number']
if spotify_data.get('disc_number'):
track['disc_number'] = spotify_data['disc_number']
spotify_tracks.append(track)
elif result.get('spotify_track') and result.get('status_class') == 'found':
track = {
'id': result.get('spotify_id', 'unknown'),
'name': result.get('spotify_track', 'Unknown Track'),
'artists': [result.get('spotify_artist', 'Unknown Artist')] if result.get('spotify_artist') else ['Unknown Artist'],
'album': result.get('spotify_album', 'Unknown Album'),
'duration_ms': 0
}
spotify_tracks.append(track)
logger.info(f"Converted {len(spotify_tracks)} Deezer matches to Spotify tracks for sync")
return spotify_tracks
return convert_results_to_spotify_tracks(discovery_results, "Deezer")
# ===================================================================
@ -22478,36 +22418,7 @@ def _run_qobuz_discovery_worker(playlist_id):
def convert_qobuz_results_to_spotify_tracks(discovery_results):
"""Convert Qobuz discovery results to Spotify tracks format for sync."""
spotify_tracks = []
for result in discovery_results:
if result.get('spotify_data'):
spotify_data = result['spotify_data']
track = {
'id': spotify_data['id'],
'name': spotify_data['name'],
'artists': spotify_data['artists'],
'album': spotify_data['album'],
'duration_ms': spotify_data.get('duration_ms', 0)
}
if spotify_data.get('track_number'):
track['track_number'] = spotify_data['track_number']
if spotify_data.get('disc_number'):
track['disc_number'] = spotify_data['disc_number']
spotify_tracks.append(track)
elif result.get('spotify_track') and result.get('status_class') == 'found':
track = {
'id': result.get('spotify_id', 'unknown'),
'name': result.get('spotify_track', 'Unknown Track'),
'artists': [result.get('spotify_artist', 'Unknown Artist')] if result.get('spotify_artist') else ['Unknown Artist'],
'album': result.get('spotify_album', 'Unknown Album'),
'duration_ms': 0
}
spotify_tracks.append(track)
logger.info(f"Converted {len(spotify_tracks)} Qobuz matches to Spotify tracks for sync")
return spotify_tracks
return convert_results_to_spotify_tracks(discovery_results, "Qobuz")
# ===================================================================
@ -22967,32 +22878,7 @@ def _build_itunes_link_state(response_data):
def _convert_link_results_to_spotify_tracks(discovery_results, label):
spotify_tracks = []
for result in discovery_results:
if result.get('spotify_data'):
spotify_data = result['spotify_data']
track = {
'id': spotify_data['id'],
'name': spotify_data['name'],
'artists': spotify_data['artists'],
'album': spotify_data['album'],
'duration_ms': spotify_data.get('duration_ms', 0)
}
if spotify_data.get('track_number'):
track['track_number'] = spotify_data['track_number']
if spotify_data.get('disc_number'):
track['disc_number'] = spotify_data['disc_number']
spotify_tracks.append(track)
elif result.get('spotify_track') and result.get('status_class') == 'found':
spotify_tracks.append({
'id': result.get('spotify_id', 'unknown'),
'name': result.get('spotify_track', 'Unknown Track'),
'artists': [result.get('spotify_artist', 'Unknown Artist')] if result.get('spotify_artist') else ['Unknown Artist'],
'album': result.get('spotify_album', 'Unknown Album'),
'duration_ms': 0
})
logger.info(f"Converted {len(spotify_tracks)} {label} matches to Spotify tracks for sync")
return spotify_tracks
return convert_results_to_spotify_tracks(discovery_results, label)
@app.route('/api/spotify/parse-public', methods=['POST'])
def parse_spotify_public_endpoint():
@ -23442,38 +23328,7 @@ def _run_spotify_public_discovery_worker(url_hash):
def convert_spotify_public_results_to_spotify_tracks(discovery_results):
"""Convert Spotify Public discovery results to Spotify tracks format for sync"""
spotify_tracks = []
for result in discovery_results:
# Support both data formats: spotify_data (manual fixes) and individual fields (automatic discovery)
if result.get('spotify_data'):
spotify_data = result['spotify_data']
track = {
'id': spotify_data['id'],
'name': spotify_data['name'],
'artists': spotify_data['artists'],
'album': spotify_data['album'],
'duration_ms': spotify_data.get('duration_ms', 0)
}
# Preserve track_number/disc_number from discovery enrichment
if spotify_data.get('track_number'):
track['track_number'] = spotify_data['track_number']
if spotify_data.get('disc_number'):
track['disc_number'] = spotify_data['disc_number']
spotify_tracks.append(track)
elif result.get('spotify_track') and result.get('status_class') == 'found':
track = {
'id': result.get('spotify_id', 'unknown'),
'name': result.get('spotify_track', 'Unknown Track'),
'artists': [result.get('spotify_artist', 'Unknown Artist')] if result.get('spotify_artist') else ['Unknown Artist'],
'album': result.get('spotify_album', 'Unknown Album'),
'duration_ms': 0
}
spotify_tracks.append(track)
logger.info(f"Converted {len(spotify_tracks)} Spotify Public matches to Spotify tracks for sync")
return spotify_tracks
return convert_results_to_spotify_tracks(discovery_results, "Spotify Public")
# ===================================================================
@ -24889,39 +24744,7 @@ def update_youtube_playlist_phase(url_hash):
def convert_youtube_results_to_spotify_tracks(discovery_results):
"""Convert YouTube discovery results to Spotify tracks format for sync"""
spotify_tracks = []
for result in discovery_results:
# Support both data formats: spotify_data (manual fixes) and individual fields (automatic discovery)
if result.get('spotify_data'):
spotify_data = result['spotify_data']
# Create track object matching the expected format
track = {
'id': spotify_data['id'],
'name': spotify_data['name'],
'artists': spotify_data['artists'],
'album': spotify_data['album'],
'duration_ms': spotify_data.get('duration_ms', 0)
}
if spotify_data.get('track_number'):
track['track_number'] = spotify_data['track_number']
if spotify_data.get('disc_number'):
track['disc_number'] = spotify_data['disc_number']
spotify_tracks.append(track)
elif result.get('spotify_track') and result.get('status_class') == 'found':
# Build from individual fields (automatic discovery format)
track = {
'id': result.get('spotify_id', 'unknown'),
'name': result.get('spotify_track', 'Unknown Track'),
'artists': [result.get('spotify_artist', 'Unknown Artist')] if result.get('spotify_artist') else ['Unknown Artist'],
'album': result.get('spotify_album', 'Unknown Album'),
'duration_ms': 0
}
spotify_tracks.append(track)
logger.info(f"Converted {len(spotify_tracks)} YouTube matches to Spotify tracks for sync")
return spotify_tracks
return convert_results_to_spotify_tracks(discovery_results, "YouTube")
# Add these new endpoints to the end of web_server.py
@ -30702,39 +30525,7 @@ def update_listenbrainz_discovery_match():
def convert_listenbrainz_results_to_spotify_tracks(discovery_results):
"""Convert ListenBrainz discovery results to Spotify tracks format for sync"""
spotify_tracks = []
for result in discovery_results:
# Support both data formats: spotify_data (manual fixes) and individual fields (automatic discovery)
if result.get('spotify_data'):
spotify_data = result['spotify_data']
# Create track object matching the expected format
track = {
'id': spotify_data['id'],
'name': spotify_data['name'],
'artists': spotify_data['artists'],
'album': spotify_data['album'],
'duration_ms': spotify_data.get('duration_ms', 0)
}
if spotify_data.get('track_number'):
track['track_number'] = spotify_data['track_number']
if spotify_data.get('disc_number'):
track['disc_number'] = spotify_data['disc_number']
spotify_tracks.append(track)
elif result.get('spotify_track') and result.get('status_class') == 'found':
# Build from individual fields (automatic discovery format)
track = {
'id': result.get('spotify_id', 'unknown'),
'name': result.get('spotify_track', 'Unknown Track'),
'artists': [result.get('spotify_artist', 'Unknown Artist')] if result.get('spotify_artist') else ['Unknown Artist'],
'album': result.get('spotify_album', 'Unknown Album'),
'duration_ms': 0
}
spotify_tracks.append(track)
logger.info(f"Converted {len(spotify_tracks)} ListenBrainz matches to Spotify tracks for sync")
return spotify_tracks
return convert_results_to_spotify_tracks(discovery_results, "ListenBrainz")
@app.route('/api/wing-it/sync', methods=['POST'])
def wing_it_sync():