Source IDs: add canonical registry, adopt at the highest-value sites

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

@@ -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",
+    )
 }
 
 

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}

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",
+    }

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,