diff --git a/src/ytdl_sub/downloaders/generic/__init__.py b/src/ytdl_sub/downloaders/generic/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/src/ytdl_sub/downloaders/generic/collection.py b/src/ytdl_sub/downloaders/generic/collection.py new file mode 100644 index 00000000..8aaf58bf --- /dev/null +++ b/src/ytdl_sub/downloaders/generic/collection.py @@ -0,0 +1,195 @@ +from typing import Dict +from typing import Generator +from typing import List +from typing import Set + +from ytdl_sub.downloaders.downloader import Downloader +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.validators.strict_dict_validator import StrictDictValidator +from ytdl_sub.validators.string_formatter_validators import DictFormatterValidator +from ytdl_sub.validators.validators import ListValidator +from ytdl_sub.validators.validators import StringValidator + + +def _entry_key(entry: Entry) -> str: + return entry.extractor + entry.uid + + +class CollectionUrlValidator(StrictDictValidator): + _required_keys = {"url"} + _optional_keys = {"variables"} + + def __init__(self, name, value): + super().__init__(name, value) + + # 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 + + @property + def url(self) -> str: + return self._url.value + + @property + def variables(self) -> Dict[str, str]: + return self._variables + + +class CollectionUrlListValidator(ListValidator[CollectionUrlValidator]): + _inner_list_type = CollectionUrlValidator + _expected_value_type_name = "collection url list" + + def __init__(self, name, value): + super().__init__(name, value) + + added_variables: Dict[str, str] = self.list[0].variables + + for idx, collection_url_validator in enumerate(self.list[1:]): + collection_variables = collection_url_validator.variables + + # see if this collection contains new added vars (it should not) + for var in collection_variables.keys(): + if var not in added_variables: + raise self._validation_exception( + f"Collection url {idx} contains the variable '{var}' that the first " + f"collection url does not. The first collection url must define all added " + f"variables." + ) + + # see if this collection is missing any added vars (if so, inherit from the top) + for var in added_variables.keys(): + if var not in collection_variables: + collection_variables[var] = added_variables[var] + + +class CollectionDownloadOptions(DownloaderValidator): + """ + Downloads from multiple URLs. If an entry is returned from more than one URL, it will + resolve to the bottom-most URL settings. + + Usage: + + .. code-block:: yaml + + presets: + my_example_preset: + generic: + # required + download_strategy: "collection" + urls: + - url: "soundcloud.com/tracks" + variables: + season: "1" + album: "{title}" + - url: "soundcloud.com/albums" + variables: + season: "1" + album: "{playlist_title}" + """ + + _required_keys = {"urls"} + + def __init__(self, name, value): + super().__init__(name, value) + self._urls = self._validate_key(key="urls", validator=CollectionUrlListValidator) + + @property + def collection_urls(self) -> CollectionUrlListValidator: + """ + Required. The Soundcloud user's url, i.e. ``soundcloud.com/the_username`` + """ + return self._urls + + +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_children_from_entry_dicts(entry_dicts=entry_dicts, child_class=EntryParent) + + for idx in range(parent.child_count): + child = 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: + entry = child.to_entry() + entry.add_variables(collection_url.variables) + parent.child_entries[idx] = entry + + leaf_children.append(entry) + + return leaf_children + + def _get_leaf_entries(self, collection_url: CollectionUrlValidator) -> List[Entry]: + # Dry-run to get the info json files + 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, + ) + + # 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 + + def _download_leaf_entry(self, entry: Entry): + download_logger.info("Downloading entry %s", entry.title) + 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 + + yield entry + + def download(self) -> Generator[Entry, None, None]: + """ + Soundcloud subscription to download albums and tracks as singles. + """ + downloaded_leaf_children: 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) diff --git a/src/ytdl_sub/entries/entry_parent.py b/src/ytdl_sub/entries/entry_parent.py index c2af6341..15596fe0 100644 --- a/src/ytdl_sub/entries/entry_parent.py +++ b/src/ytdl_sub/entries/entry_parent.py @@ -5,14 +5,18 @@ 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) class EntryParent(BaseEntry): - def __init__(self, entry_dict: Dict, working_directory: str): + def __init__( + self, entry_dict: Dict, working_directory: str, entry_parent: Optional["EntryParent"] = None + ): super().__init__(entry_dict=entry_dict, working_directory=working_directory) self._child_entries: List[TChildEntry] = [] + self._entry_parent = entry_parent # pylint: disable=no-self-use def _get_children_entry_variables_to_add( @@ -29,7 +33,7 @@ class EntryParent(BaseEntry): # pylint: enable=no-self-use def read_children_from_entry_dicts( - self, entry_dicts: List[Dict], child_class: Type[TChildEntry] + self, entry_dicts: List[Dict], child_class: Type[TChildEntry] = Entry ) -> "EntryParent": """ Parameters @@ -37,7 +41,7 @@ class EntryParent(BaseEntry): entry_dicts Entry dicts to look for children from child_class - The class to convert the entry dict to + The class to convert the entry dict to child_class. Defaults to Entry Returns ------- @@ -130,3 +134,13 @@ class EntryParent(BaseEntry): entry_dicts=entry_dicts, child_class=child_class ) 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