Skip to content

Commit

Permalink
Update extrememusic.py
Browse files Browse the repository at this point in the history
  • Loading branch information
@kclauhk
kclauhk committed Dec 15, 2024
1 parent 8f604bc commit 38e5501
Showing 1 changed file with 68 additions and 80 deletions.
148 changes: 68 additions & 80 deletions yt_dlp/extractor/extrememusic.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -57,7 +57,7 @@ def _extract_track(self, album_data, track_id=None, version_id=None):
(lambda _, v: v['id'] == int(version_id), 'track_id', {int}), get_all=False)
if track := traverse_obj(album_data['tracks'],
(lambda _, v: v['id'] == int(track_id), {dict}), get_all=False):
info = {**traverse_obj(track, {
track_info = {**traverse_obj(track, {
'track': ('title', {str}),
'track_number': ('sort_order', {lambda v: v + 1}, {int}),
'track_id': ('track_no', {str}),
Expand All @@ -80,7 +80,8 @@ def _extract_track(self, album_data, track_id=None, version_id=None):
'height': ('height', {int_or_none}),
}))
if not self._REQUIRE_VERSION:
version_id = version_id or traverse_obj(track, 'default_track_sound_id', ('track_sound_ids', 0))
version_id = (version_id
or traverse_obj(track, 'default_track_sound_id', ('track_sound_ids', 0)))
for sound_id in [version_id] if version_id else track['track_sound_ids']:
if sound := traverse_obj(album_data['track_sounds'],
(lambda _, v: v['id'] == int(sound_id) and v['track_id'] == int(track_id),
Expand All @@ -89,8 +90,8 @@ def _extract_track(self, album_data, track_id=None, version_id=None):
or 'all' in self._REQUIRE_VERSION
or any(x in sound['version_type'].lower() for x in self._REQUIRE_VERSION)):
formats = []
for audio_url in traverse_obj(sound, ('assets', 'audio', ('preview_url',
'preview_url_hls'))):
for audio_url in traverse_obj(
sound, ('assets', 'audio', ('preview_url', 'preview_url_hls'))):
if determine_ext(audio_url) == 'm3u8':
m3u8_url = re.sub(r'\.m3u8\?.*', '/HLS/128_v4.m3u8', audio_url)
for f in self._extract_m3u8_formats(m3u8_url, sound_id, 'm4a', fatal=False):
Expand All @@ -108,25 +109,14 @@ def _extract_track(self, album_data, track_id=None, version_id=None):
'id': str(sound_id),
'title': join_nonempty('title', 'version_type', from_dict=sound, delim=' - '),
'alt_title': sound['version_type'],
**info,
**track_info,
'thumbnails': thumbnails,
'duration': sound.get('duration'),
'formats': formats,
'webpage_url': f"https://www.extrememusic.com/albums/{track['album_id']}?item={track_id}&ver={sound_id}",
})

if len(entries) > 1:
return {
'id': track_id,
**info,
'entries': entries,
'_type': 'playlist',
}
elif len(entries) == 1:
return entries[0]
else:
self.raise_no_formats('Track data not found', video_id=track_id)
return []
return [entries, track_info]
self.raise_no_formats('Track data not found', video_id=track_id)


class ExtremeMusicIE(ExtremeMusicBaseIE):
Expand Down Expand Up @@ -214,10 +204,12 @@ def _real_extract(self, url):
album_id, track_id, version_id = self._match_valid_url(url).group('album', 'id', 'ver')
self._initialize(url, version_id or track_id, self.get_param('geo_bypass_country') or 'DE')
album_data = self._get_album_data(album_id, version_id or track_id)
if result := self._extract_track(album_data, track_id, version_id):
return result
else:
self.raise_no_formats('No formats were found')
if track := self._extract_track(album_data, track_id, version_id):
if len(track[0]) > 1:
return self.playlist_result(track[0], **track[1])
elif len(track[0]) == 1:
return track[0][0]
self.raise_no_formats('No formats were found')


class ExtremeMusicAIE(ExtremeMusicBaseIE):
Expand Down Expand Up @@ -247,32 +239,26 @@ def _real_extract(self, url):
album_id = self._match_id(url)
self._initialize(url, album_id, self.get_param('geo_bypass_country') or 'DE')
album_data = self._get_album_data(album_id, album_id)
subgenres = traverse_obj(album_data, ('album', 'subgenres', {str_or_none}))
album_info = merge_dicts(traverse_obj(album_data.get('album'), {
'id': ('id', {lambda v: str(v)}),
'album': ('title', {str_or_none}),
'description': ('description', {lambda v: str_or_none(v) or None}),
'artists': ('artist', {lambda v: [v] if v else None}),
'genres': ('genres', {str_or_none},
{lambda v: join_nonempty(v, subgenres, delim=', ')},
{lambda v: v.split(', ') if v else None}),
'tag': ('keywords', {lambda v: v.split(', ') if v else None}),
}), {
'description': traverse_obj(
album_data, ('bio', 'description', {lambda v: str_or_none(v) or None})),
})

entries = []
for track_id in traverse_obj(album_data, ('tracks', ..., 'id')):
if track := self._extract_track(album_data, track_id=track_id):
if track.get('entries'):
entries.extend(track['entries'])
else:
entries.append(track)

if entries:
subgenres = traverse_obj(album_data, ('album', 'subgenres', {str_or_none}))
return merge_dicts(traverse_obj(album_data.get('album'), {
'id': ('id', {lambda v: str(v)}),
'album': ('title', {str_or_none}),
'description': ('description', {lambda v: str_or_none(v) or None}),
'artists': ('artist', {lambda v: [v] if v else None}),
'genres': ('genres', {str_or_none}, {lambda v: join_nonempty(v, subgenres, delim=', ')},
{lambda v: v.split(', ') if v else None}),
'tag': ('keywords', {lambda v: v.split(', ') if v else None}),
}), {
'description': traverse_obj(album_data, ('bio', 'description', {lambda v: str_or_none(v) or None})),
'entries': entries,
'_type': 'playlist',
})
else:
self.raise_no_formats('No formats were found')
return self.playlist_result(
(entry for sound in (track[0] for track in (self._extract_track(album_data, track_id=track_id)
for track_id in traverse_obj(album_data, ('tracks', ..., 'id'))))
for entry in sound),
**album_info)


class ExtremeMusicPIE(ExtremeMusicBaseIE):
Expand Down Expand Up @@ -310,49 +296,51 @@ def playlist_query(playlist_id, offset, limit):
'range': f'{offset},{limit}',
}, headers=self._REQUEST_HEADERS)

thumbnails, entries = [], []
album_data, track_done, limit = {}, [], 50
for i in itertools.count():
playlist = playlist_query(playlist_id, i * limit, limit)
if len(playlist['playlist_items']) == 0:
break
else:
track_ids = traverse_obj(playlist, ('playlist_items', ..., 'track_id'))
for track_id in list(dict.fromkeys(track_ids)):
if track_id not in track_done:
album_id = traverse_obj(playlist,
('tracks', lambda _, v: v['id'] == track_id, 'album_id', {int}), get_all=False)
if album_id not in album_data:
album_data[album_id] = self._get_album_data(album_id, track_id, fatal=False)
playlist['album'] = traverse_obj(album_data, (album_id, 'album', {dict}))
if track := self._extract_track(playlist, track_id=track_id):
if track.get('entries'):
entries.extend(track['entries'])
else:
entries.append(track)
track_done.append(track_id)
if len(track_done) >= playlist['playlist']['playlist_items_count']:
break

if entries:
if len(track_done) < playlist['playlist']['playlist_items_count']:
self.report_warning('This playlist has geo-restricted items. Try using --xff to specify a different country code, e.g. DE')
def extract_playlist(playlist, playlist_id):
albums, tracks_done, items_count = {}, [], 0
for i in itertools.count():
if i > 0:
playlist = playlist_query(playlist_id, i * limit, limit)
if len(playlist.get('tracks', [])) == 0:
print(items_count)
if items_count < playlist['playlist']['playlist_items_count']:
self.report_warning('This playlist has geo-restricted items. Try using --xff to specify a different country code, e.g. DE')
break
else:
track_ids = traverse_obj(playlist, ('tracks', ..., 'id'))
items_count += len(track_ids)
for track_id in list(dict.fromkeys(track_ids)):
if track_id not in tracks_done:
album_id = traverse_obj(
playlist, ('tracks', lambda _, v: v['id'] == track_id,
'album_id', {int}), get_all=False)
if album_id not in albums:
albums[album_id] = self._get_album_data(album_id, track_id, fatal=False)
playlist['album'] = traverse_obj(albums, (album_id, 'album', {dict}))
if track := self._extract_track(playlist, track_id=track_id):
if len(track[0]) > 1:
for track in track[0]:
yield track
elif len(track[0]) == 1:
yield track[0][0]
tracks_done.append(track_id)
if items_count >= playlist['playlist']['playlist_items_count']:
break

playlist, thumbnails, limit = {}, [], 20
if playlist := playlist_query(playlist_id, 0, limit):
thumbnails = []
for image in traverse_obj(playlist['playlist'], ('images', 'square')):
thumbnails.append(traverse_obj(image, {
'url': ('url', {url_or_none}),
'width': ('width', {int_or_none}),
'height': ('height', {int_or_none}),
}))

return {k: v for k, v in {
playlist_info = {k: v for k, v in {
'id': playlist['playlist']['id'],
'title': playlist['playlist']['title'],
'thumbnail': traverse_obj(thumbnails, (0, 'url', {url_or_none})),
'thumbnails': thumbnails,
'uploader': playlist['playlist']['owner_name'],
'entries': entries,
'_type': 'playlist',
}.items() if v}
else:
self.raise_no_formats('No formats were found')
return self.playlist_result(extract_playlist(playlist, playlist_id), **playlist_info)

0 comments on commit 38e5501

Please sign in to comment.