Add one-off repair for source ids shared across multiple artists

core/maintenance/dedupe_source_ids.py + scripts/dedupe_source_ids.py: find
source-id clusters held by differently-named artists (the enrichment-corruption
signature) and clear the id + match-status on those rows so the now-name-checked
workers re-derive each correctly on the next enrichment pass. Same-name
duplicates (one artist across two media servers) are left untouched.

Dry-run by default; --apply to write. 8 seam tests cover detection (corrupt vs
legit), dry-run safety, apply behaviour, and the no-op case.

core/maintenance/__init__.py

@@ -0,0 +1 @@
+"""Maintenance / one-off data-repair helpers."""

core/maintenance/dedupe_source_ids.py

@@ -0,0 +1,138 @@
+"""Find and clear corrupted source-id assignments on the ``artists`` table.
+
+Background
+----------
+The metadata enrichment workers (Deezer / AudioDB / Qobuz / Tidal) historically
+"corrected" an artist's source id from an album/track match **without a name
+check**. A track our library credits to one artist but which lives on another
+artist's curated/compilation album (e.g. anyone featured on Kendrick Lamar's
+"Black Panther" album) resolved to that album, whose primary artist is someone
+else — and the worker stamped that wrong id onto our artist. The upshot: one
+source id (Kendrick's Deezer ``525046``) ends up shared across several unrelated
+artists.
+
+That bug is now fixed in the workers (they name-check before correcting). This
+module is the one-off repair for libraries that already got corrupted.
+
+What counts as corruption
+-------------------------
+A *corrupt cluster* is one source id held by artists with **different names**.
+Legitimate duplicates — the SAME artist indexed on two media servers, sharing
+one id — have identical names and are left untouched.
+
+The repair
+----------
+For every corrupt cluster, clear the source id AND its match-status column on
+each member artist, so the (now name-checked) worker re-derives each artist's id
+correctly on the next enrichment pass. Only the ``artists`` table is touched;
+album/track rows keep their match status, so the album/track correction path
+isn't re-run during re-enrichment.
+
+``clear_corrupt_source_ids`` defaults to ``dry_run=True`` — it reports exactly
+what it would change and writes nothing unless explicitly told to apply.
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+logger = logging.getLogger(__name__)
+
+# source -> (id column, match-status column) on the ``artists`` table.
+SOURCE_COLUMNS = {
+    'deezer': ('deezer_id', 'deezer_match_status'),
+    'spotify': ('spotify_artist_id', 'spotify_match_status'),
+    'itunes': ('itunes_artist_id', 'itunes_match_status'),
+    'musicbrainz': ('musicbrainz_id', 'musicbrainz_match_status'),
+    'discogs': ('discogs_id', 'discogs_match_status'),
+    'audiodb': ('audiodb_id', 'audiodb_match_status'),
+    'qobuz': ('qobuz_id', 'qobuz_match_status'),
+    'tidal': ('tidal_id', 'tidal_match_status'),
+}
+
+
+def _norm(name: str) -> str:
+    """Loose name key — lowercased, whitespace-collapsed."""
+    return ' '.join((name or '').lower().split())
+
+
+def _artists_columns(conn) -> set:
+    return {r[1] for r in conn.execute("PRAGMA table_info(artists)")}
+
+
+def find_corrupt_clusters(database: Any) -> list[dict]:
+    """Return corrupt source-id clusters across every known source column.
+
+    Each cluster is a dict: ``{source, id_column, status_column, source_id,
+    members: [(artist_id, name), ...]}``. A cluster is corrupt when one id is
+    held by artists with more than one distinct (normalized) name.
+    """
+    clusters: list[dict] = []
+    with database._get_connection() as conn:
+        existing = _artists_columns(conn)
+        for source, (id_col, status_col) in SOURCE_COLUMNS.items():
+            if id_col not in existing:
+                continue
+            rows = conn.execute(
+                f"SELECT {id_col}, id, name FROM artists "
+                f"WHERE {id_col} IS NOT NULL AND {id_col} != ''"
+            ).fetchall()
+            by_id: dict = {}
+            for sid, aid, name in rows:
+                by_id.setdefault(str(sid), []).append((aid, name))
+            for sid, members in by_id.items():
+                if len(members) > 1 and len({_norm(n) for _, n in members}) > 1:
+                    clusters.append({
+                        'source': source,
+                        'id_column': id_col,
+                        'status_column': status_col,
+                        'source_id': sid,
+                        'members': members,
+                    })
+    return clusters
+
+
+def clear_corrupt_source_ids(database: Any, dry_run: bool = True) -> dict:
+    """Clear source id + match status on every artist in a corrupt cluster.
+
+    ``dry_run=True`` (default) writes nothing — the returned report shows
+    exactly what would change so the operator can review first. Pass
+    ``dry_run=False`` to apply.
+    """
+    clusters = find_corrupt_clusters(database)
+    report = {
+        'dry_run': dry_run,
+        'cluster_count': len(clusters),
+        'artist_count': sum(len(c['members']) for c in clusters),
+        'by_source': {},
+        'clusters': [],
+    }
+    for c in clusters:
+        report['by_source'][c['source']] = (
+            report['by_source'].get(c['source'], 0) + len(c['members'])
+        )
+        report['clusters'].append({
+            'source': c['source'],
+            'source_id': c['source_id'],
+            'artists': sorted(n for _, n in c['members']),
+        })
+
+    if not dry_run and clusters:
+        with database._get_connection() as conn:
+            for c in clusters:
+                ids = [aid for aid, _ in c['members']]
+                placeholders = ','.join('?' for _ in ids)
+                conn.execute(
+                    f"UPDATE artists SET {c['id_column']} = NULL, "
+                    f"{c['status_column']} = NULL WHERE id IN ({placeholders})",
+                    ids,
+                )
+            conn.commit()
+        logger.info(
+            f"Cleared {report['artist_count']} corrupt source ids across "
+            f"{report['cluster_count']} clusters — re-run enrichment to "
+            f"re-derive them correctly"
+        )
+
+    return report

scripts/dedupe_source_ids.py

@@ -0,0 +1,59 @@
+#!/usr/bin/env python3
+"""One-off repair for source ids that enrichment wrongly shared across multiple
+artists (the Kendrick/Jorja corruption — one Deezer/AudioDB/Qobuz/Tidal id
+stamped onto several unrelated artists).
+
+Dry-run by default — shows exactly what it would clear and writes nothing.
+
+Usage:
+    python scripts/dedupe_source_ids.py            # dry-run (review first)
+    python scripts/dedupe_source_ids.py --apply    # actually clear them
+
+After --apply, run metadata enrichment so the (now name-checked) workers
+re-derive each artist's id correctly. Stop the app first so the DB isn't locked.
+"""
+
+import logging
+import os
+import sys
+
+# Allow running directly (`python scripts/dedupe_source_ids.py`) — put the repo
+# root on the path so `core` / `database` import.
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+from core.maintenance.dedupe_source_ids import clear_corrupt_source_ids  # noqa: E402
+from database.music_database import MusicDatabase  # noqa: E402
+
+if not logging.getLogger().handlers:
+    logging.basicConfig(level=logging.INFO, format="%(message)s")
+logger = logging.getLogger("dedupe_source_ids")
+
+
+def main():
+    apply = "--apply" in sys.argv[1:]
+    db = MusicDatabase()
+    report = clear_corrupt_source_ids(db, dry_run=not apply)
+
+    mode = "APPLYING" if apply else "DRY-RUN (no changes written)"
+    logger.info(f"=== Source-id corruption repair — {mode} ===")
+    logger.info(
+        f"Corrupt clusters: {report['cluster_count']}  |  "
+        f"artists affected: {report['artist_count']}"
+    )
+    if report['by_source']:
+        logger.info("By source: " + ", ".join(
+            f"{s}={n}" for s, n in sorted(report['by_source'].items())
+        ))
+    for c in report['clusters']:
+        logger.info(f"  [{c['source']}] id {c['source_id']} -> {', '.join(c['artists'])}")
+
+    if not report['cluster_count']:
+        logger.info("Nothing to clean — no shared source ids across differently-named artists.")
+    elif apply:
+        logger.info("Cleared. Now run metadata enrichment to re-derive these ids correctly.")
+    else:
+        logger.info("Re-run with --apply to clear these (then run enrichment to re-derive).")
+
+
+if __name__ == "__main__":
+    main()

tests/test_dedupe_source_ids.py

@@ -0,0 +1,131 @@
+"""Tests for core/maintenance/dedupe_source_ids.py — the one-off repair for
+source ids that enrichment wrongly shared across multiple artists.
+
+Corruption = one source id on artists with DIFFERENT names. Legit duplicates =
+the SAME artist on two media servers, same name — must be left alone.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from core.maintenance import dedupe_source_ids as dd
+from database.music_database import MusicDatabase
+
+
+@pytest.fixture
+def db(tmp_path):
+    return MusicDatabase(str(tmp_path / "music.db"))
+
+
+def _insert(db, *, artist_id, name, **extra):
+    cols = ["id", "name", "server_source"] + list(extra.keys())
+    vals = [artist_id, name, "plex"] + list(extra.values())
+    placeholders = ",".join("?" for _ in cols)
+    with db._get_connection() as conn:
+        conn.execute(
+            f"INSERT INTO artists ({','.join(cols)}) VALUES ({placeholders})", vals
+        )
+        conn.commit()
+
+
+def _get(db, artist_id, col):
+    with db._get_connection() as conn:
+        return conn.execute(f"SELECT {col} FROM artists WHERE id=?", (artist_id,)).fetchone()[0]
+
+
+# ---------------------------------------------------------------------------
+# Detection
+# ---------------------------------------------------------------------------
+
+def test_detects_different_name_cluster_as_corrupt(db):
+    _insert(db, artist_id="1", name="Kendrick Lamar", deezer_id="525046")
+    _insert(db, artist_id="2", name="Jorja Smith", deezer_id="525046")
+    _insert(db, artist_id="3", name="Vince Staples", deezer_id="525046")
+
+    clusters = dd.find_corrupt_clusters(db)
+    assert len(clusters) == 1
+    c = clusters[0]
+    assert c['source'] == 'deezer'
+    assert c['source_id'] == '525046'
+    assert {n for _, n in c['members']} == {"Kendrick Lamar", "Jorja Smith", "Vince Staples"}
+
+
+def test_same_name_duplicate_is_not_corrupt(db):
+    # Same artist on two servers — legit shared id, must be ignored.
+    _insert(db, artist_id="10", name="Radiohead", deezer_id="999")
+    _insert(db, artist_id="11", name="radiohead", deezer_id="999")  # case-insensitive
+    assert dd.find_corrupt_clusters(db) == []
+
+
+def test_unique_ids_are_not_corrupt(db):
+    _insert(db, artist_id="20", name="A", deezer_id="1")
+    _insert(db, artist_id="21", name="B", deezer_id="2")
+    assert dd.find_corrupt_clusters(db) == []
+
+
+def test_detects_corruption_across_multiple_sources(db):
+    _insert(db, artist_id="1", name="Kendrick", deezer_id="525046", spotify_artist_id="sp-x")
+    _insert(db, artist_id="2", name="Jorja", deezer_id="525046")
+    _insert(db, artist_id="3", name="Someone", spotify_artist_id="sp-x")
+    sources = {c['source'] for c in dd.find_corrupt_clusters(db)}
+    assert sources == {'deezer', 'spotify'}
+
+
+# ---------------------------------------------------------------------------
+# Repair
+# ---------------------------------------------------------------------------
+
+def test_dry_run_writes_nothing(db):
+    _insert(db, artist_id="1", name="Kendrick", deezer_id="525046", deezer_match_status="matched")
+    _insert(db, artist_id="2", name="Jorja", deezer_id="525046", deezer_match_status="matched")
+
+    report = dd.clear_corrupt_source_ids(db, dry_run=True)
+    assert report['dry_run'] is True
+    assert report['cluster_count'] == 1
+    assert report['artist_count'] == 2
+    assert report['by_source'] == {'deezer': 2}
+    # Nothing changed.
+    assert _get(db, "1", "deezer_id") == "525046"
+    assert _get(db, "2", "deezer_id") == "525046"
+
+
+def test_apply_clears_id_and_status_for_corrupt_rows(db):
+    _insert(db, artist_id="1", name="Kendrick", deezer_id="525046", deezer_match_status="matched")
+    _insert(db, artist_id="2", name="Jorja", deezer_id="525046", deezer_match_status="matched")
+
+    report = dd.clear_corrupt_source_ids(db, dry_run=False)
+    assert report['dry_run'] is False
+    assert report['artist_count'] == 2
+    # Both cleared so the (now name-checked) worker re-derives them.
+    for aid in ("1", "2"):
+        assert _get(db, aid, "deezer_id") is None
+        assert _get(db, aid, "deezer_match_status") is None
+
+
+def test_apply_leaves_legit_duplicates_untouched(db):
+    # Corrupt deezer cluster + a legit same-name spotify duplicate.
+    _insert(db, artist_id="1", name="Kendrick", deezer_id="525046")
+    _insert(db, artist_id="2", name="Jorja", deezer_id="525046")
+    _insert(db, artist_id="3", name="Radiohead", spotify_artist_id="rh", server_source="plex")
+    _insert(db, artist_id="4", name="Radiohead", spotify_artist_id="rh", server_source="jellyfin")
+
+    with db._get_connection() as conn:
+        conn.execute("UPDATE artists SET server_source='jellyfin' WHERE id='4'")
+        conn.commit()
+
+    dd.clear_corrupt_source_ids(db, dry_run=False)
+    # Corrupt deezer ids cleared…
+    assert _get(db, "1", "deezer_id") is None
+    assert _get(db, "2", "deezer_id") is None
+    # …legit same-name spotify duplicate preserved.
+    assert _get(db, "3", "spotify_artist_id") == "rh"
+    assert _get(db, "4", "spotify_artist_id") == "rh"
+
+
+def test_clean_library_is_a_noop(db):
+    _insert(db, artist_id="1", name="A", deezer_id="1")
+    _insert(db, artist_id="2", name="B", deezer_id="2")
+    report = dd.clear_corrupt_source_ids(db, dry_run=False)
+    assert report['cluster_count'] == 0
+    assert report['artist_count'] == 0