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

Code before:
        return self._search_regex(r'\"([^"]+)\"', self._download_webpage(
            'https://asobistage-api.asobistore.jp/api/v1/vspf/token', video_id,
            note='Getting token', errnote='Unable to get token'),
            name="token")

    def _get_owned_tickets(self, video_id):
        for url, name in [
            ('https://asobistage-api.asobistore.jp/api/v1/purchase_history/list', 'ticket purchase history'),
            ('https://asobistage-api.asobistore.jp/api/v1/serialcode/list', 'redemption history'),
        ]:
            for id in traverse_obj(self._download_json(
                    url, video_id, note=f'Downloading {name}', errnote=f'Unable to download {name}'),
                    ('payload', 'value', ..., 'digital_product_id', {str})):
                yield id

    def _real_extract(self, url):
        video_id = self._match_id(url)
        video_type_name = self._match_valid_url(url).group('type')
        webpage = self._download_webpage(url, video_id)

        self._check_login(video_id)

        video_type_id, live_status = {
            'archive': ['archives', 'was_live'],
            'player': ['broadcasts', 'is_live'],
        }.get(video_type_name) or [None, None]
        if not video_type_id:
            raise ExtractorError('Unknown video type')

        event_data = self._search_nextjs_data(webpage, video_id)
        event_id = traverse_obj(event_data, ('query', 'event', {str}))
        event_slug = traverse_obj(event_data, ('query', 'player_slug', {str}))
        event_title = traverse_obj(event_data, ('props', 'pageProps', 'eventCMSData', 'event_name', {str}))
        if not all((event_id, event_slug, event_title)):
            raise ExtractorError('Unable to get required event data')

        channel_list_url = functools.reduce(urljoin, [
            self._search_regex(
                r'"(?P<url>https://asobistage\.asobistore\.jp/cdn/[^/]+/)', webpage,
                'cdn endpoint url', group=('url')),
            'events/', f'{event_id}/', f'{video_type_id}.json'])
        channels_json = self._download_json(
            channel_list_url, video_id, fatal=False,
            note='Getting channel list', errnote='Unable to get channel list')
        available_channels = traverse_obj(channels_json, (
            video_type_id, lambda _, v: v['broadcast_slug'] == event_slug, 'channels',
            lambda _, v: v['chennel_vspf_id'] != '00000', {dict}))

        owned_tickets = set(self._get_owned_tickets(video_id))
        if not owned_tickets.intersection(traverse_obj(
                available_channels, (..., 'viewrights', ..., 'tickets', ..., 'digital_product_id', {str_or_none}))):

Review comment (suggestion):
I think I would just
```suggestion
        video_type_id = {
            'archive': 'archives',
            'player': 'broadcasts',
        }.get(video_type_name)
```

Code after:

    @functools.cached_property
    def _owned_tickets(self):
        owned_tickets = set()
        if not self._is_logged_in:
            return owned_tickets

        for path, name in [
            ('api/v1/purchase_history/list', 'ticket purchase history'),
            ('api/v1/serialcode/list', 'redemption history'),
        ]:
            response = self._download_json(
                f'{self._API_HOST}/{path}', None, f'Downloading {name}',
                f'Unable to download {name}', expected_status=400)
            if traverse_obj(response, ('payload', 'error_message'), 'error') == 'notlogin':
                self._is_logged_in = False
                break
            owned_tickets.update(
                traverse_obj(response, ('payload', 'value', ..., 'digital_product_id', {str_or_none})))

        return owned_tickets

    def _get_available_channel_id(self, channel):
        channel_id = traverse_obj(channel, ('chennel_vspf_id', {str}))
        if not channel_id:
            return None
        # if rights_type_id == 6, then 'No conditions (no login required - non-members are OK)'
        if traverse_obj(channel, ('viewrights', lambda _, v: v['rights_type_id'] == 6)):
            return channel_id
        available_tickets = traverse_obj(channel, (
            'viewrights', ..., ('tickets', 'serialcodes'), ..., 'digital_product_id', {str_or_none}))
        if not self._owned_tickets.intersection(available_tickets):
            self.report_warning(
                f'You are not a ticketholder for "{channel.get("channel_name") or channel_id}"')
            return None
        return channel_id

    def _real_initialize(self):
        if self._get_cookies(self._API_HOST):
            self._is_logged_in = True
        token = self._download_json(
            f'{self._API_HOST}/api/v1/vspf/token', None, 'Getting token', 'Unable to get token')
        self._HEADERS['Authorization'] = f'Bearer {token}'

    def _real_extract(self, url):
        video_id, event, type_, slug = self._match_valid_url(url).group('id', 'event', 'type', 'slug')
        video_type = {'archive': 'archives', 'player': 'broadcasts'}[type_]
        webpage = self._download_webpage(url, video_id)
        event_data = traverse_obj(
            self._search_nextjs_data(webpage, video_id, default='{}'),
            ('props', 'pageProps', 'eventCMSData', {