[BACKEND] Download and process Soundcloud tracks individually

This commit is contained in:
jbannon 2022-07-22 06:51:14 +00:00
parent 17586aa9bd
commit be491bdcf8
5 changed files with 82 additions and 78 deletions

View file

@ -1,12 +1,13 @@
from typing import Dict, Optional
from typing import Dict
from typing import Generator
from typing import List
from ytdl_sub.downloaders.soundcloud.abc import SoundcloudDownloader
from ytdl_sub.downloaders.soundcloud.abc import SoundcloudDownloaderOptions
from ytdl_sub.entries.soundcloud import SoundcloudAlbum
from ytdl_sub.entries.soundcloud import SoundcloudAlbumTrack
from ytdl_sub.entries.soundcloud import SoundcloudTrack
from ytdl_sub.validators.url_validator import SoundcloudUsernameUrlValidator
from ytdl_sub.validators.validators import BoolValidator
class SoundcloudAlbumsAndSinglesDownloadOptions(SoundcloudDownloaderOptions):
@ -36,9 +37,6 @@ class SoundcloudAlbumsAndSinglesDownloadOptions(SoundcloudDownloaderOptions):
self._url = self._validate_key(
key="url", validator=SoundcloudUsernameUrlValidator
).username_url
self._download_individually = self._validate_key_if_present(
"download_individually", BoolValidator, default=True
)
@property
def url(self) -> str:
@ -47,15 +45,6 @@ class SoundcloudAlbumsAndSinglesDownloadOptions(SoundcloudDownloaderOptions):
"""
return self._url
@property
def download_individually(self) -> Optional[bool]:
"""
Optional. Downloads files from the channel individually instead of in bulk. Setting to True
is safer when downloading large amounts of videos in case an error occurs. Downloading by
bulk (by setting to False) can increase speeds. Defaults to True.
"""
return self._download_individually.value
class SoundcloudAlbumsAndSinglesDownloader(
SoundcloudDownloader[SoundcloudAlbumsAndSinglesDownloadOptions]
@ -80,7 +69,7 @@ class SoundcloudAlbumsAndSinglesDownloader(
},
)
def _get_albums(self, entry_dicts: List[Dict]) -> List[SoundcloudAlbum]:
def _get_albums_from_entry_dicts(self, entry_dicts: List[Dict]) -> List[SoundcloudAlbum]:
"""
Parameters
----------
@ -126,36 +115,70 @@ class SoundcloudAlbumsAndSinglesDownloader(
return tracks
def download(self) -> List[SoundcloudTrack]:
"""
Soundcloud subscription to download albums and tracks as singles.
"""
def _get_albums(self) -> List[SoundcloudAlbum]:
# Dry-run to get the info json files
artist_albums_url = self.artist_albums_url(artist_url=self.download_options.url)
artist_tracks_url = self.artist_tracks_url(artist_url=self.download_options.url)
# Albums do not need to download anything since tracks contain all album tracks.
# We just need to get the metadata from the album itself
album_entry_dicts = self.extract_info_via_info_json(
url=artist_albums_url,
ytdl_options_overrides={
"skip_download": True,
"writethumbnail": False,
},
url=artist_albums_url,
)
tracks_entry_dicts = self.extract_info_via_info_json(url=artist_tracks_url)
# Get all of the artist's albums
albums = self._get_albums(entry_dicts=album_entry_dicts)
albums = self._get_albums_from_entry_dicts(entry_dicts=album_entry_dicts)
return albums
def _get_album_tracks(
self, albums: List[SoundcloudAlbum]
) -> Generator[SoundcloudAlbumTrack, None, None]:
for album in albums:
for track in album.album_tracks():
if self.download_options.skip_premiere_tracks and track.is_premiere():
continue
_ = self.extract_info(
url=album.kwargs("webpage_url"),
ytdl_options_overrides={
"playlist_items": str(track.kwargs("playlist_index")),
"writeinfojson": False,
},
)
yield track
def _get_single_tracks(
self, albums: List[SoundcloudAlbum]
) -> Generator[SoundcloudTrack, None, None]:
artist_tracks_url = self.artist_tracks_url(artist_url=self.download_options.url)
tracks_entry_dicts = self.extract_info_via_info_json(
url=artist_tracks_url,
ytdl_options_overrides={
"skip_download": True,
"writethumbnail": False,
},
)
# Then, get all singles
tracks = self._get_singles(entry_dicts=tracks_entry_dicts, albums=albums)
for track in tracks:
# Filter any premiere tracks if specified
if self.download_options.skip_premiere_tracks and track.is_premiere():
continue
# Append all album tracks as SoundcloudAlbumTrack classes to the singles
for album in albums:
tracks += album.album_tracks()
_ = self.extract_info(
url=track.kwargs("webpage_url"), ytdl_options_overrides={"writeinfojson": False}
)
# Filter any premiere tracks if specified
if self.download_options.skip_premiere_tracks:
tracks = [track for track in tracks if not track.is_premiere()]
yield track
return tracks
def download(self) -> Generator[SoundcloudTrack, None, None]:
"""
Soundcloud subscription to download albums and tracks as singles.
"""
albums = self._get_albums()
for album_track in self._get_album_tracks(albums=albums):
yield album_track
for single_track in self._get_single_tracks(albums=albums):
yield single_track

View file

@ -171,18 +171,16 @@ class YoutubeChannelDownloader(YoutubeDownloader[YoutubeChannelDownloaderOptions
)
self.channel = self._get_channel_from_entry_dicts(entry_dicts=entry_dicts)
# If downloading individually, remove the skip_download to actually download the video
if self.download_options.download_individually:
del ytdl_options_overrides["skip_download"]
del ytdl_options_overrides["writethumbnail"]
for entry_dict in entry_dicts:
if entry_dict.get("extractor") == "youtube":
# Only do the individual download if it is not dry-run and downloading individually
if not self.is_dry_run and self.download_options.download_individually:
ytdl_options_overrides["playlist_items"] = str(entry_dict.get("playlist_index"))
_ = self.extract_info(
ytdl_options_overrides=ytdl_options_overrides,
ytdl_options_overrides={
"playlist_items": str(entry_dict.get("playlist_index")),
"writeinfojson": False,
},
url=self.download_options.channel_url,
)
yield YoutubeVideo(entry_dict=entry_dict, working_directory=self.working_directory)

View file

@ -98,19 +98,16 @@ class YoutubePlaylistDownloader(
ytdl_options_overrides=ytdl_options_overrides, url=self.download_options.playlist_url
)
# If downloading individually, remove the skip_download to actually download the video
if self.download_options.download_individually:
del ytdl_options_overrides["skip_download"]
del ytdl_options_overrides["writethumbnail"]
for entry_dict in entry_dicts:
if entry_dict.get("extractor") == "youtube":
# Only do the individual download if it is not dry-run and downloading individually
if not self.is_dry_run and self.download_options.download_individually:
ytdl_options_overrides["playlist_items"] = str(entry_dict.get("playlist_index"))
_ = self.extract_info(
ytdl_options_overrides=ytdl_options_overrides,
ytdl_options_overrides={
"playlist_items": str(entry_dict.get("playlist_index")),
"writeinfojson": False,
},
url=self.download_options.playlist_url,
)

View file

@ -19,3 +19,8 @@ def music_video_config():
@pytest.fixture()
def channel_as_tv_show_config():
return ConfigFile.from_file_path(config_path="examples/kodi_tv_shows_config.yaml")
@pytest.fixture
def soundcloud_discography_config():
return ConfigFile.from_file_path(config_path="examples/soundcloud_discography_config.yaml")

View file

@ -11,22 +11,7 @@ from ytdl_sub.subscriptions.subscription import Subscription
@pytest.fixture
def config_path():
return "examples/soundcloud_discography_config.yaml"
@pytest.fixture
def subscription_name():
return "jb"
@pytest.fixture
def config(config_path):
return ConfigFile.from_file_path(config_path=config_path)
@pytest.fixture
def subscription_dict(output_directory, subscription_name):
def subscription_dict(output_directory):
return {
"preset": "sc_discography",
"soundcloud": {"url": "https://soundcloud.com/jessebannon"},
@ -40,20 +25,6 @@ def subscription_dict(output_directory, subscription_name):
}
@pytest.fixture
def discography_subscription(config, subscription_name, subscription_dict):
discography_preset = Preset.from_dict(
config=config,
preset_name=subscription_name,
preset_dict=subscription_dict,
)
return Subscription.from_preset(
preset=discography_preset,
config=config,
)
@pytest.fixture
def expected_discography_download():
# turn off black formatter here for readability
@ -96,8 +67,18 @@ class TestSoundcloudDiscography:
@pytest.mark.parametrize("dry_run", [True, False])
def test_discography_download(
self, discography_subscription, expected_discography_download, output_directory, dry_run
self,
subscription_dict,
soundcloud_discography_config,
expected_discography_download,
output_directory,
dry_run,
):
discography_subscription = Subscription.from_dict(
preset_dict=subscription_dict,
preset_name="jb",
config=soundcloud_discography_config,
)
transaction_log = discography_subscription.download(dry_run=dry_run)
assert_transaction_log_matches(
output_directory=output_directory,