diff --git a/src/ytdl_sub/downloaders/generic/collection.py b/src/ytdl_sub/downloaders/generic/collection.py index 8aaf58bf..32ed72c3 100644 --- a/src/ytdl_sub/downloaders/generic/collection.py +++ b/src/ytdl_sub/downloaders/generic/collection.py @@ -1,3 +1,5 @@ +import contextlib +import os.path from typing import Dict from typing import Generator from typing import List @@ -8,6 +10,7 @@ from ytdl_sub.downloaders.downloader import DownloaderValidator from ytdl_sub.downloaders.downloader import download_logger from ytdl_sub.entries.entry import Entry from ytdl_sub.entries.entry_parent import EntryParent +from ytdl_sub.utils.file_handler import FileHandler from ytdl_sub.validators.strict_dict_validator import StrictDictValidator from ytdl_sub.validators.string_formatter_validators import DictFormatterValidator from ytdl_sub.validators.validators import ListValidator @@ -27,16 +30,23 @@ class CollectionUrlValidator(StrictDictValidator): # TODO: url validate using yt-dlp IE self._url = self._validate_key(key="url", validator=StringValidator) - self._variables = self._validate_key_if_present( - key="variables", validator=DictFormatterValidator - ).dict_with_format_strings + variables = self._validate_key_if_present(key="variables", validator=DictFormatterValidator) + self._variables = variables.dict_with_format_strings if variables else {} @property def url(self) -> str: + """ + Returns + ------- + URL to download from + """ return self._url.value @property def variables(self) -> Dict[str, str]: + """ + Variables to add to each entry + """ return self._variables @@ -114,10 +124,10 @@ class CollectionDownloader(Downloader[CollectionDownloadOptions, Entry]): self, collection_url: CollectionUrlValidator, parent: EntryParent, entry_dicts: List[Dict] ) -> List[Entry]: leaf_children: List[Entry] = [] - parent.read_children_from_entry_dicts(entry_dicts=entry_dicts, child_class=EntryParent) + parent.read_nested_children_from_entry_dicts(entry_dicts=entry_dicts) for idx in range(parent.child_count): - child = parent.child_entries[idx] + child: EntryParent = parent.child_entries[idx] leaf_children.extend( self._recursive_child_read( @@ -127,19 +137,22 @@ class CollectionDownloader(Downloader[CollectionDownloadOptions, Entry]): # If the child has no nested children, turn it into an Entry if not child.child_count: - entry = child.to_entry() - entry.add_variables(collection_url.variables) - parent.child_entries[idx] = entry + parent.child_entries[idx] = child.to_entry() + leaf_children.append(parent.child_entries[idx]) - leaf_children.append(entry) + for leaf_child in leaf_children: + leaf_child.add_variables( + parent._get_children_entry_variables_to_add(parent.child_entries) + ) + leaf_child.add_variables(collection_url.variables) return leaf_children def _get_leaf_entries(self, collection_url: CollectionUrlValidator) -> List[Entry]: # Dry-run to get the info json files + # TODO: Mock the download entry_dicts = self.extract_info_via_info_json( only_info_json=True, - log_prefix_on_info_json_dl="Downloading metadata for", url=collection_url.url, ) @@ -160,8 +173,38 @@ class CollectionDownloader(Downloader[CollectionDownloadOptions, Entry]): return leaf_children - def _download_leaf_entry(self, entry: Entry): + @contextlib.contextmanager + def _separate_download_archives(self): + """ + Separate download archive writing between collection urls. This is so break_on_existing + does not break when downloading from subset urls. + """ + archive_path = self._enhanced_download_archive._archive_working_file_path + backup_archive_path = f"{archive_path}.backup" + + archive_file_exists = False + + # If archive path exists, maintain download archive is enable + if os.path.isfile(archive_path): + archive_file_exists = True + + # If a backup exists, it's the one prior to any downloading, use that. + if os.path.isfile(backup_archive_path): + FileHandler.copy(src_file_path=backup_archive_path, dst_file_path=archive_path) + # If not, create the backup + else: + FileHandler.copy(src_file_path=archive_path, dst_file_path=backup_archive_path) + + yield + + # If an archive path did not exist at first, but now exists, delete it + if not archive_file_exists: + FileHandler.delete(file_path=archive_path) + + def _download_leaf_entry(self, entry: Entry) -> Entry: download_logger.info("Downloading entry %s", entry.title) + + # TODO: Mock the download archive if dry-run if not self.is_dry_run: download_entry_dict = self.extract_info_with_retry( is_downloaded_fn=entry.is_downloaded, @@ -174,22 +217,32 @@ class CollectionDownloader(Downloader[CollectionDownloadOptions, Entry]): entry._kwargs["requested_subtitles"] = download_entry_dict.get("requested_subtitles") # pylint: enable=protected-access - yield entry + return entry + + def _download_collection_url( + self, collection_url: CollectionUrlValidator, downloaded_entries: Set[str] + ) -> Generator[Entry, None, None]: + with self._separate_download_archives(): + leaf_children = [ + leaf + for leaf in self._get_leaf_entries(collection_url) + if _entry_key(leaf) not in downloaded_entries + ] + + # Reverse leaf_children downloads so we download older entries first + for leaf in reversed(leaf_children): + yield self._download_leaf_entry(entry=leaf) + downloaded_entries.add(_entry_key(leaf)) def download(self) -> Generator[Entry, None, None]: """ Soundcloud subscription to download albums and tracks as singles. """ - downloaded_leaf_children: Set[str] = set() + downloaded_entries: Set[str] = set() # download the bottom-most urls first since they are top-priority for collection_url in reversed(self.download_options.collection_urls.list): - leaf_children = self._get_leaf_entries(collection_url) - - for child in leaf_children: - child_key = _entry_key(child) - if child_key in downloaded_leaf_children: - continue - - yield self._download_leaf_entry(entry=child) - downloaded_leaf_children.add(child_key) + for entry in self._download_collection_url( + collection_url=collection_url, downloaded_entries=downloaded_entries + ): + yield entry diff --git a/src/ytdl_sub/downloaders/soundcloud/albums_and_singles.py b/src/ytdl_sub/downloaders/soundcloud/albums_and_singles.py index 6b590daa..77b264d1 100644 --- a/src/ytdl_sub/downloaders/soundcloud/albums_and_singles.py +++ b/src/ytdl_sub/downloaders/soundcloud/albums_and_singles.py @@ -1,12 +1,10 @@ from typing import Dict from typing import Generator -from typing import List -from ytdl_sub.downloaders.downloader import download_logger +from ytdl_sub.downloaders.generic.collection import CollectionDownloader +from ytdl_sub.downloaders.generic.collection import CollectionDownloadOptions from ytdl_sub.downloaders.soundcloud.abc import SoundcloudDownloader from ytdl_sub.downloaders.soundcloud.abc import SoundcloudDownloaderOptions -from ytdl_sub.entries.entry_parent import EntryParent -from ytdl_sub.entries.soundcloud import SoundcloudAlbumTrack from ytdl_sub.entries.soundcloud import SoundcloudTrack from ytdl_sub.validators.url_validator import SoundcloudUsernameUrlValidator @@ -39,6 +37,36 @@ class SoundcloudAlbumsAndSinglesDownloadOptions(SoundcloudDownloaderOptions): key="url", validator=SoundcloudUsernameUrlValidator ).username_url + self.collection_validator = CollectionDownloadOptions( + name=self._name, + value={ + "urls": [ + { + "url": f"{self._url}/tracks", + "variables": { + "track_number": "1", + "track_number_padded": "01", + "track_count": "1", + "album": "{title}", + "album_sanitized": "{title_sanitized}", + "album_year": "{upload_year}", + }, + }, + { + "url": f"{self._url}/albums", + "variables": { + "track_number": "{playlist_index}", + "track_number_padded": "{playlist_index_padded}", + "track_count": "{playlist_count}", + "album": "{playlist}", + "album_sanitized": "{playlist_sanitized}", + "album_year": "{playlist_max_upload_year}", + }, + }, + ] + }, + ) + @property def url(self) -> str: """ @@ -72,104 +100,16 @@ class SoundcloudAlbumsAndSinglesDownloader( }, ) - def _get_singles( - self, entry_dicts: List[Dict], albums: List[EntryParent] - ) -> List[SoundcloudTrack]: - tracks: List[SoundcloudTrack] = [] - - # Get all tracks that are not part of an album - for entry_dict in self._filter_entry_dicts(entry_dicts): - if not any(entry_dict in album for album in albums): - tracks.append( - SoundcloudTrack(entry_dict=entry_dict, working_directory=self.working_directory) - ) - - return tracks - - def _get_albums(self) -> List[EntryParent]: - # Dry-run to get the info json files - artist_albums_url = self.artist_albums_url(artist_url=self.download_options.url) - entry_dicts = self.extract_info_via_info_json( - only_info_json=True, - log_prefix_on_info_json_dl="Downloading metadata for", - url=artist_albums_url, - ) - - albums: List[EntryParent] = [] - for entry_dict in self._filter_entry_dicts(entry_dicts, extractor="soundcloud:set"): - albums.append( - EntryParent( - entry_dict=entry_dict, working_directory=self.working_directory - ).read_children_from_entry_dicts( - entry_dicts=entry_dicts, child_class=SoundcloudAlbumTrack - ) - ) - - return albums - - def _get_album_tracks( - self, albums: List[EntryParent] - ) -> Generator[SoundcloudAlbumTrack, None, None]: - for album in albums: - if album.child_count > 0: - download_logger.info("Downloading album %s", album.title) - - for track in album.child_entries: - if self.download_options.skip_premiere_tracks and track.is_premiere(): - continue - - download_logger.info( - "Downloading album track %d/%d %s", - track.track_number, - track.track_count, - track.title, - ) - if not self.is_dry_run: - _ = self.extract_info_with_retry( - is_downloaded_fn=track.is_downloaded, - url=album.webpage_url, - ytdl_options_overrides={ - "playlist_items": str(track.kwargs("playlist_index")), - "writeinfojson": False, - }, - ) - - yield track - - def _get_single_tracks( - self, albums: List[EntryParent] - ) -> 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( - only_info_json=True, - log_prefix_on_info_json_dl="Downloading metadata for", - url=artist_tracks_url, - ) - - # 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 - - download_logger.info("Downloading single track %s", track.title) - if not self.is_dry_run: - _ = self.extract_info_with_retry( - is_downloaded_fn=track.is_downloaded, - url=track.webpage_url, - ytdl_options_overrides={"writeinfojson": False}, - ) - - yield track - 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 + downloader = CollectionDownloader( + download_options=self.download_options.collection_validator, + enhanced_download_archive=self._enhanced_download_archive, + ytdl_options_builder=self._ytdl_options_builder, + overrides=self.overrides, + ) - for single_track in self._get_single_tracks(albums=albums): - yield single_track + for entry in downloader.download(): + yield entry diff --git a/src/ytdl_sub/entries/entry_parent.py b/src/ytdl_sub/entries/entry_parent.py index 15596fe0..93b6c158 100644 --- a/src/ytdl_sub/entries/entry_parent.py +++ b/src/ytdl_sub/entries/entry_parent.py @@ -28,10 +28,32 @@ class EntryParent(BaseEntry): if not child_entries: return {} - return {"playlist_max_upload_year": max(entry.upload_year for entry in child_entries)} + return { + "playlist_max_upload_year": max( + entry.upload_year for entry in child_entries if isinstance(entry, Entry) + ) + } # pylint: enable=no-self-use + def read_nested_children_from_entry_dicts(self, entry_dicts: List[Dict]): + child_entries: List["EntryParent"] = [] + + for entry_dict in entry_dicts: + if entry_dict.get("playlist_id") == self.uid: + child_entries.append( + self.__class__( + entry_dict=entry_dict, + working_directory=self.working_directory(), + entry_parent=self, + ) + ) + + self._child_entries = sorted( + child_entries, key=lambda entry: entry.kwargs("playlist_index") + ) + return self._child_entries + def read_children_from_entry_dicts( self, entry_dicts: List[Dict], child_class: Type[TChildEntry] = Entry ) -> "EntryParent": @@ -136,11 +158,4 @@ class EntryParent(BaseEntry): return entry_parent def to_entry(self) -> Entry: - entry = Entry(entry_dict=self._kwargs, working_directory=self.working_directory()) - if self._entry_parent: - entry.add_variables( - self._entry_parent._get_children_entry_variables_to_add( - child_entries=self._entry_parent.child_entries - ) - ) - return entry + return Entry(entry_dict=self._kwargs, working_directory=self.working_directory()) diff --git a/src/ytdl_sub/entries/variables/entry_variables.py b/src/ytdl_sub/entries/variables/entry_variables.py index 155caa51..658208f6 100644 --- a/src/ytdl_sub/entries/variables/entry_variables.py +++ b/src/ytdl_sub/entries/variables/entry_variables.py @@ -1,3 +1,5 @@ +from yt_dlp.utils import sanitize_filename + from ytdl_sub.entries.base_entry import BaseEntry from ytdl_sub.entries.base_entry import BaseEntryVariables @@ -14,6 +16,72 @@ _days_in_month = [0, 31, 28, 31, 30, 31, 30, 31, 31, 30, 31, 30, 31] class EntryVariables(BaseEntryVariables): + @property + def playlist(self: BaseEntry) -> str: + """ + Returns + ------- + str + Name of its parent playlist/channel if it exists, otherwise returns its title. + """ + return self.kwargs_get("playlist", self.title) + + @property + def playlist_sanitized(self) -> str: + """ + Returns + ------- + str + The playlist name, sanitized + """ + return sanitize_filename(self.playlist) + + @property + def playlist_index(self: BaseEntry) -> int: + """ + Returns + ------- + int + Playlist index if it exists, otherwise returns ``1``. + + Note that for channels/playlists, an index of 1 implies it's the most recent + uploaded entry. It is recommended to not use this unless you know the channel/playlist + will never add new content, i.e. a music album. + """ + return self.kwargs_get("playlist_index", 1) + + @property + def playlist_index_padded(self) -> str: + """ + Returns + ------- + str + playlist_index padded two digits + """ + return _pad(self.playlist_index, width=2) + + @property + def playlist_count(self: BaseEntry) -> int: + """ + Returns + ------- + int + Playlist count if it exists, otherwise returns ``1``. + """ + return self.kwargs_get("playlist_count", 1) + + @property + def playlist_max_upload_year(self) -> int: + """ + Returns + ------- + int + Max upload_year for all entries in this entry's playlist if it exists, otherwise returns + ``upload_year`` + """ + # override in EntryParent + return self.upload_year + @property def ext(self: BaseEntry) -> str: """ diff --git a/src/ytdl_sub/ytdl_additions/enhanced_download_archive.py b/src/ytdl_sub/ytdl_additions/enhanced_download_archive.py index b39cacd7..24d0ef84 100644 --- a/src/ytdl_sub/ytdl_additions/enhanced_download_archive.py +++ b/src/ytdl_sub/ytdl_additions/enhanced_download_archive.py @@ -598,7 +598,7 @@ class DownloadArchiver: """ def __init__(self, enhanced_download_archive: EnhancedDownloadArchive): - self.__enhanced_download_archive = enhanced_download_archive + self._enhanced_download_archive = enhanced_download_archive @property def working_directory(self) -> str: @@ -607,7 +607,7 @@ class DownloadArchiver: ------- Path to the working directory """ - return self.__enhanced_download_archive.working_directory + return self._enhanced_download_archive.working_directory @property def is_dry_run(self) -> bool: @@ -616,7 +616,7 @@ class DownloadArchiver: ------- True if this session is a dry-run. False otherwise. """ - return self.__enhanced_download_archive.is_dry_run + return self._enhanced_download_archive.is_dry_run def save_file( self, @@ -640,7 +640,7 @@ class DownloadArchiver: entry Optional. Entry that the file belongs to """ - self.__enhanced_download_archive.save_file_to_output_directory( + self._enhanced_download_archive.save_file_to_output_directory( file_name=file_name, file_metadata=file_metadata, output_file_name=output_file_name,