-# coding: utf-8
-from __future__ import unicode_literals
-
import itertools
import re
import json
)
from ..compat import (
compat_HTTPError,
- compat_kwargs,
compat_str,
)
from ..utils import (
int_or_none,
KNOWN_EXTENSIONS,
mimetype2ext,
- remove_end,
parse_qs,
str_or_none,
try_get,
class SoundcloudEmbedIE(InfoExtractor):
_VALID_URL = r'https?://(?:w|player|p)\.soundcloud\.com/player/?.*?\burl=(?P<id>.+)'
+ _EMBED_REGEX = [r'<iframe[^>]+src=(["\'])(?P<url>(?:https?://)?(?:w\.)?soundcloud\.com/player.+?)\1']
_TEST = {
# from https://www.soundi.fi/uutiset/ennakkokuuntelussa-timo-kaukolammen-station-to-station-to-station-julkaisua-juhlitaan-tanaan-g-livelabissa/
'url': 'https://w.soundcloud.com/player/?visual=true&url=https%3A%2F%2Fapi.soundcloud.com%2Fplaylists%2F922213810&show_artwork=true&maxwidth=640&maxheight=960&dnt=1&secret_token=s-ziYey',
'only_matching': True,
}
- @staticmethod
- def _extract_urls(webpage):
- return [m.group('url') for m in re.finditer(
- r'<iframe[^>]+src=(["\'])(?P<url>(?:https?://)?(?:w\.)?soundcloud\.com/player.+?)\1',
- webpage)]
-
def _real_extract(self, url):
query = parse_qs(url)
api_url = query['url'][0]
return self.url_result(api_url)
-class SoundcloudIE(InfoExtractor):
+class SoundcloudBaseIE(InfoExtractor):
+ _NETRC_MACHINE = 'soundcloud'
+
+ _API_V2_BASE = 'https://api-v2.soundcloud.com/'
+ _BASE_URL = 'https://soundcloud.com/'
+ _USER_AGENT = 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/84.0.4147.105 Safari/537.36'
+ _API_AUTH_QUERY_TEMPLATE = '?client_id=%s'
+ _API_AUTH_URL_PW = 'https://api-auth.soundcloud.com/web-auth/sign-in/password%s'
+ _API_VERIFY_AUTH_TOKEN = 'https://api-auth.soundcloud.com/connect/session%s'
+ _access_token = None
+ _HEADERS = {}
+
+ def _store_client_id(self, client_id):
+ self.cache.store('soundcloud', 'client_id', client_id)
+
+ def _update_client_id(self):
+ webpage = self._download_webpage('https://soundcloud.com/', None)
+ for src in reversed(re.findall(r'<script[^>]+src="([^"]+)"', webpage)):
+ script = self._download_webpage(src, None, fatal=False)
+ if script:
+ client_id = self._search_regex(
+ r'client_id\s*:\s*"([0-9a-zA-Z]{32})"',
+ script, 'client id', default=None)
+ if client_id:
+ self._CLIENT_ID = client_id
+ self._store_client_id(client_id)
+ return
+ raise ExtractorError('Unable to extract client id')
+
+ def _download_json(self, *args, **kwargs):
+ non_fatal = kwargs.get('fatal') is False
+ if non_fatal:
+ del kwargs['fatal']
+ query = kwargs.get('query', {}).copy()
+ for _ in range(2):
+ query['client_id'] = self._CLIENT_ID
+ kwargs['query'] = query
+ try:
+ return super()._download_json(*args, **kwargs)
+ except ExtractorError as e:
+ if isinstance(e.cause, compat_HTTPError) and e.cause.code in (401, 403):
+ self._store_client_id(None)
+ self._update_client_id()
+ continue
+ elif non_fatal:
+ self.report_warning(error_to_compat_str(e))
+ return False
+ raise
+
+ def _initialize_pre_login(self):
+ self._CLIENT_ID = self.cache.load('soundcloud', 'client_id') or 'a3e059563d7fd3372b49b37f00a00bcf'
+
+ def _perform_login(self, username, password):
+ if username != 'oauth':
+ self.report_warning(
+ 'Login using username and password is not currently supported. '
+ 'Use "--username oauth --password <oauth_token>" to login using an oauth token')
+ self._access_token = password
+ query = self._API_AUTH_QUERY_TEMPLATE % self._CLIENT_ID
+ payload = {'session': {'access_token': self._access_token}}
+ token_verification = sanitized_Request(self._API_VERIFY_AUTH_TOKEN % query, json.dumps(payload).encode('utf-8'))
+ response = self._download_json(token_verification, None, note='Verifying login token...', fatal=False)
+ if response is not False:
+ self._HEADERS = {'Authorization': 'OAuth ' + self._access_token}
+ self.report_login()
+ else:
+ self.report_warning('Provided authorization token seems to be invalid. Continue as guest')
+
+ r'''
+ def genDevId():
+ def genNumBlock():
+ return ''.join([str(random.randrange(10)) for i in range(6)])
+ return '-'.join([genNumBlock() for i in range(4)])
+
+ payload = {
+ 'client_id': self._CLIENT_ID,
+ 'recaptcha_pubkey': 'null',
+ 'recaptcha_response': 'null',
+ 'credentials': {
+ 'identifier': username,
+ 'password': password
+ },
+ 'signature': self.sign(username, password, self._CLIENT_ID),
+ 'device_id': genDevId(),
+ 'user_agent': self._USER_AGENT
+ }
+
+ query = self._API_AUTH_QUERY_TEMPLATE % self._CLIENT_ID
+ login = sanitized_Request(self._API_AUTH_URL_PW % query, json.dumps(payload).encode('utf-8'))
+ response = self._download_json(login, None)
+ self._access_token = response.get('session').get('access_token')
+ if not self._access_token:
+ self.report_warning('Unable to get access token, login may has failed')
+ else:
+ self._HEADERS = {'Authorization': 'OAuth ' + self._access_token}
+ '''
+
+ # signature generation
+ def sign(self, user, pw, clid):
+ a = 33
+ i = 1
+ s = 440123
+ w = 117
+ u = 1800000
+ l = 1042
+ b = 37
+ k = 37
+ c = 5
+ n = '0763ed7314c69015fd4a0dc16bbf4b90' # _KEY
+ y = '8' # _REV
+ r = 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/84.0.4147.105 Safari/537.36' # _USER_AGENT
+ e = user # _USERNAME
+ t = clid # _CLIENT_ID
+
+ d = '-'.join([str(mInt) for mInt in [a, i, s, w, u, l, b, k]])
+ p = n + y + d + r + e + t + d + n
+ h = p
+
+ m = 8011470
+ f = 0
+
+ for f in range(f, len(h)):
+ m = (m >> 1) + ((1 & m) << 23)
+ m += ord(h[f])
+ m &= 16777215
+
+ # c is not even needed
+ out = str(y) + ':' + str(d) + ':' + format(m, 'x') + ':' + str(c)
+
+ return out
+
+ @classmethod
+ def _resolv_url(cls, url):
+ return cls._API_V2_BASE + 'resolve?url=' + url
+
+
+class SoundcloudIE(SoundcloudBaseIE):
"""Information extractor for soundcloud.com
To access the media, the uid of the song and a stream token
must be extracted from the page source and the script must make
(?!stations/track)
(?P<uploader>[\w\d-]+)/
(?!(?:tracks|albums|sets(?:/.+?)?|reposts|likes|spotlight)/?(?:$|[?#]))
- (?P<title>[\w\d-]+)/?
- (?P<token>[^?]+?)?(?:[?].*)?$)
+ (?P<title>[\w\d-]+)
+ (?:/(?P<token>(?!(?:albums|sets|recommended))[^?]+?))?
+ (?:[?].*)?$)
|(?:api(?:-v2)?\.soundcloud\.com/tracks/(?P<track_id>\d+)
(?:/?\?secret_token=(?P<secret_token>[^&]+))?)
)
},
]
- _API_V2_BASE = 'https://api-v2.soundcloud.com/'
- _BASE_URL = 'https://soundcloud.com/'
_IMAGE_REPL_RE = r'-([0-9a-z]+)\.jpg'
_ARTWORK_MAP = {
'original': 0,
}
- def _store_client_id(self, client_id):
- self._downloader.cache.store('soundcloud', 'client_id', client_id)
-
- def _update_client_id(self):
- webpage = self._download_webpage('https://soundcloud.com/', None)
- for src in reversed(re.findall(r'<script[^>]+src="([^"]+)"', webpage)):
- script = self._download_webpage(src, None, fatal=False)
- if script:
- client_id = self._search_regex(
- r'client_id\s*:\s*"([0-9a-zA-Z]{32})"',
- script, 'client id', default=None)
- if client_id:
- self._CLIENT_ID = client_id
- self._store_client_id(client_id)
- return
- raise ExtractorError('Unable to extract client id')
-
- def _download_json(self, *args, **kwargs):
- non_fatal = kwargs.get('fatal') is False
- if non_fatal:
- del kwargs['fatal']
- query = kwargs.get('query', {}).copy()
- for _ in range(2):
- query['client_id'] = self._CLIENT_ID
- kwargs['query'] = query
- try:
- return super(SoundcloudIE, self)._download_json(*args, **compat_kwargs(kwargs))
- except ExtractorError as e:
- if isinstance(e.cause, compat_HTTPError) and e.cause.code in (401, 403):
- self._store_client_id(None)
- self._update_client_id()
- continue
- elif non_fatal:
- self.report_warning(error_to_compat_str(e))
- return False
- raise
-
- def _real_initialize(self):
- self._CLIENT_ID = self._downloader.cache.load('soundcloud', 'client_id') or 'a3e059563d7fd3372b49b37f00a00bcf'
- self._login()
-
- _USER_AGENT = 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/84.0.4147.105 Safari/537.36'
- _API_AUTH_QUERY_TEMPLATE = '?client_id=%s'
- _API_AUTH_URL_PW = 'https://api-auth.soundcloud.com/web-auth/sign-in/password%s'
- _API_VERIFY_AUTH_TOKEN = 'https://api-auth.soundcloud.com/connect/session%s'
- _access_token = None
- _HEADERS = {}
- _NETRC_MACHINE = 'soundcloud'
-
- def _login(self):
- username, password = self._get_login_info()
- if username is None:
- return
-
- if username == 'oauth' and password is not None:
- self._access_token = password
- query = self._API_AUTH_QUERY_TEMPLATE % self._CLIENT_ID
- payload = {'session': {'access_token': self._access_token}}
- token_verification = sanitized_Request(self._API_VERIFY_AUTH_TOKEN % query, json.dumps(payload).encode('utf-8'))
- response = self._download_json(token_verification, None, note='Verifying login token...', fatal=False)
- if response is not False:
- self._HEADERS = {'Authorization': 'OAuth ' + self._access_token}
- self.report_login()
- else:
- self.report_warning('Provided authorization token seems to be invalid. Continue as guest')
- elif username is not None:
- self.report_warning(
- 'Login using username and password is not currently supported. '
- 'Use "--user oauth --password <oauth_token>" to login using an oauth token')
-
- r'''
- def genDevId():
- def genNumBlock():
- return ''.join([str(random.randrange(10)) for i in range(6)])
- return '-'.join([genNumBlock() for i in range(4)])
-
- payload = {
- 'client_id': self._CLIENT_ID,
- 'recaptcha_pubkey': 'null',
- 'recaptcha_response': 'null',
- 'credentials': {
- 'identifier': username,
- 'password': password
- },
- 'signature': self.sign(username, password, self._CLIENT_ID),
- 'device_id': genDevId(),
- 'user_agent': self._USER_AGENT
- }
-
- query = self._API_AUTH_QUERY_TEMPLATE % self._CLIENT_ID
- login = sanitized_Request(self._API_AUTH_URL_PW % query, json.dumps(payload).encode('utf-8'))
- response = self._download_json(login, None)
- self._access_token = response.get('session').get('access_token')
- if not self._access_token:
- self.report_warning('Unable to get access token, login may has failed')
- else:
- self._HEADERS = {'Authorization': 'OAuth ' + self._access_token}
- '''
-
- # signature generation
- def sign(self, user, pw, clid):
- a = 33
- i = 1
- s = 440123
- w = 117
- u = 1800000
- l = 1042
- b = 37
- k = 37
- c = 5
- n = '0763ed7314c69015fd4a0dc16bbf4b90' # _KEY
- y = '8' # _REV
- r = 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/84.0.4147.105 Safari/537.36' # _USER_AGENT
- e = user # _USERNAME
- t = clid # _CLIENT_ID
-
- d = '-'.join([str(mInt) for mInt in [a, i, s, w, u, l, b, k]])
- p = n + y + d + r + e + t + d + n
- h = p
-
- m = 8011470
- f = 0
-
- for f in range(f, len(h)):
- m = (m >> 1) + ((1 & m) << 23)
- m += ord(h[f])
- m &= 16777215
-
- # c is not even needed
- out = str(y) + ':' + str(d) + ':' + format(m, 'x') + ':' + str(c)
-
- return out
-
- @classmethod
- def _resolv_url(cls, url):
- return SoundcloudIE._API_V2_BASE + 'resolve?url=' + url
-
def _extract_info_dict(self, info, full_title=None, secret_token=None):
track_id = compat_str(info['id'])
title = info['title']
return self._extract_info_dict(info, full_title, token)
-class SoundcloudPlaylistBaseIE(SoundcloudIE):
+class SoundcloudPlaylistBaseIE(SoundcloudBaseIE):
def _extract_set(self, playlist, token=None):
playlist_id = compat_str(playlist['id'])
tracks = playlist.get('tracks') or []
return self._extract_set(info, token)
-class SoundcloudPagedPlaylistBaseIE(SoundcloudIE):
+class SoundcloudPagedPlaylistBaseIE(SoundcloudBaseIE):
def _extract_playlist(self, base_url, playlist_id, playlist_title):
return {
'_type': 'playlist',
'offset': 0,
}
- retries = self.get_param('extractor_retries', 3)
-
for i in itertools.count():
- attempt, last_error = -1, None
- while attempt < retries:
- attempt += 1
- if last_error:
- self.report_warning('%s. Retrying ...' % remove_end(last_error, '.'), playlist_id)
+ for retry in self.RetryManager():
try:
response = self._download_json(
url, playlist_id, query=query, headers=self._HEADERS,
- note='Downloading track page %s%s' % (i + 1, f' (retry #{attempt})' if attempt else ''))
+ note=f'Downloading track page {i + 1}')
break
except ExtractorError as e:
# Downloading page may result in intermittent 502 HTTP error
# See https://github.com/yt-dlp/yt-dlp/issues/872
- if attempt >= retries or not isinstance(e.cause, compat_HTTPError) or e.cause.code != 502:
+ if not isinstance(e.cause, compat_HTTPError) or e.cause.code != 502:
raise
- last_error = str(e.cause or e.msg)
+ retry.error = e
+ continue
def resolve_entry(*candidates):
for cand in candidates:
track_id, 'Track station: %s' % track['title'])
+class SoundcloudRelatedIE(SoundcloudPagedPlaylistBaseIE):
+ _VALID_URL = r'https?://(?:(?:www|m)\.)?soundcloud\.com/(?P<slug>[\w\d-]+/[\w\d-]+)/(?P<relation>albums|sets|recommended)'
+ IE_NAME = 'soundcloud:related'
+ _TESTS = [{
+ 'url': 'https://soundcloud.com/wajang/sexapil-pingers-5/recommended',
+ 'info_dict': {
+ 'id': '1084577272',
+ 'title': 'Sexapil - Pingers 5 (Recommended)',
+ },
+ 'playlist_mincount': 50,
+ }, {
+ 'url': 'https://soundcloud.com/wajang/sexapil-pingers-5/albums',
+ 'info_dict': {
+ 'id': '1084577272',
+ 'title': 'Sexapil - Pingers 5 (Albums)',
+ },
+ 'playlist_mincount': 1,
+ }, {
+ 'url': 'https://soundcloud.com/wajang/sexapil-pingers-5/sets',
+ 'info_dict': {
+ 'id': '1084577272',
+ 'title': 'Sexapil - Pingers 5 (Sets)',
+ },
+ 'playlist_mincount': 4,
+ }]
+
+ _BASE_URL_MAP = {
+ 'albums': 'tracks/%s/albums',
+ 'sets': 'tracks/%s/playlists_without_albums',
+ 'recommended': 'tracks/%s/related',
+ }
+
+ def _real_extract(self, url):
+ slug, relation = self._match_valid_url(url).group('slug', 'relation')
+
+ track = self._download_json(
+ self._resolv_url(self._BASE_URL + slug),
+ slug, 'Downloading track info', headers=self._HEADERS)
+
+ if track.get('errors'):
+ raise ExtractorError(f'{self.IE_NAME} said: %s' % ','.join(
+ str(err['error_message']) for err in track['errors']), expected=True)
+
+ return self._extract_playlist(
+ self._API_V2_BASE + self._BASE_URL_MAP[relation] % track['id'], str(track['id']),
+ '%s (%s)' % (track.get('title') or slug, relation.capitalize()))
+
+
class SoundcloudPlaylistIE(SoundcloudPlaylistBaseIE):
_VALID_URL = r'https?://api(?:-v2)?\.soundcloud\.com/playlists/(?P<id>[0-9]+)(?:/?\?secret_token=(?P<token>[^&]+?))?$'
IE_NAME = 'soundcloud:playlist'
return self._extract_set(data, token)
-class SoundcloudSearchIE(SearchInfoExtractor, SoundcloudIE):
+class SoundcloudSearchIE(SoundcloudBaseIE, SearchInfoExtractor):
IE_NAME = 'soundcloud:search'
IE_DESC = 'Soundcloud search'
- _MAX_RESULTS = float('inf')
+ _SEARCH_KEY = 'scsearch'
_TESTS = [{
'url': 'scsearch15:post-avant jazzcore',
'info_dict': {
'playlist_count': 15,
}]
- _SEARCH_KEY = 'scsearch'
_MAX_RESULTS_PER_PAGE = 200
_DEFAULT_RESULTS_PER_PAGE = 50
})
next_url = update_url_query(self._API_V2_BASE + endpoint, query)
- collected_results = 0
-
for i in itertools.count(1):
response = self._download_json(
- next_url, collection_id, 'Downloading page {0}'.format(i),
+ next_url, collection_id, f'Downloading page {i}',
'Unable to download API page', headers=self._HEADERS)
- collection = response.get('collection', [])
- if not collection:
- break
-
- collection = list(filter(bool, collection))
- collected_results += len(collection)
-
- for item in collection:
- yield self.url_result(item['uri'], SoundcloudIE.ie_key())
-
- if not collection or collected_results >= limit:
- break
+ for item in response.get('collection') or []:
+ if item:
+ yield self.url_result(item['uri'], SoundcloudIE.ie_key())
next_url = response.get('next_href')
if not next_url:
break
def _get_n_results(self, query, n):
- tracks = self._get_collection('search/tracks', query, limit=n, q=query)
- return self.playlist_result(tracks, playlist_title=query)
+ return self.playlist_result(itertools.islice(
+ self._get_collection('search/tracks', query, limit=n, q=query),
+ 0, None if n == float('inf') else n), query, query)
projects / yt-dlp.git / blobdiff