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

Code before:
        dmc_data = traverse_obj(api_data, (
            'media', 'delivery', 'movie', {
                'audios': 'audios',
                'videos': 'videos',
                'protocols': ('session', 'protocols'),
            }, {lambda data: data if all(key in data for key in ['audios', 'videos', 'protocols']) else None},
        ))
        if dmc_data:
            for (audio_quality, video_quality, protocol) in itertools.product(
                    dmc_data['audios'], dmc_data['videos'], dmc_data['protocols']):
                fmt = self._extract_format_for_quality(video_id, audio_quality, video_quality, protocol)
                if fmt:
                    formats.append(fmt)

        dms_data = traverse_obj(api_data, ({
            'videos': ('media', 'domand', 'videos', ..., {
                lambda item: item['id'] if item['isAvailable'] else None
            }),
            'audios': ('media', 'domand', 'audios', ..., {
                lambda item: (item['id'], {
                    'format_id': str(parse_bitrate(item['id'])),
                    'abr': float_or_none(item['bitRate'], scale=1000),
                    'asr': item['samplingRate'],
                    'acodec': 'aac',
                    'ext': 'm4a',
                }) if item['isAvailable'] else None
            }),
            'accessRightKey': ('media', 'domand', 'accessRightKey'),
            'track_id': ('client', 'watchTrackId'),
        }, {
            'videos': 'videos',
            'audios': ('audios', {lambda items: dict(items)}),
            'accessRightKey': 'accessRightKey',
            'track_id': 'track_id',
        }, {
            lambda data: data if all(key in data for key in [
                'videos', 'audios', 'accessRightKey', 'track_id',
            ]) else None
        }))
        if dms_data:
            dms_m3u8_url = traverse_obj(self._download_json(
                f'https://nvapi.nicovideo.jp/v1/watch/{video_id}/access-rights/hls', video_id,
                data=json.dumps({
                    'outputs': list(itertools.product(dms_data['videos'], dms_data['audios']))
                }).encode(), query={'actionTrackId': dms_data['track_id']}, headers={
                    'x-access-right-key': dms_data['accessRightKey'],
                    'x-frontend-id': 6,
                    'x-frontend-version': 0,
                    'x-request-with': 'https://www.nicovideo.jp',
                }), ('data', 'contentUrl'))
            dms_fmts = self._extract_m3u8_formats(dms_m3u8_url, video_id)

Review comment (bug):
I think once you have to use a transformation to construct an entire format_dict then it's time to break this up and use separate variables/traversals.

Can you show an example of what the POST data that we send with the API request should look like? Maybe I'm misreading the code, but it looks like we are sending a tuple of the `id` and the format dict, and that seems wrong

Also, you may have found that the transformation is always called regardless of whether or not the path succeeds up to that point, and that's why you included `if item['isAvailable'] else None` in the lambda function instead of just filtering. You handled it correctly here, but IMO we should avoid such complexity in transformations to prevent unintended side effects

Code after:

        # Sort before removing dupes to keep the format dicts with the lowest tbr
        video_fmts = sorted((fmt for fmt in dms_fmts if fmt['vcodec'] != 'none'), key=lambda f: f['tbr'])
        self._remove_duplicate_formats(video_fmts)
        # Calculate the true vbr/tbr by subtracting the lowest abr
        min_abr = min(traverse_obj(audios, (..., 'bitRate', {float_or_none})), default=0) / 1000
        for video_fmt in video_fmts:
            video_fmt['tbr'] -= min_abr
            video_fmt['format_id'] = f'video-{video_fmt["tbr"]:.0f}'
            yield video_fmt

    def _real_extract(self, url):
        video_id = self._match_id(url)

        try:
            webpage, handle = self._download_webpage_handle(
                'https://www.nicovideo.jp/watch/' + video_id, video_id)
            if video_id.startswith('so'):
                video_id = self._match_id(handle.url)

            api_data = self._parse_json(self._html_search_regex(
                'data-api-data="([^"]+)"', webpage,
                'API data', default='{}'), video_id)
        except ExtractorError as e:
            try:
                api_data = self._download_json(
                    'https://www.nicovideo.jp/api/watch/v3/%s?_frontendId=6&_frontendVersion=0&actionTrackId=AAAAAAAAAA_%d' % (video_id, round(time.time() * 1000)), video_id,
                    note='Downloading API JSON', errnote='Unable to fetch data')['data']
            except ExtractorError:
                if not isinstance(e.cause, HTTPError):
                    raise
                webpage = e.cause.response.read().decode('utf-8', 'replace')
                error_msg = self._html_search_regex(
                    r'(?s)<section\s+class="(?:(?:ErrorMessage|WatchExceptionPage-message)\s*)+">(.+?)</section>',
                    webpage, 'error reason', default=None)
                if not error_msg:
                    raise
                raise ExtractorError(clean_html(error_msg), expected=True)

        club_joined = traverse_obj(api_data, ('channel', 'viewer', 'follow', 'isFollowed', {bool}))
        if club_joined is None:
            fail_msg = self._html_search_regex(
                r'<p[^>]+\bclass="fail-message"[^>]*>(?P<msg>.+?)</p>',
                webpage, 'fail message', default=None, group='msg')
            if fail_msg:
                self.raise_login_required(clean_html(fail_msg), metadata_available=True)
        elif not club_joined:
            self.raise_login_required('This video is for members only', metadata_available=True)

        # Start extracting information
        tags = None