clean_html,
dict_get,
datetime_from_str,
+ error_to_compat_str,
ExtractorError,
format_field,
float_or_none,
})
def warn(message):
- self._downloader.report_warning(message)
+ self.report_warning(message)
lookup_req = [
username,
if not self._login():
return
- _YT_WEB_CLIENT_VERSION = '2.20210301.08.00'
- _DEFAULT_API_DATA = {
- 'context': {
- 'client': {
- 'clientName': 'WEB',
- 'clientVersion': _YT_WEB_CLIENT_VERSION,
- }
- },
- }
-
- _DEFAULT_BASIC_API_HEADERS = {
- 'X-YouTube-Client-Name': '1',
- 'X-YouTube-Client-Version': _YT_WEB_CLIENT_VERSION
- }
-
+ _YT_WEB_CLIENT_VERSION = '2.20210407.08.00'
+ _YT_INNERTUBE_API_KEY = 'AIzaSyAO_FJ2SlqU8Q4STEHLGCilw_Y9_11qcW8'
_YT_INITIAL_DATA_RE = r'(?:window\s*\[\s*["\']ytInitialData["\']\s*\]|ytInitialData)\s*=\s*({.+?})\s*;'
_YT_INITIAL_PLAYER_RESPONSE_RE = r'ytInitialPlayerResponse\s*=\s*({.+?})\s*;'
_YT_INITIAL_BOUNDARY_RE = r'(?:var\s+meta|</script|\n)'
return "SAPISIDHASH %s_%s" % (time_now, sapisidhash)
def _call_api(self, ep, query, video_id, fatal=True, headers=None,
- note='Downloading API JSON', errnote='Unable to download API page'):
- data = self._DEFAULT_API_DATA.copy()
+ note='Downloading API JSON', errnote='Unable to download API page',
+ context=None, api_key=None):
+
+ data = {'context': context} if context else {'context': self._extract_context()}
data.update(query)
- headers = headers or {}
- headers.update({'content-type': 'application/json'})
- auth = self._generate_sapisidhash_header()
- if auth is not None:
- headers.update({'Authorization': auth, 'X-Origin': 'https://www.youtube.com'})
+ real_headers = self._generate_api_headers()
+ real_headers.update({'content-type': 'application/json'})
+ if headers:
+ real_headers.update(headers)
return self._download_json(
'https://www.youtube.com/youtubei/v1/%s' % ep,
video_id=video_id, fatal=fatal, note=note, errnote=errnote,
- data=json.dumps(data).encode('utf8'), headers=headers,
- query={'key': 'AIzaSyAO_FJ2SlqU8Q4STEHLGCilw_Y9_11qcW8'})
+ data=json.dumps(data).encode('utf8'), headers=real_headers,
+ query={'key': api_key or self._extract_api_key()})
+
+ def _extract_api_key(self, ytcfg=None):
+ return try_get(ytcfg, lambda x: x['INNERTUBE_API_KEY'], compat_str) or self._YT_INNERTUBE_API_KEY
def _extract_yt_initial_data(self, video_id, webpage):
return self._parse_json(
@staticmethod
def _extract_account_syncid(data):
- """Extract syncId required to download private playlists of secondary channels"""
- sync_ids = (
- try_get(data, lambda x: x['responseContext']['mainAppWebResponseContext']['datasyncId'], compat_str)
- or '').split("||")
+ """
+ Extract syncId required to download private playlists of secondary channels
+ @param data Either response or ytcfg
+ """
+ sync_ids = (try_get(
+ data, (lambda x: x['responseContext']['mainAppWebResponseContext']['datasyncId'],
+ lambda x: x['DATASYNC_ID']), compat_str) or '').split("||")
if len(sync_ids) >= 2 and sync_ids[1]:
# datasyncid is of the form "channel_syncid||user_syncid" for secondary channel
# and just "user_syncid||" for primary channel. We only want the channel_syncid
return sync_ids[0]
+ # ytcfg includes channel_syncid if on secondary channel
+ return data.get('DELEGATED_SESSION_ID')
def _extract_ytcfg(self, video_id, webpage):
return self._parse_json(
self._search_regex(
r'ytcfg\.set\s*\(\s*({.+?})\s*\)\s*;', webpage, 'ytcfg',
- default='{}'), video_id, fatal=False)
+ default='{}'), video_id, fatal=False) or {}
+
+ def __extract_client_version(self, ytcfg):
+ return try_get(ytcfg, lambda x: x['INNERTUBE_CLIENT_VERSION'], compat_str) or self._YT_WEB_CLIENT_VERSION
+
+ def _extract_context(self, ytcfg=None):
+ context = try_get(ytcfg, lambda x: x['INNERTUBE_CONTEXT'], dict)
+ if context:
+ return context
+
+ # Recreate the client context (required)
+ client_version = self.__extract_client_version(ytcfg)
+ client_name = try_get(ytcfg, lambda x: x['INNERTUBE_CLIENT_NAME'], compat_str) or 'WEB'
+ context = {
+ 'client': {
+ 'clientName': client_name,
+ 'clientVersion': client_version,
+ }
+ }
+ visitor_data = try_get(ytcfg, lambda x: x['VISITOR_DATA'], compat_str)
+ if visitor_data:
+ context['client']['visitorData'] = visitor_data
+ return context
+
+ def _generate_api_headers(self, ytcfg=None, identity_token=None, account_syncid=None, visitor_data=None):
+ headers = {
+ 'X-YouTube-Client-Name': '1',
+ 'X-YouTube-Client-Version': self.__extract_client_version(ytcfg),
+ }
+ if identity_token:
+ headers['x-youtube-identity-token'] = identity_token
+ if account_syncid:
+ headers['X-Goog-PageId'] = account_syncid
+ headers['X-Goog-AuthUser'] = 0
+ if visitor_data:
+ headers['x-goog-visitor-id'] = visitor_data
+ auth = self._generate_sapisidhash_header()
+ if auth is not None:
+ headers['Authorization'] = auth
+ headers['X-Origin'] = 'https://www.youtube.com'
+ return headers
def _extract_video(self, renderer):
video_id = renderer.get('videoId')
}
def _comment_entries(self, root_continuation_data, identity_token, account_syncid,
- session_token_list, parent=None, comment_counts=None):
+ ytcfg, session_token_list, parent=None, comment_counts=None):
def extract_thread(parent_renderer):
contents = try_get(parent_renderer, lambda x: x['contents'], list) or []
if comment_replies_renderer:
comment_counts[2] += 1
comment_entries_iter = self._comment_entries(
- comment_replies_renderer, identity_token, account_syncid,
+ comment_replies_renderer, identity_token, account_syncid, ytcfg,
parent=comment.get('id'), session_token_list=session_token_list,
comment_counts=comment_counts)
if not comment_counts:
# comment so far, est. total comments, current comment thread #
comment_counts = [0, 0, 0]
- headers = self._DEFAULT_BASIC_API_HEADERS.copy()
# TODO: Generalize the download code with TabIE
- if identity_token:
- headers['x-youtube-identity-token'] = identity_token
-
- if account_syncid:
- headers['X-Goog-PageId'] = account_syncid
- headers['X-Goog-AuthUser'] = 0
-
+ context = self._extract_context(ytcfg)
+ visitor_data = try_get(context, lambda x: x['client']['visitorData'], compat_str)
continuation = YoutubeTabIE._extract_continuation(root_continuation_data) # TODO
first_continuation = False
if parent is None:
for page_num in itertools.count(0):
if not continuation:
break
+ headers = self._generate_api_headers(ytcfg, identity_token, account_syncid, visitor_data)
retries = self._downloader.params.get('extractor_retries', 3)
count = -1
last_error = None
# See: https://github.com/ytdl-org/youtube-dl/issues/28194
last_error = 'Incomplete data received'
if count >= retries:
- self._downloader.report_error(last_error)
+ raise ExtractorError(last_error)
if not response:
break
+ visitor_data = try_get(
+ response,
+ lambda x: x['responseContext']['webResponseContextExtensionData']['ytConfigData']['visitorData'],
+ compat_str) or visitor_data
known_continuation_renderers = {
'itemSectionContinuation': extract_thread,
renderer,
identity_token=self._extract_identity_token(webpage, item_id=video_id),
account_syncid=self._extract_account_syncid(ytcfg),
+ ytcfg=ytcfg,
session_token_list=[xsrf_token])
for comment in comment_iter:
player_response = self._extract_yt_initial_variable(
webpage, self._YT_INITIAL_PLAYER_RESPONSE_RE,
video_id, 'initial player response')
+
+ ytcfg = self._extract_ytcfg(video_id, webpage)
if not player_response:
player_response = self._call_api(
- 'player', {'videoId': video_id}, video_id)
+ 'player', {'videoId': video_id}, video_id, api_key=self._extract_api_key(ytcfg))
playability_status = player_response.get('playabilityStatus') or {}
if playability_status.get('reason') == 'Sign in to confirm your age':
if not formats:
if not self._downloader.params.get('allow_unplayable_formats') and streaming_data.get('licenseInfos'):
- raise ExtractorError(
+ self.raise_no_formats(
'This video is DRM protected.', expected=True)
pemr = try_get(
playability_status,
if not countries:
regions_allowed = search_meta('regionsAllowed')
countries = regions_allowed.split(',') if regions_allowed else None
- self.raise_geo_restricted(
- subreason, countries)
+ self.raise_geo_restricted(subreason, countries, metadata_available=True)
reason += '\n' + subreason
if reason:
- raise ExtractorError(reason, expected=True)
+ self.raise_no_formats(reason, expected=True)
self._sort_formats(formats)
for m in re.finditer(self._meta_regex('og:video:tag'), webpage)]
for keyword in keywords:
if keyword.startswith('yt:stretch='):
- w, h = keyword.split('=')[1].split(':')
- w, h = int(w), int(h)
+ stretch_ratio = map(
+ lambda x: int_or_none(x, default=0),
+ keyword.split('=')[1].split(':'))
+ w, h = (list(stretch_ratio) + [0])[:2]
if w > 0 and h > 0:
ratio = w / h
for f in formats:
'yt initial data')
if not initial_data:
initial_data = self._call_api(
- 'next', {'videoId': video_id}, video_id, fatal=False)
+ 'next', {'videoId': video_id}, video_id, fatal=False, api_key=self._extract_api_key(ytcfg))
if not is_live:
try:
'uploader_id': 'UCXw-G3eDE9trcvY2sBMM_aA',
},
'playlist_mincount': 21,
+ }, {
+ 'note': 'Playlist with "show unavailable videos" button',
+ 'url': 'https://www.youtube.com/playlist?list=UUTYLiWFZy8xtPwxFwX9rV7Q',
+ 'info_dict': {
+ 'title': 'Uploads from Phim Siêu Nhân Nhật Bản',
+ 'id': 'UUTYLiWFZy8xtPwxFwX9rV7Q',
+ 'uploader': 'Phim Siêu Nhân Nhật Bản',
+ 'uploader_id': 'UCTYLiWFZy8xtPwxFwX9rV7Q',
+ },
+ 'playlist_mincount': 1400,
+ 'expected_warnings': [
+ 'YouTube said: INFO - Unavailable videos are hidden',
+ ]
+ }, {
+ 'note': 'Playlist with unavailable videos in a later page',
+ 'url': 'https://www.youtube.com/playlist?list=UU8l9frL61Yl5KFOl87nIm2w',
+ 'info_dict': {
+ 'title': 'Uploads from BlankTV',
+ 'id': 'UU8l9frL61Yl5KFOl87nIm2w',
+ 'uploader': 'BlankTV',
+ 'uploader_id': 'UC8l9frL61Yl5KFOl87nIm2w',
+ },
+ 'playlist_mincount': 20000,
}, {
# https://github.com/ytdl-org/youtube-dl/issues/21844
'url': 'https://www.youtube.com/playlist?list=PLzH6n4zXuckpfMu_4Ff8E7Z1behQks5ba',
ctp = continuation_ep.get('clickTrackingParams')
return YoutubeTabIE._build_continuation_query(continuation, ctp)
- def _entries(self, tab, item_id, identity_token, account_syncid):
+ def _entries(self, tab, item_id, identity_token, account_syncid, ytcfg):
def extract_entries(parent_renderer): # this needs to called again for continuation to work with feeds
contents = try_get(parent_renderer, lambda x: x['contents'], list) or []
for entry in extract_entries(parent_renderer):
yield entry
continuation = continuation_list[0]
-
- headers = {
- 'x-youtube-client-name': '1',
- 'x-youtube-client-version': '2.20201112.04.01',
- }
- if identity_token:
- headers['x-youtube-identity-token'] = identity_token
-
- if account_syncid:
- headers['X-Goog-PageId'] = account_syncid
- headers['X-Goog-AuthUser'] = 0
+ context = self._extract_context(ytcfg)
+ visitor_data = try_get(context, lambda x: x['client']['visitorData'], compat_str)
for page_num in itertools.count(1):
if not continuation:
break
- retries = self._downloader.params.get('extractor_retries', 3)
- count = -1
- last_error = None
- while count < retries:
- count += 1
- if last_error:
- self.report_warning('%s. Retrying ...' % last_error)
- try:
- response = self._call_api(
- ep='browse', fatal=True, headers=headers,
- video_id='%s page %s' % (item_id, page_num),
- query={
- 'continuation': continuation['continuation'],
- 'clickTracking': {'clickTrackingParams': continuation['itct']},
- },
- note='Downloading API JSON%s' % (' (retry #%d)' % count if count else ''))
- except ExtractorError as e:
- if isinstance(e.cause, compat_HTTPError) and e.cause.code in (500, 503, 404):
- # Downloading page may result in intermittent 5xx HTTP error
- # Sometimes a 404 is also recieved. See: https://github.com/ytdl-org/youtube-dl/issues/28289
- last_error = 'HTTP Error %s' % e.cause.code
- if count < retries:
- continue
- raise
- else:
- # Youtube sometimes sends incomplete data
- # See: https://github.com/ytdl-org/youtube-dl/issues/28194
- if dict_get(response,
- ('continuationContents', 'onResponseReceivedActions', 'onResponseReceivedEndpoints')):
- break
-
- # Youtube may send alerts if there was an issue with the continuation page
- self._extract_alerts(response, expected=False)
-
- last_error = 'Incomplete data received'
- if count >= retries:
- self._downloader.report_error(last_error)
+ query = {
+ 'continuation': continuation['continuation'],
+ 'clickTracking': {'clickTrackingParams': continuation['itct']}
+ }
+ headers = self._generate_api_headers(ytcfg, identity_token, account_syncid, visitor_data)
+ response = self._extract_response(
+ item_id='%s page %s' % (item_id, page_num),
+ query=query, headers=headers, ytcfg=ytcfg,
+ check_get_keys=('continuationContents', 'onResponseReceivedActions', 'onResponseReceivedEndpoints'))
if not response:
break
+ visitor_data = try_get(
+ response, lambda x: x['responseContext']['visitorData'], compat_str) or visitor_data
known_continuation_renderers = {
'playlistVideoListContinuation': self._playlist_entries,
self._entries(
selected_tab, playlist_id,
self._extract_identity_token(webpage, item_id),
- self._extract_account_syncid(data)),
+ self._extract_account_syncid(data),
+ self._extract_ytcfg(item_id, webpage)),
**metadata)
- def _extract_mix_playlist(self, playlist, playlist_id):
+ def _extract_mix_playlist(self, playlist, playlist_id, data, webpage):
first_id = last_id = None
+ ytcfg = self._extract_ytcfg(playlist_id, webpage)
+ headers = self._generate_api_headers(
+ ytcfg, account_syncid=self._extract_account_syncid(data),
+ identity_token=self._extract_identity_token(webpage, item_id=playlist_id),
+ visitor_data=try_get(self._extract_context(ytcfg), lambda x: x['client']['visitorData'], compat_str))
for page_num in itertools.count(1):
videos = list(self._playlist_entries(playlist))
if not videos:
yield video
first_id = first_id or videos[0]['id']
last_id = videos[-1]['id']
-
- _, data = self._extract_webpage(
- 'https://www.youtube.com/watch?list=%s&v=%s' % (playlist_id, last_id),
- '%s page %d' % (playlist_id, page_num))
+ watch_endpoint = try_get(
+ playlist, lambda x: x['contents'][-1]['playlistPanelVideoRenderer']['navigationEndpoint']['watchEndpoint'])
+ query = {
+ 'playlistId': playlist_id,
+ 'videoId': watch_endpoint.get('videoId') or last_id,
+ 'index': watch_endpoint.get('index') or len(videos),
+ 'params': watch_endpoint.get('params') or 'OAE%3D'
+ }
+ response = self._extract_response(
+ item_id='%s page %d' % (playlist_id, page_num),
+ query=query,
+ ep='next',
+ headers=headers,
+ check_get_keys='contents'
+ )
playlist = try_get(
- data, lambda x: x['contents']['twoColumnWatchNextResults']['playlist']['playlist'], dict)
+ response, lambda x: x['contents']['twoColumnWatchNextResults']['playlist']['playlist'], dict)
- def _extract_from_playlist(self, item_id, url, data, playlist):
+ def _extract_from_playlist(self, item_id, url, data, playlist, webpage):
title = playlist.get('title') or try_get(
data, lambda x: x['titleText']['simpleText'], compat_str)
playlist_id = playlist.get('playlistId') or item_id
video_title=title)
return self.playlist_result(
- self._extract_mix_playlist(playlist, playlist_id),
+ self._extract_mix_playlist(playlist, playlist_id, data, webpage),
playlist_id=playlist_id, playlist_title=title)
def _extract_alerts(self, data, expected=False):
warnings.append([alert_type, alert_message])
for alert_type, alert_message in (warnings + errors[:-1]):
- self._downloader.report_warning('YouTube said: %s - %s' % (alert_type, alert_message))
+ self.report_warning('YouTube said: %s - %s' % (alert_type, alert_message))
if errors:
raise ExtractorError('YouTube said: %s' % errors[-1][1], expected=expected)
+ def _reload_with_unavailable_videos(self, item_id, data, webpage):
+ """
+ Get playlist with unavailable videos if the 'show unavailable videos' button exists.
+ """
+ sidebar_renderer = try_get(
+ data, lambda x: x['sidebar']['playlistSidebarRenderer']['items'], list)
+ if not sidebar_renderer:
+ return
+ browse_id = params = None
+ for item in sidebar_renderer:
+ if not isinstance(item, dict):
+ continue
+ renderer = item.get('playlistSidebarPrimaryInfoRenderer')
+ menu_renderer = try_get(
+ renderer, lambda x: x['menu']['menuRenderer']['items'], list) or []
+ for menu_item in menu_renderer:
+ if not isinstance(menu_item, dict):
+ continue
+ nav_item_renderer = menu_item.get('menuNavigationItemRenderer')
+ text = try_get(
+ nav_item_renderer, lambda x: x['text']['simpleText'], compat_str)
+ if not text or text.lower() != 'show unavailable videos':
+ continue
+ browse_endpoint = try_get(
+ nav_item_renderer, lambda x: x['navigationEndpoint']['browseEndpoint'], dict) or {}
+ browse_id = browse_endpoint.get('browseId')
+ params = browse_endpoint.get('params')
+ break
+
+ ytcfg = self._extract_ytcfg(item_id, webpage)
+ headers = self._generate_api_headers(
+ ytcfg, account_syncid=self._extract_account_syncid(ytcfg),
+ identity_token=self._extract_identity_token(webpage, item_id=item_id),
+ visitor_data=try_get(
+ self._extract_context(ytcfg), lambda x: x['client']['visitorData'], compat_str))
+ query = {
+ 'params': params or 'wgYCCAA=',
+ 'browseId': browse_id or 'VL%s' % item_id
+ }
+ return self._extract_response(
+ item_id=item_id, headers=headers, query=query,
+ check_get_keys='contents', fatal=False,
+ note='Downloading API JSON with unavailable videos')
+
+ def _extract_response(self, item_id, query, note='Downloading API JSON', headers=None,
+ ytcfg=None, check_get_keys=None, ep='browse', fatal=True):
+ response = None
+ last_error = None
+ count = -1
+ retries = self._downloader.params.get('extractor_retries', 3)
+ if check_get_keys is None:
+ check_get_keys = []
+ while count < retries:
+ count += 1
+ if last_error:
+ self.report_warning('%s. Retrying ...' % last_error)
+ try:
+ response = self._call_api(
+ ep=ep, fatal=True, headers=headers,
+ video_id=item_id, query=query,
+ context=self._extract_context(ytcfg),
+ api_key=self._extract_api_key(ytcfg),
+ note='%s%s' % (note, ' (retry #%d)' % count if count else ''))
+ except ExtractorError as e:
+ if isinstance(e.cause, compat_HTTPError) and e.cause.code in (500, 503, 404):
+ # Downloading page may result in intermittent 5xx HTTP error
+ # Sometimes a 404 is also recieved. See: https://github.com/ytdl-org/youtube-dl/issues/28289
+ last_error = 'HTTP Error %s' % e.cause.code
+ if count < retries:
+ continue
+ if fatal:
+ raise
+ else:
+ self.report_warning(error_to_compat_str(e))
+ return
+
+ else:
+ # Youtube may send alerts if there was an issue with the continuation page
+ self._extract_alerts(response, expected=False)
+ if not check_get_keys or dict_get(response, check_get_keys):
+ break
+ # Youtube sometimes sends incomplete data
+ # See: https://github.com/ytdl-org/youtube-dl/issues/28194
+ last_error = 'Incomplete data received'
+ if count >= retries:
+ if fatal:
+ raise ExtractorError(last_error)
+ else:
+ self.report_warning(last_error)
+ return
+ return response
+
def _extract_webpage(self, url, item_id):
retries = self._downloader.params.get('extractor_retries', 3)
count = -1
if data.get('contents') or data.get('currentVideoEndpoint'):
break
if count >= retries:
- self._downloader.report_error(last_error)
+ raise ExtractorError(last_error)
return webpage, data
def _real_extract(self, url):
mobj = re.match(r'(?P<pre>%s)(?P<post>/?(?![^#?]).*$)' % self._VALID_URL, url)
mobj = mobj.groupdict() if mobj else {}
if mobj and not mobj.get('not_channel'):
- self._downloader.report_warning(
+ self.report_warning(
'A channel/user page was given. All the channel\'s videos will be downloaded. '
'To download only the videos in the home page, add a "/featured" to the URL')
url = '%s/videos%s' % (mobj.get('pre'), mobj.get('post') or '')
# If there is neither video or playlist ids,
# youtube redirects to home page, which is undesirable
raise ExtractorError('Unable to recognize tab page')
- self._downloader.report_warning('A video URL was given without video ID. Trying to download playlist %s' % playlist_id)
+ self.report_warning('A video URL was given without video ID. Trying to download playlist %s' % playlist_id)
url = 'https://www.youtube.com/playlist?list=%s' % playlist_id
if video_id and playlist_id:
webpage, data = self._extract_webpage(url, item_id)
+ # YouTube sometimes provides a button to reload playlist with unavailable videos.
+ data = self._reload_with_unavailable_videos(item_id, data, webpage) or data
+
tabs = try_get(
data, lambda x: x['contents']['twoColumnBrowseResultsRenderer']['tabs'], list)
if tabs:
playlist = try_get(
data, lambda x: x['contents']['twoColumnWatchNextResults']['playlist']['playlist'], dict)
if playlist:
- return self._extract_from_playlist(item_id, url, data, playlist)
+ return self._extract_from_playlist(item_id, url, data, playlist, webpage)
video_id = try_get(
data, lambda x: x['currentVideoEndpoint']['watchEndpoint']['videoId'],
compat_str) or video_id
if video_id:
- self._downloader.report_warning('Unable to recognize playlist. Downloading just video %s' % video_id)
+ self.report_warning('Unable to recognize playlist. Downloading just video %s' % video_id)
return self.url_result(video_id, ie=YoutubeIE.ie_key(), video_id=video_id)
raise ExtractorError('Unable to recognize tab page')
ie=YoutubeTabIE.ie_key())
-class YoutubeSearchIE(SearchInfoExtractor, YoutubeBaseInfoExtractor):
+class YoutubeSearchIE(SearchInfoExtractor, YoutubeTabIE):
IE_DESC = 'YouTube.com searches, "ytsearch" keyword'
# there doesn't appear to be a real limit, for example if you search for
# 'python' you get more than 8.000.000 results
data['params'] = self._SEARCH_PARAMS
total = 0
for page_num in itertools.count(1):
- search = self._call_api(
- ep='search', video_id='query "%s"' % query, fatal=False,
- note='Downloading page %s' % page_num, query=data)
+ search = self._extract_response(
+ item_id='query "%s" page %s' % (query, page_num), ep='search', query=data,
+ check_get_keys=('contents', 'onResponseReceivedCommands')
+ )
if not search:
break
slr_contents = try_get(
projects / yt-dlp.git / blobdiff