diff --git a/core/downloads/track_link.py b/core/downloads/track_link.py index 3565a5ff..039ddcbf 100644 --- a/core/downloads/track_link.py +++ b/core/downloads/track_link.py @@ -38,6 +38,23 @@ def bubble_linked_track_first(tracks: List[Any], link_track_id: str) -> List[Any target = str(link_track_id) return sorted(tracks, key=lambda t: linked_track_id(t) != target) + +def inject_linked_track_first( + tracks: List[Any], linked_result: Any, link_track_id: str +) -> List[Any]: + """Put the EXACT linked track first. + + When ``linked_result`` is the track fetched directly by id, prepend it and + drop any search duplicate of it — so an obscure track a text search never + surfaced is still present and downloadable (#932). When it's None (the source + can't fetch one), fall back to bubbling a matching search result. Pure.""" + if not link_track_id: + return tracks + target = str(link_track_id) + if linked_result is not None: + return [linked_result] + [t for t in tracks if linked_track_id(t) != target] + return bubble_linked_track_first(tracks, target) + # host substring → download source id. Only ID-downloadable streaming sources. _HOSTS = ( ('tidal.com', 'tidal'), diff --git a/core/qobuz_client.py b/core/qobuz_client.py index e90e934a..f82e6da4 100644 --- a/core/qobuz_client.py +++ b/core/qobuz_client.py @@ -715,6 +715,26 @@ class QobuzClient(DownloadSourcePlugin): logger.error(f"Error getting Qobuz track {track_id}: {e}") return None + def get_track_result(self, track_id): + """Fetch ONE track by ID and convert it to a downloadable TrackResult. + + Used when a pasted Qobuz link must inject the EXACT track: a text search + for an obscure track's name often doesn't surface it at all, so bubbling + the matching id is a no-op (#932). Returns None on any failure so the + caller falls back to the text-search path.""" + try: + track = self.get_track(track_id) + if not track: + return None + quality_key = quality_tier_for_source('qobuz', default='lossless') + quality_info = QOBUZ_QUALITY_MAP.get(quality_key, QOBUZ_QUALITY_MAP['lossless']) + # require_streamable=False: this is the exact track the user linked — don't + # let a missing/absent 'streamable' flag in track/get drop it (#932). + return self._qobuz_to_track_result(track, quality_info, require_streamable=False) + except Exception as e: + logger.debug(f"get_track_result failed for Qobuz {track_id}: {e}") + return None + # ===================== Playlists & Favorites ===================== # # Qobuz playlist sync surface — mirrors the Tidal client contract @@ -1012,14 +1032,20 @@ class QobuzClient(DownloadSourcePlugin): traceback.print_exc() return ([], []) - def _qobuz_to_track_result(self, track: Dict, quality_info: dict) -> Optional[TrackResult]: - """Convert Qobuz track dict to TrackResult (Soulseek-compatible format).""" + def _qobuz_to_track_result(self, track: Dict, quality_info: dict, + require_streamable: bool = True) -> Optional[TrackResult]: + """Convert Qobuz track dict to TrackResult (Soulseek-compatible format). + + ``require_streamable`` gates bulk SEARCH results on the ``streamable`` flag + (filters noise). For a track the user explicitly pasted a link to, pass + False: ``track/get`` may omit the flag (which would default-False and wrongly + drop the exact track they asked for), and they should get to try it (#932).""" track_id = track.get('id') if not track_id: return None - # Check if track is streamable - if not track.get('streamable', False): + # Check if track is streamable (skipped for explicit by-id link fetches). + if require_streamable and not track.get('streamable', False): return None performer = track.get('performer', {}) diff --git a/tests/downloads/test_track_link.py b/tests/downloads/test_track_link.py index 06cbc1bb..b91e49e4 100644 --- a/tests/downloads/test_track_link.py +++ b/tests/downloads/test_track_link.py @@ -122,3 +122,112 @@ def test_bubble_noop_when_nothing_matches_or_empty(): assert bubble_linked_track_first([a, b], '296427754') == [a, b] # unchanged assert bubble_linked_track_first([], '296427754') == [] assert bubble_linked_track_first([a, b], '') == [a, b] + + +# ── inject the EXACT fetched track first (#932 reopen: text search misses obscure tracks) ── + +from core.downloads.track_link import inject_linked_track_first + + +def test_inject_puts_fetched_track_first_when_search_missed_it(): + # The reported case: the linked track isn't among the fuzzy text-search + # results at all, so the directly-fetched result is injected at the top. + fetched = _result('296427754') + search = [_result('111'), _result('222')] # unrelated lookalikes + out = inject_linked_track_first(search, fetched, '296427754') + assert out[0] is fetched + assert len(out) == 3 + + +def test_inject_dedups_a_search_copy_of_the_linked_track(): + fetched = _result('296427754') + dup = _result('296427754') # search also returned it + search = [_result('111'), dup, _result('222')] + out = inject_linked_track_first(search, fetched, '296427754') + assert out[0] is fetched + # exactly one element carries the linked id, and it's the injected one + assert [linked_track_id(t) for t in out].count('296427754') == 1 + assert all(t is not dup for t in out) # the search copy (by identity) was dropped + assert len(out) == 3 + + +def test_inject_falls_back_to_bubble_without_a_fetched_result(): + exact = _result('296427754') + out = inject_linked_track_first([_result('111'), exact, _result('222')], None, '296427754') + assert linked_track_id(out[0]) == '296427754' # bubbled, not injected + assert len(out) == 3 + + +def test_inject_is_noop_without_a_link_id(): + search = [_result('111')] + assert inject_linked_track_first(search, _result('x'), '') == search + + +def test_inject_int_track_id_is_str_safe(): + fetched = _result('296427754') + out = inject_linked_track_first([_result('111')], fetched, 296427754) + assert out[0] is fetched + + +# ── QobuzClient.get_track_result: fetch by id → downloadable TrackResult ── + +from core.qobuz_client import QobuzClient + + +def _bare_qobuz(): + return QobuzClient.__new__(QobuzClient) # no network / __init__ + + +def test_get_track_result_converts_fetched_track(monkeypatch): + client = _bare_qobuz() + sentinel = object() + monkeypatch.setattr(client, 'get_track', lambda tid: {'id': tid, 'streamable': True}) + monkeypatch.setattr(client, '_qobuz_to_track_result', + lambda track, qi, require_streamable=True: sentinel) + assert client.get_track_result('296427754') is sentinel + + +def test_get_track_result_none_when_track_unavailable(monkeypatch): + client = _bare_qobuz() + monkeypatch.setattr(client, 'get_track', lambda tid: None) + assert client.get_track_result('nope') is None + + +def test_get_track_result_none_on_exception(monkeypatch): + client = _bare_qobuz() + def _boom(_tid): + raise RuntimeError('api down') + monkeypatch.setattr(client, 'get_track', _boom) + assert client.get_track_result('x') is None + + +# ── #932 hardening: a pasted-link fetch must not be dropped by a missing +# 'streamable' flag (track/get may omit it). Exercises the REAL converter. ── + +_QOBUZ_TRACK = {'id': 296427754, 'title': 'foreign lavennew', + 'performer': {'name': 'colacola'}, 'duration': 180} + + +def test_converter_rejects_non_streamable_on_search_path(): + # Default (bulk search) still filters on streamable — unchanged behaviour. + client = _bare_qobuz() + qi = {'codec': 'flac', 'bitrate': 1411} + assert client._qobuz_to_track_result(dict(_QOBUZ_TRACK), qi) is None # no streamable flag + + +def test_converter_builds_for_link_fetch_without_streamable(): + client = _bare_qobuz() + qi = {'codec': 'flac', 'bitrate': 1411} + r = client._qobuz_to_track_result(dict(_QOBUZ_TRACK), qi, require_streamable=False) + assert r is not None + assert linked_track_id(r) == '296427754' # carries the id → injectable + downloadable + + +def test_get_track_result_survives_missing_streamable_flag(monkeypatch): + # The feared track/get shape (no 'streamable') must STILL yield the track — + # this is the end-to-end gap that couldn't be confirmed against a live API. + client = _bare_qobuz() + monkeypatch.setattr(client, 'get_track', lambda _tid: dict(_QOBUZ_TRACK)) + r = client.get_track_result('296427754') + assert r is not None + assert linked_track_id(r) == '296427754' diff --git a/web_server.py b/web_server.py index 15b5de83..4db814c2 100644 --- a/web_server.py +++ b/web_server.py @@ -7506,6 +7506,7 @@ def manual_search_for_task(task_id): link = parse_download_track_link(query) link_source = None link_track_id = None + linked_result = None # the EXACT track fetched by id, injected into results # A pasted SoundCloud link can't be turned into an "artist title" query # and searched — unlisted/private tracks aren't searchable. Instead force # the SoundCloud source and keep the URL as the query; the SoundCloud @@ -7535,6 +7536,18 @@ def manual_search_for_task(task_id): query = clean_q source = _src link_source, link_track_id = _src, _tid + # Fetch the EXACT linked track as a downloadable result to inject — + # a text search for an obscure track's name often doesn't surface it + # at all, so we can't rely on it being in the search results (#932). + # Defensive: only sources that expose get_track_result (Qobuz today); + # others fall back to the bubble path below — never worse than before. + _link_client = download_orchestrator.client(_src) if download_orchestrator else None + if _link_client is not None and hasattr(_link_client, 'get_track_result'): + try: + linked_result = _link_client.get_track_result(_tid) + except Exception as _lr_err: + logger.debug("[Manual Search] get_track_result failed for %s %s: %s", + _src, _tid, _lr_err) if source != 'all': if source not in valid_source_ids: @@ -7595,13 +7608,15 @@ def manual_search_for_task(task_id): "error": error, }) + "\n" continue - # Pasted-link exact match: bubble the track whose source id - # matches the link to the top so the user sees the exact version - # first. Reads _source_metadata['track_id'] (TrackResult has no - # top-level id) — the old getattr(t,'id') always missed (#932). - if src_name == link_source and link_track_id and tracks: - from core.downloads.track_link import bubble_linked_track_first - tracks = bubble_linked_track_first(tracks, link_track_id) + # Pasted-link exact match (#932): the linked source's search may + # not surface an obscure linked track at all, so INJECT the exact + # track (fetched by id) at the top and drop any search duplicate of + # it. If we couldn't fetch it (source has no get_track_result, or it + # failed), fall back to bubbling a matching search result — never + # worse than before. + if src_name == link_source and link_track_id: + from core.downloads.track_link import inject_linked_track_first + tracks = inject_linked_track_first(tracks, linked_result, link_track_id) serialized = [] for t in tracks: s = _serialize_candidate(t, source_override=src_name)