playlist export: resolver foundation for Spotify/Deezer targets (#945, increment 1)

First piece of "export a mirrored playlist to Spotify/Deezer" (diegocade1). Reuses the exact
machinery the ListenBrainz/JSPF export already proves out, additively:

- resolve_playlist_tracks gains an `id_key` param (default "recording_mbid" → LB/JSPF callers
  byte-for-byte unchanged). The dedup/stats/order logic is ID-agnostic; only the output field
  name differs, so service export plugs in with id_key="service_track_id".
- export_sources gains db_service_track_id(artist, title, service) + build_service_resolve_fn —
  text-matches a library track (same pattern as the MBID resolver) and returns its stored
  spotify_track_id / deezer_id. Enrichment already pinned those IDs, so export is a lookup, not a
  re-search — which is what makes the reverse direction reliable (no fuzzy guessing).

No schema change needed: get/set_playlist_export_target already key by service name, so Spotify/
Deezer targets store for free (idempotent re-export, like the LB #903 fix).

7 tests: LB default id_key unchanged, service id_key carries the id + unmatched handling, service→
column mapping, unknown-service/no-title guards, resolve_fn id+source. 38 export tests green,
ruff clean.

Remaining increments: Spotify write client (+ playlist-modify scope / re-auth), Deezer write via
the ARL gw-light gateway, the export-job branch + endpoint, the modal options.
This commit is contained in:
BoulderBadgeDad 2026-06-28 21:04:47 -07:00
parent c593a17ac2
commit ede62824ad
4 changed files with 119 additions and 5 deletions

View file

@ -72,6 +72,55 @@ def db_recording_mbid(artist: str, title: str) -> Optional[str]:
return _db_match(artist, title)[0] return _db_match(artist, title)[0]
# Service → the tracks-table column carrying that service's track ID (set by enrichment).
# Trusted constants — never user input — so safe to interpolate into the SELECT below.
_SERVICE_ID_COLUMNS = {"spotify": "spotify_track_id", "deezer": "deezer_id"}
def db_service_track_id(artist: str, title: str, service: str) -> Optional[str]:
"""The service track ID (``spotify_track_id`` / ``deezer_id``) stored on a matched
library track what lets a mirrored playlist be exported BACK to Spotify/Deezer
without re-searching, since enrichment already pinned it (#945). Text-matches by
(artist, title), same as the MBID resolver. Fail-safe: any miss/error returns None."""
column = _SERVICE_ID_COLUMNS.get((service or "").lower())
if not column or not title:
return None
try:
from database.music_database import get_database
db = get_database()
conn = db._get_connection()
try:
cur = conn.cursor()
cur.execute(
f"SELECT t.{column} FROM tracks t JOIN artists a ON t.artist_id = a.id "
"WHERE LOWER(t.title) = LOWER(?) AND LOWER(a.name) = LOWER(?) LIMIT 1",
(title, artist),
)
row = cur.fetchone()
if not row:
return None
val = row[0] if not hasattr(row, "keys") else row[column]
return val or None
finally:
try:
conn.close()
except Exception: # noqa: S110
pass
except Exception as exc:
logger.debug(f"export service-id lookup failed for '{artist} - {title}' ({service}): {exc}")
return None
def build_service_resolve_fn(service: str) -> Callable[[str, str], Tuple[Optional[str], Optional[str]]]:
"""resolve_fn for service-playlist export: ``(artist, title) -> (service_track_id, 'library')``.
Plugs into ``resolve_playlist_tracks(..., id_key='service_track_id')`` exactly like the
MBID resolver plugs in for ListenBrainz."""
def resolve_fn(artist: str, title: str) -> Tuple[Optional[str], Optional[str]]:
tid = db_service_track_id(artist, title, service)
return (tid, "library" if tid else None)
return resolve_fn
def file_recording_mbid(artist: str, title: str) -> Optional[str]: def file_recording_mbid(artist: str, title: str) -> Optional[str]:
"""Recording MBID read from the matched track's file tag (set on import post-processing).""" """Recording MBID read from the matched track's file tag (set on import post-processing)."""
_mbid, fpath = _db_match(artist, title) _mbid, fpath = _db_match(artist, title)

View file

@ -36,16 +36,22 @@ def resolve_playlist_tracks(
resolve_fn: ResolveFn, resolve_fn: ResolveFn,
*, *,
on_progress: Optional[ProgressFn] = None, on_progress: Optional[ProgressFn] = None,
id_key: str = "recording_mbid",
) -> Dict[str, Any]: ) -> Dict[str, Any]:
"""Resolve every track to a recording MBID and build the export pseudo-playlist. """Resolve every track to an ID and build the export pseudo-playlist.
``resolve_fn(artist, title) -> (id, source)`` returns whatever ID the target needs
a MusicBrainz recording MBID for ListenBrainz/JSPF (the default), or a Spotify/Deezer
track ID for service export. ``id_key`` names the field that ID lands under in each
resolved entry (defaults to ``recording_mbid`` so existing LB/JSPF callers are
untouched). The dedup + stats + ordering logic is identical regardless of ID type.
``tracks`` items may use ``artist``/``artist_name`` and ``title``/``track_name`` and ``tracks`` items may use ``artist``/``artist_name`` and ``title``/``track_name`` and
``album``/``album_name`` (both the mirrored-playlist and LB-cache shapes are accepted). ``album``/``album_name`` (both the mirrored-playlist and LB-cache shapes are accepted).
Returns ``{"resolved": [...], "stats": {...}}`` where each resolved entry is Returns ``{"resolved": [...], "stats": {...}}`` where each resolved entry is
``{artist, title, album, recording_mbid}`` (recording_mbid is None when unmatched), ``{artist, title, album, <id_key>}`` (the ID is None when unmatched), in original
in original order, and stats carries ``total, resolved, unmatched, deduped, order, and stats carries ``total, resolved, unmatched, deduped, by_source``.
by_source`` for the live display.
""" """
total = len(tracks or []) total = len(tracks or [])
memo: Dict[str, Tuple[Optional[str], Optional[str]]] = {} memo: Dict[str, Tuple[Optional[str], Optional[str]]] = {}
@ -71,7 +77,7 @@ def resolve_playlist_tracks(
memo[key] = (mbid, source) memo[key] = (mbid, source)
fresh = True fresh = True
resolved.append({"artist": artist, "title": title, "album": album, "recording_mbid": mbid}) resolved.append({"artist": artist, "title": title, "album": album, id_key: mbid})
if mbid: if mbid:
stats["resolved"] += 1 stats["resolved"] += 1

View file

@ -57,3 +57,34 @@ def test_all_miss_returns_none_and_no_write():
fn, recorded = _wire() fn, recorded = _wire()
assert fn("A", "T") == (None, None) assert fn("A", "T") == (None, None)
assert recorded == {} assert recorded == {}
# ── service track-id resolver (#945 export to Spotify/Deezer) ──
from core.exports.export_sources import (
db_service_track_id,
build_service_resolve_fn,
_SERVICE_ID_COLUMNS,
)
def test_service_id_column_mapping():
assert _SERVICE_ID_COLUMNS == {'spotify': 'spotify_track_id', 'deezer': 'deezer_id'}
def test_db_service_track_id_unknown_service_is_none():
assert db_service_track_id('A', 'X', 'tidal') is None
assert db_service_track_id('A', 'X', '') is None
def test_db_service_track_id_no_title_is_none():
assert db_service_track_id('A', '', 'spotify') is None
def test_build_service_resolve_fn_returns_id_and_source(monkeypatch):
import core.exports.export_sources as es
monkeypatch.setattr(es, 'db_service_track_id',
lambda a, t, s: 'spid-99' if t == 'Hit' else None)
fn = build_service_resolve_fn('spotify')
assert fn('Artist', 'Hit') == ('spid-99', 'library')
assert fn('Artist', 'Miss') == (None, None)

View file

@ -72,3 +72,31 @@ def test_empty_playlist():
out = resolve_playlist_tracks([], lambda a, t: (None, None)) out = resolve_playlist_tracks([], lambda a, t: (None, None))
assert out["resolved"] == [] assert out["resolved"] == []
assert out["stats"]["total"] == 0 assert out["stats"]["total"] == 0
# ── id_key generalization (#945 service export reuses the LB resolver) ──
from core.exports.playlist_export import resolve_playlist_tracks as _rpt
def _const_resolver(mapping):
return lambda artist, title: mapping.get((artist, title), (None, None))
def test_default_id_key_is_recording_mbid_unchanged():
# ListenBrainz/JSPF callers must be byte-for-byte unaffected by the generalization.
out = _rpt([{'artist': 'A', 'title': 'X'}], _const_resolver({('A', 'X'): ('mbid-1', 'db')}))
assert out['resolved'][0]['recording_mbid'] == 'mbid-1'
assert 'service_track_id' not in out['resolved'][0]
def test_custom_id_key_carries_service_id():
out = _rpt(
[{'artist': 'A', 'title': 'X'}, {'artist': 'B', 'title': 'Y'}],
_const_resolver({('A', 'X'): ('spid-1', 'library')}), # B/Y unmatched
id_key='service_track_id',
)
assert out['resolved'][0]['service_track_id'] == 'spid-1'
assert out['resolved'][1]['service_track_id'] is None
assert 'recording_mbid' not in out['resolved'][0]
assert out['stats']['resolved'] == 1 and out['stats']['unmatched'] == 1