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

simplify

Browse source
This commit is contained in:
Jesse Bannon 2022-09-14 09:58:30 -07:00
parent b72c8c781d
commit 0df1135721
5 changed files with 112 additions and 195 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -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

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

@ -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,

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

@ -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,

16
src/ytdl_sub/entries/base_entry.py
View file

@ -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)

127
src/ytdl_sub/entries/entry_parent.py
View file

@ -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)