danvics/ytdl-sub
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

[BACKEND] Use generic collection download for every playlist strategy (#233)

Browse source
This commit is contained in:
Jesse Bannon 2022-09-15 09:47:37 -07:00 committed by GitHub
parent 3894625479
commit 892c5f5e6f
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
8 changed files with 337 additions and 339 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

161
src/ytdl_sub/downloaders/downloader.py
View file

@ -9,10 +9,12 @@ from contextlib import contextmanager
from pathlib import Path
from typing import Callable
from typing import Dict
from typing import Generator
from typing import Generic
from typing import Iterable
from typing import List
from typing import Optional
from typing import Set
from typing import Tuple
from typing import Type
from typing import TypeVar
@ -23,11 +25,15 @@ from yt_dlp.utils import RejectedVideoReached
from ytdl_sub.config.preset_options import AddsVariablesMixin
from ytdl_sub.config.preset_options import Overrides
from ytdl_sub.downloaders.generic.collection_validator import CollectionUrlValidator
from ytdl_sub.downloaders.generic.collection_validator import CollectionValidator
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.thread.log_entries_downloaded_listener import LogEntriesDownloadedListener
from ytdl_sub.utils.exceptions import FileNotDownloadedException
from ytdl_sub.utils.file_handler import FileHandler
from ytdl_sub.utils.file_handler import FileMetadata
from ytdl_sub.utils.logger import Logger
from ytdl_sub.validators.strict_dict_validator import StrictDictValidator
@ -37,15 +43,60 @@ from ytdl_sub.ytdl_additions.enhanced_download_archive import EnhancedDownloadAr
download_logger = Logger.get(name="downloader")
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 DownloaderValidator(StrictDictValidator, AddsVariablesMixin, ABC):
"""
Placeholder class to define downloader options
"""
@property
@abc.abstractmethod
def collection_validator(self) -> CollectionValidator:
"""
Returns
-------
CollectionValidator
To determine how the entries are downloaded
"""
def added_source_variables(self) -> List[str]:
"""
Returns
-------
Added source variables on the collection
"""
return self.collection_validator.added_source_variables()
def validate_with_variables(
self, source_variables: List[str], override_variables: Dict[str, str]
) -> None:
"""
Validates any source variables added by the collection
"""
self.collection_validator.validate_with_variables(
source_variables=source_variables, override_variables=override_variables
)
DownloaderOptionsT = TypeVar("DownloaderOptionsT", bound=DownloaderValidator)
DownloaderEntryT = TypeVar("DownloaderEntryT", bound=Entry)
DownloaderParentEntryT = TypeVar("DownloaderParentEntryT", bound=BaseEntry)
class Downloader(DownloadArchiver, Generic[DownloaderOptionsT, DownloaderEntryT], ABC):
@ -110,6 +161,9 @@ class Downloader(DownloadArchiver, Generic[DownloaderOptionsT, DownloaderEntryT]
self.ytdl_option_defaults(), before=True
)
self.parents: List[EntryParent] = []
self.downloaded_entries: Set[str] = set()
@contextmanager
def ytdl_downloader(self, ytdl_options_overrides: Optional[Dict] = None) -> ytdl.YoutubeDL:
"""
@ -297,11 +351,114 @@ class Downloader(DownloadArchiver, Generic[DownloaderOptionsT, DownloaderEntryT]
return self._get_entry_dicts_from_info_json_files()
@abc.abstractmethod
###############################################################################################
# DOWNLOAD FUNCTIONS
@property
def collection(self) -> CollectionValidator:
"""Return the download options collection"""
return self.download_options.collection_validator
@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._ytdl_options_builder.to_dict().get("download_archive", "")
backup_archive_path = f"{archive_path}.backup"
# If archive path exists, maintain download archive is enable
if archive_file_exists := archive_path and 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 and os.path.isfile(archive_path):
FileHandler.delete(file_path=archive_path)
# If the archive file did exist, restore the backup
elif archive_file_exists:
FileHandler.copy(src_file_path=backup_archive_path, dst_file_path=archive_path)
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=None if self.is_dry_run else entry.is_downloaded,
url=entry.webpage_url,
ytdl_options_overrides={"writeinfojson": False, "skip_download": self.is_dry_run},
)
# 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"""
if parent.is_entry():
yield self._download_entry(parent.to_type(Entry))
return
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_url_metadata(self, collection_url: CollectionUrlValidator) -> List[EntryParent]:
"""
Downloads only info.json files and forms EntryParent trees
"""
with self._separate_download_archives():
entry_dicts = self.extract_info_via_info_json(
only_info_json=True,
url=collection_url.url,
log_prefix_on_info_json_dl="Downloading metadata for",
)
self.parents = EntryParent.from_entry_dicts(
entry_dicts=entry_dicts, working_directory=self.working_directory
)
return self.parents
def _download_url(
self, collection_url: CollectionUrlValidator, parents: List[EntryParent]
) -> Generator[Entry, None, None]:
"""
Downloads the leaf entries from EntryParent trees
"""
with self._separate_download_archives():
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,
) -> Iterable[DownloaderEntryT] | Iterable[Tuple[DownloaderEntryT, FileMetadata]]:
"""The function to perform the download of all media entries"""
# download the bottom-most urls first since they are top-priority
for collection_url in reversed(self.collection.collection_urls.list):
parents = self._download_url_metadata(collection_url=collection_url)
for entry in self._download_url(collection_url=collection_url, parents=parents):
yield entry
def post_download(self):
"""

256
src/ytdl_sub/downloaders/generic/collection.py
View file

@ -1,103 +1,10 @@
import contextlib
import os.path
from typing import Dict
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.downloaders.generic.collection_validator import CollectionValidator
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.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: 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"}
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)
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
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):
class CollectionDownloadOptions(CollectionValidator, DownloaderValidator):
"""
Downloads from multiple URLs. If an entry is returned from more than one URL, it will
resolve to the bottom-most URL settings.
@ -122,165 +29,12 @@ class CollectionDownloadOptions(DownloaderValidator):
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
def added_source_variables(self) -> List[str]:
"""
Returns
-------
List of variables added. The first collection url always contains all the variables.
"""
return list(self._urls.list[0].variables.keys())
def validate_with_variables(
self, source_variables: List[str], override_variables: Dict[str, str]
) -> None:
"""
Ensures new variables added are not existing variables
"""
for source_var_name in self.added_source_variables():
if source_var_name in source_variables:
raise self._validation_exception(
f"'{source_var_name}' cannot be used as a variable name because it "
f"is an existing source variable"
)
base_variables = dict(
override_variables, **{source_var: "dummy_string" for source_var in source_variables}
)
# Apply formatting to each new source variable, ensure it resolves
for collection_url in self.collection_urls.list:
for source_var_name, source_var_formatter_str in collection_url.variables.items():
_ = StringFormatterValidator(
name=f"{self._name}.{source_var_name}", value=source_var_formatter_str
).apply_formatter(base_variables)
def collection_validator(self) -> CollectionValidator:
"""Returns itself!"""
return self
class CollectionDownloader(Downloader[CollectionDownloadOptions, Entry]):
downloader_options_type = CollectionDownloadOptions
downloader_entry_type = Entry
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,
)
self.parents: List[EntryParent] = []
self.downloaded_entries: Set[str] = set()
@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._ytdl_options_builder.to_dict().get("download_archive", "")
backup_archive_path = f"{archive_path}.backup"
# If archive path exists, maintain download archive is enable
if archive_file_exists := archive_path and 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 and os.path.isfile(archive_path):
FileHandler.delete(file_path=archive_path)
# If the archive file did exist, restore the backup
elif archive_file_exists:
FileHandler.copy(src_file_path=backup_archive_path, dst_file_path=archive_path)
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=None if self.is_dry_run else entry.is_downloaded,
url=entry.webpage_url,
ytdl_options_overrides={"writeinfojson": False, "skip_download": self.is_dry_run},
)
# 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_url_metadata(self, collection_url: CollectionUrlValidator) -> List[EntryParent]:
"""
Downloads only info.json files and forms EntryParent trees
"""
with self._separate_download_archives():
entry_dicts = self.extract_info_via_info_json(
only_info_json=True,
url=collection_url.url,
log_prefix_on_info_json_dl="Downloading metadata for",
)
return EntryParent.from_entry_dicts(
entry_dicts=entry_dicts, working_directory=self.working_directory
)
def download_url(
self, collection_url: CollectionUrlValidator, parents: List[EntryParent]
) -> Generator[Entry, None, None]:
"""
Downloads the leaf entries from EntryParent trees
"""
with self._separate_download_archives():
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.
"""
# download the bottom-most urls first since they are top-priority
for collection_url in reversed(self.download_options.collection_urls.list):
parents = self.download_url_metadata(collection_url=collection_url)
for entry in self.download_url(collection_url=collection_url, parents=parents):
yield entry

117
src/ytdl_sub/downloaders/generic/collection_validator.py Normal file
View file

@ -0,0 +1,117 @@
from typing import Dict
from typing import List
from ytdl_sub.config.preset_options import AddsVariablesMixin
from ytdl_sub.validators.strict_dict_validator import StrictDictValidator
from ytdl_sub.validators.string_formatter_validators import DictFormatterValidator
from ytdl_sub.validators.string_formatter_validators import StringFormatterValidator
from ytdl_sub.validators.validators import ListValidator
from ytdl_sub.validators.validators import StringValidator
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)
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
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 CollectionValidator(StrictDictValidator, AddsVariablesMixin):
"""
Downloads from multiple URLs. If an entry is returned from more than one URL, it will
resolve to the bottom-most URL settings.
"""
_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
def added_source_variables(self) -> List[str]:
"""
Returns
-------
List of variables added. The first collection url always contains all the variables.
"""
return list(self._urls.list[0].variables.keys())
def validate_with_variables(
self, source_variables: List[str], override_variables: Dict[str, str]
) -> None:
"""
Ensures new variables added are not existing variables
"""
for source_var_name in self.added_source_variables():
if source_var_name in source_variables:
raise self._validation_exception(
f"'{source_var_name}' cannot be used as a variable name because it "
f"is an existing source variable"
)
base_variables = dict(
override_variables, **{source_var: "dummy_string" for source_var in source_variables}
)
# Apply formatting to each new source variable, ensure it resolves
for collection_url in self.collection_urls.list:
for source_var_name, source_var_formatter_str in collection_url.variables.items():
_ = StringFormatterValidator(
name=f"{self._name}.{source_var_name}", value=source_var_formatter_str
).apply_formatter(base_variables)

30
src/ytdl_sub/downloaders/soundcloud/albums_and_singles.py
View file

@ -1,11 +1,9 @@
from typing import Dict
from typing import Generator
from typing import List
from ytdl_sub.downloaders.downloader import Downloader
from ytdl_sub.downloaders.downloader import DownloaderValidator
from ytdl_sub.downloaders.generic.collection import CollectionDownloader
from ytdl_sub.downloaders.generic.collection import CollectionDownloadOptions
from ytdl_sub.downloaders.generic.collection_validator import CollectionValidator
from ytdl_sub.entries.entry import Entry
from ytdl_sub.validators.url_validator import SoundcloudUsernameUrlValidator
from ytdl_sub.validators.validators import BoolValidator
@ -43,7 +41,10 @@ class SoundcloudAlbumsAndSinglesDownloadOptions(DownloaderValidator):
"skip_premiere_tracks", BoolValidator, default=True
)
self.collection_validator = CollectionDownloadOptions(
@property
def collection_validator(self) -> CollectionValidator:
"""Downloads the album tracks first, then the tracks"""
return CollectionValidator(
name=self._name,
value={
"urls": [
@ -87,18 +88,6 @@ class SoundcloudAlbumsAndSinglesDownloadOptions(DownloaderValidator):
"""
return self._url
def added_source_variables(self) -> List[str]:
"""Validate using collection_validator"""
return self.collection_validator.added_source_variables()
def validate_with_variables(
self, source_variables: List[str], override_variables: Dict[str, str]
) -> None:
"""Validate using collection_validator"""
return self.collection_validator.validate_with_variables(
source_variables, override_variables
)
class SoundcloudAlbumsAndSinglesDownloader(
Downloader[SoundcloudAlbumsAndSinglesDownloadOptions, Entry]
@ -141,14 +130,7 @@ class SoundcloudAlbumsAndSinglesDownloader(
"""
Soundcloud subscription to download albums and tracks as singles.
"""
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 entry in downloader.download():
for entry in super().download():
if self._should_skip(entry):
continue

49
src/ytdl_sub/downloaders/youtube/channel.py
View file

@ -4,20 +4,15 @@ from typing import Generator
from typing import List
from typing import Optional
from ytdl_sub.config.preset_options import Overrides
from ytdl_sub.downloaders.downloader import DownloaderOptionsT
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.generic.collection_validator import CollectionValidator
from ytdl_sub.downloaders.youtube.abc import YoutubeDownloader
from ytdl_sub.downloaders.youtube.abc import YoutubeDownloaderOptions
from ytdl_sub.downloaders.ytdl_options_builder import YTDLOptionsBuilder
from ytdl_sub.entries.entry_parent import EntryParent
from ytdl_sub.entries.youtube import YoutubeVideo
from ytdl_sub.utils.thumbnail import convert_url_thumbnail
from ytdl_sub.validators.string_formatter_validators import OverridesStringFormatterValidator
from ytdl_sub.validators.url_validator import YoutubeChannelUrlValidator
from ytdl_sub.ytdl_additions.enhanced_download_archive import EnhancedDownloadArchive
class YoutubeChannelDownloaderOptions(YoutubeDownloaderOptions):
@ -57,9 +52,12 @@ class YoutubeChannelDownloaderOptions(YoutubeDownloaderOptions):
"channel_banner_path", OverridesStringFormatterValidator
)
self.collection_validator = CollectionDownloadOptions(
@property
def collection_validator(self) -> CollectionValidator:
"""Download from the channel url"""
return CollectionValidator(
name=self._name,
value={"urls": [{"url": self._channel_url}]},
value={"urls": [{"url": self.channel_url}]},
)
@property
@ -127,37 +125,18 @@ class YoutubeChannelDownloader(YoutubeDownloader[YoutubeChannelDownloaderOptions
# pylint: enable=line-too-long
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,
)
self.channel: Optional[EntryParent] = None
@property
def channel(self) -> EntryParent:
"""Gets the channel entry parent"""
assert len(self.parents) == 1, "Channel should be the only entry parent"
return self.parents[0]
def download(self) -> Generator[YoutubeVideo, None, None]:
"""
Downloads all videos from a channel
"""
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,
)
collection_url = self.download_options.collection_validator.collection_urls.list[0]
parents = downloader.download_url_metadata(collection_url=collection_url)
assert len(parents) == 1, "Channel should be the only entry parent"
self.channel = parents[0]
collection_url = self.collection.collection_urls.list[0]
super()._download_url_metadata(collection_url=collection_url)
# TODO: Handle this better
self.overrides.add_override_variables(
@ -168,7 +147,7 @@ class YoutubeChannelDownloader(YoutubeDownloader[YoutubeChannelDownloaderOptions
}
)
for entry in downloader.download_url(collection_url=collection_url, parents=parents):
for entry in super()._download_url(collection_url=collection_url, parents=self.parents):
# pylint: disable=protected-access
yield YoutubeVideo(entry_dict=entry._kwargs, working_directory=self.working_directory)
# pylint: enable=protected-access

6
src/ytdl_sub/downloaders/youtube/merge_playlist.py
View file

@ -147,9 +147,9 @@ class YoutubeMergePlaylistDownloader(
def download(self) -> List[Tuple[YoutubeVideo, FileMetadata]]:
"""Download a single Youtube video, then split it into multiple videos"""
merged_video = self._to_merged_video(
entry_dict=self.extract_info(url=self.download_options.playlist_url)
)
entry_dict = self.extract_info(url=self.collection.collection_urls.list[0].url)
merged_video = self._to_merged_video(entry_dict=entry_dict)
merged_video_metadata = self._get_chapters(
merged_video=merged_video, add_chapters=self.download_options.add_chapters
)

38
src/ytdl_sub/downloaders/youtube/playlist.py
View file

@ -2,10 +2,10 @@ from typing import Dict
from typing import Generator
from typing import List
from ytdl_sub.downloaders.generic.collection import CollectionDownloader
from ytdl_sub.downloaders.generic.collection import CollectionDownloadOptions
from ytdl_sub.downloaders.generic.collection_validator import CollectionValidator
from ytdl_sub.downloaders.youtube.abc import YoutubeDownloader
from ytdl_sub.downloaders.youtube.abc import YoutubeDownloaderOptions
from ytdl_sub.entries.entry_parent import EntryParent
from ytdl_sub.entries.youtube import YoutubePlaylistVideo
from ytdl_sub.validators.url_validator import YoutubePlaylistUrlValidator
@ -34,9 +34,12 @@ class YoutubePlaylistDownloaderOptions(YoutubeDownloaderOptions):
"playlist_url", YoutubePlaylistUrlValidator
).playlist_url
self.collection_validator = CollectionDownloadOptions(
@property
def collection_validator(self) -> CollectionValidator:
"""Downloads the playlist url"""
return CollectionValidator(
name=self._name,
value={"urls": [{"url": self._playlist_url}]},
value={"urls": [{"url": self.playlist_url}]},
)
@property
@ -87,32 +90,29 @@ class YoutubePlaylistDownloader(
# pylint: enable=line-too-long
@property
def playlist(self) -> EntryParent:
"""Get the playlist parent entry"""
assert len(self.parents) == 1, "Playlist should be the only entry parent"
return self.parents[0]
def download(self) -> Generator[YoutubePlaylistVideo, None, None]:
"""
Downloads all videos in a Youtube playlist.
"""
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,
)
collection_url = self.download_options.collection_validator.collection_urls.list[0]
parents = downloader.download_url_metadata(collection_url=collection_url)
assert len(parents) == 1, "Playlist should be the only entry parent"
playlist = parents[0]
collection_url = self.collection.collection_urls.list[0]
super()._download_url_metadata(collection_url)
# TODO: Handle this better
self.overrides.add_override_variables(
variables_to_add={
"source_title": playlist.title,
"source_uploader": playlist.kwargs_get("uploader", "__failed_to_scrape__"),
"source_description": playlist.kwargs_get("description", ""),
"source_title": self.playlist.title,
"source_uploader": self.playlist.kwargs_get("uploader", "__failed_to_scrape__"),
"source_description": self.playlist.kwargs_get("description", ""),
}
)
for entry in downloader.download_url(collection_url=collection_url, parents=parents):
for entry in super()._download_url(collection_url=collection_url, parents=self.parents):
# pylint: disable=protected-access
yield YoutubePlaylistVideo(
entry_dict=entry._kwargs, working_directory=self.working_directory

19
src/ytdl_sub/downloaders/youtube/video.py
View file

@ -1,6 +1,7 @@
from typing import Dict
from typing import List
from ytdl_sub.downloaders.generic.collection_validator import CollectionValidator
from ytdl_sub.downloaders.youtube.abc import YoutubeDownloader
from ytdl_sub.downloaders.youtube.abc import YoutubeDownloaderOptions
from ytdl_sub.entries.youtube import YoutubeVideo
@ -36,6 +37,14 @@ class YoutubeVideoDownloaderOptions(YoutubeDownloaderOptions):
super().__init__(name, value)
self._video_url = self._validate_key("video_url", YoutubeVideoUrlValidator).video_url
@property
def collection_validator(self) -> CollectionValidator:
"""Downloads the video url"""
return CollectionValidator(
name=self._name,
value={"urls": [{"url": self.video_url}]},
)
@property
def video_url(self) -> str:
"""
@ -64,8 +73,8 @@ class YoutubeVideoDownloader(YoutubeDownloader[YoutubeVideoDownloaderOptions, Yo
)
def download(self) -> List[YoutubeVideo]:
"""Download a single Youtube video"""
entry_dict = self.extract_info(url=self.download_options.video_url)
video = YoutubeVideo(entry_dict=entry_dict, working_directory=self.working_directory)
return [video]
"""Downloads the single video"""
for entry in super().download():
# pylint: disable=protected-access
yield YoutubeVideo(entry_dict=entry._kwargs, working_directory=self.working_directory)
# pylint: enable=protected-access