Library: single->parent-album resolution core (pure selector + injected-I/O resolver)

When a track matches a SINGLE release it carries the single's name/id and the
canonical grouping files it apart from its album-mates -> mixed cover art
(Sokhi). This re-homes it onto the album that actually contains it.

The selection is a pure, CONSERVATIVE function and the lookup loop takes injected
fetchers, so both are unit-testable without a live client. It only re-homes a
track when a real 'album'-type release's tracklist contains that EXACT track
(qualifier-tolerant) — never promotes a genuine standalone single, never guesses
(a wrong promotion would mis-home a real single, the inverse bug). Fail-safe: any
miss/error -> None (track stays as matched). 13 seam tests. Wiring next.

core/imports/single_to_album.py

@@ -0,0 +1,124 @@
+"""Single -> parent-album resolution.
+
+When a track is matched to a SINGLE release (album_type 'single', the single's
+name usually equal to the track title), it carries the single's name + the
+single's source album id. The canonical grouping in
+[core/imports/album_grouping.py] then files it under a different album row than
+its album-mates, and the album-grouped repair jobs dress that row in the
+single's art — songs of one album end up with different covers (Sokhi).
+
+This module re-homes such a track onto the ALBUM it actually belongs to, so it
+carries the album's name/id and groups with the rest of the album.
+
+Design: the SELECTION is a pure, conservative function (no I/O), and the lookup
+loop takes INJECTED fetchers, so both are unit-testable without a live metadata
+client. CONSERVATIVE by intent — it only re-homes a track when a real
+``album``-type release's tracklist *contains that exact track*. It never
+promotes a genuine standalone single and never guesses, because a wrong
+promotion would mis-home a real single onto an album (the inverse bug).
+"""
+
+from __future__ import annotations
+
+import re
+from typing import Any, Callable, Dict, List, Optional
+
+_WS = re.compile(r"\s+")
+# Trailing version qualifiers that differ between a single and its album cut but
+# don't change track identity (kept conservative — only the obvious ones).
+_QUALIFIER = re.compile(
+    r"\s*[\(\[]\s*(album version|single version|radio edit|remaster(ed)?( \d{4})?)\s*[\)\]]\s*$",
+    re.IGNORECASE,
+)
+
+
+def _norm(s: Any) -> str:
+    """Lowercase, strip a trailing '(Album Version)'-style qualifier, collapse
+    whitespace — so 'Song' matches 'Song (Album Version)'."""
+    t = str(s or "").strip().lower()
+    t = _QUALIFIER.sub("", t)
+    return _WS.sub(" ", t).strip()
+
+
+def _get(obj: Any, *keys: str, default=None):
+    for k in keys:
+        if isinstance(obj, dict):
+            if obj.get(k) is not None:
+                return obj.get(k)
+        else:
+            v = getattr(obj, k, None)
+            if v is not None:
+                return v
+    return default
+
+
+def select_parent_album(track_title: str, candidate_albums: List[Dict[str, Any]]) -> Optional[Dict[str, Any]]:
+    """Pick the parent ALBUM for ``track_title`` from normalized candidates, or
+    None. Each candidate is ``{name, album_type, tracks: [title, ...], ...}``.
+
+    Conservative rules — a candidate qualifies ONLY when:
+      * it is an ``album`` release (never single / ep / compilation), and
+      * its name is not just the track title (that IS the single), and
+      * its tracklist contains the track by exact normalized title.
+    Returns the FIRST qualifying candidate (caller passes them in priority
+    order, so the result is deterministic).
+    """
+    tgt = _norm(track_title)
+    if not tgt:
+        return None
+    for alb in candidate_albums or []:
+        if str(_get(alb, "album_type", default="album")).lower() != "album":
+            continue
+        if _norm(_get(alb, "name", "title", default="")) == tgt:
+            continue
+        tracks = _get(alb, "tracks", default=[]) or []
+        if any(_norm(t) == tgt for t in tracks):
+            return alb
+    return None
+
+
+def resolve_single_to_album(
+    track_title: str,
+    *,
+    fetch_album_candidates: Callable[[], List[Dict[str, Any]]],
+    fetch_album_tracks: Callable[[Dict[str, Any]], List[str]],
+    max_albums: int = 8,
+) -> Optional[Dict[str, Any]]:
+    """Find the parent album for a single-matched track. I/O is INJECTED so this
+    is testable without a live client:
+      * ``fetch_album_candidates()`` -> the artist's ALBUM-type releases (dicts
+        with name/album_type/id/source), in priority order.
+      * ``fetch_album_tracks(album)`` -> that album's track titles.
+    Probes at most ``max_albums`` albums, lazily (stops at the first that
+    contains the track). Fail-safe: any error / no confident match -> None
+    (the track stays as it was matched). Returns the normalized winning album
+    ``{name, album_type, album_id, source, tracks}`` or None.
+    """
+    if not _norm(track_title):
+        return None
+    try:
+        albums = fetch_album_candidates() or []
+    except Exception:
+        return None
+
+    probed = 0
+    for alb in albums:
+        if str(_get(alb, "album_type", default="album")).lower() != "album":
+            continue
+        if probed >= max_albums:
+            break
+        probed += 1
+        try:
+            tracks = fetch_album_tracks(alb) or []
+        except Exception:
+            continue
+        normalized = {
+            "name": _get(alb, "name", "title", default=""),
+            "album_type": "album",
+            "album_id": _get(alb, "id", "album_id"),
+            "source": _get(alb, "source"),
+            "tracks": list(tracks),
+        }
+        if select_parent_album(track_title, [normalized]):
+            return normalized
+    return None

tests/imports/test_single_to_album.py

@@ -0,0 +1,134 @@
+"""Seam tests for single -> parent-album resolution (Sokhi: single-matched track
+splits from its album -> mixed cover art). The selector is pure; the resolver
+takes injected fetchers, so neither needs a live metadata client.
+"""
+
+from __future__ import annotations
+
+from core.imports.single_to_album import (
+    select_parent_album,
+    resolve_single_to_album,
+)
+
+
+# ── pure selector ─────────────────────────────────────────────────────────────
+def _alb(name, tracks, album_type="album", **extra):
+    return {"name": name, "album_type": album_type, "tracks": tracks, **extra}
+
+
+def test_picks_album_that_contains_the_track():
+    got = select_parent_album("Yellow", [
+        _alb("Parachutes", ["Don't Panic", "Shiver", "Yellow", "Trouble"]),
+    ])
+    assert got and got["name"] == "Parachutes"
+
+
+def test_returns_none_when_no_album_contains_the_track():
+    assert select_parent_album("Yellow", [
+        _alb("Some Other Album", ["Track A", "Track B"]),
+    ]) is None
+
+
+def test_never_promotes_onto_a_single_release():
+    # The single's own release (album_type 'single', name == track) must be ignored.
+    assert select_parent_album("Yellow", [
+        _alb("Yellow", ["Yellow"], album_type="single"),
+    ]) is None
+
+
+def test_ignores_ep_and_compilation_types():
+    assert select_parent_album("Yellow", [
+        _alb("Yellow EP", ["Yellow", "Yellow (Live)"], album_type="ep"),
+        _alb("Greatest Hits", ["Yellow", "Clocks"], album_type="compilation"),
+    ]) is None
+
+
+def test_skips_album_named_exactly_like_the_track():
+    # An 'album' whose name IS the track title is the single dressed as an album;
+    # don't treat it as the parent.
+    assert select_parent_album("Yellow", [
+        _alb("Yellow", ["Yellow"]),
+    ]) is None
+
+
+def test_matches_through_album_version_qualifier():
+    got = select_parent_album("Yellow", [
+        _alb("Parachutes", ["Shiver", "Yellow (Album Version)"]),
+    ])
+    assert got and got["name"] == "Parachutes"
+
+
+def test_first_qualifying_candidate_wins_deterministically():
+    got = select_parent_album("Yellow", [
+        _alb("Parachutes", ["Yellow"]),
+        _alb("Parachutes (Deluxe)", ["Yellow", "Bonus"]),
+    ])
+    assert got["name"] == "Parachutes"  # input order = priority
+
+
+def test_empty_title_returns_none():
+    assert select_parent_album("", [_alb("Parachutes", ["Yellow"])]) is None
+
+
+# ── injected-I/O resolver ─────────────────────────────────────────────────────
+def test_resolver_finds_parent_album_lazily():
+    calls = {"tracks": 0}
+    albums = [
+        {"name": "Single Yellow", "album_type": "single", "id": "s1"},   # skipped (not album)
+        {"name": "Wrong Album", "album_type": "album", "id": "a1"},
+        {"name": "Parachutes", "album_type": "album", "id": "a2"},
+    ]
+
+    def fetch_tracks(alb):
+        calls["tracks"] += 1
+        return {"a1": ["Other"], "a2": ["Yellow", "Shiver"]}.get(alb["id"], [])
+
+    got = resolve_single_to_album(
+        "Yellow",
+        fetch_album_candidates=lambda: albums,
+        fetch_album_tracks=fetch_tracks,
+    )
+    assert got and got["name"] == "Parachutes" and got["album_id"] == "a2"
+    assert calls["tracks"] == 2  # probed a1 then a2, stopped; never probed the single
+
+
+def test_resolver_returns_none_when_nothing_contains_track():
+    got = resolve_single_to_album(
+        "Yellow",
+        fetch_album_candidates=lambda: [{"name": "X", "album_type": "album", "id": "a1"}],
+        fetch_album_tracks=lambda alb: ["Nope"],
+    )
+    assert got is None
+
+
+def test_resolver_is_failsafe_on_candidate_fetch_error():
+    def boom():
+        raise RuntimeError("api down")
+    assert resolve_single_to_album(
+        "Yellow", fetch_album_candidates=boom, fetch_album_tracks=lambda a: []) is None
+
+
+def test_resolver_is_failsafe_on_track_fetch_error():
+    def boom(alb):
+        raise RuntimeError("api down")
+    got = resolve_single_to_album(
+        "Yellow",
+        fetch_album_candidates=lambda: [{"name": "Parachutes", "album_type": "album", "id": "a1"}],
+        fetch_album_tracks=boom)
+    assert got is None
+
+
+def test_resolver_caps_albums_probed():
+    albums = [{"name": f"A{i}", "album_type": "album", "id": str(i)} for i in range(20)]
+    probed = {"n": 0}
+
+    def fetch_tracks(alb):
+        probed["n"] += 1
+        return ["nope"]
+
+    resolve_single_to_album(
+        "Yellow",
+        fetch_album_candidates=lambda: albums,
+        fetch_album_tracks=fetch_tracks,
+        max_albums=5)
+    assert probed["n"] == 5  # never probes more than the cap