#865: resolve pasted SoundCloud links (incl. unlisted/private share URLs) via direct yt-dlp resolve; manual-search forces the SoundCloud source
This commit is contained in:
parent
ba5d62946a
commit
c62074d54a
4 changed files with 249 additions and 2 deletions
|
|
@ -29,6 +29,7 @@ import uuid
|
|||
import time
|
||||
from typing import List, Optional, Dict, Any, Tuple, Callable
|
||||
from pathlib import Path
|
||||
from urllib.parse import urlparse
|
||||
|
||||
try:
|
||||
import yt_dlp
|
||||
|
|
@ -45,6 +46,33 @@ from core.download_plugins.types import TrackResult, AlbumResult, DownloadStatus
|
|||
logger = get_logger("soundcloud_client")
|
||||
|
||||
|
||||
def is_soundcloud_url(value: Any) -> bool:
|
||||
"""True when ``value`` is a SoundCloud URL (track, set/playlist, or a
|
||||
public-but-unlisted ``/s-<token>`` share link), incl. ``on.soundcloud.com``
|
||||
short links and scheme-less ``soundcloud.com/...`` forms.
|
||||
|
||||
Pure + network-free. Used so a pasted SoundCloud link is RESOLVED directly
|
||||
(it carries its own access token) instead of being run through scsearch,
|
||||
which can't find unlisted/private tracks (#865). A normal text query — even
|
||||
one mentioning "soundcloud" — is not a URL and returns False."""
|
||||
if not value or not isinstance(value, str):
|
||||
return False
|
||||
raw = value.strip()
|
||||
if not raw:
|
||||
return False
|
||||
candidate = raw if '://' in raw else f'https://{raw}'
|
||||
try:
|
||||
host = (urlparse(candidate).netloc or '').lower()
|
||||
except Exception:
|
||||
return False
|
||||
host = host.split('@')[-1].split(':')[0] # strip any userinfo / port
|
||||
return (
|
||||
host == 'soundcloud.com'
|
||||
or host.endswith('.soundcloud.com')
|
||||
or host.endswith('soundcloud.app.goo.gl')
|
||||
)
|
||||
|
||||
|
||||
# Quality tiers — SoundCloud anonymous access only really delivers one
|
||||
# quality, but we keep the structure consistent with other clients so
|
||||
# UI/settings can reference a familiar shape later.
|
||||
|
|
@ -173,6 +201,12 @@ class SoundcloudClient(DownloadSourcePlugin):
|
|||
logger.warning(f"Invalid SoundCloud search query: {query!r}")
|
||||
return ([], [])
|
||||
|
||||
# A pasted SoundCloud link resolves directly — scsearch can't find an
|
||||
# unlisted/private track, but its share URL carries an access token that
|
||||
# yt-dlp can resolve (#865).
|
||||
if is_soundcloud_url(query):
|
||||
return await self.resolve_url(query, timeout=timeout, progress_callback=progress_callback)
|
||||
|
||||
# SoundCloud or a transient yt-dlp parse can fail; the caller still
|
||||
# gets an empty list, never a raised exception.
|
||||
limit = min(MAX_SEARCH_LIMIT, max(1, DEFAULT_SEARCH_LIMIT))
|
||||
|
|
@ -225,6 +259,70 @@ class SoundcloudClient(DownloadSourcePlugin):
|
|||
entries = info.get('entries') or []
|
||||
return [e for e in entries if isinstance(e, dict)]
|
||||
|
||||
async def resolve_url(
|
||||
self,
|
||||
url: str,
|
||||
timeout: Optional[int] = None,
|
||||
progress_callback: Optional[Callable] = None,
|
||||
) -> Tuple[List[TrackResult], List[AlbumResult]]:
|
||||
"""Resolve a direct SoundCloud track/set URL into downloadable
|
||||
TrackResult(s) — including public-but-unlisted/private share links that
|
||||
scsearch can't find, because the URL itself carries the access token
|
||||
(#865). A set/playlist link resolves to all its tracks. Returns
|
||||
``(tracks, [])`` and never raises (empty on any failure)."""
|
||||
if not self.is_available():
|
||||
logger.warning("SoundCloud not available for URL resolve (yt-dlp missing)")
|
||||
return ([], [])
|
||||
url = (url or '').strip()
|
||||
if not url:
|
||||
return ([], [])
|
||||
|
||||
logger.info(f"Resolving SoundCloud URL: {url}")
|
||||
loop = asyncio.get_event_loop()
|
||||
try:
|
||||
info = await loop.run_in_executor(None, self._extract_url_info, url)
|
||||
except Exception as exc:
|
||||
logger.error(f"SoundCloud URL resolve failed: {exc}")
|
||||
return ([], [])
|
||||
if not isinstance(info, dict):
|
||||
return ([], [])
|
||||
|
||||
# A set/playlist link yields `entries`; a single track is the dict itself.
|
||||
entries = info.get('entries')
|
||||
raw_entries = [e for e in entries if isinstance(e, dict)] if entries else [info]
|
||||
|
||||
track_results: List[TrackResult] = []
|
||||
for entry in raw_entries:
|
||||
# Full (non-flat) extraction puts a transient media-stream URL in
|
||||
# `url`; the stable permalink lives in `webpage_url`. Pin `url` to the
|
||||
# permalink so the download encoding matches what search() produces
|
||||
# (the worker re-resolves the permalink at download time).
|
||||
permalink = entry.get('webpage_url') or entry.get('url')
|
||||
if permalink:
|
||||
entry = {**entry, 'url': permalink}
|
||||
try:
|
||||
converted = self._sc_to_track_result(entry)
|
||||
if converted is not None:
|
||||
track_results.append(converted)
|
||||
except Exception as exc:
|
||||
logger.debug(f"Skipping SoundCloud URL entry conversion error: {exc}")
|
||||
|
||||
logger.info(f"Resolved {len(track_results)} SoundCloud track(s) from URL")
|
||||
return (track_results, [])
|
||||
|
||||
def _extract_url_info(self, url: str) -> Optional[Dict[str, Any]]:
|
||||
"""Full (non-flat) yt-dlp extraction of a direct SoundCloud URL — needed
|
||||
to get the real id/permalink/title for a single track or set. Anonymous;
|
||||
the share token (if any) is already in the URL. Injectable for tests."""
|
||||
opts = {
|
||||
'quiet': True,
|
||||
'no_warnings': True,
|
||||
'skip_download': True,
|
||||
'noplaylist': False,
|
||||
}
|
||||
with yt_dlp.YoutubeDL(opts) as ydl:
|
||||
return ydl.extract_info(url, download=False)
|
||||
|
||||
def _sc_to_track_result(self, entry: Dict[str, Any]) -> Optional[TrackResult]:
|
||||
"""Convert a yt-dlp SoundCloud entry into the standard TrackResult.
|
||||
|
||||
|
|
|
|||
|
|
@ -130,6 +130,10 @@ def manual_search_client(monkeypatch):
|
|||
'qobuz': _make_plugin(),
|
||||
'hifi': _make_plugin(),
|
||||
'deezer': _make_plugin(),
|
||||
# Present in the registry but deliberately NOT in the default
|
||||
# hybrid_order, so the #865 SoundCloud-link test can select it
|
||||
# without changing the 'all' fan-out tests.
|
||||
'soundcloud': _make_plugin(),
|
||||
}
|
||||
|
||||
class _FakeSpec:
|
||||
|
|
@ -217,6 +221,46 @@ def manual_search_client(monkeypatch):
|
|||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_manual_search_soundcloud_link_forces_soundcloud_source(manual_search_client):
|
||||
"""#865: pasting a SoundCloud URL forces the SoundCloud source and passes the
|
||||
URL straight through (so its search resolves the link, incl. unlisted/private)
|
||||
— it must NOT be turned into a text query or fan out to other sources."""
|
||||
client, ctx = manual_search_client
|
||||
# Hybrid order = soundcloud only, so SoundCloud is the lone available source.
|
||||
from config.settings import config_manager
|
||||
original = config_manager.get
|
||||
|
||||
def _cfg(key, default=None):
|
||||
if key == 'download_source.mode':
|
||||
return 'hybrid'
|
||||
if key == 'download_source.hybrid_order':
|
||||
return ['soundcloud']
|
||||
return original(key, default)
|
||||
ctx['config_get_setter'](_cfg)
|
||||
|
||||
url = 'https://soundcloud.com/artist/secret-track/s-AbC123'
|
||||
resp = client.post('/api/downloads/task/task-abc/manual-search',
|
||||
json={'query': url, 'source': 'all'})
|
||||
assert resp.status_code == 200
|
||||
msgs = _consume_ndjson(resp)
|
||||
header = next(m for m in msgs if m.get('type') == 'header')
|
||||
# The link forced the SoundCloud source, with the raw URL kept as the query.
|
||||
assert header['sources_queried'] == ['soundcloud']
|
||||
assert header['query'] == url
|
||||
|
||||
|
||||
def test_manual_search_soundcloud_link_errors_when_not_connected(manual_search_client):
|
||||
"""A SoundCloud link with SoundCloud not configured → clear 400, not a
|
||||
useless text search of the raw URL."""
|
||||
client, ctx = manual_search_client
|
||||
# Default hybrid_order has no soundcloud → it's not an available source.
|
||||
resp = client.post('/api/downloads/task/task-abc/manual-search',
|
||||
json={'query': 'https://soundcloud.com/artist/track', 'source': 'all'})
|
||||
assert resp.status_code == 400
|
||||
assert 'soundcloud' in resp.get_data(as_text=True).lower()
|
||||
ctx['plugins']['soundcloud'].search.assert_not_called()
|
||||
|
||||
|
||||
def test_manual_search_validates_query_length(manual_search_client):
|
||||
"""Empty / 1-char query returns 400 — frontend hint says ≥2 chars."""
|
||||
client, _ctx = manual_search_client
|
||||
|
|
|
|||
|
|
@ -248,6 +248,99 @@ def test_search_handles_empty_entries(tmp_dl: Path) -> None:
|
|||
assert tracks == []
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# URL resolution (#865): paste a SoundCloud link, incl. unlisted/private
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_is_soundcloud_url_accepts_real_links() -> None:
|
||||
ok = [
|
||||
'https://soundcloud.com/artist/track-name',
|
||||
'http://soundcloud.com/artist/sets/my-set',
|
||||
'https://soundcloud.com/artist/track/s-AbC123', # private share token
|
||||
'soundcloud.com/artist/track-name', # scheme-less
|
||||
'https://m.soundcloud.com/artist/track',
|
||||
'https://on.soundcloud.com/aBcDe', # short link
|
||||
]
|
||||
for u in ok:
|
||||
assert soundcloud_client.is_soundcloud_url(u) is True, u
|
||||
|
||||
|
||||
def test_is_soundcloud_url_rejects_non_links() -> None:
|
||||
no = [
|
||||
'daft punk around the world',
|
||||
'best soundcloud tracks 2024', # mentions soundcloud, not a URL
|
||||
'https://youtube.com/watch?v=abc',
|
||||
'https://open.spotify.com/track/xyz',
|
||||
'', None, 42,
|
||||
]
|
||||
for u in no:
|
||||
assert soundcloud_client.is_soundcloud_url(u) is False, repr(u)
|
||||
|
||||
|
||||
def test_search_routes_a_url_to_resolve(tmp_dl: Path) -> None:
|
||||
"""A pasted SoundCloud URL must go through resolve_url, NOT scsearch."""
|
||||
client = SoundcloudClient(download_path=str(tmp_dl))
|
||||
sentinel = ([TrackResult(username='soundcloud', filename='1||u||n', size=0,
|
||||
bitrate=128, duration=None, quality='mp3',
|
||||
free_upload_slots=999, upload_speed=1, queue_length=0)], [])
|
||||
# resolve_url is async → patch.object gives an AsyncMock, so return_value is
|
||||
# what `await` yields.
|
||||
with patch.object(client, 'resolve_url', return_value=sentinel) as rv, \
|
||||
patch.object(client, '_extract_search_entries') as scsearch:
|
||||
tracks, albums = _run(client.search('https://soundcloud.com/x/private/s-tok'))
|
||||
rv.assert_called_once()
|
||||
scsearch.assert_not_called()
|
||||
assert len(tracks) == 1
|
||||
|
||||
|
||||
def test_resolve_url_single_track_uses_permalink(tmp_dl: Path) -> None:
|
||||
"""Full extraction puts a transient stream URL in `url`; the stable permalink
|
||||
is `webpage_url`. The filename must encode the permalink."""
|
||||
info = {
|
||||
'id': '999',
|
||||
'title': 'Artist Name - Secret Song',
|
||||
'uploader': 'artistname',
|
||||
'url': 'https://cf-media.sndcdn.com/transient-stream.mp3', # expires
|
||||
'webpage_url': 'https://soundcloud.com/artistname/secret-song/s-Tok3n',
|
||||
'duration': 200.0,
|
||||
}
|
||||
client = SoundcloudClient(download_path=str(tmp_dl))
|
||||
with patch.object(client, '_extract_url_info', return_value=info):
|
||||
tracks, albums = _run(client.resolve_url('https://soundcloud.com/artistname/secret-song/s-Tok3n'))
|
||||
assert albums == []
|
||||
assert len(tracks) == 1
|
||||
parts = tracks[0].filename.split('||')
|
||||
assert parts[0] == '999'
|
||||
assert parts[1] == 'https://soundcloud.com/artistname/secret-song/s-Tok3n' # permalink, not stream
|
||||
assert tracks[0]._source_metadata['permalink_url'] == 'https://soundcloud.com/artistname/secret-song/s-Tok3n'
|
||||
assert tracks[0].artist == 'Artist Name'
|
||||
assert tracks[0].duration == 200000
|
||||
|
||||
|
||||
def test_resolve_url_set_yields_all_tracks(tmp_dl: Path) -> None:
|
||||
info = {
|
||||
'entries': [
|
||||
{'id': '1', 'title': 'A - One', 'webpage_url': 'https://soundcloud.com/a/one', 'duration': 100},
|
||||
{'id': '2', 'title': 'A - Two', 'webpage_url': 'https://soundcloud.com/a/two', 'duration': 120},
|
||||
],
|
||||
}
|
||||
client = SoundcloudClient(download_path=str(tmp_dl))
|
||||
with patch.object(client, '_extract_url_info', return_value=info):
|
||||
tracks, _ = _run(client.resolve_url('https://soundcloud.com/a/sets/my-set'))
|
||||
assert {t._source_metadata['track_id'] for t in tracks} == {'1', '2'}
|
||||
assert all('soundcloud.com' in t.filename.split('||')[1] for t in tracks)
|
||||
|
||||
|
||||
def test_resolve_url_empty_on_failure(tmp_dl: Path) -> None:
|
||||
client = SoundcloudClient(download_path=str(tmp_dl))
|
||||
with patch.object(client, '_extract_url_info', side_effect=RuntimeError('blocked')):
|
||||
assert _run(client.resolve_url('https://soundcloud.com/x/y')) == ([], [])
|
||||
with patch.object(client, '_extract_url_info', return_value=None):
|
||||
assert _run(client.resolve_url('https://soundcloud.com/x/y')) == ([], [])
|
||||
assert _run(client.resolve_url('')) == ([], [])
|
||||
|
||||
|
||||
def test_search_handles_malformed_entries_individually(tmp_dl: Path) -> None:
|
||||
"""One bad entry shouldn't poison the entire result set."""
|
||||
fake_entries = [
|
||||
|
|
|
|||
|
|
@ -7367,10 +7367,22 @@ def manual_search_for_task(task_id):
|
|||
# source isn't connected or the link can't be resolved — so the user is
|
||||
# never worse off than typing the query themselves.
|
||||
from core.downloads.track_link import parse_download_track_link
|
||||
from core.soundcloud_client import is_soundcloud_url
|
||||
link = parse_download_track_link(query)
|
||||
link_source = None
|
||||
link_track_id = None
|
||||
if link:
|
||||
# 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
|
||||
# client resolves the link directly (#865).
|
||||
if is_soundcloud_url(query):
|
||||
if 'soundcloud' not in valid_source_ids:
|
||||
return jsonify({
|
||||
"error": "SoundCloud isn't connected — enable it in Settings to "
|
||||
"resolve a SoundCloud link."
|
||||
}), 400
|
||||
source = 'soundcloud'
|
||||
elif link:
|
||||
_src, _tid = link
|
||||
# A parsed link is unambiguously a Tidal/Qobuz track URL, never a
|
||||
# name a user would type — so if we can't use it, say why clearly
|
||||
|
|
@ -14744,7 +14756,7 @@ def _youtube_cookie_opts():
|
|||
cb = config_manager.get('youtube.cookies_browser', '')
|
||||
if cb:
|
||||
opts['cookiesfrombrowser'] = (cb,)
|
||||
except Exception:
|
||||
except Exception: # noqa: S110 - cookie config is best-effort; resolve still works without it
|
||||
pass
|
||||
return opts
|
||||
|
||||
|
|
|
|||
Loading…
Reference in a new issue