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

working aside from variables in deep trees

Browse source
This commit is contained in:
Jesse Bannon 2022-09-16 12:46:16 -07:00
parent 3719b14103
commit 05e69d4d3b
9 changed files with 282 additions and 74 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
src/ytdl_sub/config/preset_class_mappings.py
View file

@ -3,6 +3,7 @@ from typing import List
from typing import Type
from ytdl_sub.downloaders.downloader import Downloader
from ytdl_sub.downloaders.generic.collection import CollectionDownloader
from ytdl_sub.downloaders.soundcloud.albums_and_singles import SoundcloudAlbumsAndSinglesDownloader
from ytdl_sub.downloaders.youtube.channel import YoutubeChannelDownloader
from ytdl_sub.downloaders.youtube.merge_playlist import YoutubeMergePlaylistDownloader
@ -36,6 +37,9 @@ class DownloadStrategyMapping:
"soundcloud": {
"albums_and_singles": SoundcloudAlbumsAndSinglesDownloader,
},
"generic": {
"collection": CollectionDownloader,
},
}
@classmethod

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

@ -404,23 +404,22 @@ class Downloader(DownloadArchiver, Generic[DownloaderOptionsT, DownloaderEntryT]
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
# Download the parent's entries first, in reverse order
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))
yield self._download_entry(entry_child)
self.downloaded_entries.add(_entry_key(entry_child))
# Recursion the parent's parent entries
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]:
def _download_url_metadata(
self, collection_url: CollectionUrlValidator
) -> Tuple[List[EntryParent], List[Entry]]:
"""
Downloads only info.json files and forms EntryParent trees
"""
@ -434,14 +433,24 @@ class Downloader(DownloadArchiver, Generic[DownloaderOptionsT, DownloaderEntryT]
self.parents = EntryParent.from_entry_dicts(
entry_dicts=entry_dicts, working_directory=self.working_directory
)
return self.parents
orphans = EntryParent.from_entry_dicts_with_no_parents(
parents=self.parents, entry_dicts=entry_dicts, working_directory=self.working_directory
)
return self.parents, orphans
def _download_url(
self, collection_url: CollectionUrlValidator, parents: List[EntryParent]
self,
collection_url: CollectionUrlValidator,
parents: List[EntryParent],
orphans: List[Entry] = None,
) -> Generator[Entry, None, None]:
"""
Downloads the leaf entries from EntryParent trees
"""
if orphans is None:
orphans = []
with self._separate_download_archives():
for parent in parents:
for entry_child in self._download_parent_entry(parent=parent):
@ -450,14 +459,19 @@ class Downloader(DownloadArchiver, Generic[DownloaderOptionsT, DownloaderEntryT]
)
yield entry_child
for orphan in orphans:
yield self._download_entry(orphan)
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):
parents, orphan_entries = self._download_url_metadata(collection_url=collection_url)
for entry in self._download_url(
collection_url=collection_url, parents=parents, orphans=orphan_entries
):
yield entry
def post_download(self):

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

@ -136,7 +136,8 @@ class YoutubeChannelDownloader(YoutubeDownloader[YoutubeChannelDownloaderOptions
Downloads all videos from a channel
"""
collection_url = self.collection.collection_urls.list[0]
super()._download_url_metadata(collection_url=collection_url)
_, orphans = super()._download_url_metadata(collection_url=collection_url)
assert not orphans
# TODO: Handle this better
self.overrides.add_override_variables(
@ -148,9 +149,7 @@ class YoutubeChannelDownloader(YoutubeDownloader[YoutubeChannelDownloaderOptions
)
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
yield entry.to_type(YoutubeVideo)
def _download_thumbnail(
self,

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

@ -101,7 +101,8 @@ class YoutubePlaylistDownloader(
Downloads all videos in a Youtube playlist.
"""
collection_url = self.collection.collection_urls.list[0]
super()._download_url_metadata(collection_url)
_, orphans = super()._download_url_metadata(collection_url)
assert not orphans
# TODO: Handle this better
self.overrides.add_override_variables(
@ -113,8 +114,4 @@ class YoutubePlaylistDownloader(
)
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
)
# pylint: enable=protected-access
yield entry.to_type(YoutubePlaylistVideo)

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

@ -75,6 +75,4 @@ class YoutubeVideoDownloader(YoutubeDownloader[YoutubeVideoDownloaderOptions, Yo
def download(self) -> List[YoutubeVideo]:
"""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
yield entry.to_type(YoutubeVideo)

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

@ -1,8 +1,11 @@
from abc import ABC
from pathlib import Path
from typing import Any
from typing import Dict
from typing import List
from typing import Optional
from typing import Type
from typing import TypeVar
from typing import final
from yt_dlp.utils import sanitize_filename
@ -45,9 +48,9 @@ class BaseEntryVariables:
Returns
-------
str
The title of the entry
The title of the entry. If a title does not exist, returns its unique ID.
"""
return self.kwargs("title")
return self.kwargs_get("title", self.uid)
@property
def title_sanitized(self) -> str:
@ -83,6 +86,9 @@ class BaseEntryVariables:
# pylint: enable=no-member
TBaseEntry = TypeVar("TBaseEntry", bound="BaseEntry")
class BaseEntry(BaseEntryVariables, ABC):
"""
Abstract entry object to represent anything download from ytdl (playlist metadata, media, etc).
@ -160,6 +166,22 @@ class BaseEntry(BaseEntryVariables, ABC):
self._additional_variables = dict(self._additional_variables, **variables_to_add)
return self
def get_download_info_json_name(self) -> str:
"""
Returns
-------
The download info json's file name
"""
return f"{self.uid}.{self.info_json_ext}"
def get_download_info_json_path(self) -> str:
"""
Returns
-------
Entry's downloaded info json file path
"""
return str(Path(self.working_directory()) / self.get_download_info_json_name())
def _added_variables(self) -> Dict[str, str]:
"""
Returns
@ -189,3 +211,42 @@ 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())
@final
def to_type(self, entry_type: Type[TBaseEntry]) -> TBaseEntry:
"""
Returns
-------
Converted EntryParent to Entry-like class
"""
return entry_type(entry_dict=self._kwargs, working_directory=self._working_directory)
@classmethod
def is_entry_parent(cls, entry_dict: Dict | TBaseEntry):
"""
Returns
-------
True if it is a parent. False otherwise
"""
entry_type: Optional[str] = None
if isinstance(entry_dict, cls):
entry_type = entry_dict.kwargs_get("_type")
if isinstance(entry_dict, dict):
entry_type = entry_dict.get("_type")
return entry_type == "playlist"
@classmethod
def is_entry(cls, entry_dict: Dict | TBaseEntry):
"""
Returns
-------
True if it is an entry. False otherwise
"""
entry_ext: Optional[str] = None
if isinstance(entry_dict, cls):
entry_ext = entry_dict.kwargs_get("ext")
if isinstance(entry_dict, dict):
entry_ext = entry_dict.get("ext")
return entry_ext is not None

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

@ -57,22 +57,6 @@ class Entry(EntryVariables, BaseEntry):
return None
def get_download_info_json_name(self) -> str:
"""
Returns
-------
The download info json's file name
"""
return f"{self.uid}.{self.info_json_ext}"
def get_download_info_json_path(self) -> str:
"""
Returns
-------
Entry's downloaded info json file path
"""
return str(Path(self.working_directory()) / self.get_download_info_json_name())
def write_info_json(self) -> None:
"""
Write the entry's _kwargs back into the info.json file as well as its source variables

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

@ -1,12 +1,10 @@
import os
from typing import Dict
from typing import List
from typing import Optional
from typing import Type
from typing import TypeVar
from ytdl_sub.entries.base_entry import BaseEntry
TBaseEntry = TypeVar("TBaseEntry", bound=BaseEntry)
from ytdl_sub.entries.entry import Entry
class EntryParent(BaseEntry):
@ -14,21 +12,13 @@ class EntryParent(BaseEntry):
super().__init__(entry_dict=entry_dict, working_directory=working_directory)
self.child_entries: List["EntryParent"] = []
def is_entry(self) -> bool:
"""
Returns
-------
True if the entry contains a media file. False otherwise.
"""
return self.kwargs_contains("ext")
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]
return [child for child in self.child_entries if self.is_entry_parent(child)]
def entry_children(self) -> List["EntryParent"]:
def entry_children(self) -> List[Entry]:
"""This parent's children that are entries"""
return [child for child in self.child_entries if child.is_entry()]
return [child.to_type(Entry) for child in self.child_entries if self.is_entry(child)]
def read_children_from_entry_dicts(self, entry_dicts: List[Dict]) -> "EntryParent":
"""
@ -42,21 +32,12 @@ class EntryParent(BaseEntry):
self.__class__(
entry_dict=entry_dict,
working_directory=self.working_directory(),
)
).read_children_from_entry_dicts(entry_dicts)
)
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
def child_count(self) -> int:
"""
Returns
-------
Number of child entries
"""
return len(self.child_entries)
def get_thumbnail_url(self, thumbnail_id: str) -> Optional[str]:
"""
Downloads a specific thumbnail from a YTDL entry's thumbnail list
@ -75,6 +56,20 @@ class EntryParent(BaseEntry):
return thumbnail["url"]
return None
def __contains__(self, item: Dict | BaseEntry) -> bool:
playlist_id: Optional[str] = None
if isinstance(item, dict):
playlist_id = item.get("playlist_id")
elif isinstance(item, BaseEntry):
playlist_id = item.kwargs_get("playlist_id")
if not playlist_id:
return False
return self.uid == playlist_id or any(
child.__contains__(item) for child in self.child_entries
)
@classmethod
def from_entry_dicts(
cls, entry_dicts: List[Dict], working_directory: str
@ -82,18 +77,42 @@ class EntryParent(BaseEntry):
"""
Reads all entry dicts and builds a tree of EntryParents
"""
return [
parents = [
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
if cls.is_entry_parent(entry_dict)
]
def to_type(self, entry_type: Type[TBaseEntry]) -> TBaseEntry:
if not parents:
return []
# find disconnected root parent if one exists
first_parent = min(
parents, key=lambda x: os.stat(x.get_download_info_json_path()).st_ctime_ns
)
if len(first_parent.child_entries) == 0:
parents.remove(first_parent)
first_parent.child_entries = parents
return [first_parent]
return parents
@classmethod
def from_entry_dicts_with_no_parents(
cls, parents: List["EntryParent"], entry_dicts: List[Dict], working_directory: str
) -> List[Entry]:
"""
Returns
-------
Converted EntryParent to Entry-like class
Reads all entries that do not have any parents
"""
return entry_type(entry_dict=self._kwargs, working_directory=self._working_directory)
def _in_any_parents(entry_dict: Dict):
return any(entry_dict in parent for parent in parents)
return [
Entry(entry_dict=entry_dict, working_directory=working_directory)
for entry_dict in entry_dicts
if cls.is_entry(entry_dict) and not _in_any_parents(entry_dict)
]

132
tests/e2e/youtube/test_generic_search.py Normal file
View file

@ -0,0 +1,132 @@
# import pytest
# from conftest import assert_debug_log
# from e2e.conftest import mock_run_from_cli
# from e2e.expected_download import assert_expected_downloads
# from e2e.expected_transaction_log import assert_transaction_log_matches
#
# import ytdl_sub.downloaders.downloader
# from ytdl_sub.subscriptions.subscription import Subscription
#
#
# @pytest.fixture
# def playlist_preset_dict(output_directory):
# return {
# "generic": {
# "download_strategy": "collection",
# "urls": {"url": "https://www.youtube.com/c/Polyphia/featured"},
# },
# # "date_range": {
# # "after": "today-4months"
# # },
# # override the output directory with our fixture-generated dir
# "output_options": {
# "file_name": "{title_sanitized}.{ext}",
# "output_directory": output_directory,
# },
# # download the worst format so it is fast
# "ytdl_options": {
# "break_per_url": True,
# "format": "worst[ext=mp4]",
# },
# "subtitles": {
# "subtitles_name": "{title_sanitized}.{lang}.{subtitles_ext}",
# "allow_auto_generated_subtitles": True,
# },
# "overrides": {"artist": "JMC"},
# }
#
#
# class TestPlaylist:
# """
# Downloads my old minecraft youtube channel, pretends they are music videos. Ensure the above
# files exist and have the expected md5 file hashes.
# """
#
# @pytest.mark.parametrize("dry_run", [True, False])
# def test_playlist_download(
# self,
# music_video_config,
# playlist_preset_dict,
# output_directory,
# dry_run,
# ):
# playlist_subscription = Subscription.from_dict(
# config=music_video_config,
# preset_name="music_video_playlist_test",
# preset_dict=playlist_preset_dict,
# )
#
# transaction_log = playlist_subscription.download(dry_run=dry_run)
# assert_transaction_log_matches(
# output_directory=output_directory,
# transaction_log=transaction_log,
# transaction_log_summary_file_name="youtube/test_playlist.txt",
# )
# assert_expected_downloads(
# output_directory=output_directory,
# dry_run=dry_run,
# expected_download_summary_file_name="youtube/test_playlist.json",
# )
#
# # Ensure another invocation will hit ExistingVideoReached
# if not dry_run:
# with assert_debug_log(
# logger=ytdl_sub.downloaders.downloader.download_logger,
# expected_message="ExistingVideoReached, stopping additional downloads",
# ):
# _ = playlist_subscription.download()
#
# # TODO: output_directory_nfo is always rewritten, fix!
# # assert transaction_log.is_empty
# assert_expected_downloads(
# output_directory=output_directory,
# dry_run=dry_run,
# expected_download_summary_file_name="youtube/test_playlist.json",
# )
#
# @pytest.mark.parametrize("dry_run", [True, False])
# def test_playlist_download_from_cli_sub(
# self,
# preset_dict_to_subscription_yaml_generator,
# music_video_config_path,
# playlist_preset_dict,
# output_directory,
# dry_run,
# ):
# with preset_dict_to_subscription_yaml_generator(
# subscription_name="music_video_playlist_test", preset_dict=playlist_preset_dict
# ) as subscription_path:
# args = "--dry-run " if dry_run else ""
# args += f"--config {music_video_config_path} "
# args += f"sub {subscription_path}"
# subscription_transaction_log = mock_run_from_cli(args=args)
#
# assert len(subscription_transaction_log) == 1
# transaction_log = subscription_transaction_log[0][1]
#
# assert_transaction_log_matches(
# output_directory=output_directory,
# transaction_log=transaction_log,
# transaction_log_summary_file_name="youtube/test_playlist.txt",
# )
# assert_expected_downloads(
# output_directory=output_directory,
# dry_run=dry_run,
# expected_download_summary_file_name="youtube/test_playlist.json",
# )
#
# if not dry_run:
# # Ensure another invocation will hit ExistingVideoReached
# with assert_debug_log(
# logger=ytdl_sub.downloaders.downloader.download_logger,
# expected_message="ExistingVideoReached, stopping additional downloads",
# ):
# _ = mock_run_from_cli(args=args)[0][1]
#
# # TODO: output_directory_nfo is always rewritten, fix!
# # assert transaction_log.is_empty
# assert_expected_downloads(
# output_directory=output_directory,
# dry_run=dry_run,
# expected_download_summary_file_name="youtube/test_playlist.json",
# )