Repository: yt-dlp/yt-dlp
File: yt_dlp/downloader/youtube_live_chat.py (Python)

Code before:

        live_offset = [0]  # Python 2 doesnot support nonlocal

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

        if info_dict['protocol'] == 'youtube_live_chat_replay':
            parse_actions = parse_actions_replay
        elif info_dict['protocol'] == 'youtube_live_chat':
            parse_actions = parse_actions_live

        def download_and_parse_fragment(url, frag_index, request_data):
            count = 0
            while count <= fragment_retries:
                try:
                    success, raw_fragment = dl_fragment(url, request_data, {'content-type': 'application/json'})
                    if not success:
                        return False, None, None
                    try:
                        data = ie._extract_yt_initial_data(video_id, raw_fragment.decode('utf-8', 'replace'))
                    except RegexNotFoundError:
                        data = None
                    if not data:

Review comment (suggestion):
Either use a single `try_get`, or something like:
``` py
           continuation_data = dict_get(
               try_get(live_chat_continuation, lambda x: x['continuations'][0], dict),
               ('invalidationContinuationData', 'timedContinuationData'))
```

Code after:
                    live_chat_continuation = try_get(
                        data,
                        lambda x: x['continuationContents']['liveChatContinuation'], dict) or {}
                    continuation_id, offset = parse_actions(live_chat_continuation)
                    return True, continuation_id, offset
                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

        self._prepare_and_start_frag_download(ctx)

        success, raw_fragment = dl_fragment(info_dict['url'])
        if not success:
            return False
        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
        elif info_dict['protocol'] == 'youtube_live_chat':
            url = 'https://www.youtube.com/youtubei/v1/live_chat/get_live_chat?key=' + api_key

        frag_index = offset = 0
        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))}