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

Code before:
                        metadata['ride']['vod_stream_url'],
                        compat_urllib_parse.quote(compat_urllib_parse.quote(token))))
            elif metadata['ride'].get('live_stream_url'):
                url = self._MANIFEST_URL_TEMPLATE % (metadata['ride'].get('live_stream_url'), compat_urllib_parse.quote(token))
                is_live = True
            else:
                raise ExtractorError('Missing video URL')
            formats = self._extract_m3u8_formats(url, video_id, 'mp4')
            if formats:
                for i, f in enumerate(formats, start=2):
                    if f.get('vcodec') == 'none':
                        f['source_preference'] = -i
                    else:
                        break
                self._sort_formats(formats)
            ext = 'mp4'
            m3u8_doc = self._download_webpage(url, video_id, note='Downloading m3u8', fatal=False)
            if m3u8_doc:
                for line in m3u8_doc.splitlines():
                    if line.startswith('#EXT-X-MEDIA:'):
                        media = parse_m3u8_attributes(line)
                        if media.get('TYPE') == 'SUBTITLES':
                            subtitles.setdefault(media.get('LANGUAGE'), []).append({
                                'url': media.get('URI'),
                                'ext': 'vtt'
                            })

        if metadata.get('instructor_cues'):
            subtitles['cues'] = [{
                'data': json.dumps(metadata.get('instructor_cues')),
                'ext': 'json'
            }]

        release_date = None
        if metadata['ride'].get('original_air_time'):
            release_datetime = datetime.datetime.utcfromtimestamp(metadata['ride'].get('original_air_time'))
            if release_datetime:
                release_date = release_datetime.strftime('%Y%m%d')
        creator = None
        if metadata['ride'].get('instructor') and metadata['ride'].get('instructor').get('name'):
            creator = metadata['ride'].get('instructor').get('name')
        categories = None
        if metadata['ride'].get('fitness_discipline_display_name'):
            categories = [metadata['ride'].get('fitness_discipline_display_name')]
        tags = None
        if metadata['ride'].get('equipment_tags'):
            tags = []
            for equipment in metadata['ride'].get('equipment_tags'):
                if equipment.get('name'):
                    tags.append(equipment.get('name'))
        chapters = None

Review comment (refactor):
It would be nice if this can be generalized and made available in `common.py`. There are other extractors that can probably benefit from this

Code after:
                self._start_session(video_id)
            else:
                raise

        metadata = self._download_json('https://api.onepeloton.com/api/ride/%s/details?stream_source=multichannel' % video_id, video_id)
        ride_data = metadata.get('ride')
        if not ride_data:
            raise ExtractorError('Missing stream metadata')
        token = self._get_token(video_id)

        is_live = False
        if ride_data.get('content_format') == 'audio':
            url = self._MANIFEST_URL_TEMPLATE % (ride_data.get('vod_stream_url'), compat_urllib_parse.quote(token))
            formats = [{
                'url': url,
                'ext': 'm4a',
                'format_id': 'audio',
                'vcodec': 'none',
            }]
            subtitles = {}
        else:
            if ride_data.get('vod_stream_url'):
                url = 'https://members.onepeloton.com/.netlify/functions/m3u8-proxy?displayLanguage=en&acceptedSubtitles=%s&url=%s?hdnea=%s' % (
                    ','.join([re.sub('^([a-z]+)-([A-Z]+)$', r'\1', caption) for caption in ride_data['captions']]),
                    ride_data['vod_stream_url'],
                    compat_urllib_parse.quote(compat_urllib_parse.quote(token)))
            elif ride_data.get('live_stream_url'):
                url = self._MANIFEST_URL_TEMPLATE % (ride_data.get('live_stream_url'), compat_urllib_parse.quote(token))
                is_live = True
            else:
                raise ExtractorError('Missing video URL')
            formats, subtitles = self._extract_m3u8_formats_and_subtitles(url, video_id, 'mp4')

        if metadata.get('instructor_cues'):
            subtitles['cues'] = [{
                'data': json.dumps(metadata.get('instructor_cues')),
                'ext': 'json'
            }]

        category = ride_data.get('fitness_discipline_display_name')
        chapters = [{
            'start_time': segment.get('start_time_offset'),
            'end_time': segment.get('start_time_offset') + segment.get('length'),
            'title': segment.get('name')
        } for segment in traverse_obj(metadata, ('segments', 'segment_list'))]

        self._sort_formats(formats)
        return {
            'id': video_id,
            'title': ride_data.get('title'),
            'formats': formats,