Library: group imported albums by canonical release id, not just the name string

Sokhi: songs in one album get mismatched cover art. Root cause is upstream of
the repair jobs (which correctly apply one cover per album_id): the standalone
import grouped albums by the album NAME hash (artist::album_name), so the SAME
release split into multiple album rows whenever the name string drifted, and the
cover-art/re-tag jobs then dressed each split row in its own art.

Foundation (new imports only; existing rows untouched): a pure, seam-testable
helper find_existing_soulsync_album_id() resolves the album row by precedence
name-hash id -> source RELEASE id -> (title, artist). When an import carries a
metadata-source album id, a differently-named import of the SAME release now
unifies into one row instead of splitting. Source-column lookup is allow-listed
(it's spliced into SQL) and guarded so a source without a dedicated album column
(Deezer) falls through to the name match instead of breaking the import.

Deliberate scope: this does NOT merge a track that genuinely matched a SINGLE
(a different release id) into its parent album — that needs single->album
resolution upstream and is the next step; this is the grouping substrate it will
feed. 10 seam tests (canonical unify, single-vs-album stays separate, precedence,
allowlist, server-source scope, missing-column fallthrough).

core/imports/album_grouping.py

@@ -0,0 +1,94 @@
+"""Canonical album grouping for the SoulSync standalone import.
+
+SoulSync grouped imported tracks into albums by the album NAME string
+(``_stable_soulsync_id("artist::album_name")``). That splits one release into
+several album rows whenever the name string drifts between imports (case,
+punctuation, ``(Deluxe Edition)`` suffixes, source-A-vs-B spelling), and every
+downstream tool (Library Re-tag, Cover-Art Filler) then dresses each split row
+in its own cover — so songs that belong to one album end up with different art
+(Sokhi).
+
+This module is the pure, seam-testable heart of "group by canonical id, not
+name": when an imported track carries a metadata-source RELEASE id, prefer
+matching an existing album row by that id over the fragile name string, so the
+SAME release always lands in ONE album row regardless of how its name was typed.
+
+Scope (deliberate): this unifies differently-named imports of the SAME release.
+It does NOT merge a track that genuinely matched a SINGLE release (a different
+release id) into its parent album — that needs single->album resolution upstream
+and is a separate change. New imports only; existing rows are left untouched.
+
+Pure SQL-over-a-cursor; no app singletons, so it tests against an in-memory DB.
+"""
+
+from __future__ import annotations
+
+from typing import Any, Optional
+
+# Album source-id columns this grouping may key on. An allowlist (not arbitrary
+# interpolation) — the column name IS spliced into SQL, so it must be a known,
+# trusted identifier. Mirrors get_library_source_id_columns()' 'album' values.
+ALLOWED_ALBUM_SOURCE_COLS = frozenset({
+    "spotify_album_id",
+    "itunes_album_id",
+    "deezer_id",
+    "soul_id",
+    "discogs_id",
+    "musicbrainz_release_id",
+})
+
+
+def find_existing_soulsync_album_id(
+    cursor: Any,
+    *,
+    name_key_id: str,
+    artist_id: str,
+    album_name: str,
+    album_source_col: Optional[str] = None,
+    album_source_id: Optional[str] = None,
+) -> Optional[str]:
+    """Resolve the existing ``soulsync`` album row a track should join, or None
+    (caller inserts a new row keyed by ``name_key_id``).
+
+    Match precedence:
+      1. ``name_key_id`` — the exact prior stable-name-hash id (unchanged
+         behaviour: a re-import with the identical name hits its own row).
+      2. ``album_source_col == album_source_id`` — CANONICAL grouping: an
+         existing row already carrying THIS release's source id, so a
+         differently-named import of the same release unifies instead of
+         splitting. Only when the column is allow-listed and the id is non-empty.
+      3. ``(title, artist_id)`` — the legacy name match (kept so nothing that
+         grouped before stops grouping now).
+    """
+    cursor.execute(
+        "SELECT id FROM albums WHERE id = ? AND server_source = 'soulsync'",
+        (name_key_id,),
+    )
+    row = cursor.fetchone()
+    if row:
+        return row[0]
+
+    if album_source_col in ALLOWED_ALBUM_SOURCE_COLS and album_source_id:
+        try:
+            cursor.execute(
+                f"SELECT id FROM albums WHERE {album_source_col} = ? "
+                "AND server_source = 'soulsync' LIMIT 1",
+                (album_source_id,),
+            )
+            row = cursor.fetchone()
+            if row:
+                return row[0]
+        except Exception:
+            # That source has no dedicated album column on this DB (e.g. Deezer
+            # doesn't split per-entity id columns) — fall through to the name
+            # match rather than break the import. Mirrors the guarded source-id
+            # UPDATE the caller already does on insert.
+            pass
+
+    cursor.execute(
+        "SELECT id FROM albums WHERE title COLLATE NOCASE = ? AND artist_id = ? "
+        "AND server_source = 'soulsync' LIMIT 1",
+        (album_name, artist_id),
+    )
+    row = cursor.fetchone()
+    return row[0] if row else None

core/imports/side_effects.py

@@ -565,19 +565,22 @@ def record_soulsync_library_entry(context: Dict[str, Any], artist_context: Dict[
             # ── Album row: same insert-or-fill-empty-fields shape ──
             album_source_col = source_columns.get("album")
 
-            cursor.execute(
-                "SELECT id FROM albums WHERE id = ? AND server_source = 'soulsync'",
-                (album_id,),
+            # Group by CANONICAL release id when we have one (not just the name
+            # string), so differently-named imports of the SAME release land in
+            # one album row instead of splitting — which left the repair jobs
+            # dressing each split row in its own cover art (Sokhi). Precedence:
+            # name-hash id -> source release id -> (title, artist). Falls back to
+            # the legacy name match, so nothing that grouped before stops now.
+            from core.imports.album_grouping import find_existing_soulsync_album_id
+            existing_album_id = find_existing_soulsync_album_id(
+                cursor, name_key_id=album_id, artist_id=artist_id, album_name=album_name,
+                album_source_col=album_source_col, album_source_id=album_source_id,
             )
-            row = cursor.fetchone()
-            if not row:
-                cursor.execute(
-                    "SELECT id FROM albums WHERE title COLLATE NOCASE = ? AND artist_id = ? AND server_source = 'soulsync' LIMIT 1",
-                    (album_name, artist_id),
-                )
-                row = cursor.fetchone()
-                if row:
-                    album_id = row[0]
+            if existing_album_id is not None:
+                album_id = existing_album_id
+                row = (album_id,)
+            else:
+                row = None
 
             if row:
                 _fill_empty_columns(

tests/imports/test_album_grouping.py

@@ -0,0 +1,138 @@
+"""Seam tests for canonical album grouping (Sokhi: split album rows -> mixed
+cover art). Drives find_existing_soulsync_album_id against a real in-memory
+SQLite albums table — no app singletons, no I/O.
+"""
+
+from __future__ import annotations
+
+import sqlite3
+
+import pytest
+
+from core.imports.album_grouping import (
+    find_existing_soulsync_album_id,
+    ALLOWED_ALBUM_SOURCE_COLS,
+)
+
+
+@pytest.fixture()
+def cur():
+    conn = sqlite3.connect(":memory:")
+    conn.execute(
+        """CREATE TABLE albums (
+            id TEXT PRIMARY KEY,
+            artist_id TEXT,
+            title TEXT,
+            server_source TEXT,
+            spotify_album_id TEXT,
+            itunes_album_id TEXT,
+            deezer_id TEXT,
+            soul_id TEXT,
+            discogs_id TEXT,
+            musicbrainz_release_id TEXT
+        )"""
+    )
+    yield conn.cursor()
+    conn.close()
+
+
+def _add(cur, *, id, title, artist_id="art1", server_source="soulsync", **source_ids):
+    cols = ["id", "artist_id", "title", "server_source"] + list(source_ids)
+    vals = [id, artist_id, title, server_source] + list(source_ids.values())
+    cur.execute(
+        f"INSERT INTO albums ({', '.join(cols)}) VALUES ({', '.join(['?'] * len(cols))})",
+        vals,
+    )
+
+
+def test_empty_db_returns_none(cur):
+    assert find_existing_soulsync_album_id(
+        cur, name_key_id="nk", artist_id="art1", album_name="Parachutes",
+        album_source_col="spotify_album_id", album_source_id="SP1") is None
+
+
+def test_exact_name_hash_id_wins_first(cur):
+    _add(cur, id="nk", title="Parachutes")
+    assert find_existing_soulsync_album_id(
+        cur, name_key_id="nk", artist_id="art1", album_name="Parachutes") == "nk"
+
+
+def test_canonical_source_id_unifies_differently_named_imports(cur):
+    # Existing row for release SP1 named "Parachutes". A second import of the
+    # SAME release id but a drifted name must JOIN it, not split.
+    _add(cur, id="existing", title="Parachutes", spotify_album_id="SP1")
+    got = find_existing_soulsync_album_id(
+        cur, name_key_id="different_hash", artist_id="art1",
+        album_name="Parachutes (Deluxe Edition)",
+        album_source_col="spotify_album_id", album_source_id="SP1")
+    assert got == "existing"
+
+
+def test_different_release_id_stays_separate(cur):
+    # The single-vs-album case: a genuinely different release id must NOT merge
+    # (documents the known limit — single->album resolution is a separate step).
+    _add(cur, id="album_row", title="Parachutes", spotify_album_id="SP_ALBUM")
+    got = find_existing_soulsync_album_id(
+        cur, name_key_id="single_hash", artist_id="art1", album_name="Yellow",
+        album_source_col="spotify_album_id", album_source_id="SP_SINGLE")
+    assert got is None
+
+
+def test_legacy_name_match_still_groups_without_a_source_id(cur):
+    _add(cur, id="byname", title="Parachutes")
+    got = find_existing_soulsync_album_id(
+        cur, name_key_id="other_hash", artist_id="art1", album_name="parachutes",
+        album_source_col=None, album_source_id=None)
+    assert got == "byname"  # case-insensitive title + artist
+
+
+def test_source_id_match_is_scoped_to_soulsync_rows(cur):
+    _add(cur, id="plexrow", title="Parachutes", server_source="plex", spotify_album_id="SP1")
+    got = find_existing_soulsync_album_id(
+        cur, name_key_id="nk", artist_id="art1", album_name="X",
+        album_source_col="spotify_album_id", album_source_id="SP1")
+    assert got is None  # the matching row belongs to Plex, not soulsync
+
+
+def test_non_allowlisted_column_is_ignored(cur):
+    # A column not on the allowlist must never be spliced into SQL.
+    assert "title" not in ALLOWED_ALBUM_SOURCE_COLS
+    _add(cur, id="row", title="Parachutes")
+    got = find_existing_soulsync_album_id(
+        cur, name_key_id="nk", artist_id="art1", album_name="nope",
+        album_source_col="title", album_source_id="Parachutes")
+    assert got is None  # 'title' ignored as a source col; name 'nope' doesn't match
+
+
+def test_empty_source_id_skips_canonical_match(cur):
+    _add(cur, id="row", title="Parachutes", spotify_album_id="")
+    got = find_existing_soulsync_album_id(
+        cur, name_key_id="nk", artist_id="art1", album_name="Other",
+        album_source_col="spotify_album_id", album_source_id="")
+    assert got is None
+
+
+def test_missing_album_column_falls_through_not_raises(cur):
+    # Some sources (Deezer) don't have a dedicated album id column on the albums
+    # table; an allow-listed-but-absent column must NOT raise (it broke the whole
+    # import once) — it falls through to the name match.
+    cur.execute("CREATE TABLE albums_min (id TEXT, artist_id TEXT, title TEXT, server_source TEXT)")
+    cur.execute("INSERT INTO albums_min VALUES ('byname','art1','DZ Album','soulsync')")
+    # Point the helper at a table missing deezer_id by aliasing via a fresh cursor.
+    conn2 = sqlite3.connect(":memory:")
+    conn2.execute("CREATE TABLE albums (id TEXT, artist_id TEXT, title TEXT, server_source TEXT)")
+    conn2.execute("INSERT INTO albums VALUES ('byname','art1','DZ Album','soulsync')")
+    c2 = conn2.cursor()
+    got = find_existing_soulsync_album_id(
+        c2, name_key_id="nk", artist_id="art1", album_name="DZ Album",
+        album_source_col="deezer_id", album_source_id="67890")
+    conn2.close()
+    assert got == "byname"   # deezer_id column absent -> fell through to name match
+
+
+def test_musicbrainz_release_id_grouping(cur):
+    _add(cur, id="mbrow", title="Album", musicbrainz_release_id="mb-123")
+    got = find_existing_soulsync_album_id(
+        cur, name_key_id="nk2", artist_id="art1", album_name="Album (Remaster)",
+        album_source_col="musicbrainz_release_id", album_source_id="mb-123")
+    assert got == "mbrow"