diff options
Diffstat (limited to 'youtube/yt_data_extract/watch_extraction.py')
| -rw-r--r-- | youtube/yt_data_extract/watch_extraction.py | 323 |
1 files changed, 248 insertions, 75 deletions
diff --git a/youtube/yt_data_extract/watch_extraction.py b/youtube/yt_data_extract/watch_extraction.py index db53581..e09e2d3 100644 --- a/youtube/yt_data_extract/watch_extraction.py +++ b/youtube/yt_data_extract/watch_extraction.py @@ -2,7 +2,8 @@ from .common import (get, multi_get, deep_get, multi_deep_get, liberal_update, conservative_update, remove_redirect, normalize_url, extract_str, extract_formatted_text, extract_int, extract_approx_int, extract_date, check_missing_keys, extract_item_info, extract_items, - extract_response, concat_or_none) + extract_response, concat_or_none, liberal_dict_update, + conservative_dict_update) import json import urllib.parse @@ -116,7 +117,99 @@ _formats = { '397': {'vcodec': 'av01.0.05M.08'}, } -def _extract_metadata_row_info(video_renderer_info): + +def _extract_from_video_information_renderer(renderer_content): + subtitle = extract_str(renderer_content.get('expandedSubtitle'), + default='') + info = { + 'title': extract_str(renderer_content.get('title')), + 'view_count': extract_int(subtitle), + 'unlisted': False, + 'live': 'watching' in subtitle, + } + for badge in renderer_content.get('badges', []): + if deep_get(badge, 'metadataBadgeRenderer', 'label') == 'Unlisted': + info['unlisted'] = True + return info + +def _extract_likes_dislikes(renderer_content): + def extract_button_count(toggle_button_renderer): + # all the digits can be found in the accessibility data + count = extract_int(multi_deep_get( + toggle_button_renderer, + ['defaultText', 'accessibility', 'accessibilityData', 'label'], + ['accessibility', 'label'], + ['accessibilityData', 'accessibilityData', 'label'], + ['accessibilityText'], + )) + + # this count doesn't have all the digits, it's like 53K for instance + dumb_count = extract_int(extract_str(multi_get( + toggle_button_renderer, ['defaultText', 'title']))) + + # The accessibility text will be "No likes" or "No dislikes" or + # something like that, but dumb count will be 0 + if dumb_count == 0: + count = 0 + return count + + info = { + 'like_count': None, + 'dislike_count': None, + } + for button in renderer_content.get('buttons', ()): + if 'slimMetadataToggleButtonRenderer' in button: + button_renderer = button['slimMetadataToggleButtonRenderer'] + count = extract_button_count(deep_get(button_renderer, + 'button', + 'toggleButtonRenderer')) + if 'isLike' in button_renderer: + info['like_count'] = count + elif 'isDislike' in button_renderer: + info['dislike_count'] = count + elif 'slimMetadataButtonRenderer' in button: + button_renderer = button['slimMetadataButtonRenderer'] + liberal_update(info, 'like_count', extract_button_count( + multi_deep_get(button_renderer, + ['button', 'segmentedLikeDislikeButtonRenderer', + 'likeButton', 'toggleButtonRenderer'], + ['button', 'segmentedLikeDislikeButtonViewModel', + 'likeButtonViewModel', 'likeButtonViewModel', + 'toggleButtonViewModel', 'toggleButtonViewModel', + 'defaultButtonViewModel', 'buttonViewModel'] + ) + )) + '''liberal_update(info, 'dislike_count', extract_button_count( + deep_get( + button_renderer, 'button', + 'segmentedLikeDislikeButtonRenderer', + 'dislikeButton', 'toggleButtonRenderer' + ) + ))''' + return info + +def _extract_from_owner_renderer(renderer_content): + return { + 'author': extract_str(renderer_content.get('title')), + 'author_id': deep_get( + renderer_content, + 'navigationEndpoint', 'browseEndpoint', 'browseId'), + } + +def _extract_from_video_header_renderer(renderer_content): + return { + 'title': extract_str(renderer_content.get('title')), + 'time_published': extract_date(extract_str( + renderer_content.get('publishDate'))), + } + +def _extract_from_description_renderer(renderer_content): + return { + 'description': extract_str( + renderer_content.get('descriptionBodyText'), recover_urls=True), + } + +def _extract_metadata_row_info(renderer_content): # extract category and music list info = { 'category': None, @@ -124,7 +217,7 @@ def _extract_metadata_row_info(video_renderer_info): } current_song = {} - for row in deep_get(video_renderer_info, 'metadataRowContainer', 'metadataRowContainerRenderer', 'rows', default=[]): + for row in deep_get(renderer_content, 'rows', default=[]): row_title = extract_str(deep_get(row, 'metadataRowRenderer', 'title'), default='') row_content = extract_str(deep_get(row, 'metadataRowRenderer', 'contents', 0)) if row_title == 'Category': @@ -146,18 +239,69 @@ def _extract_metadata_row_info(video_renderer_info): return info -def _extract_watch_info_mobile(top_level): - info = {} - microformat = deep_get(top_level, 'playerResponse', 'microformat', 'playerMicroformatRenderer', default={}) +def _extract_from_music_renderer(renderer_content): + # latest format for the music list + info = { + 'music_list': [], + } - family_safe = microformat.get('isFamilySafe') - if family_safe is None: - info['age_restricted'] = None - else: - info['age_restricted'] = not family_safe - info['allowed_countries'] = microformat.get('availableCountries', []) - info['time_published'] = microformat.get('publishDate') + for carousel in renderer_content.get('carouselLockups', []): + song = {} + carousel = carousel.get('carouselLockupRenderer', {}) + video_renderer = carousel.get('videoLockup', {}) + video_renderer_info = extract_item_info(video_renderer) + video_id = video_renderer_info.get('id') + song['url'] = concat_or_none('https://www.youtube.com/watch?v=', + video_id) + song['title'] = video_renderer_info.get('title') + for row in carousel.get('infoRows', []): + row = row.get('infoRowRenderer', {}) + title = extract_str(row.get('title')) + data = extract_str(row.get('defaultMetadata')) + if title == 'SONG': + song['title'] = data + elif title == 'ARTIST': + song['artist'] = data + elif title == 'ALBUM': + song['album'] = data + elif title == 'WRITERS': + song['writers'] = data + info['music_list'].append(song) + return info +def _extract_from_video_metadata(renderer_content): + info = _extract_from_video_information_renderer(renderer_content) + liberal_dict_update(info, _extract_likes_dislikes(renderer_content)) + liberal_dict_update(info, _extract_from_owner_renderer(renderer_content)) + liberal_dict_update(info, _extract_metadata_row_info(deep_get( + renderer_content, 'metadataRowContainer', + 'metadataRowContainerRenderer', default={} + ))) + liberal_update(info, 'title', extract_str(renderer_content.get('title'))) + liberal_update( + info, 'description', + extract_str(renderer_content.get('description'), recover_urls=True) + ) + liberal_update(info, 'time_published', + extract_date(renderer_content.get('dateText'))) + return info + +visible_extraction_dispatch = { + # Either these ones spread around in various places + 'slimVideoInformationRenderer': _extract_from_video_information_renderer, + 'slimVideoActionBarRenderer': _extract_likes_dislikes, + 'slimOwnerRenderer': _extract_from_owner_renderer, + 'videoDescriptionHeaderRenderer': _extract_from_video_header_renderer, + 'videoDescriptionMusicSectionRenderer': _extract_from_music_renderer, + 'expandableVideoDescriptionRenderer': _extract_from_description_renderer, + 'metadataRowContainerRenderer': _extract_metadata_row_info, + # OR just this one, which contains SOME of the above inside it + 'slimVideoMetadataRenderer': _extract_from_video_metadata, +} + +def _extract_watch_info_mobile(top_level): + '''Scrapes information from the visible page''' + info = {} response = top_level.get('response', {}) # this renderer has the stuff visible on the page @@ -190,47 +334,24 @@ def _extract_watch_info_mobile(top_level): else: info['playlist'] = None - # Holds the visible video info. It is inside singleColumnWatchNextResults - # but use our convenience function instead - items, _ = extract_items(response, item_types={'slimVideoMetadataRenderer'}) - if items: - video_info = items[0]['slimVideoMetadataRenderer'] - else: - print('Failed to extract video metadata') - video_info = {} - - info.update(_extract_metadata_row_info(video_info)) - info['description'] = extract_str(video_info.get('description'), recover_urls=True) - info['view_count'] = extract_int(extract_str(video_info.get('expandedSubtitle'))) - info['author'] = extract_str(deep_get(video_info, 'owner', 'slimOwnerRenderer', 'title')) - info['author_id'] = deep_get(video_info, 'owner', 'slimOwnerRenderer', 'navigationEndpoint', 'browseEndpoint', 'browseId') - info['title'] = extract_str(video_info.get('title')) - info['live'] = 'watching' in extract_str(video_info.get('expandedSubtitle'), default='') - info['unlisted'] = False - for badge in video_info.get('badges', []): - if deep_get(badge, 'metadataBadgeRenderer', 'label') == 'Unlisted': - info['unlisted'] = True - info['like_count'] = None - info['dislike_count'] = None - if not info['time_published']: - info['time_published'] = extract_date(extract_str(video_info.get('dateText', None))) - for button in video_info.get('buttons', ()): - button_renderer = button.get('slimMetadataToggleButtonRenderer', {}) - - # all the digits can be found in the accessibility data - count = extract_int(deep_get(button_renderer, 'button', 'toggleButtonRenderer', 'defaultText', 'accessibility', 'accessibilityData', 'label')) - - # this count doesn't have all the digits, it's like 53K for instance - dumb_count = extract_int(extract_str(deep_get(button_renderer, 'button', 'toggleButtonRenderer', 'defaultText'))) - - # the accessibility text will be "No likes" or "No dislikes" or something like that, but dumb count will be 0 - if dumb_count == 0: - count = 0 - - if 'isLike' in button_renderer: - info['like_count'] = count - elif 'isDislike' in button_renderer: - info['dislike_count'] = count + # use dispatch table to get information scattered in various renderers + items, _ = extract_items( + response, + item_types=visible_extraction_dispatch.keys(), + search_engagement_panels=True + ) + found = set() + for renderer in items: + name, renderer_content = list(renderer.items())[0] + found.add(name) + liberal_dict_update( + info, + visible_extraction_dispatch[name](renderer_content) + ) + # Call the function on blank dict for any that weren't found + # so that the empty keys get added + for name in visible_extraction_dispatch.keys() - found: + liberal_dict_update(info, visible_extraction_dispatch[name]({})) # comment section info items, _ = extract_items(response, item_types={ @@ -244,17 +365,18 @@ def _extract_watch_info_mobile(top_level): # https://www.androidpolice.com/2019/10/31/google-youtube-app-comment-section-below-videos/ # https://www.youtube.com/watch?v=bR5Q-wD-6qo if header_type == 'commentsEntryPointHeaderRenderer': - comment_count_text = extract_str(comment_info.get('headerText')) + comment_count_text = extract_str(multi_get( + comment_info, 'commentCount', 'headerText')) else: comment_count_text = extract_str(deep_get(comment_info, 'header', 'commentSectionHeaderRenderer', 'countText')) if comment_count_text == 'Comments': # just this with no number, means 0 comments - info['comment_count'] = 0 + info['comment_count'] = '0' else: - info['comment_count'] = extract_int(comment_count_text) + info['comment_count'] = extract_approx_int(comment_count_text) info['comments_disabled'] = False else: # no comment section present means comments are disabled - info['comment_count'] = 0 + info['comment_count'] = '0' info['comments_disabled'] = True # check for limited state @@ -274,7 +396,6 @@ def _extract_watch_info_desktop(top_level): info = { 'comment_count': None, 'comments_disabled': None, - 'allowed_countries': [], 'limited_state': None, 'playlist': None, } @@ -307,26 +428,28 @@ def _extract_watch_info_desktop(top_level): return info def update_format_with_codec_info(fmt, codec): - if (codec.startswith('av') - or codec in ('vp9', 'vp8', 'vp8.0', 'h263', 'h264', 'mp4v')): + if any(codec.startswith(c) for c in ('av', 'vp', 'h263', 'h264', 'mp4v')): if codec == 'vp8.0': codec = 'vp8' conservative_update(fmt, 'vcodec', codec) elif (codec.startswith('mp4a') - or codec in ('opus', 'mp3', 'aac', 'dtse', 'ec-3', 'vorbis')): + or codec in ('opus', 'mp3', 'aac', 'dtse', 'ec-3', 'vorbis', + 'ac-3')): conservative_update(fmt, 'acodec', codec) else: print('Warning: unrecognized codec: ' + codec) fmt_type_re = re.compile( - r'(text|audio|video)/([\w0-9]+); codecs="([\w0-9\.]+(?:, [\w0-9\.]+)*)"') + r'(text|audio|video)/([\w0-9]+); codecs="([^"]+)"') def update_format_with_type_info(fmt, yt_fmt): # 'type' for invidious api format mime_type = multi_get(yt_fmt, 'mimeType', 'type') if mime_type is None: return match = re.fullmatch(fmt_type_re, mime_type) - + if match is None: + print('Warning: Could not read mimetype', mime_type) + return type, fmt['ext'], codecs = match.groups() codecs = codecs.split(', ') for codec in codecs: @@ -349,17 +472,32 @@ def _extract_formats(info, player_response): for yt_fmt in yt_formats: itag = yt_fmt.get('itag') + # Translated audio track + # Example: https://www.youtube.com/watch?v=gF9kkB0UWYQ + # Only get the original language for now so a foreign + # translation will not be picked just because it comes first + if deep_get(yt_fmt, 'audioTrack', 'audioIsDefault') is False: + continue + fmt = {} fmt['itag'] = itag fmt['ext'] = None fmt['audio_bitrate'] = None + fmt['bitrate'] = yt_fmt.get('bitrate') fmt['acodec'] = None fmt['vcodec'] = None fmt['width'] = yt_fmt.get('width') fmt['height'] = yt_fmt.get('height') - fmt['file_size'] = yt_fmt.get('contentLength') - fmt['audio_sample_rate'] = yt_fmt.get('audioSampleRate') + fmt['file_size'] = extract_int(yt_fmt.get('contentLength')) + fmt['audio_sample_rate'] = extract_int(yt_fmt.get('audioSampleRate')) + fmt['duration_ms'] = yt_fmt.get('approxDurationMs') fmt['fps'] = yt_fmt.get('fps') + fmt['init_range'] = yt_fmt.get('initRange') + fmt['index_range'] = yt_fmt.get('indexRange') + for key in ('init_range', 'index_range'): + if fmt[key]: + fmt[key]['start'] = int(fmt[key]['start']) + fmt[key]['end'] = int(fmt[key]['end']) update_format_with_type_info(fmt, yt_fmt) cipher = dict(urllib.parse.parse_qsl(multi_get(yt_fmt, 'cipher', 'signatureCipher', default=''))) @@ -373,8 +511,16 @@ def _extract_formats(info, player_response): # update with information from big table hardcoded_itag_info = _formats.get(str(itag), {}) for key, value in hardcoded_itag_info.items(): - conservative_update(fmt, key, value) # prefer info from Youtube + conservative_update(fmt, key, value) # prefer info from YouTube fmt['quality'] = hardcoded_itag_info.get('height') + conservative_update( + fmt, 'quality', + extract_int(yt_fmt.get('quality'), whole_word=False) + ) + conservative_update( + fmt, 'quality', + extract_int(yt_fmt.get('qualityLabel'), whole_word=False) + ) info['formats'].append(fmt) @@ -397,7 +543,7 @@ def extract_hls_formats(hls_manifest): if lines[i].startswith('#EXT-X-STREAM-INF'): fmt = {'acodec': None, 'vcodec': None, 'height': None, 'width': None, 'fps': None, 'audio_bitrate': None, - 'itag': None, 'file_size': None, + 'itag': None, 'file_size': None, 'duration_ms': None, 'audio_sample_rate': None, 'url': None} properties = lines[i].split(':')[1] properties += ',' # make regex work for last key-value pair @@ -484,6 +630,25 @@ def extract_watch_info(polymer_json): info['translation_languages'] = [] captions_info = player_response.get('captions', {}) info['_captions_base_url'] = normalize_url(deep_get(captions_info, 'playerCaptionsRenderer', 'baseUrl')) + # Sometimes the above playerCaptionsRender is randomly missing + # Extract base_url from one of the captions by removing lang specifiers + if not info['_captions_base_url']: + base_url = normalize_url(deep_get( + captions_info, + 'playerCaptionsTracklistRenderer', + 'captionTracks', + 0, + 'baseUrl' + )) + if base_url: + url_parts = urllib.parse.urlparse(base_url) + qs = urllib.parse.parse_qs(url_parts.query) + for key in ('tlang', 'lang', 'name', 'kind', 'fmt'): + if key in qs: + del qs[key] + base_url = urllib.parse.urlunparse(url_parts._replace( + query=urllib.parse.urlencode(qs, doseq=True))) + info['_captions_base_url'] = base_url for caption_track in deep_get(captions_info, 'playerCaptionsTracklistRenderer', 'captionTracks', default=()): lang_code = caption_track.get('languageCode') if not lang_code: @@ -564,9 +729,17 @@ def extract_watch_info(polymer_json): liberal_update(info, 'category', mf.get('category')) liberal_update(info, 'time_published', mf.get('publishDate')) liberal_update(info, 'time_uploaded', mf.get('uploadDate')) + family_safe = mf.get('isFamilySafe') + if family_safe is None: + conservative_update(info, 'age_restricted', None) + else: + conservative_update(info, 'age_restricted', not family_safe) + info['allowed_countries'] = mf.get('availableCountries', []) # other stuff info['author_url'] = 'https://www.youtube.com/channel/' + info['author_id'] if info['author_id'] else None + info['storyboard_spec_url'] = deep_get(player_response, 'storyboards', 'playerStoryboardSpecRenderer', 'spec') + return info single_char_codes = { @@ -646,10 +819,15 @@ def extract_watch_info_from_html(watch_html): return extract_watch_info(fake_polymer_json) +def captions_available(info): + return bool(info['_captions_base_url']) + def get_caption_url(info, language, format, automatic=False, translation_language=None): '''Gets the url for captions with the given language and format. If automatic is True, get the automatic captions for that language. If translation_language is given, translate the captions from `language` to `translation_language`. If automatic is true and translation_language is given, the automatic captions will be translated.''' url = info['_captions_base_url'] + if not url: + return None url += '&lang=' + language url += '&fmt=' + format if automatic: @@ -661,15 +839,10 @@ def get_caption_url(info, language, format, automatic=False, translation_languag url += '&tlang=' + translation_language return url -def update_with_age_restricted_info(info, video_info_page): - '''Inserts urls from 'player_response' in get_video_info page''' +def update_with_new_urls(info, player_response): + '''Inserts urls from player_response json''' ERROR_PREFIX = 'Error getting missing player or bypassing age-restriction: ' - video_info = urllib.parse.parse_qs(video_info_page) - player_response = deep_get(video_info, 'player_response', 0) - if player_response is None: - info['playability_error'] = ERROR_PREFIX + 'Could not find player_response in video_info_page' - return try: player_response = json.loads(player_response) except json.decoder.JSONDecodeError: |