[yt-dlp.git] / yt_dlp / downloader / youtube_live_chat.py

import json
import time

from .fragment import FragmentFD
from ..compat import compat_urllib_error
from ..extractor.youtube import YoutubeBaseInfoExtractor as YT_BaseIE
from ..utils import RegexNotFoundError, dict_get, int_or_none, try_get


class YoutubeLiveChatFD(FragmentFD):
    """ Downloads YouTube live chats fragment by fragment """

    def real_download(self, filename, info_dict):
        video_id = info_dict['video_id']
        self.to_screen('[%s] Downloading live chat' % self.FD_NAME)
        if not self.params.get('skip_download') and info_dict['protocol'] == 'youtube_live_chat':
            self.report_warning('Live chat download runs until the livestream ends. '
                                'If you wish to download the video simultaneously, run a separate yt-dlp instance')

        fragment_retries = self.params.get('fragment_retries', 0)
        test = self.params.get('test', False)

        ctx = {
            'filename': filename,
            'live': True,
            'total_frags': None,
        }

        ie = YT_BaseIE(self.ydl)

        start_time = int(time.time() * 1000)

        def dl_fragment(url, data=None, headers=None):
            http_headers = info_dict.get('http_headers', {})
            if headers:
                http_headers = http_headers.copy()
                http_headers.update(headers)
            return self._download_fragment(ctx, url, info_dict, http_headers, data)

        def parse_actions_replay(live_chat_continuation):
            offset = continuation_id = click_tracking_params = None
            processed_fragment = bytearray()
            for action in live_chat_continuation.get('actions', []):
                if 'replayChatItemAction' in action:
                    replay_chat_item_action = action['replayChatItemAction']
                    offset = int(replay_chat_item_action['videoOffsetTimeMsec'])
                processed_fragment.extend(
                    json.dumps(action, ensure_ascii=False).encode() + b'\n')
            if offset is not None:
                continuation = try_get(
                    live_chat_continuation,
                    lambda x: x['continuations'][0]['liveChatReplayContinuationData'], dict)
                if continuation:
                    continuation_id = continuation.get('continuation')
                    click_tracking_params = continuation.get('clickTrackingParams')
            self._append_fragment(ctx, processed_fragment)
            return continuation_id, offset, click_tracking_params

        def try_refresh_replay_beginning(live_chat_continuation):
            # choose the second option that contains the unfiltered live chat replay
            refresh_continuation = try_get(
                live_chat_continuation,
                lambda x: x['header']['liveChatHeaderRenderer']['viewSelector']['sortFilterSubMenuRenderer']['subMenuItems'][1]['continuation']['reloadContinuationData'], dict)
            if refresh_continuation:
                # no data yet but required to call _append_fragment
                self._append_fragment(ctx, b'')
                refresh_continuation_id = refresh_continuation.get('continuation')
                offset = 0
                click_tracking_params = refresh_continuation.get('trackingParams')
                return refresh_continuation_id, offset, click_tracking_params
            return parse_actions_replay(live_chat_continuation)

        live_offset = 0

        def parse_actions_live(live_chat_continuation):
            nonlocal live_offset
            continuation_id = click_tracking_params = None
            processed_fragment = bytearray()
            for action in live_chat_continuation.get('actions', []):
                timestamp = self.parse_live_timestamp(action)
                if timestamp is not None:
                    live_offset = timestamp - start_time
                # compatibility with replay format
                pseudo_action = {
                    'replayChatItemAction': {'actions': [action]},
                    'videoOffsetTimeMsec': str(live_offset),
                    'isLive': True,
                }
                processed_fragment.extend(
                    json.dumps(pseudo_action, ensure_ascii=False).encode() + b'\n')
            continuation_data_getters = [
                lambda x: x['continuations'][0]['invalidationContinuationData'],
                lambda x: x['continuations'][0]['timedContinuationData'],
            ]
            continuation_data = try_get(live_chat_continuation, continuation_data_getters, dict)
            if continuation_data:
                continuation_id = continuation_data.get('continuation')
                click_tracking_params = continuation_data.get('clickTrackingParams')
                timeout_ms = int_or_none(continuation_data.get('timeoutMs'))
                if timeout_ms is not None:
                    time.sleep(timeout_ms / 1000)
            self._append_fragment(ctx, processed_fragment)
            return continuation_id, live_offset, click_tracking_params

        def download_and_parse_fragment(url, frag_index, request_data=None, headers=None):
            count = 0
            while count <= fragment_retries:
                try:
                    success = dl_fragment(url, request_data, headers)
                    if not success:
                        return False, None, None, None
                    raw_fragment = self._read_fragment(ctx)
                    try:
                        data = ie.extract_yt_initial_data(video_id, raw_fragment.decode('utf-8', 'replace'))
                    except RegexNotFoundError:
                        data = None
                    if not data:
                        data = json.loads(raw_fragment)
                    live_chat_continuation = try_get(
                        data,
                        lambda x: x['continuationContents']['liveChatContinuation'], dict) or {}
                    if info_dict['protocol'] == 'youtube_live_chat_replay':
                        if frag_index == 1:
                            continuation_id, offset, click_tracking_params = try_refresh_replay_beginning(live_chat_continuation)
                        else:
                            continuation_id, offset, click_tracking_params = parse_actions_replay(live_chat_continuation)
                    elif info_dict['protocol'] == 'youtube_live_chat':
                        continuation_id, offset, click_tracking_params = parse_actions_live(live_chat_continuation)
                    return True, continuation_id, offset, click_tracking_params
                except compat_urllib_error.HTTPError as err:
                    count += 1
                    if count <= fragment_retries:
                        self.report_retry_fragment(err, frag_index, count, fragment_retries)
            if count > fragment_retries:
                self.report_error('giving up after %s fragment retries' % fragment_retries)
                return False, None, None, None

        self._prepare_and_start_frag_download(ctx, info_dict)

        success = dl_fragment(info_dict['url'])
        if not success:
            return False
        raw_fragment = self._read_fragment(ctx)
        try:
            data = ie.extract_yt_initial_data(video_id, raw_fragment.decode('utf-8', 'replace'))
        except RegexNotFoundError:
            return False
        continuation_id = try_get(
            data,
            lambda x: x['contents']['twoColumnWatchNextResults']['conversationBar']['liveChatRenderer']['continuations'][0]['reloadContinuationData']['continuation'])
        # no data yet but required to call _append_fragment
        self._append_fragment(ctx, b'')

        ytcfg = ie.extract_ytcfg(video_id, raw_fragment.decode('utf-8', 'replace'))

        if not ytcfg:
            return False
        api_key = try_get(ytcfg, lambda x: x['INNERTUBE_API_KEY'])
        innertube_context = try_get(ytcfg, lambda x: x['INNERTUBE_CONTEXT'])
        if not api_key or not innertube_context:
            return False
        visitor_data = try_get(innertube_context, lambda x: x['client']['visitorData'], str)
        if info_dict['protocol'] == 'youtube_live_chat_replay':
            url = 'https://www.youtube.com/youtubei/v1/live_chat/get_live_chat_replay?key=' + api_key
            chat_page_url = 'https://www.youtube.com/live_chat_replay?continuation=' + continuation_id
        elif info_dict['protocol'] == 'youtube_live_chat':
            url = 'https://www.youtube.com/youtubei/v1/live_chat/get_live_chat?key=' + api_key
            chat_page_url = 'https://www.youtube.com/live_chat?continuation=' + continuation_id

        frag_index = offset = 0
        click_tracking_params = None
        while continuation_id is not None:
            frag_index += 1
            request_data = {
                'context': innertube_context,
                'continuation': continuation_id,
            }
            if frag_index > 1:
                request_data['currentPlayerState'] = {'playerOffsetMs': str(max(offset - 5000, 0))}
                if click_tracking_params:
                    request_data['context']['clickTracking'] = {'clickTrackingParams': click_tracking_params}
                headers = ie.generate_api_headers(ytcfg=ytcfg, visitor_data=visitor_data)
                headers.update({'content-type': 'application/json'})
                fragment_request_data = json.dumps(request_data, ensure_ascii=False).encode() + b'\n'
                success, continuation_id, offset, click_tracking_params = download_and_parse_fragment(
                    url, frag_index, fragment_request_data, headers)
            else:
                success, continuation_id, offset, click_tracking_params = download_and_parse_fragment(
                    chat_page_url, frag_index)
            if not success:
                return False
            if test:
                break

        self._finish_frag_download(ctx, info_dict)
        return True

    @staticmethod
    def parse_live_timestamp(action):
        action_content = dict_get(
            action,
            ['addChatItemAction', 'addLiveChatTickerItemAction', 'addBannerToLiveChatCommand'])
        if not isinstance(action_content, dict):
            return None
        item = dict_get(action_content, ['item', 'bannerRenderer'])
        if not isinstance(item, dict):
            return None
        renderer = dict_get(item, [
            # text
            'liveChatTextMessageRenderer', 'liveChatPaidMessageRenderer',
            'liveChatMembershipItemRenderer', 'liveChatPaidStickerRenderer',
            # ticker
            'liveChatTickerPaidMessageItemRenderer',
            'liveChatTickerSponsorItemRenderer',
            # banner
            'liveChatBannerRenderer',
        ])
        if not isinstance(renderer, dict):
            return None
        parent_item_getters = [
            lambda x: x['showItemEndpoint']['showLiveChatItemEndpoint']['renderer'],
            lambda x: x['contents'],
        ]
        parent_item = try_get(renderer, parent_item_getters, dict)
        if parent_item:
            renderer = dict_get(parent_item, [
                'liveChatTextMessageRenderer', 'liveChatPaidMessageRenderer',
                'liveChatMembershipItemRenderer', 'liveChatPaidStickerRenderer',
            ])
            if not isinstance(renderer, dict):
                return None
        return int_or_none(renderer.get('timestampUsec'), 1000)
Commit	Line	Data
a78e3a57	1	import json
c60ee3a2	2	import time
a78e3a57	3
a78e3a57	4	from .fragment import FragmentFD
82e3f6eb	5	from ..compat import compat_urllib_error
82e3f6eb	6	from ..extractor.youtube import YoutubeBaseInfoExtractor as YT_BaseIE
f8271158	7	from ..utils import RegexNotFoundError, dict_get, int_or_none, try_get
a78e3a57	8
a78e3a57	9
c60ee3a2	10	class YoutubeLiveChatFD(FragmentFD):
c60ee3a2	11	""" Downloads YouTube live chats fragment by fragment """
a78e3a57	12
a78e3a57	13	def real_download(self, filename, info_dict):
	14	video_id = info_dict['video_id']
	15	self.to_screen('[%s] Downloading live chat' % self.FD_NAME)
592b7485	16	if not self.params.get('skip_download') and info_dict['protocol'] == 'youtube_live_chat':
08d30158	17	self.report_warning('Live chat download runs until the livestream ends. '
08d30158	18	'If you wish to download the video simultaneously, run a separate yt-dlp instance')
a78e3a57	19
82e3f6eb	20	fragment_retries = self.params.get('fragment_retries', 0)
a78e3a57	21	test = self.params.get('test', False)
	22
	23	ctx = {
	24	'filename': filename,
	25	'live': True,
	26	'total_frags': None,
	27	}
	28
273762c8	29	ie = YT_BaseIE(self.ydl)
a78e3a57	30
c60ee3a2	31	start_time = int(time.time() * 1000)
c60ee3a2	32
273762c8	33	def dl_fragment(url, data=None, headers=None):
	34	http_headers = info_dict.get('http_headers', {})
	35	if headers:
	36	http_headers = http_headers.copy()
	37	http_headers.update(headers)
	38	return self._download_fragment(ctx, url, info_dict, http_headers, data)
a78e3a57	39
c60ee3a2	40	def parse_actions_replay(live_chat_continuation):
c2603313	41	offset = continuation_id = click_tracking_params = None
c60ee3a2	42	processed_fragment = bytearray()
	43	for action in live_chat_continuation.get('actions', []):
	44	if 'replayChatItemAction' in action:
	45	replay_chat_item_action = action['replayChatItemAction']
	46	offset = int(replay_chat_item_action['videoOffsetTimeMsec'])
	47	processed_fragment.extend(
0f06bcd7	48	json.dumps(action, ensure_ascii=False).encode() + b'\n')
c60ee3a2	49	if offset is not None:
c2603313	50	continuation = try_get(
c60ee3a2	51	live_chat_continuation,
c2603313	52	lambda x: x['continuations'][0]['liveChatReplayContinuationData'], dict)
	53	if continuation:
	54	continuation_id = continuation.get('continuation')
	55	click_tracking_params = continuation.get('clickTrackingParams')
c60ee3a2	56	self._append_fragment(ctx, processed_fragment)
c2603313	57	return continuation_id, offset, click_tracking_params
c60ee3a2	58
d534c452	59	def try_refresh_replay_beginning(live_chat_continuation):
d534c452	60	# choose the second option that contains the unfiltered live chat replay
c2603313	61	refresh_continuation = try_get(
d534c452	62	live_chat_continuation,
c2603313	63	lambda x: x['header']['liveChatHeaderRenderer']['viewSelector']['sortFilterSubMenuRenderer']['subMenuItems'][1]['continuation']['reloadContinuationData'], dict)
c2603313	64	if refresh_continuation:
d534c452	65	# no data yet but required to call _append_fragment
d534c452	66	self._append_fragment(ctx, b'')
c2603313	67	refresh_continuation_id = refresh_continuation.get('continuation')
	68	offset = 0
	69	click_tracking_params = refresh_continuation.get('trackingParams')
	70	return refresh_continuation_id, offset, click_tracking_params
d534c452	71	return parse_actions_replay(live_chat_continuation)
d534c452	72
c60ee3a2	73	live_offset = 0
	74
	75	def parse_actions_live(live_chat_continuation):
	76	nonlocal live_offset
c2603313	77	continuation_id = click_tracking_params = None
c60ee3a2	78	processed_fragment = bytearray()
	79	for action in live_chat_continuation.get('actions', []):
	80	timestamp = self.parse_live_timestamp(action)
	81	if timestamp is not None:
	82	live_offset = timestamp - start_time
	83	# compatibility with replay format
	84	pseudo_action = {
	85	'replayChatItemAction': {'actions': [action]},
	86	'videoOffsetTimeMsec': str(live_offset),
	87	'isLive': True,
	88	}
	89	processed_fragment.extend(
0f06bcd7	90	json.dumps(pseudo_action, ensure_ascii=False).encode() + b'\n')
c60ee3a2	91	continuation_data_getters = [
	92	lambda x: x['continuations'][0]['invalidationContinuationData'],
	93	lambda x: x['continuations'][0]['timedContinuationData'],
	94	]
	95	continuation_data = try_get(live_chat_continuation, continuation_data_getters, dict)
	96	if continuation_data:
	97	continuation_id = continuation_data.get('continuation')
c2603313	98	click_tracking_params = continuation_data.get('clickTrackingParams')
c60ee3a2	99	timeout_ms = int_or_none(continuation_data.get('timeoutMs'))
	100	if timeout_ms is not None:
	101	time.sleep(timeout_ms / 1000)
	102	self._append_fragment(ctx, processed_fragment)
c2603313	103	return continuation_id, live_offset, click_tracking_params
c60ee3a2	104
d534c452	105	def download_and_parse_fragment(url, frag_index, request_data=None, headers=None):
82e3f6eb	106	count = 0
	107	while count <= fragment_retries:
	108	try:
d71fd412	109	success = dl_fragment(url, request_data, headers)
82e3f6eb	110	if not success:
c2603313	111	return False, None, None, None
d71fd412	112	raw_fragment = self._read_fragment(ctx)
d534c452	113	try:
11f9be09	114	data = ie.extract_yt_initial_data(video_id, raw_fragment.decode('utf-8', 'replace'))
d534c452	115	except RegexNotFoundError:
	116	data = None
	117	if not data:
	118	data = json.loads(raw_fragment)
82e3f6eb	119	live_chat_continuation = try_get(
	120	data,
	121	lambda x: x['continuationContents']['liveChatContinuation'], dict) or {}
d534c452	122	if info_dict['protocol'] == 'youtube_live_chat_replay':
d534c452	123	if frag_index == 1:
c2603313	124	continuation_id, offset, click_tracking_params = try_refresh_replay_beginning(live_chat_continuation)
d534c452	125	else:
c2603313	126	continuation_id, offset, click_tracking_params = parse_actions_replay(live_chat_continuation)
d534c452	127	elif info_dict['protocol'] == 'youtube_live_chat':
c2603313	128	continuation_id, offset, click_tracking_params = parse_actions_live(live_chat_continuation)
c2603313	129	return True, continuation_id, offset, click_tracking_params
82e3f6eb	130	except compat_urllib_error.HTTPError as err:
	131	count += 1
	132	if count <= fragment_retries:
	133	self.report_retry_fragment(err, frag_index, count, fragment_retries)
	134	if count > fragment_retries:
	135	self.report_error('giving up after %s fragment retries' % fragment_retries)
c2603313	136	return False, None, None, None
82e3f6eb	137
3ba7740d	138	self._prepare_and_start_frag_download(ctx, info_dict)
a78e3a57	139
d71fd412	140	success = dl_fragment(info_dict['url'])
a78e3a57	141	if not success:
a78e3a57	142	return False
d71fd412	143	raw_fragment = self._read_fragment(ctx)
273762c8	144	try:
11f9be09	145	data = ie.extract_yt_initial_data(video_id, raw_fragment.decode('utf-8', 'replace'))
273762c8	146	except RegexNotFoundError:
273762c8	147	return False
82e3f6eb	148	continuation_id = try_get(
	149	data,
	150	lambda x: x['contents']['twoColumnWatchNextResults']['conversationBar']['liveChatRenderer']['continuations'][0]['reloadContinuationData']['continuation'])
a78e3a57	151	# no data yet but required to call _append_fragment
	152	self._append_fragment(ctx, b'')
	153
11f9be09	154	ytcfg = ie.extract_ytcfg(video_id, raw_fragment.decode('utf-8', 'replace'))
273762c8	155
	156	if not ytcfg:
	157	return False
	158	api_key = try_get(ytcfg, lambda x: x['INNERTUBE_API_KEY'])
	159	innertube_context = try_get(ytcfg, lambda x: x['INNERTUBE_CONTEXT'])
	160	if not api_key or not innertube_context:
	161	return False
c60ee3a2	162	visitor_data = try_get(innertube_context, lambda x: x['client']['visitorData'], str)
	163	if info_dict['protocol'] == 'youtube_live_chat_replay':
	164	url = 'https://www.youtube.com/youtubei/v1/live_chat/get_live_chat_replay?key=' + api_key
d534c452	165	chat_page_url = 'https://www.youtube.com/live_chat_replay?continuation=' + continuation_id
c60ee3a2	166	elif info_dict['protocol'] == 'youtube_live_chat':
c60ee3a2	167	url = 'https://www.youtube.com/youtubei/v1/live_chat/get_live_chat?key=' + api_key
d534c452	168	chat_page_url = 'https://www.youtube.com/live_chat?continuation=' + continuation_id
273762c8	169
82e3f6eb	170	frag_index = offset = 0
c2603313	171	click_tracking_params = None
a78e3a57	172	while continuation_id is not None:
82e3f6eb	173	frag_index += 1
273762c8	174	request_data = {
	175	'context': innertube_context,
	176	'continuation': continuation_id,
	177	}
	178	if frag_index > 1:
	179	request_data['currentPlayerState'] = {'playerOffsetMs': str(max(offset - 5000, 0))}
c2603313	180	if click_tracking_params:
c2603313	181	request_data['context']['clickTracking'] = {'clickTrackingParams': click_tracking_params}
99e9e001	182	headers = ie.generate_api_headers(ytcfg=ytcfg, visitor_data=visitor_data)
d534c452	183	headers.update({'content-type': 'application/json'})
0f06bcd7	184	fragment_request_data = json.dumps(request_data, ensure_ascii=False).encode() + b'\n'
c2603313	185	success, continuation_id, offset, click_tracking_params = download_and_parse_fragment(
d534c452	186	url, frag_index, fragment_request_data, headers)
d534c452	187	else:
c2603313	188	success, continuation_id, offset, click_tracking_params = download_and_parse_fragment(
c2603313	189	chat_page_url, frag_index)
82e3f6eb	190	if not success:
	191	return False
	192	if test:
a78e3a57	193	break
a78e3a57	194
3ba7740d	195	self._finish_frag_download(ctx, info_dict)
a78e3a57	196	return True
c60ee3a2	197
	198	@staticmethod
	199	def parse_live_timestamp(action):
	200	action_content = dict_get(
	201	action,
	202	['addChatItemAction', 'addLiveChatTickerItemAction', 'addBannerToLiveChatCommand'])
	203	if not isinstance(action_content, dict):
	204	return None
	205	item = dict_get(action_content, ['item', 'bannerRenderer'])
	206	if not isinstance(item, dict):
	207	return None
	208	renderer = dict_get(item, [
	209	# text
	210	'liveChatTextMessageRenderer', 'liveChatPaidMessageRenderer',
	211	'liveChatMembershipItemRenderer', 'liveChatPaidStickerRenderer',
	212	# ticker
	213	'liveChatTickerPaidMessageItemRenderer',
	214	'liveChatTickerSponsorItemRenderer',
	215	# banner
	216	'liveChatBannerRenderer',
	217	])
	218	if not isinstance(renderer, dict):
	219	return None
	220	parent_item_getters = [
	221	lambda x: x['showItemEndpoint']['showLiveChatItemEndpoint']['renderer'],
	222	lambda x: x['contents'],
	223	]
	224	parent_item = try_get(renderer, parent_item_getters, dict)
	225	if parent_item:
	226	renderer = dict_get(parent_item, [
	227	'liveChatTextMessageRenderer', 'liveChatPaidMessageRenderer',
	228	'liveChatMembershipItemRenderer', 'liveChatPaidStickerRenderer',
	229	])
	230	if not isinstance(renderer, dict):
	231	return None
	232	return int_or_none(renderer.get('timestampUsec'), 1000)