Mirroring
/
yt-dlp
mirror of https://github.com/yt-dlp/yt-dlp
1
0
Fork 0
Code Issues Releases Wiki Activity

Split Parti Extractor into Live and Video

Browse Source
pull/12769/head
Ben Faerber 4 months ago
parent c6dd86af14
commit 5f632986db
No known key found for this signature in database
GPG Key ID: 6D33349087BCDC6E
2 changed files with 84 additions and 54 deletions
Show Stats Download Patch File Download Diff File

5
yt_dlp/extractor/_extractors.py
Unescape Escape View File

@ -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,

133
yt_dlp/extractor/parti.py
Unescape Escape View File

@ -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<id>\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<id>[\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'])

Write Preview
Loading…
Cancel
Save