Deezer playlists: tag the REAL album track number, not the playlist index

A downloaded track was getting the wrong track number (e.g. 'Apologize' from Shock
Value tagged track 1 instead of 16). Root cause: Deezer PLAYLIST track objects don't
carry `track_position` (only `/track/<id>` and `/album/<id>/tracks` do — even the
album object's embedded tracks omit it, verified against the live API). Both Deezer
playlist builders numbered tracks by their enumerate INDEX, which poisoned the real
album track number — and that value rides into the wishlist and onto the file tag.

Resolve the authoritative position from each album's `/album/<id>/tracks` (cache-first,
handles multi-disc) via a shared resolve_album_track_positions() helper, used by both
deezer_client.get_playlist and deezer_download_client.get_playlist_tracks. Falls back
to the index only if the album lookup fails (no regression). Seam tests for the helper.
This commit is contained in:
BoulderBadgeDad 2026-06-21 15:47:13 -07:00
parent cc7d48b736
commit 7bbd069147
3 changed files with 146 additions and 17 deletions

View file

@ -117,6 +117,48 @@ def _is_full_track_payload(payload: Optional[Dict[str, Any]]) -> bool:
return 'track_position' in payload and 'contributors' in payload
def resolve_album_track_positions(session, base_url, album_ids, cache=None, sleep_s=0.2):
"""Build ``{str(track_id): track_position}`` for a set of Deezer album ids.
Deezer PLAYLIST and SEARCH track objects (and even the album object's embedded
``tracks.data``) omit ``track_position`` only ``/album/<id>/tracks`` and
``/track/<id>`` carry it. So numbering playlist tracks by their playlist index
silently poisons the real album track number, which then rides onto the
downloaded file's tag. This resolves the authoritative position per album
(cache-first, best-effort a failed album just isn't in the map)."""
import time as _time
positions: Dict[str, int] = {}
for aid in album_ids:
aid = str(aid)
at_list = None
if cache:
try:
ct = cache.get_entity('deezer', 'album_tracks', aid)
if ct and ct.get('data'):
at_list = ct['data']
except Exception: # noqa: BLE001 - cache is best-effort
at_list = None
if at_list is None:
try:
if sleep_s:
_time.sleep(sleep_s) # respect Deezer rate limits
r = session.get(f"{base_url}/album/{aid}/tracks", params={'limit': 500}, timeout=10)
if getattr(r, 'ok', False):
at_list = (r.json() or {}).get('data', [])
if cache and at_list is not None:
try:
cache.store_entity('deezer', 'album_tracks', aid, {'data': at_list})
except Exception: # noqa: BLE001
pass
except Exception: # noqa: BLE001 - never let metadata resolution break the fetch
at_list = None
for at in (at_list or []):
tp = at.get('track_position')
if at.get('id') and tp:
positions[str(at['id'])] = tp
return positions
# ==================== Dataclasses (match iTunesClient / SpotifyClient format) ====================
@dataclass
@ -1357,6 +1399,16 @@ class DeezerClient:
raw_tracks.extend(page_tracks)
# Real album track positions — playlist tracks don't carry track_position,
# so numbering by playlist index would poison the downloaded file's tag.
album_ids = {str(t.get('album', {}).get('id')) for t in raw_tracks if t.get('album', {}).get('id')}
try:
from core.metadata.cache import get_metadata_cache
_cache = get_metadata_cache()
except Exception:
_cache = None
track_positions = resolve_album_track_positions(self.session, self.BASE_URL, album_ids, _cache)
# Normalize tracks
tracks: List[Dict[str, Any]] = []
for i, t in enumerate(raw_tracks, start=1):
@ -1368,7 +1420,8 @@ class DeezerClient:
'artists': [artist_name],
'album': t.get('album', {}).get('title', ''),
'duration_ms': t.get('duration', 0) * 1000,
'track_number': i,
# REAL album position; the playlist index is a last resort only.
'track_number': track_positions.get(str(t.get('id'))) or i,
})
result = {

View file

@ -417,6 +417,12 @@ class DeezerDownloadClient(DownloadSourcePlugin):
if aid:
album_ids.add(str(aid))
album_release_dates = {}
# Deezer PLAYLIST tracks do NOT carry `track_position` (only `/track/<id>`
# and `/album/<id>/tracks` do), so numbering them by their playlist index
# poisons the real album track number — which then rides into the wishlist
# and onto the downloaded file's tag (e.g. 'Apologize' tagged track 1 instead
# of 16). Resolve the REAL position from each album's track list (cache-first).
track_positions: Dict[str, int] = {} # str(track_id) -> album track_position
try:
from core.metadata.cache import get_metadata_cache
cache = get_metadata_cache()
@ -429,24 +435,32 @@ class DeezerDownloadClient(DownloadSourcePlugin):
cached = cache.get_entity('deezer', 'album', aid)
if cached and cached.get('release_date'):
album_release_dates[aid] = cached['release_date']
continue
except Exception as e:
logger.debug("cache get_entity album release_date: %s", e)
# Cache miss — fetch from API
try:
time.sleep(0.3) # Respect rate limits
a_resp = self._session.get(f'https://api.deezer.com/album/{aid}', timeout=10)
if a_resp.ok:
a_data = a_resp.json()
album_release_dates[aid] = a_data.get('release_date', '')
# Store in metadata cache for future use
if cache:
try:
cache.store_entity('deezer', 'album', aid, a_data)
except Exception as e:
logger.debug("cache store_entity album release_date: %s", e)
except Exception as e:
logger.debug("fetch deezer album release_date %s: %s", aid, e)
if aid not in album_release_dates:
try:
time.sleep(0.3) # Respect rate limits
a_resp = self._session.get(f'https://api.deezer.com/album/{aid}', timeout=10)
if a_resp.ok:
a_data = a_resp.json()
album_release_dates[aid] = a_data.get('release_date', '')
# Store in metadata cache for future use
if cache:
try:
cache.store_entity('deezer', 'album', aid, a_data)
except Exception as e:
logger.debug("cache store_entity album release_date: %s", e)
except Exception as e:
logger.debug("fetch deezer album release_date %s: %s", aid, e)
# Real album track positions (separate endpoint — playlist tracks AND the
# album object's embedded tracks both omit track_position). Cache-first.
try:
from core.deezer_client import resolve_album_track_positions
track_positions = resolve_album_track_positions(
self._session, 'https://api.deezer.com', album_ids, cache)
except Exception as e:
logger.debug("resolve deezer album track positions: %s", e)
tracks = []
for i, t in enumerate(raw_tracks, start=1):
@ -467,7 +481,9 @@ class DeezerDownloadClient(DownloadSourcePlugin):
'id': album_id,
},
'duration_ms': t.get('duration', 0) * 1000,
'track_number': i,
# REAL album position (resolved above); the playlist index is a last
# resort only when the album lookup failed, never the default.
'track_number': track_positions.get(str(t.get('id'))) or t.get('track_position') or i,
})
return {

View file

@ -0,0 +1,60 @@
"""Deezer playlist tracks must carry the REAL album track_position, not their
playlist index otherwise the downloaded file is tagged with the wrong track
number (e.g. 'Apologize' from Shock Value tagged track 1 instead of 16)."""
from __future__ import annotations
from core.deezer_client import resolve_album_track_positions
class _Resp:
def __init__(self, data, ok=True):
self._d, self.ok = data, ok
def json(self):
return self._d
class _Session:
"""Fake requests session returning /album/<id>/tracks payloads."""
def __init__(self, by_album, fail_for=()):
self.by_album, self.fail_for, self.calls = by_album, set(fail_for), []
def get(self, url, params=None, timeout=None):
aid = url.rstrip('/').split('/')[-2] # …/album/<aid>/tracks
self.calls.append(aid)
if aid in self.fail_for:
return _Resp(None, ok=False)
return _Resp({'data': self.by_album.get(aid, [])})
def test_maps_track_id_to_real_album_position():
sess = _Session({'119606': [
{'id': 100, 'track_position': 16}, {'id': 101, 'track_position': 2}]})
pos = resolve_album_track_positions(sess, 'https://api.deezer.com', {'119606'}, sleep_s=0)
assert pos == {'100': 16, '101': 2} # real positions, not 1/2 enumerate
def test_cache_first_skips_the_network():
class _Cache:
def __init__(self): self.stored = {}
def get_entity(self, src, kind, aid):
return {'data': [{'id': 7, 'track_position': 9}]} if kind == 'album_tracks' else None
def store_entity(self, *a, **k): pass
sess = _Session({})
pos = resolve_album_track_positions(sess, 'https://api.deezer.com', {'42'}, cache=_Cache(), sleep_s=0)
assert pos == {'7': 9} and sess.calls == [] # served from cache, no HTTP
def test_failed_album_is_simply_absent_not_fatal():
sess = _Session({'1': [{'id': 5, 'track_position': 3}]}, fail_for={'2'})
pos = resolve_album_track_positions(sess, 'https://api.deezer.com', {'1', '2'}, sleep_s=0)
assert pos == {'5': 3} # album 2 failed → just missing
def test_zero_position_is_ignored():
# Deezer sometimes returns 0/None for odd entries — don't poison the map with them
sess = _Session({'1': [{'id': 5, 'track_position': 0}, {'id': 6}, {'id': 7, 'track_position': 4}]})
pos = resolve_album_track_positions(sess, 'https://api.deezer.com', {'1'}, sleep_s=0)
assert pos == {'7': 4}