#767-2: reorganize finds the right album edition instead of mislabeling singles as deluxe
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.
This commit is contained in:
BoulderBadgeDad
parent
2921e80d58
commit
69fc21d6b2
5 changed files with 628 additions and 27 deletions
|
|
@ -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'
|
||||
|
|
|
|||
|
|
@ -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",
|
||||
]
|
||||
|
|
|
|||
106
tests/test_canonical_alternates.py
Normal file
106
tests/test_canonical_alternates.py
Normal file
|
|
@ -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
|
||||
|
|
@ -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
|
||||
|
|
|
|||
119
tests/test_reorganize_alternate_edition.py
Normal file
119
tests/test_reorganize_alternate_edition.py
Normal file
|
|
@ -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"
|
||||
Loading…
Reference in a new issue