From 4fcc461616442d3b622d5585dfc0ecc04df959a7 Mon Sep 17 00:00:00 2001 From: BoulderBadgeDad Date: Fri, 29 May 2026 12:19:59 -0700 Subject: [PATCH] Source IDs: add canonical registry, adopt at the highest-value sites MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The same provider ID is stored under inconsistent column names across tables (deezer_id vs deezer_artist_id vs album_deezer_id vs similar_artist_deezer_id; spotify/itunes keep an entity qualifier, others don't; musicbrainz uses three nouns), so code checks 2-5 name variants everywhere. Add core/source_ids.py as the single source of truth for (provider, entity) -> column, with accessors that read an ID from a dict/sqlite3.Row robustly (canonical column first, then known aliases). NO database columns are renamed — these are the real names today; the registry just centralizes the knowledge. Targeted adoption (behavior-identical, verified): - core/artist_source_lookup.SOURCE_ID_FIELD now derives from the registry instead of duplicating the mapping (values unchanged). - web_server.py artist-detail builds artist_source_ids via source_id_map(...) instead of a hand-rolled per-source .get() dict. Broader call-site adoption deferred as clearly-scoped follow-up. Tests: tests/test_source_ids_registry.py (canonical columns, alias fallback, canonical-preferred, sqlite3.Row, source_id_map, SOURCE_ID_FIELD unchanged). Existing artist_source_lookup + artist_full_detail suites still green. --- core/artist_source_lookup.py | 16 ++-- core/source_ids.py | 144 ++++++++++++++++++++++++++++++ tests/test_source_ids_registry.py | 99 ++++++++++++++++++++ web_server.py | 16 ++-- 4 files changed, 260 insertions(+), 15 deletions(-) create mode 100644 core/source_ids.py create mode 100644 tests/test_source_ids_registry.py diff --git a/core/artist_source_lookup.py b/core/artist_source_lookup.py index 507d09e6..60fff775 100644 --- a/core/artist_source_lookup.py +++ b/core/artist_source_lookup.py @@ -21,6 +21,8 @@ from __future__ import annotations import logging from typing import Optional +from core.source_ids import id_column as _artist_id_column + logger = logging.getLogger("artist_source_lookup") @@ -29,14 +31,14 @@ SOURCE_ONLY_ARTIST_SOURCES = frozenset({ }) +# The per-source column on the ``artists`` table, derived from the canonical +# source-ID registry (the single source of truth). Values are unchanged from the +# previous hardcoded map — this just stops duplicating that knowledge here. SOURCE_ID_FIELD = { - "spotify": "spotify_artist_id", - "itunes": "itunes_artist_id", - "deezer": "deezer_id", - "discogs": "discogs_id", - "hydrabase": "soul_id", - "musicbrainz": "musicbrainz_id", - "amazon": "amazon_id", + source: _artist_id_column(source, "artist") + for source in ( + "spotify", "itunes", "deezer", "discogs", "hydrabase", "musicbrainz", "amazon", + ) } diff --git a/core/source_ids.py b/core/source_ids.py new file mode 100644 index 00000000..00168784 --- /dev/null +++ b/core/source_ids.py @@ -0,0 +1,144 @@ +"""Canonical registry of external/source ID column names. + +SoulSync stores each metadata provider's ID for an artist/album/track under a +column whose NAME is inconsistent across tables — e.g. Deezer's artist id is +``deezer_id`` on the ``artists`` table but ``deezer_artist_id`` on +``watchlist_artists`` and ``album_deezer_id`` / ``similar_artist_deezer_id`` on +the discovery tables. Spotify/iTunes keep an entity qualifier on the core tables +while Deezer/Amazon/Tidal/... don't, and MusicBrainz uses three different nouns. +The result is code that checks 2–5 property-name variants everywhere. + +This module is the single source of truth for "(provider, entity) → column". +It does NOT rename any database column — these ARE the real names today; the +registry just centralizes the knowledge and offers accessors that read an ID +from a dict / sqlite3.Row robustly (canonical column first, then known aliases), +so callers stop hand-rolling variant checks. +""" + +from __future__ import annotations + +from typing import Any, Dict, Iterable, Optional + +# Entity types this registry knows about. +ENTITIES = ("artist", "album", "track") + +# Canonical column name on the CORE table (artists / albums / tracks) for each +# (entity, provider). This is the name to prefer when reading/writing. +_CORE_ID_COLUMNS: Dict[str, Dict[str, str]] = { + "artist": { + "spotify": "spotify_artist_id", + "itunes": "itunes_artist_id", + "deezer": "deezer_id", + "musicbrainz": "musicbrainz_id", + "discogs": "discogs_id", + "amazon": "amazon_id", + "tidal": "tidal_id", + "qobuz": "qobuz_id", + "audiodb": "audiodb_id", + "genius": "genius_id", + "hydrabase": "soul_id", + }, + "album": { + "spotify": "spotify_album_id", + "itunes": "itunes_album_id", + "deezer": "deezer_id", + "musicbrainz": "musicbrainz_release_id", + "discogs": "discogs_id", + "amazon": "amazon_id", + "tidal": "tidal_id", + "qobuz": "qobuz_id", + "audiodb": "audiodb_id", + "hydrabase": "soul_id", + }, + "track": { + "spotify": "spotify_track_id", + "itunes": "itunes_track_id", + "deezer": "deezer_id", + "musicbrainz": "musicbrainz_recording_id", + "amazon": "amazon_id", + "tidal": "tidal_id", + "qobuz": "qobuz_id", + "audiodb": "audiodb_id", + "genius": "genius_id", + "hydrabase": "soul_id", + }, +} + +# Other column / dict-key names the SAME (entity, provider) ID appears under +# elsewhere (satellite tables, API payloads). Accessors check the canonical +# column first, then these, so a read works regardless of where the row/dict +# came from. Keyed by (entity, provider). +_ALIASES: Dict[tuple, tuple] = { + ("artist", "spotify"): ("similar_artist_spotify_id",), + ("artist", "itunes"): ("artist_itunes_id", "similar_artist_itunes_id"), + ("artist", "deezer"): ("deezer_artist_id", "artist_deezer_id", "similar_artist_deezer_id"), + ("artist", "musicbrainz"): ("musicbrainz_artist_id", "similar_artist_musicbrainz_id"), + ("artist", "discogs"): ("discogs_artist_id",), + ("artist", "amazon"): ("amazon_artist_id",), + ("album", "spotify"): ("album_spotify_id",), + ("album", "itunes"): ("album_itunes_id",), + ("album", "deezer"): ("deezer_album_id", "album_deezer_id"), + ("album", "discogs"): ("discogs_release_id",), + ("track", "deezer"): ("deezer_track_id",), +} + + +def id_column(provider: str, entity: str = "artist") -> Optional[str]: + """Canonical core-table column for this provider + entity, or None if the + provider isn't tracked for that entity.""" + return _CORE_ID_COLUMNS.get(entity, {}).get(provider) + + +def id_keys(provider: str, entity: str = "artist") -> tuple: + """All known key names (canonical first, then aliases) the ID may live + under. Useful for code that needs the full variant list explicitly.""" + keys = [] + canon = id_column(provider, entity) + if canon: + keys.append(canon) + for alias in _ALIASES.get((entity, provider), ()): # preserve order, no dups + if alias not in keys: + keys.append(alias) + return tuple(keys) + + +def _read(data: Any, key: str) -> Any: + """Read ``key`` from a dict or sqlite3.Row, returning None if absent.""" + try: + keys = data.keys() # dict and sqlite3.Row both support .keys() + except AttributeError: + return None + if key in keys: + try: + return data[key] + except (KeyError, IndexError): + return None + return None + + +def get_id(data: Any, provider: str, entity: str = "artist") -> Optional[str]: + """Read this provider's ID for ``entity`` from a dict / sqlite3.Row. + + Tries the canonical column first, then every known alias, and returns the + first non-empty value (or None). Replaces hand-rolled + ``row.get('deezer_id') or row.get('deezer_artist_id')`` chains. + """ + for key in id_keys(provider, entity): + value = _read(data, key) + if value: + return value + return None + + +def source_id_map( + data: Any, + entity: str = "artist", + providers: Optional[Iterable[str]] = None, +) -> Dict[str, Optional[str]]: + """Build a ``{provider: id}`` dict for ``entity`` from a row/dict — the + common "artist_source_ids" pattern. Defaults to every provider known for the + entity; pass ``providers`` to restrict/order the result. + """ + if providers is None: + providers = list(_CORE_ID_COLUMNS.get(entity, {}).keys()) + return {p: get_id(data, p, entity) for p in providers} diff --git a/tests/test_source_ids_registry.py b/tests/test_source_ids_registry.py new file mode 100644 index 00000000..212abf61 --- /dev/null +++ b/tests/test_source_ids_registry.py @@ -0,0 +1,99 @@ +"""Tests for the canonical source-ID registry (core/source_ids.py).""" + +from __future__ import annotations + +import sqlite3 + +import pytest + +from core import source_ids as sid + + +def test_canonical_columns_match_real_schema(): + # Spot-check the canonical names against the actual DB columns. + assert sid.id_column("spotify", "artist") == "spotify_artist_id" + assert sid.id_column("deezer", "artist") == "deezer_id" + assert sid.id_column("musicbrainz", "artist") == "musicbrainz_id" + assert sid.id_column("hydrabase", "artist") == "soul_id" + assert sid.id_column("spotify", "album") == "spotify_album_id" + assert sid.id_column("musicbrainz", "album") == "musicbrainz_release_id" + assert sid.id_column("deezer", "album") == "deezer_id" + assert sid.id_column("spotify", "track") == "spotify_track_id" + assert sid.id_column("musicbrainz", "track") == "musicbrainz_recording_id" + + +def test_id_column_unknown_returns_none(): + assert sid.id_column("nonesuch", "artist") is None + assert sid.id_column("spotify", "playlist") is None + + +def test_id_keys_canonical_first_then_aliases(): + keys = sid.id_keys("deezer", "artist") + assert keys[0] == "deezer_id" # canonical first + assert "deezer_artist_id" in keys # watchlist/pool alias + assert "similar_artist_deezer_id" in keys + + +def test_get_id_reads_canonical_column(): + row = {"deezer_id": "525046", "name": "Artist"} + assert sid.get_id(row, "deezer", "artist") == "525046" + + +def test_get_id_falls_back_to_alias(): + # A watchlist-shaped row uses deezer_artist_id, not deezer_id. + row = {"deezer_artist_id": "999", "artist_name": "X"} + assert sid.get_id(row, "deezer", "artist") == "999" + + +def test_get_id_prefers_canonical_over_alias(): + row = {"deezer_id": "canon", "deezer_artist_id": "alias"} + assert sid.get_id(row, "deezer", "artist") == "canon" + + +def test_get_id_missing_and_empty_return_none(): + assert sid.get_id({"name": "X"}, "deezer", "artist") is None + assert sid.get_id({"deezer_id": ""}, "deezer", "artist") is None + assert sid.get_id({"deezer_id": None}, "deezer", "artist") is None + + +def test_get_id_works_with_sqlite_row(): + conn = sqlite3.connect(":memory:") + conn.row_factory = sqlite3.Row + conn.execute("CREATE TABLE t (spotify_artist_id TEXT, name TEXT)") + conn.execute("INSERT INTO t VALUES ('spfy123', 'Artist')") + row = conn.execute("SELECT * FROM t").fetchone() + conn.close() + assert sid.get_id(row, "spotify", "artist") == "spfy123" + # A column not present on the row must not raise — just None. + assert sid.get_id(row, "deezer", "artist") is None + + +def test_source_id_map_builds_provider_dict(): + row = { + "spotify_artist_id": "s1", + "deezer_id": "d1", + "itunes_artist_id": None, + } + result = sid.source_id_map(row, "artist", providers=["spotify", "deezer", "itunes"]) + assert result == {"spotify": "s1", "deezer": "d1", "itunes": None} + + +def test_source_id_map_default_covers_all_providers(): + result = sid.source_id_map({"deezer_id": "d1"}, "artist") + assert result["deezer"] == "d1" + assert "spotify" in result and result["spotify"] is None + + +def test_source_id_field_unchanged_after_registry_refactor(): + """artist_source_lookup.SOURCE_ID_FIELD must keep its exact prior mapping + after being folded into the registry (no behavior change).""" + from core.artist_source_lookup import SOURCE_ID_FIELD + assert SOURCE_ID_FIELD == { + "spotify": "spotify_artist_id", + "itunes": "itunes_artist_id", + "deezer": "deezer_id", + "discogs": "discogs_id", + "hydrabase": "soul_id", + "musicbrainz": "musicbrainz_id", + "amazon": "amazon_id", + } diff --git a/web_server.py b/web_server.py index 27bca092..7c51d140 100644 --- a/web_server.py +++ b/web_server.py @@ -7601,15 +7601,15 @@ def get_artist_detail(artist_id): try: from core.metadata.lookup import MetadataLookupOptions from core.metadata_service import get_artist_detail_discography as _get_artist_detail_discography + from core.source_ids import source_id_map - artist_source_ids = { - 'spotify': artist_info.get('spotify_artist_id'), - 'deezer': artist_info.get('deezer_id'), - 'itunes': artist_info.get('itunes_artist_id'), - 'discogs': artist_info.get('discogs_id'), - 'hydrabase': artist_info.get('soul_id'), - 'amazon': artist_info.get('amazon_id'), - } + # Per-source artist IDs, read via the canonical source-ID registry + # (same columns as before: spotify_artist_id / deezer_id / + # itunes_artist_id / discogs_id / soul_id / amazon_id). + artist_source_ids = source_id_map( + artist_info, 'artist', + providers=('spotify', 'deezer', 'itunes', 'discogs', 'hydrabase', 'amazon'), + ) artist_detail_discography = _get_artist_detail_discography( artist_id,