diff --git a/core/imports/single_to_album.py b/core/imports/single_to_album.py new file mode 100644 index 00000000..e6fc5f2d --- /dev/null +++ b/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 diff --git a/tests/imports/test_single_to_album.py b/tests/imports/test_single_to_album.py new file mode 100644 index 00000000..7933b709 --- /dev/null +++ b/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