Auto-download: resolve an album track's real position from its album (not 1/1)

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

@@ -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"]

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

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)