From b95a8f539edcea930f6cc0b354f05f825d16aa32 Mon Sep 17 00:00:00 2001 From: BoulderBadgeDad Date: Mon, 22 Jun 2026 13:32:24 -0700 Subject: [PATCH] Auto-download: resolve an album track's real position from its album (not 1/1) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Tracks auto-downloaded from the playlist pipeline / wishlist / watchlist landed as 01/1 even though they belong to multi-track albums (wolf/Sokhi; verified live — Deezer says 'Obelisk' is track 9 of The Grand Mirage, Olives is 3/4, etc.). Root cause, located in code: discovery doesn't carry a per-track position for sources whose search/track endpoint omits it (Deezer search, MusicBrainz recordings — only their ALBUM endpoint has it). detect_album_info_web then set 'track_number': track_info.get('track_number') (= None) and never looked it up from the album it HAD identified (context.py); the pipeline floored it to 1. The one helper that does an album lookup only ran for the no-album-context branch and is gated off by default. Not isolated to Deezer — the gap is source-agnostic. Fix: when the album is known (album_id present) but the position is missing, resolve the REAL (track_number, disc_number) from the album's own track list via the source-agnostic get_album_tracks_for_source — using the album id discovery already picked (no re-search, no edition guessing). Matches by ISRC -> source track id -> title. Fail-safe: any miss/error leaves the number untouched, so it still falls through to the filename exactly as before — never worse than today. kettui: pure seam core/imports/album_position.resolve_track_position_in_album (I/O-free, ISRC>id>title priority, skips position-less entries) + a fail-safe integration wrapper, both covered — 11 tests incl. the 'Obelisk = 9/12' case, priority resolution, and never-raises-on-fetch-error. 788 import/context/pipeline tests green, ruff clean. --- core/imports/album_position.py | 94 +++++++++++++++++++++ core/imports/context.py | 62 ++++++++++++-- tests/imports/test_album_position.py | 119 +++++++++++++++++++++++++++ 3 files changed, 269 insertions(+), 6 deletions(-) create mode 100644 core/imports/album_position.py create mode 100644 tests/imports/test_album_position.py diff --git a/core/imports/album_position.py b/core/imports/album_position.py new file mode 100644 index 00000000..7da1b14a --- /dev/null +++ b/core/imports/album_position.py @@ -0,0 +1,94 @@ +"""Resolve a track's position WITHIN its album's track list. + +The bug this fixes: a track auto-downloaded from the playlist pipeline / wishlist / +watchlist is identified as belonging to an album, but the per-track position is +unknown — Deezer's search/track and MusicBrainz's recording lookups don't carry a +track position (only their album endpoint does). ``detect_album_info_web`` then +leaves ``track_number = None``, the import pipeline falls through to the default-1 +floor, and the file lands as ``01/1`` even though the album is known +(``core/imports/context.py``). Verified live: e.g. Deezer says "Obelisk" is track +9 of *The Grand Mirage*, but it was tagged 1/1. + +This is the pure matcher: given the album's track list (fetched by the caller via +``core.metadata.album_tracks.get_album_tracks_for_source`` — so this stays +source-agnostic and I/O-free) plus the track's own identifiers, return its real +``(track_number, disc_number)``. Match priority is by reliability: + +1. **ISRC** — an exact recording identity; trusted immediately. +2. **source track id** — exact within this album. +3. **normalized title** — last resort. + +Returns ``(None, None)`` on no confident match, so the caller keeps its existing +behaviour (never worse than today). +""" + +from __future__ import annotations + +import re +from typing import Any, List, Optional, Tuple + + +def _norm_title(value: Any) -> str: + """Lower, strip punctuation, collapse whitespace — for tolerant title match.""" + s = re.sub(r"[^\w\s]", "", str(value or "").lower()) + return re.sub(r"\s+", " ", s).strip() + + +def _pos_int(value: Any) -> Optional[int]: + try: + n = int(value) + except (TypeError, ValueError): + return None + return n if n >= 1 else None + + +def resolve_track_position_in_album( + album_tracks: List[dict], + *, + title: str = "", + track_id: str = "", + isrc: str = "", +) -> Tuple[Optional[int], Optional[int]]: + """Return ``(track_number, disc_number)`` for this track within ``album_tracks``, + or ``(None, None)`` when no confident match is found. + + ``album_tracks`` is the list under ``get_album_tracks_for_source(...)['tracks']`` + — each entry has ``track_number`` / ``disc_number`` / ``id`` / ``name`` / ``isrc``. + Entries without a valid positive ``track_number`` are skipped. Pure: no I/O. + """ + if not album_tracks: + return (None, None) + + want_isrc = str(isrc or "").strip().upper() + want_id = str(track_id or "").strip() + want_title = _norm_title(title) + + by_id: Optional[Tuple[int, int]] = None + by_title: Optional[Tuple[int, int]] = None + + for t in album_tracks: + if not isinstance(t, dict): + continue + tn = _pos_int(t.get("track_number")) + if tn is None: + continue + dn = _pos_int(t.get("disc_number")) or 1 + + # 1) ISRC — exact recording. Win immediately. + if want_isrc and str(t.get("isrc") or "").strip().upper() == want_isrc: + return (tn, dn) + # 2) source track id — exact within the album. + if by_id is None and want_id and str(t.get("id") or "").strip() == want_id: + by_id = (tn, dn) + # 3) normalized title — last resort. + if by_title is None and want_title and _norm_title(t.get("name")) == want_title: + by_title = (tn, dn) + + if by_id is not None: + return by_id + if by_title is not None: + return by_title + return (None, None) + + +__all__ = ["resolve_track_position_in_album"] diff --git a/core/imports/context.py b/core/imports/context.py index 9ed2a909..4e15c918 100644 --- a/core/imports/context.py +++ b/core/imports/context.py @@ -433,16 +433,24 @@ def detect_album_info_web(context, artist_context=None): track_name.strip().lower(), artist_name.strip().lower(), }: + _tn = track_info.get("track_number") + _dn = track_info.get("disc_number") + # The album is identified but discovery often doesn't carry the per-track + # POSITION — Deezer's search/track and MusicBrainz's recording lookups omit + # it (only their album endpoint has it). Without a position the pipeline + # falls through to the default-1 floor and files an album track as 01/1 + # (e.g. Deezer says "Obelisk" is track 9 of The Grand Mirage). Resolve the + # REAL position from the album's own track list when we have its id. + # Fail-safe: leaves the numbers untouched on any miss, so behaviour is + # never worse than the old preserve-None-and-fall-through. + if _tn is None: + _tn, _dn = _resolve_album_position_from_source(context, artist_context, _dn) return build_import_album_info( context, album_info={ "album_name": album_name, - # Preserve missing numbers as None so the import pipeline - # can fall through to ``extract_track_number_from_filename`` - # at ``core/imports/pipeline.py:652`` instead of locking - # to track/disc 01 for every wishlist re-attempt. - "track_number": track_info.get("track_number"), - "disc_number": track_info.get("disc_number"), + "track_number": _tn, + "disc_number": _dn, "album_image_url": album_ctx.get("image_url", ""), "confidence": 0.5, }, @@ -454,6 +462,48 @@ def detect_album_info_web(context, artist_context=None): return _resolve_single_to_parent_album(context, artist_context) +def _resolve_album_position_from_source(context, artist_context, current_disc): + """Look up a track's real ``(track_number, disc_number)`` from its album's track + list, for the case where the album is known but discovery didn't carry a + position (Deezer/MusicBrainz search omit it). + + Uses the SAME album id discovery already resolved (``get_import_source_ids`` → + ``album_id``), so it re-homes the track onto its own album with no re-search and + no edition guessing. Matches by ISRC → source track id → title via the pure + ``core.imports.album_position`` seam. Returns ``(None, current_disc)`` on any + miss/error so the caller falls back exactly as before — never worse than today. + """ + try: + source = get_import_source(context) + ids = get_import_source_ids(context) + album_id = str(ids.get("album_id") or "") + if not source or not album_id: + return None, current_disc + + from core.metadata.album_tracks import get_album_tracks_for_source + payload = get_album_tracks_for_source(source, album_id) or {} + tracks = payload.get("tracks") or [] + if not tracks: + return None, current_disc + + track_info = get_import_track_info(context) + original_search = get_import_original_search(context) + title = (track_info.get("name") or original_search.get("title") or "").strip() + isrc = str(track_info.get("isrc") or original_search.get("isrc") or "") + + from core.imports.album_position import resolve_track_position_in_album + tn, dn = resolve_track_position_in_album( + tracks, title=title, track_id=str(ids.get("track_id") or ""), isrc=isrc) + if tn is not None: + logger.info("album-position: resolved '%s' to track %s/disc %s from album %s (%s)", + title, tn, dn, album_id, source) + return tn, (dn if dn is not None else current_disc) + return None, current_disc + except Exception as e: + logger.debug("album-position resolution failed: %s", e) + return None, current_disc + + def _resolve_single_to_parent_album(context, artist_context): """A single-matched track -> a promoted album_info for its parent album, or None. GATED by ``metadata_enhancement.single_to_album`` (default OFF — it's a diff --git a/tests/imports/test_album_position.py b/tests/imports/test_album_position.py new file mode 100644 index 00000000..d22b7ab7 --- /dev/null +++ b/tests/imports/test_album_position.py @@ -0,0 +1,119 @@ +"""A track auto-downloaded from the playlist pipeline / wishlist / watchlist is +identified as belonging to an album, but Deezer's search/track and MusicBrainz's +recording lookups don't carry a track POSITION — so detect_album_info_web left +track_number=None, the pipeline floored it to 1, and album tracks landed as 01/1 +(verified live: Deezer says "Obelisk" is track 9 of The Grand Mirage, tagged 1/1). + +The fix resolves the real position from the album's OWN track list. These pin the +pure matcher and the (fail-safe) integration wrapper. +""" + +from __future__ import annotations + +import pytest + +from core.imports.album_position import resolve_track_position_in_album + + +def _album_12(): + # a realistic 12-track album payload shape (get_album_tracks_for_source -> 'tracks') + return [ + {"id": f"t{i}", "name": n, "track_number": i, "disc_number": 1, + "isrc": f"ISRC{i:03d}"} + for i, n in enumerate( + ["Intro", "Drift", "Mirage", "Haze", "Pulse", "Glow", "Echo", "Tide", + "Obelisk", "Comet", "Dawn", "Outro"], start=1) + ] + + +# ── pure matcher ───────────────────────────────────────────────────────────── + +def test_resolves_position_by_title(): + tn, dn = resolve_track_position_in_album(_album_12(), title="Obelisk") + assert (tn, dn) == (9, 1) + + +def test_title_match_is_case_and_punctuation_insensitive(): + tracks = [{"id": "t1", "name": "Lueur Déclinante!!!", "track_number": 3, "disc_number": 1}] + tn, _ = resolve_track_position_in_album(tracks, title="lueur déclinante") + assert tn == 3 + + +def test_resolves_by_isrc_exactly(): + tn, dn = resolve_track_position_in_album(_album_12(), isrc="isrc009") # case-insensitive + assert (tn, dn) == (9, 1) + + +def test_resolves_by_track_id(): + tn, _ = resolve_track_position_in_album(_album_12(), track_id="t9") + assert tn == 9 + + +def test_isrc_beats_id_beats_title_on_conflict(): + # craft a list where ISRC, id, and title each point at a DIFFERENT track + tracks = [ + {"id": "byid", "name": "other", "track_number": 2, "disc_number": 1, "isrc": "X"}, + {"id": "z", "name": "WANT", "track_number": 3, "disc_number": 1, "isrc": "Y"}, + {"id": "z2", "name": "other2", "track_number": 4, "disc_number": 1, "isrc": "WANTISRC"}, + ] + # all three signals provided -> ISRC wins (track 4) + tn, _ = resolve_track_position_in_album(tracks, title="WANT", track_id="byid", isrc="wantisrc") + assert tn == 4 + # no ISRC -> id wins (track 2) + tn, _ = resolve_track_position_in_album(tracks, title="WANT", track_id="byid") + assert tn == 2 + # only title -> title wins (track 3) + tn, _ = resolve_track_position_in_album(tracks, title="want") + assert tn == 3 + + +def test_carries_disc_number(): + tracks = [{"id": "t1", "name": "B-Side", "track_number": 2, "disc_number": 2}] + assert resolve_track_position_in_album(tracks, title="B-Side") == (2, 2) + + +def test_no_match_returns_none(): + assert resolve_track_position_in_album(_album_12(), title="Not On This Album") == (None, None) + assert resolve_track_position_in_album([], title="x") == (None, None) + assert resolve_track_position_in_album(None, title="x") == (None, None) + + +def test_skips_entries_without_a_valid_position(): + tracks = [ + {"id": "t1", "name": "Obelisk", "track_number": 0}, # 0 -> skip + {"id": "t2", "name": "Obelisk", "track_number": None}, # None -> skip + {"id": "t3", "name": "Obelisk", "track_number": "junk"}, # junk -> skip + ] + assert resolve_track_position_in_album(tracks, title="Obelisk") == (None, None) + + +# ── integration wrapper (fail-safe, real album lookup) ─────────────────────── + +def test_wrapper_resolves_from_source_album(monkeypatch): + import core.imports.context as ctx + monkeypatch.setattr("core.metadata.album_tracks.get_album_tracks_for_source", + lambda source, album_id: {"tracks": _album_12()}) + context = {"source": "deezer", + "track_info": {"id": "t9", "name": "Obelisk", "deezer_album_id": "232620572"}} + tn, dn = ctx._resolve_album_position_from_source(context, {}, 1) + assert tn == 9 and dn == 1 + + +def test_wrapper_is_failsafe_on_empty_or_missing(monkeypatch): + import core.imports.context as ctx + # no album id at all -> keeps current disc, no number + assert ctx._resolve_album_position_from_source({"source": "deezer", "track_info": {}}, {}, 1) == (None, 1) + # fetcher returns nothing -> fail-safe + monkeypatch.setattr("core.metadata.album_tracks.get_album_tracks_for_source", + lambda source, album_id: {"tracks": []}) + context = {"source": "deezer", "track_info": {"name": "Obelisk", "deezer_album_id": "x"}} + assert ctx._resolve_album_position_from_source(context, {}, 2) == (None, 2) + + +def test_wrapper_never_raises_on_fetch_error(monkeypatch): + import core.imports.context as ctx + def _boom(source, album_id): + raise RuntimeError("deezer down") + monkeypatch.setattr("core.metadata.album_tracks.get_album_tracks_for_source", _boom) + context = {"source": "deezer", "track_info": {"name": "Obelisk", "deezer_album_id": "x"}} + assert ctx._resolve_album_position_from_source(context, {}, 1) == (None, 1)