From 69fc21d6b22b5a09d92d0550afba62b056d3aa47 Mon Sep 17 00:00:00 2001 From: BoulderBadgeDad Date: Sat, 6 Jun 2026 10:52:58 -0700 Subject: [PATCH] #767-2: reorganize finds the right album edition instead of mislabeling singles as deluxe MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A single enriched against the deluxe gets every source ID pointing at the deluxe, so the organizer filed it as e.g. track 2 of a 10-track album. Root cause: the canonical resolver only ever scored the editions already linked — the correct single was never even a candidate, and the misfit deluxe scored so low (0.1, below the 0.5 floor) that nothing got pinned and the priority-walk grabbed the deluxe anyway. Fix, in three tested layers: - resolve_canonical_for_album gains a fetch_alternates seam: when no linked edition clears the floor, it scores the source's OTHER editions of the same release and re-picks by best fit (dedup, injected, pure). - default_fetch_alternates lists the artist's editions and keeps the same-release ones (edition-blind name match: Deluxe / - Single / [Remastered] all collapse), returning their tracklists. Favors recall; the scorer is the precision gate. - _resolve_source does the misfit check inline: it fit-scores the walked edition and only on a clear misfit searches for a better edition, then persists the pin on apply (Track Number Repair + future runs agree). Cost-neutral and behavior- identical for well-fitting albums (no extra API calls); strict_source and the #758 manual lock are never overridden. Tests: +4 resolver (expand/no-expand/dedupe/back-compat), +7 alternates (name matcher + fetcher over fake APIs + cap), +3 organizer end-to-end (misfit->single +pin, well-fit->no-expand, strict->no-expand). 300 passed across the reorganize + canonical family, lint clean. --- core/library_reorganize.py | 130 +++++++++++++- core/metadata/canonical_resolver.py | 200 ++++++++++++++++++--- tests/test_canonical_alternates.py | 106 +++++++++++ tests/test_canonical_resolver.py | 100 +++++++++++ tests/test_reorganize_alternate_edition.py | 119 ++++++++++++ 5 files changed, 628 insertions(+), 27 deletions(-) create mode 100644 tests/test_canonical_alternates.py create mode 100644 tests/test_reorganize_alternate_edition.py diff --git a/core/library_reorganize.py b/core/library_reorganize.py index 02509c11..e2991682 100644 --- a/core/library_reorganize.py +++ b/core/library_reorganize.py @@ -213,7 +213,79 @@ def _unresolvable_reason(album_data: dict, primary_source: str, strict_source: b return "No metadata source ID for this album" -def _resolve_source(album_data: dict, primary_source: str, strict_source: bool = False): +# #767-2: a walked edition scoring below this against the on-disk files is treated +# as the WRONG edition (e.g. a 1-track single vs the 10-track deluxe scores 0.1), +# triggering the alternate-edition search. Matches the resolver's min_score. +_CANONICAL_FIT_FLOOR = 0.5 + + +def _score_edition_items(file_tracks: List[dict], items: List[dict]) -> float: + """Score a fetched provider tracklist (raw ``items``) against the on-disk + ``file_tracks`` using the canonical scorer. Normalises the provider's varied + shapes (``name``/``title``, ``duration_ms``/``duration`` seconds) first.""" + from core.metadata.canonical_version import score_release_against_files + rel = [] + for it in items or []: + dur = it.get('duration_ms') + if dur is None: + secs = it.get('duration') + dur = int(secs * 1000) if isinstance(secs, (int, float)) and secs else None + rel.append({'title': it.get('name') or it.get('title') or '', 'duration_ms': dur}) + return score_release_against_files(file_tracks, rel) if rel else 0.0 + + +def _resolve_better_edition(album_data, source_ids, file_tracks, primary_source): + """Misfit path: run the canonical resolver WITH alternate-edition expansion and, + if it lands on a genuinely different edition than the linked ones, fetch it for + organizing. Returns ``(source, album_id, api_album, items, score)`` or ``None``.""" + from core.metadata.canonical_resolver import ( + default_fetch_alternates, + default_fetch_tracklist, + resolve_canonical_for_album, + ) + art_id = str(album_data.get('artist_id') or '') + art_name = album_data.get('artist_name') or '' + title = album_data.get('title') or '' + + def _alts(source, aid): + return default_fetch_alternates( + source, aid, artist_id=art_id, artist_name=art_name, album_title=title, + ) + + try: + result = resolve_canonical_for_album( + album_source_ids=source_ids, + file_tracks=file_tracks, + fetch_tracklist=default_fetch_tracklist, + fetch_alternates=_alts, + source_priority=get_source_priority(primary_source), + primary_source=primary_source, + ) + except Exception as e: + logger.warning(f"[Reorganize] canonical resolve raised: {e}") + return None + if not result: + return None + linked = source_ids.get(result['source']) + if str(result['album_id']) == str(linked or ''): + return None # resolver chose a linked edition the walk already considered + try: + b_album = get_album_for_source(result['source'], result['album_id']) + b_items = _normalize_album_tracks( + get_album_tracks_for_source(result['source'], result['album_id']) + ) + except Exception as e: + logger.warning(f"[Reorganize] alternate edition fetch raised: {e}") + return None + if not b_album or not b_items: + return None + return result['source'], result['album_id'], b_album, b_items, result.get('score') or 0.0 + + +def _resolve_source( + album_data: dict, primary_source: str, strict_source: bool = False, + *, file_tracks: Optional[List[dict]] = None, on_better_edition=None, +): """Walk the configured source priority looking for the first source we have an ID for AND that returns a usable tracklist. @@ -223,6 +295,11 @@ def _resolve_source(album_data: dict, primary_source: str, strict_source: bool = means "use Spotify or fail", not "use Spotify and silently fall back to Deezer". + When ``file_tracks`` is supplied (and not ``strict_source``), the walked + edition is fit-scored against the on-disk files; a clear misfit triggers an + alternate-edition search (#767-2). ``on_better_edition(source, album_id, + score)`` is invoked to persist the pin when a better edition is chosen. + Returns ``(source_name, album_meta, tracks_list)`` or ``(None, None, None)``. """ source_ids = _extract_source_ids(album_data) @@ -251,6 +328,7 @@ def _resolve_source(album_data: dict, primary_source: str, strict_source: bool = else: sources_to_try = get_source_priority(primary_source) + walk_source = walk_album = walk_items = None for source in sources_to_try: sid = source_ids.get(source) or '' if not sid: @@ -264,8 +342,36 @@ def _resolve_source(album_data: dict, primary_source: str, strict_source: bool = items = _normalize_album_tracks(api_tracks) if not items or not api_album: continue - return source, api_album, items + walk_source, walk_album, walk_items = source, api_album, items + break + # #767-2: the walk takes the first source we have an ID for, but that ID can + # point at the WRONG edition (a single enriched against the deluxe → it'd file + # the track as #2 of a 10-track album). With the on-disk tracklist in hand, + # fit-score the walked edition; only a clear misfit looks for a better-fitting + # edition. Well-fitting albums keep today's exact behavior + make no extra calls. + if not strict_source and file_tracks: + walk_fit = _score_edition_items(file_tracks, walk_items) if walk_items else 0.0 + if walk_fit < _CANONICAL_FIT_FLOOR: + better = _resolve_better_edition( + album_data, source_ids, file_tracks, primary_source, + ) + if better is not None: + b_source, b_id, b_album, b_items, b_score = better + if on_better_edition: + try: + on_better_edition(b_source, b_id, b_score) + except Exception as e: + logger.warning(f"[Reorganize] canonical pin persist failed: {e}") + logger.info( + "[Reorganize] %s: walked edition fit %.2f below floor — using " + "better-fit %s edition %s (fit %.2f)", + album_data.get('title', '?'), walk_fit, b_source, b_id, b_score, + ) + return b_source, b_album, b_items + + if walk_source: + return walk_source, walk_album, walk_items return None, None, None @@ -682,6 +788,7 @@ def plan_album_reorganize( strict_source: bool = False, metadata_source: str = 'api', resolve_file_path_fn: Optional[Callable[[Optional[str]], Optional[str]]] = None, + on_better_edition: Optional[Callable[[str, str, float], None]] = None, ) -> dict: """Compute the per-track plan for an album reorganize without doing any file IO. Both the actual reorganize orchestrator and the preview @@ -731,8 +838,14 @@ def plan_album_reorganize( except Exception: primary_source = 'deezer' + # On-disk track shape for the #767-2 fit check (duration stored in ms). + file_tracks = [ + {'duration_ms': t.get('duration') or 0, 'title': t.get('title') or ''} + for t in tracks + ] source, api_album, api_tracks = _resolve_source( - album_data, primary_source, strict_source=strict_source + album_data, primary_source, strict_source=strict_source, + file_tracks=file_tracks, on_better_edition=on_better_edition, ) if not source: reason = _unresolvable_reason(album_data, primary_source, strict_source) @@ -1475,12 +1588,23 @@ def reorganize_album( summary['total'] = len(tracks) _emit(total=len(tracks)) + # #767-2: persist the canonical pin when the resolver lands on a better-fit + # edition than the linked one, so Track Number Repair + future runs agree and + # we don't re-resolve every time. Never overrides a manually-locked pin (the + # set_album_canonical SQL guard from #758 enforces that). + def _persist_canonical(source, alt_album_id, score): + try: + db.set_album_canonical(album_id, source, alt_album_id, score) + except Exception as e: + logger.warning("[Reorganize] set_album_canonical failed: %s", e) + # Build the per-track plan (same logic the preview uses). plan = plan_album_reorganize( album_data, tracks, primary_source=primary_source, strict_source=strict_source, metadata_source=metadata_source, resolve_file_path_fn=resolve_file_path_fn, + on_better_edition=_persist_canonical, ) if plan['status'] == 'no_source_id': summary['status'] = 'no_source_id' diff --git a/core/metadata/canonical_resolver.py b/core/metadata/canonical_resolver.py index a18b4b8b..78426591 100644 --- a/core/metadata/canonical_resolver.py +++ b/core/metadata/canonical_resolver.py @@ -38,6 +38,9 @@ def resolve_canonical_for_album( min_score: float = 0.5, mode: str = MODE_ACTIVE_PREFERRED, primary_source: Optional[str] = None, + fetch_alternates: Optional[ + Callable[[str, str], Optional[List[Dict[str, Any]]]] + ] = None, ) -> Optional[Dict[str, Any]]: """Pick the canonical release for one album, honoring the source-selection mode. @@ -68,48 +71,90 @@ def resolve_canonical_for_album( if not file_tracks: return None primary = primary_source or (source_priority[0] if source_priority else None) - scored: List[Dict[str, Any]] = [] # every source we actually scored + scored: List[Dict[str, Any]] = [] # every edition we actually scored + seen: set = set() # (source, album_id) already scored — dedup linked + alternates - def _score(source: Optional[str]) -> Optional[Dict[str, Any]]: - if not source or any(e['source'] == source for e in scored): - return next((e for e in scored if e['source'] == source), None) - album_id = album_source_ids.get(source) - if not album_id: + def _score_edition( + source: Optional[str], album_id: Any, + tracks: Optional[List[Dict[str, Any]]] = None, + ) -> Optional[Dict[str, Any]]: + """Score one concrete (source, album_id) edition, deduped by that pair. + Fetches the tracklist when not pre-supplied. Returns the entry (existing + on a repeat) or None when it has no resolvable tracklist.""" + if not source or not album_id: return None - try: - tracks = fetch_tracklist(source, str(album_id)) - except Exception: - tracks = None + key = (source, str(album_id)) + if key in seen: + return next((e for e in scored if (e['source'], e['album_id']) == key), None) + if tracks is None: + try: + tracks = fetch_tracklist(source, str(album_id)) + except Exception: + tracks = None if not tracks: return None entry = { 'source': source, 'album_id': str(album_id), - 'track_count': len(tracks), 'score': round(score_release_against_files(file_tracks, tracks), 4), + 'track_count': len(tracks), + 'score': round(score_release_against_files(file_tracks, tracks), 4), '_tracks': tracks, } scored.append(entry) + seen.add(key) return entry + def _score_linked(source: Optional[str]) -> Optional[Dict[str, Any]]: + return _score_edition(source, album_source_ids.get(source)) if source else None + + def _best_clearing_floor( + entries: List[Dict[str, Any]], + ) -> Optional[Dict[str, Any]]: + best = None + for e in entries: # priority-ordered -> strictly-greater = priority tiebreak + if best is None or e['score'] > best['score'] + 1e-9: + best = e + return best if (best and best['score'] >= min_score) else None + winner: Optional[Dict[str, Any]] = None - # Active-source modes: try the primary first. + # Active-source modes: try the primary's linked edition first. if mode in (MODE_ACTIVE_ONLY, MODE_ACTIVE_PREFERRED): - p = _score(primary) + p = _score_linked(primary) if p and p['score'] >= min_score: winner = p - elif mode == MODE_ACTIVE_ONLY: - return None # never consider other sources - # best_fit, or active_preferred fallback: score the rest and pick the best. - if winner is None: + # best_fit, or active_preferred fallback: score the rest of the linked editions. + if winner is None and mode != MODE_ACTIVE_ONLY: for source in source_priority: - _score(source) - best = None - for e in scored: # source_priority order -> strictly-greater = priority tiebreak - if best is None or e['score'] > best['score'] + 1e-9: - best = e - if best and best['score'] >= min_score: - winner = best + _score_linked(source) + winner = _best_clearing_floor(scored) + + # #767-2 expansion: no LINKED edition cleared the floor — e.g. a 1-track single + # linked only to the 10-track deluxe, whose count_fit tanks its score to 0.1. + # Fetch the source's OTHER editions of the same release and score those too, + # then re-pick. Gated on winner-is-None so a well-fitting library never + # triggers a fetch (zero behaviour change + no API cost for the common case). + if winner is None and fetch_alternates is not None: + if mode == MODE_ACTIVE_ONLY: + expand_sources = [primary] if primary else [] + else: + expand_sources = [s for s in source_priority if album_source_ids.get(s)] + for s in album_source_ids: # any linked source not in the priority list + if s not in expand_sources: + expand_sources.append(s) + for source in expand_sources: + linked_id = album_source_ids.get(source) + if not linked_id: + continue + try: + alts = fetch_alternates(source, str(linked_id)) or [] + except Exception: + alts = [] + for alt in alts: + _score_edition(source, alt.get('album_id'), alt.get('tracks')) + # active_only stays on-source; other modes re-pick across everything scored. + pool = [e for e in scored if e['source'] == primary] if mode == MODE_ACTIVE_ONLY else scored + winner = _best_clearing_floor(pool) if winner is None: return None @@ -172,6 +217,98 @@ def default_fetch_tracklist(source: str, album_id: str) -> Optional[List[Dict[st return out or None +# Edition/format qualifiers stripped when deciding whether two album titles name +# the SAME underlying release (so "Scatterbrain", "Scatterbrain (Deluxe)" and +# "Scatterbrain - Single" all collapse to one key). Generous on purpose: the +# scorer is the real precision gate, so over-including an edition is harmless — +# it just won't win. Under-including is what hides the right single. +_EDITION_TOKENS = frozenset({ + "deluxe", "expanded", "edition", "remaster", "remastered", "single", "ep", + "anniversary", "special", "bonus", "explicit", "clean", "version", "extended", + "complete", "collectors", "reissue", "original", "standard", +}) + + +def _release_name_key(name: str) -> str: + """Normalise an album title to a comparison key for 'same release': lowercase, + drop bracketed qualifiers, strip punctuation, and remove edition/format words. + Pure — unit-tested directly.""" + import re + if not name: + return "" + t = str(name).lower() + t = re.sub(r"[\(\[].*?[\)\]]", " ", t) # (Deluxe Edition), [Remastered] + t = re.sub(r"[^a-z0-9 ]", " ", t) # punctuation -> space ("- Single") + toks = [w for w in t.split() if w not in _EDITION_TOKENS] + return " ".join(toks) + + +def _same_release(a: str, b: str) -> bool: + """True when two album titles name the same underlying release (edition-blind).""" + ka, kb = _release_name_key(a), _release_name_key(b) + return bool(ka) and ka == kb + + +def default_fetch_alternates( + source: str, album_id: str, *, + artist_id: str = "", artist_name: str = "", album_title: str = "", + max_editions: int = 6, +) -> Optional[List[Dict[str, Any]]]: + """Production ``fetch_alternates``: list a release's OTHER editions on a source + and return ``[{album_id, tracks}, ...]`` for the canonical resolver to score. + + Strategy: discover the album's artist + title (from supplied context, else one + ``get_album_for_source`` call), list the artist's albums+singles, keep the ones + whose title is the same release (edition-blind), and fetch each one's tracklist. + Best-effort throughout — returns ``[]`` on any miss so the resolver simply + finds no alternates rather than erroring. Only ever called on the misfit path, + so the artist-albums + per-edition fetches don't run for a well-fitting library.""" + try: + from core.metadata.album_tracks import ( + get_album_for_source, + get_artist_albums_for_source, + ) + except Exception: + return [] + + title = album_title + a_id, a_name = str(artist_id or ""), str(artist_name or "") + if not (title and (a_id or a_name)): + try: + meta = get_album_for_source(source, str(album_id)) or {} + except Exception: + meta = {} + title = title or (_item_get(meta, "name") or _item_get(meta, "title") or "") + a_id = a_id or str(_item_get(meta, "artist_id") or _item_get(meta, "artistId") or "") + a_name = a_name or str(_item_get(meta, "artist") or _item_get(meta, "artist_name") or "") + if not title or not (a_id or a_name): + return [] + + try: + albums = get_artist_albums_for_source( + source, a_id, a_name, album_type="album,single", limit=50, + ) or [] + except Exception: + albums = [] + + out: List[Dict[str, Any]] = [] + seen_ids: set = set() + for alb in albums: + if len(out) >= max_editions: + break + alb_title = _item_get(alb, "name") or _item_get(alb, "title") or "" + if not _same_release(title, alb_title): + continue + alb_id = _item_get(alb, "id") or _item_get(alb, "album_id") + if not alb_id or str(alb_id) in seen_ids: + continue + seen_ids.add(str(alb_id)) + tracks = default_fetch_tracklist(source, str(alb_id)) + if tracks: + out.append({"album_id": str(alb_id), "tracks": tracks}) + return out + + def _lookup_artist_thumb(db, artist_id) -> Optional[str]: """Best-effort artist thumb URL by id. Returns None on missing column / any error (the artists table doesn't have thumb_url in every schema).""" @@ -199,6 +336,7 @@ def resolve_and_store_canonical_for_album( album_id, *, fetch_tracklist: Optional[Callable[[str, str], Any]] = None, + fetch_alternates: Optional[Callable[[str, str], Any]] = None, source_priority: Optional[List[str]] = None, min_score: float = 0.5, store: bool = True, @@ -230,6 +368,18 @@ def resolve_and_store_canonical_for_album( if fetch_tracklist is None: fetch_tracklist = default_fetch_tracklist + if fetch_alternates is None: + # Default alternates fetcher, primed with the artist/title we already + # loaded (no extra get_album call). Only fires on the misfit path. + _art_id = str(album_data.get('artist_id') or '') + _art_name = album_data.get('artist_name') or '' + _title = album_data.get('title') or '' + + def fetch_alternates(source, aid): # noqa: ANN001 + return default_fetch_alternates( + source, aid, + artist_id=_art_id, artist_name=_art_name, album_title=_title, + ) primary_source = None if source_priority is None: try: @@ -243,6 +393,7 @@ def resolve_and_store_canonical_for_album( album_source_ids=source_ids, file_tracks=file_tracks, fetch_tracklist=fetch_tracklist, + fetch_alternates=fetch_alternates, source_priority=source_priority, min_score=min_score, mode=mode, @@ -278,4 +429,5 @@ __all__ = [ "resolve_canonical_for_album", "resolve_and_store_canonical_for_album", "default_fetch_tracklist", + "default_fetch_alternates", ] diff --git a/tests/test_canonical_alternates.py b/tests/test_canonical_alternates.py new file mode 100644 index 00000000..e242d342 --- /dev/null +++ b/tests/test_canonical_alternates.py @@ -0,0 +1,106 @@ +"""Tests for #767-2 alternate-edition fetching: the pure same-release name +matcher, and the production default_fetch_alternates wired over fake source APIs.""" + +from __future__ import annotations + +import core.metadata.canonical_resolver as cr +from core.metadata.canonical_resolver import ( + _release_name_key, + _same_release, + default_fetch_alternates, +) + + +# ── pure same-release name matching ─────────────────────────────────────────── + +def test_release_name_key_strips_editions_and_punctuation(): + assert _release_name_key("Scatterbrain") == "scatterbrain" + assert _release_name_key("Scatterbrain (Deluxe Edition)") == "scatterbrain" + assert _release_name_key("Scatterbrain - Single") == "scatterbrain" + assert _release_name_key("Scatterbrain [Remastered]") == "scatterbrain" + assert _release_name_key("Scatterbrain (Expanded)") == "scatterbrain" + + +def test_same_release_matches_editions_of_one_album(): + assert _same_release("Scatterbrain", "Scatterbrain (Deluxe)") + assert _same_release("Scatterbrain - Single", "Scatterbrain (Deluxe Edition)") + assert _same_release("The Wall", "The Wall [Remastered]") + + +def test_same_release_rejects_different_albums(): + assert not _same_release("Scatterbrain", "Brain Scatter") + assert not _same_release("Yellow", "Parachutes") + assert not _same_release("", "Anything") # empty key never matches + + +# ── production fetcher over fake source APIs ────────────────────────────────── + +SINGLE = [{"name": "Scatterbrain", "track_number": 1, "duration_ms": 129_000}] +DELUXE = [{"name": "Intro", "track_number": 1, "duration_ms": 200_000}] + [ + {"name": f"Track {i}", "track_number": i + 1, "duration_ms": 180_000} + for i in range(1, 10) +] + + +def _install_fake_apis(monkeypatch, *, artist_albums, tracklists, album_meta=None): + """Patch the album_tracks module functions default_fetch_alternates imports.""" + import core.metadata.album_tracks as at + + monkeypatch.setattr(at, "get_album_for_source", + lambda s, aid: (album_meta or {}).get(aid), raising=True) + monkeypatch.setattr(at, "get_artist_albums_for_source", + lambda s, a_id, a_name, **kw: artist_albums, raising=True) + # default_fetch_alternates pulls per-edition tracklists via default_fetch_tracklist, + # which calls get_album_tracks_for_source in the metadata_service module. + monkeypatch.setattr( + "core.metadata_service.get_album_tracks_for_source", + lambda s, aid: tracklists.get(aid), raising=True, + ) + + +def test_default_fetch_alternates_finds_the_single(monkeypatch): + artist_albums = [ + {"id": "sp_deluxe", "name": "Scatterbrain (Deluxe)"}, + {"id": "sp_single", "name": "Scatterbrain - Single"}, + {"id": "other", "name": "A Different Album"}, + ] + tracklists = {"sp_deluxe": DELUXE, "sp_single": SINGLE, "other": SINGLE} + _install_fake_apis(monkeypatch, artist_albums=artist_albums, tracklists=tracklists) + + out = default_fetch_alternates( + "spotify", "sp_deluxe", + artist_id="art1", artist_name="The Band", album_title="Scatterbrain", + ) + ids = {e["album_id"] for e in out} + assert ids == {"sp_deluxe", "sp_single"} # the unrelated album is excluded + single = next(e for e in out if e["album_id"] == "sp_single") + assert len(single["tracks"]) == 1 and single["tracks"][0]["duration_ms"] == 129_000 + + +def test_default_fetch_alternates_discovers_artist_from_album_meta(monkeypatch): + # No artist context supplied -> it must call get_album_for_source to learn it. + album_meta = {"sp_deluxe": {"title": "Scatterbrain", "artist_id": "art1", "artist": "The Band"}} + artist_albums = [{"id": "sp_single", "name": "Scatterbrain (Single)"}] + tracklists = {"sp_single": SINGLE} + _install_fake_apis(monkeypatch, artist_albums=artist_albums, + tracklists=tracklists, album_meta=album_meta) + + out = default_fetch_alternates("spotify", "sp_deluxe") + assert [e["album_id"] for e in out] == ["sp_single"] + + +def test_default_fetch_alternates_empty_when_no_artist(monkeypatch): + _install_fake_apis(monkeypatch, artist_albums=[], tracklists={}) + out = default_fetch_alternates("spotify", "x", album_title="Scatterbrain") + assert out == [] # no artist id/name and no album meta -> nothing to search + + +def test_default_fetch_alternates_caps_editions(monkeypatch): + # 10 same-release editions, cap is 6. + artist_albums = [{"id": f"e{i}", "name": "Scatterbrain (Version %d)" % i} for i in range(10)] + tracklists = {f"e{i}": SINGLE for i in range(10)} + _install_fake_apis(monkeypatch, artist_albums=artist_albums, tracklists=tracklists) + out = default_fetch_alternates( + "spotify", "e0", artist_id="a", album_title="Scatterbrain", max_editions=6, + ) + assert len(out) == 6 diff --git a/tests/test_canonical_resolver.py b/tests/test_canonical_resolver.py index 516eac11..7b7ab461 100644 --- a/tests/test_canonical_resolver.py +++ b/tests/test_canonical_resolver.py @@ -193,3 +193,103 @@ def test_score_is_rounded(): source_priority=PRIORITY, ) assert out["score"] == round(out["score"], 4) + + +# ── #767-2: expand to alternate editions when the linked one clearly misfits ── +# +# The reported bug: a 1-track single was enriched against the *deluxe* album, so +# every linked source ID points at the 10-track deluxe. The single is never a +# candidate, so it can never be chosen — and the deluxe scores so badly (0.1, +# below the 0.5 floor) that nothing gets pinned and the organizer falls back to +# the stored deluxe ID. The fix: when no LINKED edition clears the floor, fetch +# the source's other editions of the same release and score those too. + +SINGLE_FILE = [{"duration_ms": 129_000, "title": "Scatterbrain"}] # owns the 2:09 single +# 10-track deluxe; "Scatterbrain" sits at track 2 (2:10, within ±3s of the file). +DELUXE_10 = ( + [{"duration_ms": 200_000, "title": "Intro"}] + + [{"duration_ms": 130_000, "title": "Scatterbrain"}] + + [{"duration_ms": 180_000 + i * 5_000, "title": f"Deluxe {i+1}"} for i in range(8)] +) +SINGLE_RELEASE = [{"duration_ms": 129_000, "title": "Scatterbrain", "track_number": 1}] + + +def _alternates(table): + """fetch_alternates backed by a {(source, album_id): [release, ...]} table. + Each release is {'album_id', 'tracks'} — a sibling edition from that source.""" + def fetch(source, album_id): + return table.get((source, album_id)) or [] + return fetch + + +def test_expands_to_alternate_edition_when_linked_misfits(): + # Only linked id is the deluxe; user actually owns the single. + alt_table = { + ("spotify", "sp_deluxe"): [ + {"album_id": "sp_single", "tracks": SINGLE_RELEASE}, + {"album_id": "sp_deluxe", "tracks": DELUXE_10}, # the edition we already have + ], + } + out = resolve_canonical_for_album( + album_source_ids={"spotify": "sp_deluxe"}, + file_tracks=list(SINGLE_FILE), + fetch_tracklist=_fetcher({("spotify", "sp_deluxe"): DELUXE_10}), + fetch_alternates=_alternates(alt_table), + source_priority=PRIORITY, + ) + assert out is not None, "deluxe alone scores 0.1 (below floor) — must expand" + assert out["source"] == "spotify" and out["album_id"] == "sp_single" + assert out["score"] > 0.9 + + +def test_does_not_expand_when_linked_edition_fits(): + # Linked edition fits the files (score ~1.0) -> NEVER fetch alternates (cost + # guard + zero behaviour change for the 95% common case). + calls = [] + + def spy(source, album_id): + calls.append((source, album_id)) + return [{"album_id": "sp_other", "tracks": DELUXE_10}] + + out = resolve_canonical_for_album( + album_source_ids={"spotify": "sp_std"}, + file_tracks=list(STD), + fetch_tracklist=_fetcher({("spotify", "sp_std"): STD}), + fetch_alternates=spy, + source_priority=PRIORITY, + ) + assert out["source"] == "spotify" and out["album_id"] == "sp_std" + assert calls == [], "must not fetch alternates when the linked edition already fits" + + +def test_expansion_dedupes_alternates_against_linked(): + # The alternates list re-offers the linked deluxe; it must not be double-scored + # and must not crash. The exact-fit single still wins. + alt_table = { + ("spotify", "sp_deluxe"): [ + {"album_id": "sp_deluxe", "tracks": DELUXE_10}, # dup of the linked one + {"album_id": "sp_single", "tracks": SINGLE_RELEASE}, + ], + } + out = resolve_canonical_for_album( + album_source_ids={"spotify": "sp_deluxe"}, + file_tracks=list(SINGLE_FILE), + fetch_tracklist=_fetcher({("spotify", "sp_deluxe"): DELUXE_10}), + fetch_alternates=_alternates(alt_table), + source_priority=PRIORITY, + ) + assert out["album_id"] == "sp_single" + ids = [c["album_id"] for c in out["candidates"]] + assert ids.count("sp_deluxe") == 1, "linked edition must be scored exactly once" + + +def test_no_expansion_when_alternates_not_provided(): + # Backward-compat: without a fetch_alternates callable, behaviour is exactly + # as before — the misfit deluxe stays unresolved (None). + out = resolve_canonical_for_album( + album_source_ids={"spotify": "sp_deluxe"}, + file_tracks=list(SINGLE_FILE), + fetch_tracklist=_fetcher({("spotify", "sp_deluxe"): DELUXE_10}), + source_priority=PRIORITY, + ) + assert out is None diff --git a/tests/test_reorganize_alternate_edition.py b/tests/test_reorganize_alternate_edition.py new file mode 100644 index 00000000..045df645 --- /dev/null +++ b/tests/test_reorganize_alternate_edition.py @@ -0,0 +1,119 @@ +"""#767-2: the reorganizer's on-demand alternate-edition path. + +When the walked edition (the first source we have an ID for) clearly misfits the +on-disk files — e.g. a 1-track single whose only ID points at the 10-track deluxe +— `_resolve_source` must find a better-fitting edition, use it for the plan, and +(on apply) persist the canonical pin. A well-fitting album must keep today's exact +behavior and never trigger an alternate fetch.""" + +from __future__ import annotations + +import core.library_reorganize as lr +import core.metadata.canonical_resolver as cr + +# Provider-shaped raw tracklists (what get_album_tracks_for_source returns). +SINGLE_RAW = [{"name": "Scatterbrain", "track_number": 1, "duration_ms": 129_000}] +DELUXE_RAW = [{"name": "Intro", "track_number": 1, "duration_ms": 200_000}] + [ + {"name": "Scatterbrain", "track_number": 2, "duration_ms": 130_000} +] + [ + {"name": f"Bonus {i}", "track_number": i + 2, "duration_ms": 180_000} + for i in range(1, 9) +] +# Resolver-normalised shape (what default_fetch_tracklist returns). +SINGLE_NORM = [{"title": "Scatterbrain", "track_number": 1, "duration_ms": 129_000}] +DELUXE_NORM = [{"title": t["name"], "duration_ms": t["duration_ms"]} for t in DELUXE_RAW] + +ALBUM_META = { + "sp_deluxe": {"name": "Scatterbrain (Deluxe)"}, + "sp_single": {"name": "Scatterbrain - Single"}, +} +TRACKLISTS = {"sp_deluxe": DELUXE_RAW, "sp_single": SINGLE_RAW} + + +def _wire(monkeypatch, *, alternates): + """Patch the source-API seams the reorganizer + resolver funnel through.""" + monkeypatch.setattr(lr, "get_source_priority", lambda primary: ["spotify"]) + monkeypatch.setattr(lr, "get_album_for_source", lambda s, aid: ALBUM_META.get(aid)) + monkeypatch.setattr(lr, "get_album_tracks_for_source", lambda s, aid: TRACKLISTS.get(aid)) + # Resolver-internal fetchers (imported by name inside _resolve_better_edition). + norm = {"sp_deluxe": DELUXE_NORM, "sp_single": SINGLE_NORM} + monkeypatch.setattr(cr, "default_fetch_tracklist", lambda s, aid: norm.get(aid)) + monkeypatch.setattr(cr, "default_fetch_alternates", alternates) + + +def test_misfit_single_resolves_to_the_single_edition(monkeypatch): + alt_calls = [] + + def alternates(source, aid, **kw): + alt_calls.append((source, aid)) + return [ + {"album_id": "sp_single", "tracks": SINGLE_NORM}, + {"album_id": "sp_deluxe", "tracks": DELUXE_NORM}, + ] + + _wire(monkeypatch, alternates=alternates) + pins = [] + album_data = { + "spotify_album_id": "sp_deluxe", "title": "Scatterbrain", + "artist_id": "a1", "artist_name": "The Band", + } + file_tracks = [{"duration_ms": 129_000, "title": "Scatterbrain"}] # owns the single + + source, api_album, items = lr._resolve_source( + album_data, "spotify", + file_tracks=file_tracks, + on_better_edition=lambda s, aid, sc: pins.append((s, aid, sc)), + ) + + assert source == "spotify" + assert api_album == ALBUM_META["sp_single"] # used the single, not the deluxe + assert len(items) == 1 + assert alt_calls, "misfit must trigger an alternate-edition fetch" + assert pins and pins[0][1] == "sp_single", "apply must persist the better pin" + + +def test_well_fitting_album_keeps_walk_and_never_expands(monkeypatch): + alt_calls = [] + + def alternates(source, aid, **kw): + alt_calls.append((source, aid)) + return [{"album_id": "sp_single", "tracks": SINGLE_NORM}] + + _wire(monkeypatch, alternates=alternates) + pins = [] + # The library actually IS the deluxe (10 matching tracks) -> walk fits -> no expand. + album_data = { + "spotify_album_id": "sp_deluxe", "title": "Scatterbrain (Deluxe)", + "artist_id": "a1", "artist_name": "The Band", + } + file_tracks = [{"duration_ms": t["duration_ms"], "title": t["name"]} for t in DELUXE_RAW] + + source, api_album, items = lr._resolve_source( + album_data, "spotify", + file_tracks=file_tracks, + on_better_edition=lambda s, aid, sc: pins.append((s, aid, sc)), + ) + + assert source == "spotify" and api_album == ALBUM_META["sp_deluxe"] + assert alt_calls == [], "a well-fitting edition must not trigger any alternate fetch" + assert pins == [], "no pin written when the walk already fits" + + +def test_strict_source_never_expands(monkeypatch): + # User explicitly picked the source in the modal -> their choice wins, even on + # a misfit. No alternate search. + alt_calls = [] + + def alternates(source, aid, **kw): + alt_calls.append((source, aid)) + return [{"album_id": "sp_single", "tracks": SINGLE_NORM}] + + _wire(monkeypatch, alternates=alternates) + album_data = {"spotify_album_id": "sp_deluxe", "title": "Scatterbrain"} + file_tracks = [{"duration_ms": 129_000, "title": "Scatterbrain"}] + + source, api_album, items = lr._resolve_source( + album_data, "spotify", strict_source=True, file_tracks=file_tracks, + ) + assert source == "spotify" and api_album == ALBUM_META["sp_deluxe"] + assert alt_calls == [], "strict_source must not trigger alternate expansion"