Repository: yt-dlp/yt-dlp
File: yt_dlp/extractor/youtube/_video.py (Python)

Code before:
            self.write_debug(f'Detected a {length}s skippable ad')
            return length_ms
        length_ms = traverse_obj(instream_renderer, ('playerVars', {urllib.parse.parse_qs}, 'length_seconds', -1, {int_or_none(invscale=1000)}))
        if length_ms is not None:
            length = length_ms / 1000 if length_ms % 1000 else length_ms // 1000
            self.write_debug(f'Detected a {length}s unskippable ad')
            return length_ms
        return None

    def _get_preroll_length(self, ad_slot_lists):
        for slot_renderer in traverse_obj(ad_slot_lists, ('adSlots', ..., 'adSlotRenderer', {dict})):
            if traverse_obj(slot_renderer, ('adSlotMetadata', 'triggerEvent')) != 'SLOT_TRIGGER_EVENT_BEFORE_CONTENT':
                continue
            rendering_content = traverse_obj(slot_renderer, (
                'fulfillmentContent', 'fulfilledLayout', 'playerBytesAdLayoutRenderer',
                'renderingContent', {dict})) or {}
            if 'instreamVideoAdRenderer' in rendering_content:
                length = self._parse_instream_ad_renderer(rendering_content)
                if length is not None:
                    return length
            if 'playerBytesSequentialLayoutRenderer' in rendering_content:
                total = 0
                for layout in traverse_obj(rendering_content, ('playerBytesSequentialLayoutRenderer', 'sequentialLayouts'), []):
                    total += self._parse_instream_ad_renderer(traverse_obj(layout, ('playerBytesAdLayoutRenderer', 'renderingContent', {dict}))) or 0
                return total
        return 0

    def _list_formats(self, video_id, microformats, video_details, player_responses, player_url, duration=None):
        live_broadcast_details = traverse_obj(microformats, (..., 'liveBroadcastDetails'))
        is_live = get_first(video_details, 'isLive')
        if is_live is None:
            is_live = get_first(live_broadcast_details, 'isLiveNow')
        live_content = get_first(video_details, 'isLiveContent')
        is_upcoming = get_first(video_details, 'isUpcoming')
        post_live = get_first(video_details, 'isPostLiveDvr')
        live_status = ('post_live' if post_live
                       else 'is_live' if is_live
                       else 'is_upcoming' if is_upcoming
                       else 'was_live' if live_content
                       else 'not_live' if False in (is_live, live_content)
                       else None)
        *formats, subtitles = self._extract_formats_and_subtitles(video_id, player_responses, player_url, live_status, duration)
        if all(f.get('has_drm') for f in formats):
            # If there are no formats that definitely don't have DRM, all have DRM
            for f in formats:
                f['has_drm'] = True

        return live_broadcast_details, live_status, formats, subtitles

    def _download_initial_data(self, video_id, webpage, webpage_client, webpage_ytcfg):
        initial_data = None

Review comment (bug):
```suggestion
    def _get_available_at_timestamp(self, player_response, video_id, client):
        now = time.time()
        wait_seconds = 0

        instream_video_ad_renderers = traverse_obj(player_response, (
            'adSlots', lambda _, v: v['adSlotRenderer']['adSlotMetadata']['triggerEvent'] == 'SLOT_TRIGGER_EVENT_BEFORE_CONTENT',
            'adSlotRenderer', 'fulfillmentContent', 'fulfilledLayout', 'playerBytesAdLayoutRenderer', 'renderingContent', (
                None,
                ('playerBytesSequentialLayoutRenderer', 'sequentialLayouts', ..., 'playerBytesAdLayoutRenderer', 'renderingContent'),
            ), 'instreamVideoAdRenderer', {dict}))

        for renderer in instream_video_ad_renderers:
            duration = traverse_obj(renderer, ('skipOffsetMilliseconds', {float_or_none(scale=1000)}))
            if duration is not None:
                duration = duration if duration % 1 else int(duration)
                self.write_debug(f'{video_id}: Detected a {duration}s skippable ad for {client} client')
                # This ad is skippable so only accrue the required waiting time before we can skip
                wait_seconds += duration
                continue
            duration = traverse_obj(renderer, ('playerVars', {urllib.parse.parse_qs}, 'length_seconds', -1, {int_or_none}))
            if duration is not None:
                self.write_debug(f'{video_id}: Detected a {duration}s unskippable ad for {client} client')
                # This ad is unskippable so accrue the full ad duration
                wait_seconds += duration

        if wait_seconds:
            return math.ceil(now) + wait_seconds

        return int(now)
```

Code after:
            duration = traverse_obj(renderer, ('playerVars', {urllib.parse.parse_qs}, 'length_seconds', -1, {int_or_none}))
            ad = 'an ad' if duration is None else f'a {duration}s ad'

            skip_time = traverse_obj(renderer, ('skipOffsetMilliseconds', {float_or_none(scale=1000)}))
            if skip_time is not None:
                # YT allows skipping this ad; use the wait-until-skip time instead of full ad duration
                skip_time = skip_time if skip_time % 1 else int(skip_time)
                ad += f' skippable after {skip_time}s'
                duration = skip_time

            if duration is not None:
                self.write_debug(f'{video_id}: Detected {ad} for {client}')
                wait_seconds += duration

        if wait_seconds:
            return math.ceil(now) + wait_seconds

        return int(now)

    def _list_formats(self, video_id, microformats, video_details, player_responses, player_url, duration=None):
        live_broadcast_details = traverse_obj(microformats, (..., 'liveBroadcastDetails'))
        is_live = get_first(video_details, 'isLive')
        if is_live is None:
            is_live = get_first(live_broadcast_details, 'isLiveNow')
        live_content = get_first(video_details, 'isLiveContent')
        is_upcoming = get_first(video_details, 'isUpcoming')
        post_live = get_first(video_details, 'isPostLiveDvr')
        live_status = ('post_live' if post_live
                       else 'is_live' if is_live
                       else 'is_upcoming' if is_upcoming
                       else 'was_live' if live_content
                       else 'not_live' if False in (is_live, live_content)
                       else None)
        *formats, subtitles = self._extract_formats_and_subtitles(video_id, player_responses, player_url, live_status, duration)
        if all(f.get('has_drm') for f in formats):
            # If there are no formats that definitely don't have DRM, all have DRM
            for f in formats:
                f['has_drm'] = True

        return live_broadcast_details, live_status, formats, subtitles

    def _download_initial_data(self, video_id, webpage, webpage_client, webpage_ytcfg):
        initial_data = None
        if webpage and 'initial_data' not in self._configuration_arg('webpage_skip'):
            initial_data = self.extract_yt_initial_data(video_id, webpage, fatal=False)
            if not traverse_obj(initial_data, 'contents'):
                self.report_warning('Incomplete data received in embedded initial data; re-fetching using API.')
                initial_data = None
        if not initial_data and 'initial_data' not in self._configuration_arg('player_skip'):
            query = {'videoId': video_id}
            query.update(self._get_checkok_params())