X-Git-Url: https://jfr.im/git/yt-dlp.git/blobdiff_plain/9a9006ba20f1f9f34183e1bde098c75502a018f8..61edf57f8f13f6dfd81154174e647eb5fdd26089:/yt_dlp/extractor/twitcasting.py diff --git a/yt_dlp/extractor/twitcasting.py b/yt_dlp/extractor/twitcasting.py index 2548dae04..53b408469 100644 --- a/yt_dlp/extractor/twitcasting.py +++ b/yt_dlp/extractor/twitcasting.py @@ -5,8 +5,9 @@ from .common import InfoExtractor from ..dependencies import websockets from ..utils import ( - clean_html, ExtractorError, + UserNotLive, + clean_html, float_or_none, get_element_by_class, get_element_by_id, @@ -22,7 +23,7 @@ class TwitCastingIE(InfoExtractor): - _VALID_URL = r'https?://(?:[^/]+\.)?twitcasting\.tv/(?P[^/]+)/(?:movie|twplayer)/(?P\d+)' + _VALID_URL = r'https?://(?:[^/?#]+\.)?twitcasting\.tv/(?P[^/?#]+)/(?:movie|twplayer)/(?P\d+)' _M3U8_HEADERS = { 'Origin': 'https://twitcasting.tv', 'Referer': 'https://twitcasting.tv/', @@ -107,9 +108,9 @@ def _real_extract(self, url): url, video_id, data=request_data, headers={'Origin': 'https://twitcasting.tv'}, note='Trying video password') - if urlh.geturl() != url and request_data: + if urlh.url != url and request_data: webpage = self._download_webpage( - urlh.geturl(), video_id, data=request_data, + urlh.url, video_id, data=request_data, headers={'Origin': 'https://twitcasting.tv'}, note='Retrying authentication') # has to check here as the first request can contain password input form even if the password is correct @@ -138,10 +139,10 @@ def _real_extract(self, url): webpage, 'datetime', None)) stream_server_data = self._download_json( - 'https://twitcasting.tv/streamserver.php?target=%s&mode=client' % uploader_id, video_id, + f'https://twitcasting.tv/streamserver.php?target={uploader_id}&mode=client', video_id, 'Downloading live info', fatal=False) - is_live = 'data-status="online"' in webpage + is_live = any(f'data-{x}' in webpage for x in ['is-onlive="true"', 'live-type="live"', 'status="online"']) if not traverse_obj(stream_server_data, 'llfmp4') and is_live: self.raise_login_required(method='cookies') @@ -188,7 +189,7 @@ def find_dmu(x): for mode, ws_url in streams.items(): formats.append({ 'url': ws_url, - 'format_id': 'ws-%s' % mode, + 'format_id': f'ws-{mode}', 'ext': 'mp4', 'quality': qq(mode), 'source_preference': -10, @@ -231,54 +232,64 @@ def find_dmu(x): class TwitCastingLiveIE(InfoExtractor): - _VALID_URL = r'https?://(?:[^/]+\.)?twitcasting\.tv/(?P[^/]+)/?(?:[#?]|$)' + _VALID_URL = r'https?://(?:[^/?#]+\.)?twitcasting\.tv/(?P[^/?#]+)/?(?:[#?]|$)' _TESTS = [{ 'url': 'https://twitcasting.tv/ivetesangalo', 'only_matching': True, + }, { + 'url': 'https://twitcasting.tv/c:unusedlive', + 'expected_exception': 'UserNotLive', }] def _real_extract(self, url): uploader_id = self._match_id(url) self.to_screen( - 'Downloading live video of user {0}. ' - 'Pass "https://twitcasting.tv/{0}/show" to download the history'.format(uploader_id)) + f'Downloading live video of user {uploader_id}. ' + f'Pass "https://twitcasting.tv/{uploader_id}/show" to download the history') - webpage = self._download_webpage(url, uploader_id) + is_live = traverse_obj(self._download_json( + f'https://frontendapi.twitcasting.tv/watch/user/{uploader_id}', + uploader_id, 'Checking live status', data=b'', fatal=False), ('is_live', {bool})) + if is_live is False: # only raise here if API response was as expected + raise UserNotLive(video_id=uploader_id) + + # Use /show/ page so that password-protected and members-only livestreams can be found + webpage = self._download_webpage( + f'https://twitcasting.tv/{uploader_id}/show/', uploader_id, 'Downloading live history') + is_live = is_live or self._search_regex( + r'(?s)(\s*LIVE)', + webpage, 'is live?', default=False) + # Current live is always the first match current_live = self._search_regex( - (r'data-type="movie" data-id="(\d+)">', - r'tw-sound-flag-open-link" data-id="(\d+)" style=',), - webpage, 'current live ID', default=None) - if not current_live: - # fetch unfiltered /show to find running livestreams; we can't get ID of the password-protected livestream above - webpage = self._download_webpage( - f'https://twitcasting.tv/{uploader_id}/show/', uploader_id, - note='Downloading live history') - is_live = self._search_regex(r'(?s)(\s*LIVE)', webpage, 'is live?', default=None) - if is_live: - # get the first live; running live is always at the first - current_live = self._search_regex( - r'(?s)\d+)"\s*>.+?', - webpage, 'current live ID 2', default=None, group='video_id') - if not current_live: - raise ExtractorError('The user is not currently live') - return self.url_result('https://twitcasting.tv/%s/movie/%s' % (uploader_id, current_live)) + r'(?s)\d+)"', + webpage, 'current live ID', default=None, group='video_id') + if not is_live or not current_live: + raise UserNotLive(video_id=uploader_id) + + return self.url_result(f'https://twitcasting.tv/{uploader_id}/movie/{current_live}', TwitCastingIE) class TwitCastingUserIE(InfoExtractor): - _VALID_URL = r'https?://(?:[^/]+\.)?twitcasting\.tv/(?P[^/]+)/show/?(?:[#?]|$)' + _VALID_URL = r'https?://(?:[^/?#]+\.)?twitcasting\.tv/(?P[^/?#]+)/(:?show|archive)/?(?:[#?]|$)' _TESTS = [{ + 'url': 'https://twitcasting.tv/natsuiromatsuri/archive/', + 'info_dict': { + 'id': 'natsuiromatsuri', + 'title': 'natsuiromatsuri - Live History', + }, + 'playlist_mincount': 235, + }, { 'url': 'https://twitcasting.tv/noriyukicas/show', 'only_matching': True, }] def _entries(self, uploader_id): - base_url = next_url = 'https://twitcasting.tv/%s/show' % uploader_id + base_url = next_url = f'https://twitcasting.tv/{uploader_id}/show' for page_num in itertools.count(1): webpage = self._download_webpage( - next_url, uploader_id, query={'filter': 'watchable'}, note='Downloading page %d' % page_num) + next_url, uploader_id, query={'filter': 'watchable'}, note=f'Downloading page {page_num}') matches = re.finditer( - r'''(?isx)/[^/]+/movie/\d+)"\s*>.+?''', - webpage) + r'(?s)/[^/"]+/movie/\d+)"', webpage) for mobj in matches: yield self.url_result(urljoin(base_url, mobj.group('url'))) @@ -292,4 +303,4 @@ def _entries(self, uploader_id): def _real_extract(self, url): uploader_id = self._match_id(url) return self.playlist_result( - self._entries(uploader_id), uploader_id, '%s - Live History' % uploader_id) + self._entries(uploader_id), uploader_id, f'{uploader_id} - Live History')