From 3e554f8274263fde1486ee3e1f7c87fde7212668 Mon Sep 17 00:00:00 2001
From: BoulderBadgeDad <batmanesignal@gmail.com>
Date: Thu, 18 Jun 2026 15:31:33 -0700
Subject: [PATCH] =?UTF-8?q?#889=20Phase=203:=20re-identify=20search=20?=
 =?UTF-8?q?=E2=80=94=20multi-source=20track=E2=86=92release=20lookup=20+?=
 =?UTF-8?q?=20API?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Search any configured source (tabs, default active) and surface the SAME song
across its collections (single/EP/album) so the user can pick which release a
track should be filed under.

- core/imports/rematch_search.py: pure normalize + injected client factory.
  search_release_candidates() → lightweight display rows from typed search_tracks
  (title/artist/release/type badge/year/count/art/isrc/track_id); resolve_hint_fields()
  runs ONCE on the picked row via get_track_details to pull the album_id (+ isrc/
  track#/disc) the hint needs. infer_release_type() handles Spotify's missing 'EP'
  (multi-track 'single' → EP badge); filing is driven by real album_id, not the label.
- GET /api/reidentify/sources (tabs) + GET /api/reidentify/search (rows). Graceful
  empty on dead source / blank query / client error — never raises.

14 tests. Inert until the modal (Phase 4) calls it.
---
 core/imports/rematch_search.py       | 247 +++++++++++++++++++++++++++
 tests/imports/test_rematch_search.py | 121 +++++++++++++
 web_server.py                        |  41 +++++
 3 files changed, 409 insertions(+)
 create mode 100644 core/imports/rematch_search.py
 create mode 100644 tests/imports/test_rematch_search.py

diff --git a/core/imports/rematch_search.py b/core/imports/rematch_search.py
new file mode 100644
index 00000000..da32c373
--- /dev/null
+++ b/core/imports/rematch_search.py
@@ -0,0 +1,247 @@
+"""#889 Phase 3: search a metadata source for the releases a track appears on.
+
+The Re-identify modal lets the user search ANY configured source (tabs, defaulting
+to the active one) and shows the SAME song across its different collections —
+single / EP / album — so they can pick which release the track should be filed
+under. Two steps, deliberately split:
+
+  * ``search_release_candidates(source, query)`` — lightweight DISPLAY rows from the
+    normal typed ``search_tracks`` (title, artist, release name, type badge, year,
+    track count, art, ISRC, track_id). No album_id needed to draw the list.
+  * ``resolve_hint_fields(source, track_id)`` — runs ONCE, on the row the user
+    picks: ``get_track_details`` yields the album_id / isrc / track#/disc the hint
+    needs. We don't pay that lookup for every search result, only the chosen one.
+
+Pure normalization + injected client factory, so the search/normalize/resolve seam
+is unit-tested with a fake client and no network.
+"""
+
+from __future__ import annotations
+
+from typing import Any, Callable, Dict, List, Optional
+
+
+def _get(obj: Any, key: str, default=None):
+    """Read ``key`` from either an object (attr) or a mapping (item)."""
+    if obj is None:
+        return default
+    if isinstance(obj, dict):
+        return obj.get(key, default)
+    return getattr(obj, key, default)
+
+
+def _year(release_date: Any) -> Optional[str]:
+    s = str(release_date or "").strip()
+    return s[:4] if len(s) >= 4 and s[:4].isdigit() else None
+
+
+def infer_release_type(album_type: Any, total_tracks: Any) -> str:
+    """Normalize a source's release type to one of album / ep / single / compilation.
+
+    Sources disagree: Spotify has no 'EP' — EPs come back as ``album_type='single'``
+    with several tracks; MusicBrainz/Deezer label EPs properly. So when a 'single'
+    carries more than a handful of tracks, call it an EP for the badge. The actual
+    filing is unaffected — that's driven by the real album_id, not this label."""
+    t = str(album_type or "").strip().lower()
+    try:
+        n = int(total_tracks) if total_tracks is not None else 0
+    except (TypeError, ValueError):
+        n = 0
+    if t in ("compilation", "comp"):
+        return "compilation"
+    if t == "ep":
+        return "ep"
+    if t == "album":
+        return "album"   # an explicit album stays an album; only 'single' gets promoted to EP
+    if t == "single":
+        # 1–3 tracks → single; 4+ → almost always an EP in practice.
+        return "ep" if n >= 4 else "single"
+    # Unknown type: infer purely from track count.
+    if n >= 7:
+        return "album"
+    if n >= 4:
+        return "ep"
+    if n >= 1:
+        return "single"
+    return t or "album"
+
+
+def normalize_search_result(result: Any, source: str) -> Optional[Dict[str, Any]]:
+    """One typed search Track (or raw dict) → a display row, or ``None`` if it has
+    no usable id/title. ``album`` is just a name at search time; album_id is
+    resolved later for the picked row only."""
+    track_id = _get(result, "id") or _get(result, "track_id")
+    title = _get(result, "name") or _get(result, "title")
+    if not track_id or not title:
+        return None
+
+    artists = _get(result, "artists")
+    if isinstance(artists, list):
+        artist_name = ", ".join(str(_get(a, "name", a) if not isinstance(a, str) else a) for a in artists)
+    else:
+        artist_name = str(artists or _get(result, "artist") or "")
+
+    album = _get(result, "album")
+    album_name = album if isinstance(album, str) else (_get(album, "name") or "")
+    raw_type = _get(result, "album_type")
+    total = _get(result, "total_tracks")
+    ext = _get(result, "external_ids") or {}
+    isrc = _get(result, "isrc") or (ext.get("isrc") if isinstance(ext, dict) else None)
+
+    return {
+        "source": source,
+        "track_id": str(track_id),
+        "track_title": str(title),
+        "artist_name": artist_name,
+        "album_name": str(album_name or ""),
+        "album_type": infer_release_type(raw_type, total),
+        "raw_album_type": str(raw_type or ""),
+        "total_tracks": int(total) if isinstance(total, int) else None,
+        "year": _year(_get(result, "release_date")),
+        "image_url": _get(result, "image_url") or "",
+        "isrc": isrc or None,
+    }
+
+
+def search_release_candidates(
+    source: str,
+    query: str,
+    *,
+    limit: int = 25,
+    client_factory: Optional[Callable[[str], Any]] = None,
+) -> List[Dict[str, Any]]:
+    """Search ``source`` for tracks matching ``query`` → normalized display rows.
+
+    Returns ``[]`` (never raises) when the source has no client or errors — the UI
+    just shows an empty tab. Rows keep duplicate releases; the UI groups them."""
+    query = (query or "").strip()
+    if not query:
+        return []
+    factory = client_factory or _default_client_factory
+    try:
+        client = factory(source)
+    except Exception:
+        client = None
+    if client is None or not hasattr(client, "search_tracks"):
+        return []
+    try:
+        results = client.search_tracks(query, limit=limit)
+    except TypeError:
+        results = client.search_tracks(query)   # clients with no limit kwarg
+    except Exception:
+        return []
+
+    rows: List[Dict[str, Any]] = []
+    for r in results or []:
+        row = normalize_search_result(r, source)
+        if row is not None:
+            rows.append(row)
+    return rows
+
+
+def resolve_hint_fields(
+    source: str,
+    track_id: str,
+    *,
+    client_factory: Optional[Callable[[str], Any]] = None,
+) -> Optional[Dict[str, Any]]:
+    """Resolve the picked track to the fields a hint needs (album_id critically,
+    plus isrc / track# / disc# / album name+type). One lookup for one chosen row.
+    Returns ``None`` if it can't be resolved (caller surfaces an error)."""
+    factory = client_factory or _default_client_factory
+    try:
+        client = factory(source)
+    except Exception:
+        client = None
+    if client is None or not hasattr(client, "get_track_details"):
+        return None
+    try:
+        details = client.get_track_details(track_id)
+    except Exception:
+        return None
+    if not details:
+        return None
+
+    album = _get(details, "album") or {}
+    album_id = _get(album, "id") if not isinstance(album, str) else None
+    album_name = _get(album, "name") if not isinstance(album, str) else album
+    album_type = _get(album, "album_type") or _get(details, "album_type")
+    total = _get(album, "total_tracks") or _get(details, "total_tracks")
+
+    artists = _get(details, "artists") or []
+    artist_id = None
+    artist_name = ""
+    if isinstance(artists, list) and artists:
+        artist_id = _get(artists[0], "id")
+        artist_name = ", ".join(str(_get(a, "name", a) if not isinstance(a, str) else a) for a in artists)
+
+    ext = _get(details, "external_ids") or {}
+    isrc = _get(details, "isrc") or (ext.get("isrc") if isinstance(ext, dict) else None)
+
+    if not album_id:
+        return None   # without an album_id the import can't fetch the tracklist
+
+    return {
+        "source": source,
+        "track_id": str(track_id),
+        "album_id": str(album_id),
+        "artist_id": str(artist_id) if artist_id else None,
+        "track_title": _get(details, "name") or _get(details, "title") or "",
+        "album_name": str(album_name or ""),
+        "artist_name": artist_name,
+        "album_type": infer_release_type(album_type, total),
+        "track_number": _get(details, "track_number"),
+        "disc_number": _get(details, "disc_number") or 1,
+        "isrc": isrc or None,
+    }
+
+
+def _default_client_factory(source: str):
+    from core.metadata.registry import get_client_for_source
+    return get_client_for_source(source)
+
+
+def available_sources() -> List[Dict[str, Any]]:
+    """The source tabs for the modal: every metadata source with a live client,
+    the primary one flagged ``active`` so the UI selects it by default."""
+    from core.metadata.registry import (
+        METADATA_SOURCE_PRIORITY,
+        get_client_for_source,
+        get_primary_source,
+    )
+
+    try:
+        primary = get_primary_source()
+    except Exception:
+        primary = None
+
+    out: List[Dict[str, Any]] = []
+    seen = set()
+    for src in METADATA_SOURCE_PRIORITY:
+        if src in seen:
+            continue
+        seen.add(src)
+        try:
+            client = get_client_for_source(src)
+        except Exception:
+            client = None
+        if client is None or not hasattr(client, "search_tracks"):
+            continue
+        out.append({
+            "source": src,
+            "label": src.replace("_", " ").title(),
+            "active": src == primary,
+        })
+    # Guarantee the primary is selectable + first even if priority ordering missed it.
+    if primary and not any(s["active"] for s in out):
+        out.insert(0, {"source": primary, "label": primary.replace("_", " ").title(), "active": True})
+    return out
+
+
+__all__ = [
+    "infer_release_type",
+    "normalize_search_result",
+    "search_release_candidates",
+    "resolve_hint_fields",
+    "available_sources",
+]
diff --git a/tests/imports/test_rematch_search.py b/tests/imports/test_rematch_search.py
new file mode 100644
index 00000000..dd962bbe
--- /dev/null
+++ b/tests/imports/test_rematch_search.py
@@ -0,0 +1,121 @@
+"""#889 Phase 3: re-identify search — normalize results across sources, infer the
+release-type badge, and resolve the picked row's album_id.
+
+Locks down: same song surfaces as multiple rows (single/EP/album), the EP
+inference from a multi-track 'single', graceful empty on a dead source, and that
+resolve_hint_fields pulls album_id (and refuses a result without one).
+"""
+
+from __future__ import annotations
+
+import types
+
+from core.imports.rematch_search import (
+    available_sources,
+    infer_release_type,
+    normalize_search_result,
+    resolve_hint_fields,
+    search_release_candidates,
+)
+
+
+# typed-Track-ish object (mirrors core.metadata.types.Track attrs the modal reads)
+def _track(tid, title, album, album_type, total, isrc=None, year="2020"):
+    return types.SimpleNamespace(
+        id=tid, name=title, artists=["Artist"], album=album,
+        album_type=album_type, total_tracks=total, release_date=year + "-01-01",
+        image_url="http://img/" + tid, isrc=isrc, external_ids={},
+    )
+
+
+# ── release-type inference ────────────────────────────────────────────────────
+def test_infer_album_stays_album():
+    assert infer_release_type("album", 12) == "album"
+
+
+def test_infer_single_one_track_is_single():
+    assert infer_release_type("single", 1) == "single"
+
+
+def test_infer_multitrack_single_promoted_to_ep():
+    # Spotify labels EPs as album_type='single' — promote on track count.
+    assert infer_release_type("single", 5) == "ep"
+
+
+def test_infer_compilation():
+    assert infer_release_type("compilation", 40) == "compilation"
+
+
+def test_infer_unknown_falls_back_to_count():
+    assert infer_release_type(None, 10) == "album"
+    assert infer_release_type("", 4) == "ep"
+    assert infer_release_type(None, 1) == "single"
+
+
+# ── normalization ─────────────────────────────────────────────────────────────
+def test_normalize_builds_display_row():
+    row = normalize_search_result(_track("t1", "Song", "Album1", "album", 12, isrc="US1234567890"), "spotify")
+    assert row["track_id"] == "t1"
+    assert row["album_name"] == "Album1" and row["album_type"] == "album"
+    assert row["artist_name"] == "Artist"
+    assert row["year"] == "2020" and row["isrc"] == "US1234567890"
+
+
+def test_normalize_skips_result_without_id_or_title():
+    assert normalize_search_result(types.SimpleNamespace(id="", name="X"), "spotify") is None
+    assert normalize_search_result(types.SimpleNamespace(id="t", name=""), "spotify") is None
+
+
+def test_same_song_multiple_collections():
+    """The headline case: one song, three releases, three distinct rows + badges."""
+    results = [
+        _track("t_alb", "Song", "Album1", "album", 12),
+        _track("t_ep", "Song", "EP1", "single", 5),       # multi-track single → EP
+        _track("t_sgl", "Song", "Song (Single)", "single", 1),
+    ]
+    client = types.SimpleNamespace(search_tracks=lambda q, limit=25: results)
+    rows = search_release_candidates("spotify", "Song", client_factory=lambda s: client)
+    badges = {r["album_name"]: r["album_type"] for r in rows}
+    assert badges == {"Album1": "album", "EP1": "ep", "Song (Single)": "single"}
+
+
+def test_search_empty_on_missing_client():
+    assert search_release_candidates("spotify", "x", client_factory=lambda s: None) == []
+
+
+def test_search_empty_on_blank_query():
+    called = []
+    search_release_candidates("spotify", "   ", client_factory=lambda s: called.append(1))
+    assert called == []   # never even fetches a client for an empty query
+
+
+def test_search_swallows_client_error():
+    def boom(q, limit=25):
+        raise RuntimeError("rate limited")
+    client = types.SimpleNamespace(search_tracks=boom)
+    assert search_release_candidates("spotify", "x", client_factory=lambda s: client) == []
+
+
+# ── resolve on select ─────────────────────────────────────────────────────────
+def test_resolve_pulls_album_id_and_fields():
+    details = {
+        "name": "Song", "track_number": 5, "disc_number": 1, "isrc": "US1234567890",
+        "album": {"id": "alb_album1", "name": "Album1", "album_type": "album", "total_tracks": 12},
+        "artists": [{"id": "art_1", "name": "Artist"}],
+    }
+    client = types.SimpleNamespace(get_track_details=lambda tid: details)
+    out = resolve_hint_fields("spotify", "t_alb", client_factory=lambda s: client)
+    assert out["album_id"] == "alb_album1"
+    assert out["artist_id"] == "art_1"
+    assert out["track_number"] == 5 and out["disc_number"] == 1
+    assert out["album_type"] == "album" and out["isrc"] == "US1234567890"
+
+
+def test_resolve_refuses_result_without_album_id():
+    details = {"name": "Song", "album": {"name": "NoId Album"}}   # no album id
+    client = types.SimpleNamespace(get_track_details=lambda tid: details)
+    assert resolve_hint_fields("spotify", "t", client_factory=lambda s: client) is None
+
+
+def test_resolve_none_on_missing_client():
+    assert resolve_hint_fields("spotify", "t", client_factory=lambda s: None) is None
diff --git a/web_server.py b/web_server.py
index 6d9cdb3d..d5234210 100644
--- a/web_server.py
+++ b/web_server.py
@@ -10044,6 +10044,47 @@ def get_artist_enhanced_detail(artist_id):
     except Exception as e:
         return jsonify({"success": False, "error": str(e)}), 500
 
+
+# ── Re-identify an imported track (#889) ──
+@app.route('/api/reidentify/sources', methods=['GET'])
+def reidentify_sources():
+    """Source tabs for the Re-identify modal — every metadata source with a live
+    client, the active one flagged so the UI selects it by default."""
+    try:
+        from core.imports.rematch_search import available_sources
+        return jsonify({"success": True, "sources": available_sources()})
+    except Exception as e:
+        return jsonify({"success": False, "error": str(e), "sources": []}), 500
+
+
+@app.route('/api/reidentify/search', methods=['GET'])
+def reidentify_search():
+    """Search one metadata source for the releases a track appears on.
+
+    Query params: ``source`` (defaults to the active source), ``q`` (the query),
+    ``limit``. Returns display rows — the SAME song across single/EP/album, each
+    with a type badge — for the user to pick. album_id is resolved later, only for
+    the chosen row."""
+    try:
+        from core.imports.rematch_search import available_sources, search_release_candidates
+        query = (request.args.get('q') or '').strip()
+        if not query:
+            return jsonify({"success": True, "results": []})
+        source = (request.args.get('source') or '').strip()
+        if not source:
+            actives = [s for s in available_sources() if s.get('active')]
+            source = actives[0]['source'] if actives else 'spotify'
+        try:
+            limit = max(1, min(50, int(request.args.get('limit', 25))))
+        except (TypeError, ValueError):
+            limit = 25
+        rows = search_release_candidates(source, query, limit=limit)
+        return jsonify({"success": True, "source": source, "results": rows})
+    except Exception as e:
+        logger.error(f"Re-identify search error: {e}")
+        return jsonify({"success": False, "error": str(e), "results": []}), 500
+
+
 @app.route('/api/library/artist/<artist_id>/quality-analysis')
 def get_artist_quality_analysis(artist_id):
     """Analyze track quality for an artist — returns tier classification for each track."""