from __future__ import unicode_literals
+import base64
import calendar
import copy
+import datetime
import hashlib
import itertools
import json
from ..jsinterp import JSInterpreter
from ..utils import (
bool_or_none,
+ bytes_to_intlist,
clean_html,
dict_get,
datetime_from_str,
format_field,
float_or_none,
int_or_none,
+ intlist_to_bytes,
mimetype2ext,
parse_codecs,
+ parse_count,
parse_duration,
qualities,
remove_start,
update_url_query,
url_or_none,
urlencode_postdata,
- urljoin
+ urljoin,
+ variadic
)
if username:
warn('Logging in using username and password is broken. %s' % self._LOGIN_HINTS['cookies'])
return
- # Everything below this is broken!
+ # Everything below this is broken!
+ r'''
# No authentication to be performed
if username is None:
if self._LOGIN_REQUIRED and self.get_param('cookiefile') is None:
return False
return True
+ '''
def _initialize_consent(self):
cookies = self._get_cookies('https://www.youtube.com/')
'hl': 'en',
}
},
- 'INNERTUBE_CONTEXT_CLIENT_NAME': 'ANDROID'
+ 'INNERTUBE_CONTEXT_CLIENT_NAME': 3
},
'ANDROID_EMBEDDED_PLAYER': {
'INNERTUBE_API_VERSION': 'v1',
'hl': 'en',
}
},
- 'INNERTUBE_CONTEXT_CLIENT_NAME': 'ANDROID_EMBEDDED_PLAYER'
+ 'INNERTUBE_CONTEXT_CLIENT_NAME': 55
},
'ANDROID_MUSIC': {
'INNERTUBE_API_VERSION': 'v1',
'hl': 'en',
}
},
- 'INNERTUBE_CONTEXT_CLIENT_NAME': 'ANDROID_MUSIC'
+ 'INNERTUBE_CONTEXT_CLIENT_NAME': 21
}
}
def _extract_client_name(self, ytcfg, default_client='WEB'):
return self._ytcfg_get_safe(ytcfg, lambda x: x['INNERTUBE_CLIENT_NAME'], compat_str, default_client)
+ @staticmethod
+ def _extract_session_index(ytcfg):
+ return int_or_none(try_get(ytcfg, lambda x: x['SESSION_INDEX']))
+
def _extract_client_version(self, ytcfg, default_client='WEB'):
return self._ytcfg_get_safe(ytcfg, lambda x: x['INNERTUBE_CLIENT_VERSION'], compat_str, default_client)
'identity token', default=None)
@staticmethod
- def _extract_account_syncid(data):
+ def _extract_account_syncid(*args):
"""
Extract syncId required to download private playlists of secondary channels
- @param data Either response or ytcfg
+ @params response and/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')
+ for data in args:
+ # ytcfg includes channel_syncid if on secondary channel
+ delegated_sid = try_get(data, lambda x: x['DELEGATED_SESSION_ID'], compat_str)
+ if delegated_sid:
+ return delegated_sid
+ 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]
def _extract_ytcfg(self, video_id, webpage):
if not webpage:
default='{}'), video_id, fatal=False) or {}
def _generate_api_headers(self, ytcfg=None, identity_token=None, account_syncid=None,
- visitor_data=None, api_hostname=None, client='WEB'):
+ visitor_data=None, api_hostname=None, client='WEB', session_index=None):
origin = 'https://' + (api_hostname if api_hostname else self._get_innertube_host(client))
headers = {
'X-YouTube-Client-Name': compat_str(
'X-YouTube-Client-Version': self._extract_client_version(ytcfg, client),
'Origin': origin
}
+ if not visitor_data and ytcfg:
+ visitor_data = try_get(
+ self._extract_context(ytcfg, client), lambda x: x['client']['visitorData'], compat_str)
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 session_index is None and ytcfg:
+ session_index = self._extract_session_index(ytcfg)
+ if account_syncid or session_index is not None:
+ headers['X-Goog-AuthUser'] = session_index if session_index is not None else 0
if visitor_data:
headers['X-Goog-Visitor-Id'] = visitor_data
auth = self._generate_sapisidhash_header(origin)
return headers
@staticmethod
- def _extract_alerts(data):
+ def _build_api_continuation_query(continuation, ctp=None):
+ query = {
+ 'continuation': continuation
+ }
+ # TODO: Inconsistency with clickTrackingParams.
+ # Currently we have a fixed ctp contained within context (from ytcfg)
+ # and a ctp in root query for continuation.
+ if ctp:
+ query['clickTracking'] = {'clickTrackingParams': ctp}
+ return query
+
+ @classmethod
+ def _extract_next_continuation_data(cls, renderer):
+ next_continuation = try_get(
+ renderer, (lambda x: x['continuations'][0]['nextContinuationData'],
+ lambda x: x['continuation']['reloadContinuationData']), dict)
+ if not next_continuation:
+ return
+ continuation = next_continuation.get('continuation')
+ if not continuation:
+ return
+ ctp = next_continuation.get('clickTrackingParams')
+ return cls._build_api_continuation_query(continuation, ctp)
+
+ @classmethod
+ def _extract_continuation_ep_data(cls, continuation_ep: dict):
+ if isinstance(continuation_ep, dict):
+ continuation = try_get(
+ continuation_ep, lambda x: x['continuationCommand']['token'], compat_str)
+ if not continuation:
+ return
+ ctp = continuation_ep.get('clickTrackingParams')
+ return cls._build_api_continuation_query(continuation, ctp)
+
+ @classmethod
+ def _extract_continuation(cls, renderer):
+ next_continuation = cls._extract_next_continuation_data(renderer)
+ if next_continuation:
+ return next_continuation
+
+ contents = []
+ for key in ('contents', 'items'):
+ contents.extend(try_get(renderer, lambda x: x[key], list) or [])
+
+ for content in contents:
+ if not isinstance(content, dict):
+ continue
+ continuation_ep = try_get(
+ content, (lambda x: x['continuationItemRenderer']['continuationEndpoint'],
+ lambda x: x['continuationItemRenderer']['button']['buttonRenderer']['command']),
+ dict)
+ continuation = cls._extract_continuation_ep_data(continuation_ep)
+ if continuation:
+ return continuation
+
+ @classmethod
+ def _extract_alerts(cls, data):
for alert_dict in try_get(data, lambda x: x['alerts'], list) or []:
if not isinstance(alert_dict, dict):
continue
alert_type = alert.get('type')
if not alert_type:
continue
- message = try_get(alert, lambda x: x['text']['simpleText'], compat_str) or ''
- if message:
- yield alert_type, message
- for run in try_get(alert, lambda x: x['text']['runs'], list) or []:
- message += try_get(run, lambda x: x['text'], compat_str)
+ message = cls._get_text(alert.get('text'))
if message:
yield alert_type, message
def _extract_and_report_alerts(self, data, *args, **kwargs):
return self._report_alerts(self._extract_alerts(data), *args, **kwargs)
+ def _extract_badges(self, renderer: dict):
+ badges = set()
+ for badge in try_get(renderer, lambda x: x['badges'], list) or []:
+ label = try_get(badge, lambda x: x['metadataBadgeRenderer']['label'], compat_str)
+ if label:
+ badges.add(label.lower())
+ return badges
+
+ @staticmethod
+ def _get_text(data, getter=None, max_runs=None):
+ for get in variadic(getter):
+ d = try_get(data, get) if get is not None else data
+ text = try_get(d, lambda x: x['simpleText'], compat_str)
+ if text:
+ return text
+ runs = try_get(d, lambda x: x['runs'], list) or []
+ if not runs and isinstance(d, list):
+ runs = d
+
+ def get_runs(runs):
+ for run in runs[:min(len(runs), max_runs or len(runs))]:
+ yield try_get(run, lambda x: x['text'], compat_str) or ''
+
+ text = ''.join(get_runs(runs))
+ if text:
+ return text
+
def _extract_response(self, item_id, query, note='Downloading API JSON', headers=None,
ytcfg=None, check_get_keys=None, ep='browse', fatal=True, api_hostname=None,
default_client='WEB'):
def _extract_video(self, renderer):
video_id = renderer.get('videoId')
- title = try_get(
- renderer,
- (lambda x: x['title']['runs'][0]['text'],
- lambda x: x['title']['simpleText']), compat_str)
- description = try_get(
- renderer, lambda x: x['descriptionSnippet']['runs'][0]['text'],
- compat_str)
- duration = parse_duration(try_get(
- renderer, lambda x: x['lengthText']['simpleText'], compat_str))
- view_count_text = try_get(
- renderer, lambda x: x['viewCountText']['simpleText'], compat_str) or ''
+ title = self._get_text(renderer.get('title'))
+ description = self._get_text(renderer.get('descriptionSnippet'))
+ duration = parse_duration(self._get_text(renderer.get('lengthText')))
+ view_count_text = self._get_text(renderer.get('viewCountText')) or ''
view_count = str_to_int(self._search_regex(
r'^([\d,]+)', re.sub(r'\s', '', view_count_text),
'view count', default=None))
- uploader = try_get(
- renderer,
- (lambda x: x['ownerText']['runs'][0]['text'],
- lambda x: x['shortBylineText']['runs'][0]['text']), compat_str)
+
+ uploader = self._get_text(renderer, (lambda x: x['ownerText'], lambda x: x['shortBylineText']))
+
return {
'_type': 'url',
'ie_key': YoutubeIE.ie_key(),
if len(time_text_split) >= 3:
return datetime_from_str('now-%s%s' % (time_text_split[0], time_text_split[1]), precision='auto')
- @staticmethod
- def _join_text_entries(runs):
- text = None
- for run in runs:
- if not isinstance(run, dict):
- continue
- sub_text = try_get(run, lambda x: x['text'], compat_str)
- if sub_text:
- if not text:
- text = sub_text
- continue
- text += sub_text
- return text
-
def _extract_comment(self, comment_renderer, parent=None):
comment_id = comment_renderer.get('commentId')
if not comment_id:
return
- comment_text_runs = try_get(comment_renderer, lambda x: x['contentText']['runs']) or []
- text = self._join_text_entries(comment_text_runs) or ''
- comment_time_text = try_get(comment_renderer, lambda x: x['publishedTimeText']['runs']) or []
- time_text = self._join_text_entries(comment_time_text)
- timestamp = calendar.timegm(self.parse_time_text(time_text).timetuple())
- author = try_get(comment_renderer, lambda x: x['authorText']['simpleText'], compat_str)
+
+ text = self._get_text(comment_renderer.get('contentText'))
+
+ # note: timestamp is an estimate calculated from the current time and time_text
+ time_text = self._get_text(comment_renderer.get('publishedTimeText')) or ''
+ time_text_dt = self.parse_time_text(time_text)
+ if isinstance(time_text_dt, datetime.datetime):
+ timestamp = calendar.timegm(time_text_dt.timetuple())
+ author = self._get_text(comment_renderer.get('authorText'))
author_id = try_get(comment_renderer,
lambda x: x['authorEndpoint']['browseEndpoint']['browseId'], compat_str)
- votes = str_to_int(try_get(comment_renderer, (lambda x: x['voteCount']['simpleText'],
- lambda x: x['likeCount']), compat_str)) or 0
+
+ votes = parse_count(try_get(comment_renderer, (lambda x: x['voteCount']['simpleText'],
+ lambda x: x['likeCount']), compat_str)) or 0
author_thumbnail = try_get(comment_renderer,
lambda x: x['authorThumbnail']['thumbnails'][-1]['url'], compat_str)
author_is_uploader = try_get(comment_renderer, lambda x: x['authorIsChannelOwner'], bool)
- is_liked = try_get(comment_renderer, lambda x: x['isLiked'], bool)
+ is_favorited = 'creatorHeart' in (try_get(
+ comment_renderer, lambda x: x['actionButtons']['commentActionButtonsRenderer'], dict) or {})
return {
'id': comment_id,
'text': text,
'timestamp': timestamp,
'time_text': time_text,
'like_count': votes,
- 'is_favorited': is_liked,
+ 'is_favorited': is_favorited,
'author': author,
'author_id': author_id,
'author_thumbnail': author_thumbnail,
}
def _comment_entries(self, root_continuation_data, identity_token, account_syncid,
- ytcfg, session_token_list, parent=None, comment_counts=None):
+ ytcfg, video_id, parent=None, comment_counts=None):
- def extract_thread(parent_renderer):
- contents = try_get(parent_renderer, lambda x: x['contents'], list) or []
+ def extract_header(contents):
+ _total_comments = 0
+ _continuation = None
+ for content in contents:
+ comments_header_renderer = try_get(content, lambda x: x['commentsHeaderRenderer'])
+ expected_comment_count = parse_count(self._get_text(
+ comments_header_renderer, (lambda x: x['countText'], lambda x: x['commentsCount']), max_runs=1))
+
+ if expected_comment_count:
+ comment_counts[1] = expected_comment_count
+ self.to_screen('Downloading ~%d comments' % expected_comment_count)
+ _total_comments = comment_counts[1]
+ sort_mode_str = self._configuration_arg('comment_sort', [''])[0]
+ comment_sort_index = int(sort_mode_str != 'top') # 1 = new, 0 = top
+
+ sort_menu_item = try_get(
+ comments_header_renderer,
+ lambda x: x['sortMenu']['sortFilterSubMenuRenderer']['subMenuItems'][comment_sort_index], dict) or {}
+ sort_continuation_ep = sort_menu_item.get('serviceEndpoint') or {}
+
+ _continuation = self._extract_continuation_ep_data(sort_continuation_ep) or self._extract_continuation(sort_menu_item)
+ if not _continuation:
+ continue
+
+ sort_text = sort_menu_item.get('title')
+ if isinstance(sort_text, compat_str):
+ sort_text = sort_text.lower()
+ else:
+ sort_text = 'top comments' if comment_sort_index == 0 else 'newest first'
+ self.to_screen('Sorting comments by %s' % sort_text)
+ break
+ return _total_comments, _continuation
+
+ def extract_thread(contents):
if not parent:
comment_counts[2] = 0
for content in contents:
comment_counts[2] += 1
comment_entries_iter = self._comment_entries(
comment_replies_renderer, identity_token, account_syncid, ytcfg,
- parent=comment.get('id'), session_token_list=session_token_list,
- comment_counts=comment_counts)
+ video_id, parent=comment.get('id'), comment_counts=comment_counts)
for reply_comment in comment_entries_iter:
yield reply_comment
+ # YouTube comments have a max depth of 2
+ max_depth = int_or_none(self._configuration_arg('max_comment_depth', [''])[0]) or float('inf')
+ if max_depth == 1 and parent:
+ return
if not comment_counts:
# comment so far, est. total comments, current comment thread #
comment_counts = [0, 0, 0]
- # TODO: Generalize the download code with TabIE
- 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:
- first_continuation = True
+ continuation = self._extract_continuation(root_continuation_data)
+ if continuation and len(continuation['continuation']) < 27:
+ self.write_debug('Detected old API continuation token. Generating new API compatible token.')
+ continuation_token = self._generate_comment_continuation(video_id)
+ continuation = self._build_api_continuation_query(continuation_token, None)
+
+ visitor_data = None
+ is_first_continuation = 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.get_param('extractor_retries', 3)
- count = -1
- last_error = None
-
- while count < retries:
- count += 1
- if last_error:
- self.report_warning('%s. Retrying ...' % last_error)
- try:
- query = {
- 'ctoken': continuation['ctoken'],
- 'pbj': 1,
- 'type': 'next',
- }
- if 'itct' in continuation:
- query['itct'] = continuation['itct']
- if parent:
- query['action_get_comment_replies'] = 1
- else:
- query['action_get_comments'] = 1
-
- comment_prog_str = '(%d/%d)' % (comment_counts[0], comment_counts[1])
- if page_num == 0:
- if first_continuation:
- note_prefix = 'Downloading initial comment continuation page'
- else:
- note_prefix = ' Downloading comment reply thread %d %s' % (comment_counts[2], comment_prog_str)
- else:
- note_prefix = '%sDownloading comment%s page %d %s' % (
- ' ' if parent else '',
- ' replies' if parent else '',
- page_num,
- comment_prog_str)
-
- browse = self._download_json(
- 'https://www.youtube.com/comment_service_ajax', None,
- '%s %s' % (note_prefix, '(retry #%d)' % count if count else ''),
- headers=headers, query=query,
- data=urlencode_postdata({
- 'session_token': session_token_list[0]
- }))
- except ExtractorError as e:
- if isinstance(e.cause, compat_HTTPError) and e.cause.code in (500, 503, 404, 413):
- if e.cause.code == 413:
- self.report_warning('Assumed end of comments (received HTTP Error 413)')
- return
- # 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 e.cause.code == 404:
- last_error = last_error + ' (this API is probably deprecated)'
- if count < retries:
- continue
- raise
+ comment_prog_str = '(%d/%d)' % (comment_counts[0], comment_counts[1])
+ if page_num == 0:
+ if is_first_continuation:
+ note_prefix = 'Downloading comment section API JSON'
else:
- session_token = try_get(browse, lambda x: x['xsrf_token'], compat_str)
- if session_token:
- session_token_list[0] = session_token
-
- response = try_get(browse,
- (lambda x: x['response'],
- lambda x: x[1]['response']), dict) or {}
-
- if response.get('continuationContents'):
- break
-
- # YouTube sometimes gives reload: now json if something went wrong (e.g. bad auth)
- if isinstance(browse, dict):
- if browse.get('reload'):
- raise ExtractorError('Invalid or missing params in continuation request', expected=False)
-
- # TODO: not tested, merged from old extractor
- err_msg = browse.get('externalErrorMessage')
- if err_msg:
- last_error = err_msg
- continue
-
- response_error = try_get(response, lambda x: x['responseContext']['errors']['error'][0], dict) or {}
- err_msg = response_error.get('externalErrorMessage')
- if err_msg:
- last_error = err_msg
- continue
-
- # Youtube sometimes sends incomplete data
- # See: https://github.com/ytdl-org/youtube-dl/issues/28194
- last_error = 'Incomplete data received'
- if count >= retries:
- raise ExtractorError(last_error)
+ note_prefix = ' Downloading comment API JSON reply thread %d %s' % (
+ comment_counts[2], comment_prog_str)
+ else:
+ note_prefix = '%sDownloading comment%s API JSON page %d %s' % (
+ ' ' if parent else '', ' replies' if parent else '',
+ page_num, comment_prog_str)
+ response = self._extract_response(
+ item_id=None, query=continuation,
+ ep='next', ytcfg=ytcfg, headers=headers, note=note_prefix,
+ check_get_keys=('onResponseReceivedEndpoints', 'continuationContents'))
if not response:
break
visitor_data = try_get(
lambda x: x['responseContext']['webResponseContextExtensionData']['ytConfigData']['visitorData'],
compat_str) or visitor_data
- known_continuation_renderers = {
- 'itemSectionContinuation': extract_thread,
- 'commentRepliesContinuation': extract_thread
- }
+ continuation_contents = dict_get(response, ('onResponseReceivedEndpoints', 'continuationContents'))
- # extract next root continuation from the results
- continuation_contents = try_get(
- response, lambda x: x['continuationContents'], dict) or {}
-
- for key, value in continuation_contents.items():
- if key not in known_continuation_renderers:
- continue
- continuation_renderer = value
-
- if first_continuation:
- first_continuation = False
- expected_comment_count = try_get(
- continuation_renderer,
- (lambda x: x['header']['commentsHeaderRenderer']['countText']['runs'][0]['text'],
- lambda x: x['header']['commentsHeaderRenderer']['commentsCount']['runs'][0]['text']),
- compat_str)
-
- if expected_comment_count:
- comment_counts[1] = str_to_int(expected_comment_count)
- self.to_screen('Downloading ~%d comments' % str_to_int(expected_comment_count))
- yield comment_counts[1]
-
- # TODO: cli arg.
- # 1/True for newest, 0/False for popular (default)
- comment_sort_index = int(True)
- sort_continuation_renderer = try_get(
- continuation_renderer,
- lambda x: x['header']['commentsHeaderRenderer']['sortMenu']['sortFilterSubMenuRenderer']['subMenuItems']
- [comment_sort_index]['continuation']['reloadContinuationData'], dict)
- # If this fails, the initial continuation page
- # starts off with popular anyways.
- if sort_continuation_renderer:
- continuation = YoutubeTabIE._build_continuation_query(
- continuation=sort_continuation_renderer.get('continuation'),
- ctp=sort_continuation_renderer.get('clickTrackingParams'))
- self.to_screen('Sorting comments by %s' % ('popular' if comment_sort_index == 0 else 'newest'))
+ continuation = None
+ if isinstance(continuation_contents, list):
+ for continuation_section in continuation_contents:
+ if not isinstance(continuation_section, dict):
+ continue
+ continuation_items = try_get(
+ continuation_section,
+ (lambda x: x['reloadContinuationItemsCommand']['continuationItems'],
+ lambda x: x['appendContinuationItemsAction']['continuationItems']),
+ list) or []
+ if is_first_continuation:
+ total_comments, continuation = extract_header(continuation_items)
+ if total_comments:
+ yield total_comments
+ is_first_continuation = False
+ if continuation:
+ break
+ continue
+ count = 0
+ for count, entry in enumerate(extract_thread(continuation_items)):
+ yield entry
+ continuation = self._extract_continuation({'contents': continuation_items})
+ if continuation:
+ # Sometimes YouTube provides a continuation without any comments
+ # In most cases we end up just downloading these with very little comments to come.
+ if count == 0:
+ if not parent:
+ self.report_warning('No comments received - assuming end of comments')
+ continuation = None
break
- for entry in known_continuation_renderers[key](continuation_renderer):
- yield entry
+ # Deprecated response structure
+ elif isinstance(continuation_contents, dict):
+ known_continuation_renderers = ('itemSectionContinuation', 'commentRepliesContinuation')
+ for key, continuation_renderer in continuation_contents.items():
+ if key not in known_continuation_renderers:
+ continue
+ if not isinstance(continuation_renderer, dict):
+ continue
+ if is_first_continuation:
+ header_continuation_items = [continuation_renderer.get('header') or {}]
+ total_comments, continuation = extract_header(header_continuation_items)
+ if total_comments:
+ yield total_comments
+ is_first_continuation = False
+ if continuation:
+ break
- continuation = YoutubeTabIE._extract_continuation(continuation_renderer) # TODO
- break
+ # Sometimes YouTube provides a continuation without any comments
+ # In most cases we end up just downloading these with very little comments to come.
+ count = 0
+ for count, entry in enumerate(extract_thread(continuation_renderer.get('contents') or {})):
+ yield entry
+ continuation = self._extract_continuation(continuation_renderer)
+ if count == 0:
+ if not parent:
+ self.report_warning('No comments received - assuming end of comments')
+ continuation = None
+ break
- def _extract_comments(self, ytcfg, video_id, contents, webpage, xsrf_token):
+ @staticmethod
+ def _generate_comment_continuation(video_id):
+ """
+ Generates initial comment section continuation token from given video id
+ """
+ b64_vid_id = base64.b64encode(bytes(video_id.encode('utf-8')))
+ parts = ('Eg0SCw==', b64_vid_id, 'GAYyJyIRIgs=', b64_vid_id, 'MAB4AjAAQhBjb21tZW50cy1zZWN0aW9u')
+ new_continuation_intlist = list(itertools.chain.from_iterable(
+ [bytes_to_intlist(base64.b64decode(part)) for part in parts]))
+ return base64.b64encode(intlist_to_bytes(new_continuation_intlist)).decode('utf-8')
+
+ def _extract_comments(self, ytcfg, video_id, contents, webpage):
"""Entry for comment extraction"""
+ def _real_comment_extract(contents):
+ if isinstance(contents, list):
+ for entry in contents:
+ for key, renderer in entry.items():
+ if key not in known_entry_comment_renderers:
+ continue
+ yield from self._comment_entries(
+ renderer, video_id=video_id, ytcfg=ytcfg,
+ identity_token=self._extract_identity_token(webpage, item_id=video_id),
+ account_syncid=self._extract_account_syncid(ytcfg))
+ break
comments = []
- known_entry_comment_renderers = (
- 'itemSectionRenderer',
- )
+ known_entry_comment_renderers = ('itemSectionRenderer',)
estimated_total = 0
- for entry in contents:
- for key, renderer in entry.items():
- if key not in known_entry_comment_renderers:
- continue
-
- comment_iter = self._comment_entries(
- 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])
+ max_comments = int_or_none(self._configuration_arg('max_comments', [''])[0]) or float('inf')
- for comment in comment_iter:
- if isinstance(comment, int):
- estimated_total = comment
- continue
- comments.append(comment)
- break
+ try:
+ for comment in _real_comment_extract(contents):
+ if len(comments) >= max_comments:
+ break
+ if isinstance(comment, int):
+ estimated_total = comment
+ continue
+ comments.append(comment)
+ except KeyboardInterrupt:
+ self.to_screen('Interrupted by user')
self.to_screen('Downloaded %d/%d comments' % (len(comments), estimated_total))
return {
'comments': comments,
}
@staticmethod
- def _get_video_info_params(video_id):
- return {
+ def _get_video_info_params(video_id, client='TVHTML5'):
+ GVI_CLIENTS = {
+ 'ANDROID': {
+ 'c': 'ANDROID',
+ 'cver': '16.20',
+ },
+ 'TVHTML5': {
+ 'c': 'TVHTML5',
+ 'cver': '6.20180913',
+ }
+ }
+ query = {
'video_id': video_id,
'eurl': 'https://youtube.googleapis.com/v/' + video_id,
- 'html5': '1',
- 'c': 'TVHTML5',
- 'cver': '6.20180913',
+ 'html5': '1'
}
+ query.update(GVI_CLIENTS.get(client))
+ return query
def _real_extract(self, url):
url, smuggled_data = unsmuggle_url(url, {})
ytcfg = self._extract_ytcfg(video_id, webpage) or self._get_default_ytcfg()
identity_token = self._extract_identity_token(webpage, video_id)
- syncid = self._extract_account_syncid(ytcfg)
- headers = self._generate_api_headers(ytcfg, identity_token, syncid)
-
+ session_index = self._extract_session_index(ytcfg)
player_url = self._extract_player_url(ytcfg, webpage)
- player_client = (self._configuration_arg('player_client') or [''])[0]
+ player_client = self._configuration_arg('player_client', [''])[0]
if player_client not in ('web', 'android', ''):
- self.report_warning(f'Invalid player_client {player_client} given. Falling back to WEB')
- force_mobile_client = player_client == 'android'
+ self.report_warning(f'Invalid player_client {player_client} given. Falling back to android client.')
+ force_mobile_client = player_client != 'web'
player_skip = self._configuration_arg('player_skip')
+ player_response = None
+ if webpage:
+ player_response = self._extract_yt_initial_variable(
+ webpage, self._YT_INITIAL_PLAYER_RESPONSE_RE,
+ video_id, 'initial player response')
- def get_text(x):
- if not x:
- return
- text = x.get('simpleText')
- if text and isinstance(text, compat_str):
- return text
- runs = x.get('runs')
- if not isinstance(runs, list):
- return
- return ''.join([r['text'] for r in runs if isinstance(r.get('text'), compat_str)])
+ syncid = self._extract_account_syncid(ytcfg, player_response)
+ headers = self._generate_api_headers(ytcfg, identity_token, syncid, session_index=session_index)
ytm_streaming_data = {}
if is_music_url:
if sts and not force_mobile_client and 'configs' not in player_skip:
ytm_webpage = self._download_webpage(
'https://music.youtube.com',
- video_id, fatal=False, note="Downloading remix client config")
+ video_id, fatal=False, note='Downloading remix client config')
ytm_cfg = self._extract_ytcfg(video_id, ytm_webpage) or {}
ytm_client = 'WEB_REMIX'
# Android client already has signature descrambled
# See: https://github.com/TeamNewPipe/NewPipeExtractor/issues/562
if not sts:
- self.report_warning('Falling back to mobile remix client for player API.')
+ self.report_warning('Falling back to android remix client for player API.')
ytm_client = 'ANDROID_MUSIC'
ytm_cfg = {}
ytm_headers = self._generate_api_headers(
ytm_cfg, identity_token, syncid,
- client=ytm_client)
+ client=ytm_client, session_index=session_index)
ytm_query = {'videoId': video_id}
ytm_query.update(self._generate_player_context(sts))
item_id=video_id, ep='player', query=ytm_query,
ytcfg=ytm_cfg, headers=ytm_headers, fatal=False,
default_client=ytm_client,
- note='Downloading %sremix player API JSON' % ('mobile ' if force_mobile_client else ''))
-
- ytm_streaming_data = try_get(ytm_player_response, lambda x: x['streamingData']) or {}
- player_response = None
- if webpage:
- player_response = self._extract_yt_initial_variable(
- webpage, self._YT_INITIAL_PLAYER_RESPONSE_RE,
- video_id, 'initial player response')
+ note='Downloading %sremix player API JSON' % ('android ' if force_mobile_client else ''))
+ ytm_streaming_data = try_get(ytm_player_response, lambda x: x['streamingData'], dict) or {}
if not player_response or force_mobile_client:
sts = self._extract_signature_timestamp(video_id, player_url, ytcfg, fatal=False)
# Android client already has signature descrambled
# See: https://github.com/TeamNewPipe/NewPipeExtractor/issues/562
if not sts:
- self.report_warning('Falling back to mobile client for player API.')
+ self.report_warning('Falling back to android client for player API.')
yt_client = 'ANDROID'
ytpcfg = {}
- ytp_headers = self._generate_api_headers(ytpcfg, identity_token, syncid, yt_client)
+ ytp_headers = self._generate_api_headers(ytpcfg, identity_token, syncid,
+ client=yt_client, session_index=session_index)
yt_query = {'videoId': video_id}
yt_query.update(self._generate_player_context(sts))
item_id=video_id, ep='player', query=yt_query,
ytcfg=ytpcfg, headers=ytp_headers, fatal=False,
default_client=yt_client,
- note='Downloading %splayer API JSON' % ('mobile ' if force_mobile_client else '')
- )
+ note='Downloading %splayer API JSON' % ('android ' if force_mobile_client else '')
+ ) or player_response
# Age-gate workarounds
playability_status = player_response.get('playabilityStatus') or {}
if playability_status.get('reason') in self._AGE_GATE_REASONS:
- pr = self._parse_json(try_get(compat_parse_qs(
- self._download_webpage(
- base_url + 'get_video_info', video_id,
- 'Refetching age-gated info webpage', 'unable to download video info webpage',
- query=self._get_video_info_params(video_id), fatal=False)),
- lambda x: x['player_response'][0],
- compat_str) or '{}', video_id)
+ gvi_clients = ('ANDROID', 'TVHTML5') if force_mobile_client else ('TVHTML5', 'ANDROID')
+ for gvi_client in gvi_clients:
+ pr = self._parse_json(try_get(compat_parse_qs(
+ self._download_webpage(
+ base_url + 'get_video_info', video_id,
+ 'Refetching age-gated %s info webpage' % gvi_client.lower(),
+ 'unable to download video info webpage', fatal=False,
+ query=self._get_video_info_params(video_id, client=gvi_client))),
+ lambda x: x['player_response'][0],
+ compat_str) or '{}', video_id)
+ if pr:
+ break
if not pr:
self.report_warning('Falling back to embedded-only age-gate workaround.')
embed_webpage = None
# See: https://github.com/TeamNewPipe/NewPipeExtractor/issues/562
if not sts:
self.report_warning(
- 'Falling back to mobile embedded client for player API (note: some formats may be missing).')
+ 'Falling back to android embedded client for player API (note: some formats may be missing).')
yt_client = 'ANDROID_EMBEDDED_PLAYER'
ytcfg_age = {}
ytage_headers = self._generate_api_headers(
- ytcfg_age, identity_token, syncid, client=yt_client)
+ ytcfg_age, identity_token, syncid,
+ client=yt_client, session_index=session_index)
yt_age_query = {'videoId': video_id}
yt_age_query.update(self._generate_player_context(sts))
pr = self._extract_response(
item_id=video_id, ep='player', query=yt_age_query,
ytcfg=ytcfg_age, headers=ytage_headers, fatal=False,
default_client=yt_client,
- note='Downloading %sage-gated player API JSON' % ('mobile ' if force_mobile_client else '')
+ note='Downloading %sage-gated player API JSON' % ('android ' if force_mobile_client else '')
) or {}
if pr:
lambda x: x['microformat']['playerMicroformatRenderer'],
dict) or {}
video_title = video_details.get('title') \
- or get_text(microformat.get('title')) \
+ or self._get_text(microformat.get('title')) \
or search_meta(['og:title', 'twitter:title', 'title'])
video_description = video_details.get('shortDescription')
formats, itags, stream_ids = [], [], []
itag_qualities = {}
q = qualities([
+ # "tiny" is the smallest video-only format. But some audio-only formats
+ # was also labeled "tiny". It is not clear if such formats still exist
'tiny', 'audio_quality_low', 'audio_quality_medium', 'audio_quality_high', # Audio only formats
'small', 'medium', 'large', 'hd720', 'hd1080', 'hd1440', 'hd2160', 'hd2880', 'highres'
])
'width': fmt.get('width'),
'language': audio_track.get('id', '').split('.')[0],
}
- mimetype = fmt.get('mimeType')
- if mimetype:
- mobj = re.match(
- r'((?:[^/]+)/(?:[^;]+))(?:;\s*codecs="([^"]+)")?', mimetype)
- if mobj:
- dct['ext'] = mimetype2ext(mobj.group(1))
- dct.update(parse_codecs(mobj.group(2)))
+ mime_mobj = re.match(
+ r'((?:[^/]+)/(?:[^;]+))(?:;\s*codecs="([^"]+)")?', fmt.get('mimeType') or '')
+ if mime_mobj:
+ dct['ext'] = mimetype2ext(mime_mobj.group(1))
+ dct.update(parse_codecs(mime_mobj.group(2)))
+ # The 3gp format in android client has a quality of "small",
+ # but is actually worse than all other formats
+ if dct['ext'] == '3gp':
+ dct['quality'] = q('tiny')
no_audio = dct.get('acodec') == 'none'
no_video = dct.get('vcodec') == 'none'
if no_audio:
playability_status,
lambda x: x['errorScreen']['playerErrorMessageRenderer'],
dict) or {}
- reason = get_text(pemr.get('reason')) or playability_status.get('reason')
+ reason = self._get_text(pemr.get('reason')) or playability_status.get('reason')
subreason = pemr.get('subreason')
if subreason:
- subreason = clean_html(get_text(subreason))
+ subreason = clean_html(self._get_text(subreason))
if subreason == 'The uploader has not made this video available in your country.':
countries = microformat.get('availableCountries')
if not countries:
continue
process_language(
subtitles, base_url, lang_code,
- try_get(caption_track, lambda x: x.get('name').get('simpleText')),
+ try_get(caption_track, lambda x: x['name']['simpleText']),
{})
continue
automatic_captions = {}
continue
process_language(
automatic_captions, base_url, translation_language_code,
- try_get(translation_language, (
- lambda x: x['languageName']['simpleText'],
- lambda x: x['languageName']['runs'][0]['text'])),
+ self._get_text(translation_language.get('languageName'), max_runs=1),
{'tlang': translation_language_code})
info['automatic_captions'] = automatic_captions
info['subtitles'] = subtitles
def chapter_time(mmlir):
return parse_duration(
- get_text(mmlir.get('timeDescription')))
+ self._get_text(mmlir.get('timeDescription')))
chapters = []
for next_num, content in enumerate(contents, start=1):
chapters.append({
'start_time': start_time,
'end_time': end_time,
- 'title': get_text(mmlir.get('title')),
+ 'title': self._get_text(mmlir.get('title')),
})
if chapters:
break
if vpir:
stl = vpir.get('superTitleLink')
if stl:
- stl = get_text(stl)
+ stl = self._get_text(stl)
if try_get(
vpir,
lambda x: x['superTitleIcon']['iconType']) == 'LOCATION_PIN':
})
vsir = content.get('videoSecondaryInfoRenderer')
if vsir:
- info['channel'] = get_text(try_get(
+ info['channel'] = self._get_text(try_get(
vsir,
lambda x: x['owner']['videoOwnerRenderer']['title'],
dict))
mrr_title = mrr.get('title')
if not mrr_title:
continue
- mrr_title = get_text(mrr['title'])
- mrr_contents_text = get_text(mrr['contents'][0])
+ mrr_title = self._get_text(mrr['title'])
+ mrr_contents_text = self._get_text(mrr['contents'][0])
if mrr_title == 'License':
info['license'] = mrr_contents_text
elif not multiple_songs:
if initial_data and is_private is not None:
is_membersonly = False
is_premium = False
- contents = try_get(initial_data, lambda x: x['contents']['twoColumnWatchNextResults']['results']['results']['contents'], list)
- for content in contents or []:
- badges = try_get(content, lambda x: x['videoPrimaryInfoRenderer']['badges'], list)
- for badge in badges or []:
- label = try_get(badge, lambda x: x['metadataBadgeRenderer']['label']) or ''
- if label.lower() == 'members only':
- is_membersonly = True
- break
- elif label.lower() == 'premium':
- is_premium = True
- break
- if is_membersonly or is_premium:
- break
+ contents = try_get(initial_data, lambda x: x['contents']['twoColumnWatchNextResults']['results']['results']['contents'], list) or []
+ badge_labels = set()
+ for content in contents:
+ if not isinstance(content, dict):
+ continue
+ badge_labels.update(self._extract_badges(content.get('videoPrimaryInfoRenderer')))
+ for badge_label in badge_labels:
+ if badge_label.lower() == 'members only':
+ is_membersonly = True
+ elif badge_label.lower() == 'premium':
+ is_premium = True
+ elif badge_label.lower() == 'unlisted':
+ is_unlisted = True
- # TODO: Add this for playlists
info['availability'] = self._availability(
is_private=is_private,
needs_premium=is_premium,
data=urlencode_postdata({xsrf_field_name: xsrf_token}))
if get_comments:
- info['__post_extractor'] = lambda: self._extract_comments(ytcfg, video_id, contents, webpage, xsrf_token)
+ info['__post_extractor'] = lambda: self._extract_comments(ytcfg, video_id, contents, webpage)
self.mark_watched(video_id, player_response)
'title': 'Album - Royalty Free Music Library V2 (50 Songs)',
},
'playlist_count': 50,
+ }, {
+ 'note': 'unlisted single video playlist',
+ 'url': 'https://www.youtube.com/playlist?list=PLwL24UFy54GrB3s2KMMfjZscDi1x5Dajf',
+ 'info_dict': {
+ 'uploader_id': 'UC9zHu_mHU96r19o-wV5Qs1Q',
+ 'uploader': 'colethedj',
+ 'id': 'PLwL24UFy54GrB3s2KMMfjZscDi1x5Dajf',
+ 'title': 'yt-dlp unlisted playlist test',
+ 'availability': 'unlisted'
+ },
+ 'playlist_count': 1,
}]
@classmethod
renderer = self._extract_basic_item_renderer(item)
if not isinstance(renderer, dict):
continue
- title = try_get(
- renderer, (lambda x: x['title']['runs'][0]['text'],
- lambda x: x['title']['simpleText']), compat_str)
+ title = self._get_text(renderer.get('title'))
+
# playlist
playlist_id = renderer.get('playlistId')
if playlist_id:
# channel
channel_id = renderer.get('channelId')
if channel_id:
- title = try_get(
- renderer, lambda x: x['title']['simpleText'], compat_str)
yield self.url_result(
'https://www.youtube.com/channel/%s' % channel_id,
ie=YoutubeTabIE.ie_key(), video_title=title)
# will not work
if skip_channels and '/channels?' in shelf_url:
return
- title = try_get(
- shelf_renderer, lambda x: x['title']['runs'][0]['text'], compat_str)
+ title = self._get_text(shelf_renderer, lambda x: x['title'])
yield self.url_result(shelf_url, video_title=title)
# Shelf may not contain shelf URL, fallback to extraction from content
for entry in self._shelf_entries_from_content(shelf_renderer):
if entry:
yield entry
'''
-
- @staticmethod
- def _build_continuation_query(continuation, ctp=None):
- query = {
- 'ctoken': continuation,
- 'continuation': continuation,
- }
- if ctp:
- query['itct'] = ctp
- return query
-
- @staticmethod
- def _extract_next_continuation_data(renderer):
- next_continuation = try_get(
- renderer, lambda x: x['continuations'][0]['nextContinuationData'], dict)
- if not next_continuation:
- return
- continuation = next_continuation.get('continuation')
- if not continuation:
- return
- ctp = next_continuation.get('clickTrackingParams')
- return YoutubeTabIE._build_continuation_query(continuation, ctp)
-
- @classmethod
- def _extract_continuation(cls, renderer):
- next_continuation = cls._extract_next_continuation_data(renderer)
- if next_continuation:
- return next_continuation
- contents = []
- for key in ('contents', 'items'):
- contents.extend(try_get(renderer, lambda x: x[key], list) or [])
- for content in contents:
- if not isinstance(content, dict):
- continue
- continuation_ep = try_get(
- content, lambda x: x['continuationItemRenderer']['continuationEndpoint'],
- dict)
- if not continuation_ep:
- continue
- continuation = try_get(
- continuation_ep, lambda x: x['continuationCommand']['token'], compat_str)
- if not continuation:
- continue
- ctp = continuation_ep.get('clickTrackingParams')
- return YoutubeTabIE._build_continuation_query(continuation, ctp)
-
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
for entry in extract_entries(parent_renderer):
yield entry
continuation = continuation_list[0]
- context = self._extract_context(ytcfg)
- visitor_data = try_get(context, lambda x: x['client']['visitorData'], compat_str)
+ visitor_data = None
for page_num in itertools.count(1):
if not continuation:
break
- 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,
+ query=continuation, headers=headers, ytcfg=ytcfg,
check_get_keys=('continuationContents', 'onResponseReceivedActions', 'onResponseReceivedEndpoints'))
if not response:
else:
raise ExtractorError('Unable to find selected tab')
- @staticmethod
- def _extract_uploader(data):
+ @classmethod
+ def _extract_uploader(cls, data):
uploader = {}
- sidebar_renderer = try_get(
- data, lambda x: x['sidebar']['playlistSidebarRenderer']['items'], list)
- if sidebar_renderer:
- for item in sidebar_renderer:
- if not isinstance(item, dict):
- continue
- renderer = item.get('playlistSidebarSecondaryInfoRenderer')
- if not isinstance(renderer, dict):
- continue
- owner = try_get(
- renderer, lambda x: x['videoOwner']['videoOwnerRenderer']['title']['runs'][0], dict)
- if owner:
- uploader['uploader'] = owner.get('text')
- uploader['uploader_id'] = try_get(
- owner, lambda x: x['navigationEndpoint']['browseEndpoint']['browseId'], compat_str)
- uploader['uploader_url'] = urljoin(
- 'https://www.youtube.com/',
- try_get(owner, lambda x: x['navigationEndpoint']['browseEndpoint']['canonicalBaseUrl'], compat_str))
+ renderer = cls._extract_sidebar_info_renderer(data, 'playlistSidebarSecondaryInfoRenderer') or {}
+ owner = try_get(
+ renderer, lambda x: x['videoOwner']['videoOwnerRenderer']['title']['runs'][0], dict)
+ if owner:
+ uploader['uploader'] = owner.get('text')
+ uploader['uploader_id'] = try_get(
+ owner, lambda x: x['navigationEndpoint']['browseEndpoint']['browseId'], compat_str)
+ uploader['uploader_url'] = urljoin(
+ 'https://www.youtube.com/',
+ try_get(owner, lambda x: x['navigationEndpoint']['browseEndpoint']['canonicalBaseUrl'], compat_str))
return {k: v for k, v in uploader.items() if v is not None}
def _extract_from_tabs(self, item_id, webpage, data, tabs):
thumbnails_list = (
try_get(renderer, lambda x: x['avatar']['thumbnails'], list)
or try_get(
- data,
- lambda x: x['sidebar']['playlistSidebarRenderer']['items'][0]['playlistSidebarPrimaryInfoRenderer']['thumbnailRenderer']['playlistVideoThumbnailRenderer']['thumbnail']['thumbnails'],
+ self._extract_sidebar_info_renderer(data, 'playlistSidebarPrimaryInfoRenderer'),
+ lambda x: x['thumbnailRenderer']['playlistVideoThumbnailRenderer']['thumbnail']['thumbnails'],
list)
or [])
or playlist_id)
title += format_field(selected_tab, 'title', ' - %s')
title += format_field(selected_tab, 'expandedText', ' - %s')
-
metadata = {
'playlist_id': playlist_id,
'playlist_title': title,
'thumbnails': thumbnails,
'tags': tags,
}
+ availability = self._extract_availability(data)
+ if availability:
+ metadata['availability'] = availability
if not channel_id:
metadata.update(self._extract_uploader(data))
metadata.update({
'channel': metadata['uploader'],
'channel_id': metadata['uploader_id'],
'channel_url': metadata['uploader_url']})
+ ytcfg = self._extract_ytcfg(item_id, webpage)
return self.playlist_result(
self._entries(
selected_tab, playlist_id,
self._extract_identity_token(webpage, item_id),
- self._extract_account_syncid(data),
- self._extract_ytcfg(item_id, webpage)),
+ self._extract_account_syncid(ytcfg, data), ytcfg),
**metadata)
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))
+ ytcfg, account_syncid=self._extract_account_syncid(ytcfg, data),
+ identity_token=self._extract_identity_token(webpage, item_id=playlist_id))
for page_num in itertools.count(1):
videos = list(self._playlist_entries(playlist))
if not videos:
}
response = self._extract_response(
item_id='%s page %d' % (playlist_id, page_num),
- query=query,
- ep='next',
- headers=headers,
+ query=query, ep='next', headers=headers, ytcfg=ytcfg,
check_get_keys='contents'
)
playlist = try_get(
self._extract_mix_playlist(playlist, playlist_id, data, webpage),
playlist_id=playlist_id, playlist_title=title)
+ def _extract_availability(self, data):
+ """
+ Gets the availability of a given playlist/tab.
+ Note: Unless YouTube tells us explicitly, we do not assume it is public
+ @param data: response
+ """
+ is_private = is_unlisted = None
+ renderer = self._extract_sidebar_info_renderer(data, 'playlistSidebarPrimaryInfoRenderer') or {}
+ badge_labels = self._extract_badges(renderer)
+
+ # Personal playlists, when authenticated, have a dropdown visibility selector instead of a badge
+ privacy_dropdown_entries = try_get(
+ renderer, lambda x: x['privacyForm']['dropdownFormFieldRenderer']['dropdown']['dropdownRenderer']['entries'], list) or []
+ for renderer_dict in privacy_dropdown_entries:
+ is_selected = try_get(
+ renderer_dict, lambda x: x['privacyDropdownItemRenderer']['isSelected'], bool) or False
+ if not is_selected:
+ continue
+ label = self._get_text(
+ try_get(renderer_dict, lambda x: x['privacyDropdownItemRenderer']['label'], dict) or [])
+ if label:
+ badge_labels.add(label.lower())
+ break
+
+ for badge_label in badge_labels:
+ if badge_label == 'unlisted':
+ is_unlisted = True
+ elif badge_label == 'private':
+ is_private = True
+ elif badge_label == 'public':
+ is_unlisted = is_private = False
+ return self._availability(is_private, False, False, False, is_unlisted)
+
+ @staticmethod
+ def _extract_sidebar_info_renderer(data, info_renderer, expected_type=dict):
+ sidebar_renderer = try_get(
+ data, lambda x: x['sidebar']['playlistSidebarRenderer']['items'], list) or []
+ for item in sidebar_renderer:
+ renderer = try_get(item, lambda x: x[info_renderer], expected_type)
+ if renderer:
+ return renderer
+
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):
+ renderer = self._extract_sidebar_info_renderer(data, 'playlistSidebarPrimaryInfoRenderer')
+ if not renderer:
+ return
+ 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
- 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
+ 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')
+ ytcfg = self._extract_ytcfg(item_id, webpage)
+ headers = self._generate_api_headers(
+ ytcfg, account_syncid=self._extract_account_syncid(ytcfg, data),
+ 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, ytcfg=ytcfg,
+ note='Downloading API JSON with unavailable videos')
def _extract_webpage(self, url, item_id):
retries = self.get_param('extractor_retries', 3)
if 'no-youtube-unavailable-videos' not in compat_opts:
data = self._reload_with_unavailable_videos(item_id, data, webpage) or data
self._extract_and_report_alerts(data)
-
tabs = try_get(
data, lambda x: x['contents']['twoColumnBrowseResultsRenderer']['tabs'], list)
if tabs:
if self._SEARCH_PARAMS:
data['params'] = self._SEARCH_PARAMS
total = 0
+ continuation = {}
for page_num in itertools.count(1):
+ data.update(continuation)
search = self._extract_response(
item_id='query "%s" page %s' % (query, page_num), ep='search', query=data,
check_get_keys=('contents', 'onResponseReceivedCommands')
# Youtube sometimes adds promoted content to searches,
# changing the index location of videos and token.
# So we search through all entries till we find them.
- continuation_token = None
+ continuation = None
for slr_content in slr_contents:
- if continuation_token is None:
- continuation_token = try_get(
- slr_content,
- lambda x: x['continuationItemRenderer']['continuationEndpoint']['continuationCommand']['token'],
- compat_str)
+ if not continuation:
+ continuation = self._extract_continuation({'contents': [slr_content]})
isr_contents = try_get(
slr_content,
if total == n:
return
- if not continuation_token:
+ if not continuation:
break
- data['continuation'] = continuation_token
def _get_n_results(self, query, n):
"""Get a specified number of results for a query"""