diff --git a/src/ytdl_sub/downloaders/generic/collection.py b/src/ytdl_sub/downloaders/generic/collection.py index 3488b672..b9585383 100644 --- a/src/ytdl_sub/downloaders/generic/collection.py +++ b/src/ytdl_sub/downloaders/generic/collection.py @@ -5,9 +5,13 @@ from typing import Generator from typing import List from typing import Set +from ytdl_sub.config.preset_options import Overrides from ytdl_sub.downloaders.downloader import Downloader +from ytdl_sub.downloaders.downloader import DownloaderOptionsT from ytdl_sub.downloaders.downloader import DownloaderValidator from ytdl_sub.downloaders.downloader import download_logger +from ytdl_sub.downloaders.ytdl_options_builder import YTDLOptionsBuilder +from ytdl_sub.entries.base_entry import BaseEntry from ytdl_sub.entries.entry import Entry from ytdl_sub.entries.entry_parent import EntryParent from ytdl_sub.utils.file_handler import FileHandler @@ -16,12 +20,27 @@ from ytdl_sub.validators.string_formatter_validators import DictFormatterValidat from ytdl_sub.validators.string_formatter_validators import StringFormatterValidator from ytdl_sub.validators.validators import ListValidator from ytdl_sub.validators.validators import StringValidator +from ytdl_sub.ytdl_additions.enhanced_download_archive import EnhancedDownloadArchive -def _entry_key(entry: Entry) -> str: +def _entry_key(entry: BaseEntry) -> str: return entry.extractor + entry.uid +def _get_parent_entry_variables(parent: EntryParent) -> Dict[str, str | int]: + """ + Adds source variables to the child entry derived from the parent entry. + """ + if not parent.child_entries: + return {} + + return { + "playlist_max_upload_year": max( + child_entry.to_type(Entry).upload_year for child_entry in parent.child_entries + ) + } + + class CollectionUrlValidator(StrictDictValidator): _required_keys = {"url"} _optional_keys = {"variables"} @@ -153,56 +172,22 @@ class CollectionDownloader(Downloader[CollectionDownloadOptions, Entry]): downloader_options_type = CollectionDownloadOptions downloader_entry_type = Entry - def _recursive_child_read( - self, collection_url: CollectionUrlValidator, parent: EntryParent, entry_dicts: List[Dict] - ) -> List[Entry]: - leaf_children: List[Entry] = [] - parent.read_nested_children_from_entry_dicts(entry_dicts=entry_dicts) - - for idx in range(parent.child_count): - child: EntryParent = parent.child_entries[idx] - - leaf_children.extend( - self._recursive_child_read( - collection_url=collection_url, parent=child, entry_dicts=entry_dicts - ) - ) - - # If the child has no nested children, turn it into an Entry - if not child.child_count: - parent.child_entries[idx] = child.to_entry() - leaf_children.append(parent.child_entries[idx]) - - for leaf_child in leaf_children: - leaf_child.add_variables(parent.get_children_entry_variables_to_add()) - 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, - url=collection_url.url, + def __init__( + self, + download_options: DownloaderOptionsT, + enhanced_download_archive: EnhancedDownloadArchive, + ytdl_options_builder: YTDLOptionsBuilder, + overrides: Overrides, + ): + super().__init__( + download_options=download_options, + enhanced_download_archive=enhanced_download_archive, + ytdl_options_builder=ytdl_options_builder, + overrides=overrides, ) - # initialize top-level parents, determined by whether a playlist_id exists in the entry dict - parents: List[EntryParent] = [ - EntryParent(entry_dict=entry_dict, working_directory=self.working_directory) - for entry_dict in entry_dicts - if "playlist_id" not in entry_dict - ] - - leaf_children: List[Entry] = [] - for parent in parents: - leaf_children.extend( - self._recursive_child_read( - collection_url=collection_url, parent=parent, entry_dicts=entry_dicts - ) - ) - - return leaf_children + self.parents: List[EntryParent] = [] + self.downloaded_entries: Set[str] = set() @contextlib.contextmanager def _separate_download_archives(self): @@ -233,48 +218,58 @@ class CollectionDownloader(Downloader[CollectionDownloadOptions, Entry]): elif archive_file_exists: FileHandler.copy(src_file_path=backup_archive_path, dst_file_path=archive_path) - def _download_leaf_entry(self, entry: Entry) -> Entry: + def _download_entry(self, entry: Entry) -> Entry: download_logger.info("Downloading entry %s", entry.title) + download_entry_dict = self.extract_info_with_retry( + is_downloaded_fn=entry.is_downloaded, + url=entry.webpage_url, + ytdl_options_overrides={"writeinfojson": False, "skip_download": self.is_dry_run}, + ) - # 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, - url=entry.webpage_url, - ytdl_options_overrides={"writeinfojson": False}, - ) - - # Workaround for the ytdlp issue - # pylint: disable=protected-access - entry._kwargs["requested_subtitles"] = download_entry_dict.get("requested_subtitles") - # pylint: enable=protected-access + # Workaround for the ytdlp issue + # pylint: disable=protected-access + entry._kwargs["requested_subtitles"] = download_entry_dict.get("requested_subtitles") + # pylint: enable=protected-access return entry + def _download_parent_entry(self, parent: EntryParent) -> Generator[Entry, None, None]: + """Download in reverse order, that way we download older entries ones first""" + for entry_child in reversed(parent.entry_children()): + if _entry_key(entry_child) in self.downloaded_entries: + continue + + yield self._download_entry(entry_child.to_type(Entry)) + self.downloaded_entries.add(_entry_key(entry_child)) + + for parent_child in reversed(parent.parent_children()): + for entry_child in self._download_parent_entry(parent=parent_child): + yield entry_child + def _download_collection_url( - self, collection_url: CollectionUrlValidator, downloaded_entries: Set[str] + self, collection_url: CollectionUrlValidator ) -> 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 - ] + entry_dicts = self.extract_info_via_info_json( + only_info_json=True, + url=collection_url.url, + ) - # 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)) + parents = EntryParent.from_entry_dicts( + entry_dicts=entry_dicts, working_directory=self.working_directory + ) + for parent in parents: + for entry_child in self._download_parent_entry(parent=parent): + entry_child.add_variables( + dict(_get_parent_entry_variables(parent), **collection_url.variables) + ) + yield entry_child def download(self) -> Generator[Entry, None, None]: """ Soundcloud subscription to download albums and tracks as singles. """ - 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): - for entry in self._download_collection_url( - collection_url=collection_url, downloaded_entries=downloaded_entries - ): + for entry in self._download_collection_url(collection_url=collection_url): yield entry diff --git a/src/ytdl_sub/downloaders/youtube/channel.py b/src/ytdl_sub/downloaders/youtube/channel.py index d185a71a..08e38cb0 100644 --- a/src/ytdl_sub/downloaders/youtube/channel.py +++ b/src/ytdl_sub/downloaders/youtube/channel.py @@ -180,12 +180,12 @@ class YoutubeChannelDownloader(YoutubeDownloader[YoutubeChannelDownloaderOptions url=self.download_options.channel_url, ) - self.channel = EntryParent.from_entry_dicts_with_children( + parents: List[EntryParent] = EntryParent.from_entry_dicts( entry_dicts=entry_dicts, working_directory=self.working_directory, - child_class=YoutubeVideo, - extractor="youtube:tab", ) + assert len(parents) == 1, "Channel should be the only parent" + self.channel = parents[0] self.overrides.add_override_variables( variables_to_add={ @@ -199,6 +199,7 @@ class YoutubeChannelDownloader(YoutubeDownloader[YoutubeChannelDownloaderOptions # the channel must be redownloaded, it will fetch most recent metadata first, and break # on the older video that's been processed and is in the download archive. for idx, video in enumerate(reversed(self.channel.child_entries), start=1): + video = video.to_type(YoutubeVideo) download_logger.info("Downloading %d/%d %s", idx, self.channel.child_count, video.title) # Re-download the contents even if it's a dry-run as a single video. At this time, diff --git a/src/ytdl_sub/downloaders/youtube/playlist.py b/src/ytdl_sub/downloaders/youtube/playlist.py index 25d08eba..6b0d03f9 100644 --- a/src/ytdl_sub/downloaders/youtube/playlist.py +++ b/src/ytdl_sub/downloaders/youtube/playlist.py @@ -95,12 +95,13 @@ class YoutubePlaylistDownloader( url=self.download_options.playlist_url, ) - playlist: EntryParent = EntryParent.from_entry_dicts_with_children( + parents: List[EntryParent] = EntryParent.from_entry_dicts( entry_dicts=entry_dicts, working_directory=self.working_directory, - child_class=YoutubePlaylistVideo, - extractor="youtube:tab", ) + assert len(parents) == 1, "Playlist should be the only parent" + playlist = parents[0] + self.overrides.add_override_variables( variables_to_add={ "source_title": playlist.title, @@ -113,6 +114,7 @@ class YoutubePlaylistDownloader( # the playlist must be redownloaded, it will fetch most recent metadata first, and break # on the older video that's been processed and is in the download archive. for idx, video in enumerate(reversed(playlist.child_entries), start=1): + video = video.to_type(YoutubePlaylistVideo) download_logger.info("Downloading %d/%d %s", idx, len(entry_dicts), video.title) # Re-download the contents even if it's a dry-run as a single video. At this time, diff --git a/src/ytdl_sub/entries/base_entry.py b/src/ytdl_sub/entries/base_entry.py index c861827f..1dc6f7ca 100644 --- a/src/ytdl_sub/entries/base_entry.py +++ b/src/ytdl_sub/entries/base_entry.py @@ -187,19 +187,3 @@ class BaseEntry(BaseEntryVariables, ABC): source_var: getattr(self, source_var) for source_var in self.source_variables() } return dict(source_variable_dict, **self._added_variables()) - - # TODO: super typing - @classmethod - def from_entry_dicts( - cls, entry_dicts: List[Dict], working_directory: str, extractor: Optional[str] - ): - """ - Load the entry from a list of dicts. There should only be one entry with the extractor - type - """ - extractor = extractor or cls.extractor - output = [ - entry_dict for entry_dict in entry_dicts if entry_dict.get("extractor") == extractor - ] - assert len(output) == 1, f"Expected a single entry with extractor of type '{extractor}'" - return cls(entry_dict=output[0], working_directory=working_directory) diff --git a/src/ytdl_sub/entries/entry_parent.py b/src/ytdl_sub/entries/entry_parent.py index 830e2f94..966682f0 100644 --- a/src/ytdl_sub/entries/entry_parent.py +++ b/src/ytdl_sub/entries/entry_parent.py @@ -5,36 +5,34 @@ from typing import Type from typing import TypeVar from ytdl_sub.entries.base_entry import BaseEntry -from ytdl_sub.entries.entry import Entry -TChildEntry = TypeVar("TChildEntry", bound=BaseEntry) +TBaseEntry = TypeVar("TBaseEntry", bound=BaseEntry) class EntryParent(BaseEntry): def __init__( - self, entry_dict: Dict, working_directory: str, entry_parent: Optional["EntryParent"] = None + self, entry_dict: Dict, working_directory: str ): super().__init__(entry_dict=entry_dict, working_directory=working_directory) - self._child_entries: List[TChildEntry] = [] - self._entry_parent = entry_parent + self.child_entries: List["EntryParent"] = [] - # pylint: disable=no-self-use - def get_children_entry_variables_to_add(self) -> Dict[str, str | int]: + def is_entry(self) -> bool: """ - Adds source variables to the child entry derived from the parent entry. + Returns + ------- + True if the entry contains a media file. False otherwise. """ - if not self.child_entries: - return {} + return self.kwargs_contains("ext") - return { - "playlist_max_upload_year": max( - entry.upload_year for entry in self.child_entries if isinstance(entry, Entry) - ) - } + def parent_children(self) -> List["EntryParent"]: + """This parent's children that are also parents""" + return [child for child in self.child_entries if child.child_count() > 0] - # pylint: enable=no-self-use + def entry_children(self) -> List["EntryParent"]: + """This parent's children that are entries""" + return [child for child in self.child_entries if child.is_entry()] - def read_nested_children_from_entry_dicts(self, entry_dicts: List[Dict]): + def read_children_from_entry_dicts(self, entry_dicts: List[Dict]) -> "EntryParent": """ Populates a tree of EntryParents that belong to this instance """ @@ -46,56 +44,13 @@ class EntryParent(BaseEntry): self.__class__( entry_dict=entry_dict, working_directory=self.working_directory(), - entry_parent=self, ) ) + child_entries[-1].read_children_from_entry_dicts(entry_dicts) - 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": - """ - Parameters - ---------- - entry_dicts - Entry dicts to look for children from - child_class - The class to convert the entry dict to child_class. Defaults to Entry - - Returns - ------- - List of children - """ - child_entries: List[TChildEntry] = [] - - for entry_dict in entry_dicts: - if entry_dict.get("playlist_id") == self.uid: - child_entries.append( - child_class(entry_dict=entry_dict, working_directory=self.working_directory()) - ) - - self._child_entries = sorted( - child_entries, key=lambda entry: entry.kwargs("playlist_index") - ) - for child_entry in self._child_entries: - child_entry.add_variables(variables_to_add=self.get_children_entry_variables_to_add()) - + self.child_entries = sorted(child_entries, key=lambda entry: entry.kwargs("playlist_index")) return self - @property - def child_entries(self) -> List[TChildEntry]: - """ - Returns - ------- - List of child entities - """ - return self._child_entries - - @property def child_count(self) -> int: """ Returns @@ -122,45 +77,25 @@ class EntryParent(BaseEntry): return thumbnail["url"] return None - def __contains__(self, item): - """ - Returns - ------- - True if the the item (entry_dict) has the same id as one of the tracks. False otherwise. - """ - uid: Optional[str] = None - if isinstance(item, BaseEntry): - uid = item.uid - elif isinstance(item, dict): - uid = item.get("id") - - if uid is not None: - return any(uid == child_entry.uid for child_entry in self._child_entries) - return False - @classmethod - def from_entry_dicts_with_children( - cls, - entry_dicts: List[Dict], - working_directory: str, - child_class: Type[TChildEntry], - extractor: Optional[str], - ) -> "EntryParent": + def from_entry_dicts( + cls, entry_dicts: List[Dict], working_directory: str + ) -> List["EntryParent"]: """ - Load the parent entry and its children + Reads all entry dicts and builds a tree of EntryParents """ - entry_parent: EntryParent = cls.from_entry_dicts( - entry_dicts=entry_dicts, working_directory=working_directory, extractor=extractor - ) - entry_parent.read_children_from_entry_dicts( - entry_dicts=entry_dicts, child_class=child_class - ) - return entry_parent + return [ + EntryParent( + entry_dict=entry_dict, working_directory=working_directory + ).read_children_from_entry_dicts(entry_dicts) + for entry_dict in entry_dicts + if "playlist_id" not in entry_dict + ] - def to_entry(self) -> Entry: + def to_type(self, entry_type: Type[TBaseEntry]) -> TBaseEntry: """ Returns ------- - EntryParent converted to Entry + Converted EntryParent to Entry-like class """ - return Entry(entry_dict=self._kwargs, working_directory=self._working_directory) + return entry_type(entry_dict=self._kwargs, working_directory=self._working_directory)