From 5f632986dbde9c21890706d8ab43cc8c0a4d3234 Mon Sep 17 00:00:00 2001 From: Ben Faerber Date: Thu, 27 Mar 2025 16:55:27 -0600 Subject: [PATCH] Split Parti Extractor into Live and Video --- yt_dlp/extractor/_extractors.py | 5 +- yt_dlp/extractor/parti.py | 133 +++++++++++++++++++------------- 2 files changed, 84 insertions(+), 54 deletions(-) diff --git a/yt_dlp/extractor/_extractors.py b/yt_dlp/extractor/_extractors.py index 36cf46127e..fa8242faa7 100644 --- a/yt_dlp/extractor/_extractors.py +++ b/yt_dlp/extractor/_extractors.py @@ -1492,7 +1492,10 @@ from .paramountplus import ( ) from .parler import ParlerIE from .parlview import ParlviewIE -from .parti import PartiIE +from .parti import ( + PartiLivestreamIE, + PartiVideoIE, +) from .patreon import ( PatreonCampaignIE, PatreonIE, diff --git a/yt_dlp/extractor/parti.py b/yt_dlp/extractor/parti.py index 59106446f2..8cd5e5690a 100644 --- a/yt_dlp/extractor/parti.py +++ b/yt_dlp/extractor/parti.py @@ -1,5 +1,7 @@ import datetime +from yt_dlp.utils._utils import UserNotLive + from ..utils import ( int_or_none, traverse_obj, @@ -7,28 +9,86 @@ from ..utils import ( from .common import InfoExtractor -class PartiIE(InfoExtractor): +class PartiBaseIE(InfoExtractor): + _RECORDING_BASE_URL = 'https://watch.parti.com' + _GET_LIVESTREAM_API = 'https://api-backend.parti.com/parti_v2/profile/get_livestream_channel_info' + _PLAYBACK_VERSION = '1.17.0' + + def _get_formats(self, stream_url, creator, is_live): + return self._extract_m3u8_formats(stream_url, creator, 'mp4', live=is_live) + + def _build_recording_url(self, path): + return self._RECORDING_BASE_URL + '/' + path + + +class PartiVideoIE(PartiBaseIE): + IE_NAME = 'parti:video' + IE_DESC = 'Download a video from parti.com' + _VALID_URL = r'https://parti\.com/video/(?P\d+)' + _TESTS = [ + { + 'url': 'https://parti.com/video/66284', + 'info_dict': { + 'id': '66284', + 'ext': 'mp4', + 'title': str, + 'upload_date': str, + 'is_live': False, + 'categories': list, + 'thumbnail': str, + 'channel': str, + }, + 'params': {'skip_download': 'm3u8'}, + }, + ] + + def _get_video_info(self, video_id): + url = self._GET_LIVESTREAM_API + '/recent/' + video_id + data = self._download_json(url, video_id) + return traverse_obj(data, { + 'channel': ('user_name', {str}), + 'thumbnail': ('event_file', {str}), + 'categories': ('category', {lambda c: [c]}), + 'url': ('livestream_recording', {str}), + 'title': ('event_title', {str}), + 'upload_date': ('event_start_ts', {lambda ts: datetime.date.fromtimestamp(ts).strftime('%Y%m%d')}), + }) + + def _real_extract(self, url): + video_id = self._match_id(url) + video_info = self._get_video_info(video_id) + full_recording_url = self._build_recording_url(video_info['url']) + formats = self._get_formats(full_recording_url, video_info['channel'], is_live=False) + + return { + 'id': video_id, + 'url': url, + 'formats': formats, + 'is_live': False, + **video_info, + } + + +class PartiLivestreamIE(PartiBaseIE): + IE_NAME = 'parti:livestream' IE_DESC = 'Download a stream from parti.com' _VALID_URL = r'https://parti\.com/creator/(parti|discord|telegram)/(?P[\w-]+)' _TESTS = [ { - 'url': 'https://parti.com/creator/parti/ItZTMGG', + 'url': 'https://parti.com/creator/parti/SpartanTheDog', 'info_dict': { - 'id': 'ItZTMGG', + 'id': 'SpartanTheDog', 'ext': 'mp4', 'title': str, 'description': str, 'upload_date': str, - 'is_live': False, + 'is_live': True, + 'live_status': 'is_live', }, 'params': {'skip_download': 'm3u8'}, }, ] _CREATOR_API = 'https://api-backend.parti.com/parti_v2/profile/get_user_by_social_media/parti' - _GET_LIVESTREAM_API = 'https://api-backend.parti.com/parti_v2/profile/get_livestream_channel_info' - _GET_USER_FEED_API = 'https://api-backend.parti.com/parti_v2/profile/user_profile_feed/' - _RECORDING_BASE_URL = 'https://watch.parti.com' - _PLAYBACK_VERSION = '1.17.0' def _get_creator_id(self, creator): """ The creator ID is a number returned as plain text """ @@ -61,57 +121,24 @@ class PartiIE(InfoExtractor): **extracted, } - def _get_user_feed(self, creator_id): - """ The user feed are VODs listed below the main stream """ - url = self._GET_USER_FEED_API + '/' + creator_id + '?limit=10' - vods = self._download_json(url, None, 'Fetching user feed') - if not vods: - raise Exception('No vods found!') - return list(vods) - - def _download_vod(self, url, creator, creator_id): - """ Download the VOD visible on the creators feed """ - feed = self._get_user_feed(creator_id) - vod = feed[0] - vod_url = self._RECORDING_BASE_URL + '/' + vod['livestream_recording'] - created_at = datetime.date.fromtimestamp(vod['created_at']) - upload_date = str(created_at).replace('-', '') - - formats = self._extract_m3u8_formats(vod_url, creator, 'mp4', live=False) - return { - 'id': creator, - 'url': url, - 'title': f'{creator}\'s Parti VOD - {upload_date}', - 'description': vod['post_content'], - 'upload_date': upload_date, - 'is_live': False, - 'formats': formats, - } + def _real_extract(self, url): + creator = self._match_id(url) - def _download_livestream(self, url, creator, stream_url): - """ Download a currently active livestream """ - formats = self._extract_m3u8_formats(stream_url, creator, 'mp4', live=True) + creator_id = self._get_creator_id(creator) + playback_data = self._get_live_playback_data(creator_id) + if not playback_data['is_live']: + raise UserNotLive + + formats = self._get_formats(playback_data['url'], creator, is_live=True) created_at = datetime.datetime.now() - upload_date = str(created_at.date()).replace('-', '') - pretty_timestamp = str(created_at).replace(':', '_') + streamed_at = created_at.strftime('%Y%m%d') return { 'id': creator, 'url': url, - 'title': f'{creator}\'s Parti Live - {pretty_timestamp}', - 'description': f'A livestream from {pretty_timestamp}', - 'upload_date': upload_date, + 'title': f'{creator}\'s Parti Livestream', + 'description': f'A livestream from {created_at}', + 'upload_date': streamed_at, 'is_live': True, 'formats': formats, } - - def _real_extract(self, url): - creator = self._match_id(url) - - creator_id = self._get_creator_id(creator) - playback_data = self._get_live_playback_data(creator_id) - if not playback_data['is_live']: - return self._download_vod(url, creator, creator_id) - else: - return self._download_livestream(url, creator, playback_data['url']) -