more wip
This commit is contained in:
Jesse Bannon
parent
8fbec3be2a
commit
a6052160bf
5 changed files with 211 additions and 135 deletions
|
|
@ -1,3 +1,5 @@
|
|||
import contextlib
|
||||
import os.path
|
||||
from typing import Dict
|
||||
from typing import Generator
|
||||
from typing import List
|
||||
|
|
@ -8,6 +10,7 @@ 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.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.validators import ListValidator
|
||||
|
|
@ -27,16 +30,23 @@ class CollectionUrlValidator(StrictDictValidator):
|
|||
|
||||
# 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
|
||||
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
|
||||
|
||||
|
||||
|
|
@ -114,10 +124,10 @@ class CollectionDownloader(Downloader[CollectionDownloadOptions, Entry]):
|
|||
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)
|
||||
parent.read_nested_children_from_entry_dicts(entry_dicts=entry_dicts)
|
||||
|
||||
for idx in range(parent.child_count):
|
||||
child = parent.child_entries[idx]
|
||||
child: EntryParent = parent.child_entries[idx]
|
||||
|
||||
leaf_children.extend(
|
||||
self._recursive_child_read(
|
||||
|
|
@ -127,19 +137,22 @@ class CollectionDownloader(Downloader[CollectionDownloadOptions, Entry]):
|
|||
|
||||
# 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
|
||||
parent.child_entries[idx] = child.to_entry()
|
||||
leaf_children.append(parent.child_entries[idx])
|
||||
|
||||
leaf_children.append(entry)
|
||||
for leaf_child in leaf_children:
|
||||
leaf_child.add_variables(
|
||||
parent._get_children_entry_variables_to_add(parent.child_entries)
|
||||
)
|
||||
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,
|
||||
log_prefix_on_info_json_dl="Downloading metadata for",
|
||||
url=collection_url.url,
|
||||
)
|
||||
|
||||
|
|
@ -160,8 +173,38 @@ class CollectionDownloader(Downloader[CollectionDownloadOptions, Entry]):
|
|||
|
||||
return leaf_children
|
||||
|
||||
def _download_leaf_entry(self, entry: Entry):
|
||||
@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._enhanced_download_archive._archive_working_file_path
|
||||
backup_archive_path = f"{archive_path}.backup"
|
||||
|
||||
archive_file_exists = False
|
||||
|
||||
# If archive path exists, maintain download archive is enable
|
||||
if 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:
|
||||
FileHandler.delete(file_path=archive_path)
|
||||
|
||||
def _download_leaf_entry(self, entry: Entry) -> Entry:
|
||||
download_logger.info("Downloading entry %s", entry.title)
|
||||
|
||||
# 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,
|
||||
|
|
@ -174,22 +217,32 @@ class CollectionDownloader(Downloader[CollectionDownloadOptions, Entry]):
|
|||
entry._kwargs["requested_subtitles"] = download_entry_dict.get("requested_subtitles")
|
||||
# pylint: enable=protected-access
|
||||
|
||||
yield entry
|
||||
return entry
|
||||
|
||||
def _download_collection_url(
|
||||
self, collection_url: CollectionUrlValidator, downloaded_entries: Set[str]
|
||||
) -> 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
|
||||
]
|
||||
|
||||
# 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))
|
||||
|
||||
def download(self) -> Generator[Entry, None, None]:
|
||||
"""
|
||||
Soundcloud subscription to download albums and tracks as singles.
|
||||
"""
|
||||
downloaded_leaf_children: Set[str] = set()
|
||||
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):
|
||||
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)
|
||||
for entry in self._download_collection_url(
|
||||
collection_url=collection_url, downloaded_entries=downloaded_entries
|
||||
):
|
||||
yield entry
|
||||
|
|
|
|||
|
|
@ -1,12 +1,10 @@
|
|||
from typing import Dict
|
||||
from typing import Generator
|
||||
from typing import List
|
||||
|
||||
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.soundcloud.abc import SoundcloudDownloader
|
||||
from ytdl_sub.downloaders.soundcloud.abc import SoundcloudDownloaderOptions
|
||||
from ytdl_sub.entries.entry_parent import EntryParent
|
||||
from ytdl_sub.entries.soundcloud import SoundcloudAlbumTrack
|
||||
from ytdl_sub.entries.soundcloud import SoundcloudTrack
|
||||
from ytdl_sub.validators.url_validator import SoundcloudUsernameUrlValidator
|
||||
|
||||
|
|
@ -39,6 +37,36 @@ class SoundcloudAlbumsAndSinglesDownloadOptions(SoundcloudDownloaderOptions):
|
|||
key="url", validator=SoundcloudUsernameUrlValidator
|
||||
).username_url
|
||||
|
||||
self.collection_validator = CollectionDownloadOptions(
|
||||
name=self._name,
|
||||
value={
|
||||
"urls": [
|
||||
{
|
||||
"url": f"{self._url}/tracks",
|
||||
"variables": {
|
||||
"track_number": "1",
|
||||
"track_number_padded": "01",
|
||||
"track_count": "1",
|
||||
"album": "{title}",
|
||||
"album_sanitized": "{title_sanitized}",
|
||||
"album_year": "{upload_year}",
|
||||
},
|
||||
},
|
||||
{
|
||||
"url": f"{self._url}/albums",
|
||||
"variables": {
|
||||
"track_number": "{playlist_index}",
|
||||
"track_number_padded": "{playlist_index_padded}",
|
||||
"track_count": "{playlist_count}",
|
||||
"album": "{playlist}",
|
||||
"album_sanitized": "{playlist_sanitized}",
|
||||
"album_year": "{playlist_max_upload_year}",
|
||||
},
|
||||
},
|
||||
]
|
||||
},
|
||||
)
|
||||
|
||||
@property
|
||||
def url(self) -> str:
|
||||
"""
|
||||
|
|
@ -72,104 +100,16 @@ class SoundcloudAlbumsAndSinglesDownloader(
|
|||
},
|
||||
)
|
||||
|
||||
def _get_singles(
|
||||
self, entry_dicts: List[Dict], albums: List[EntryParent]
|
||||
) -> List[SoundcloudTrack]:
|
||||
tracks: List[SoundcloudTrack] = []
|
||||
|
||||
# Get all tracks that are not part of an album
|
||||
for entry_dict in self._filter_entry_dicts(entry_dicts):
|
||||
if not any(entry_dict in album for album in albums):
|
||||
tracks.append(
|
||||
SoundcloudTrack(entry_dict=entry_dict, working_directory=self.working_directory)
|
||||
)
|
||||
|
||||
return tracks
|
||||
|
||||
def _get_albums(self) -> List[EntryParent]:
|
||||
# Dry-run to get the info json files
|
||||
artist_albums_url = self.artist_albums_url(artist_url=self.download_options.url)
|
||||
entry_dicts = self.extract_info_via_info_json(
|
||||
only_info_json=True,
|
||||
log_prefix_on_info_json_dl="Downloading metadata for",
|
||||
url=artist_albums_url,
|
||||
)
|
||||
|
||||
albums: List[EntryParent] = []
|
||||
for entry_dict in self._filter_entry_dicts(entry_dicts, extractor="soundcloud:set"):
|
||||
albums.append(
|
||||
EntryParent(
|
||||
entry_dict=entry_dict, working_directory=self.working_directory
|
||||
).read_children_from_entry_dicts(
|
||||
entry_dicts=entry_dicts, child_class=SoundcloudAlbumTrack
|
||||
)
|
||||
)
|
||||
|
||||
return albums
|
||||
|
||||
def _get_album_tracks(
|
||||
self, albums: List[EntryParent]
|
||||
) -> Generator[SoundcloudAlbumTrack, None, None]:
|
||||
for album in albums:
|
||||
if album.child_count > 0:
|
||||
download_logger.info("Downloading album %s", album.title)
|
||||
|
||||
for track in album.child_entries:
|
||||
if self.download_options.skip_premiere_tracks and track.is_premiere():
|
||||
continue
|
||||
|
||||
download_logger.info(
|
||||
"Downloading album track %d/%d %s",
|
||||
track.track_number,
|
||||
track.track_count,
|
||||
track.title,
|
||||
)
|
||||
if not self.is_dry_run:
|
||||
_ = self.extract_info_with_retry(
|
||||
is_downloaded_fn=track.is_downloaded,
|
||||
url=album.webpage_url,
|
||||
ytdl_options_overrides={
|
||||
"playlist_items": str(track.kwargs("playlist_index")),
|
||||
"writeinfojson": False,
|
||||
},
|
||||
)
|
||||
|
||||
yield track
|
||||
|
||||
def _get_single_tracks(
|
||||
self, albums: List[EntryParent]
|
||||
) -> Generator[SoundcloudTrack, None, None]:
|
||||
artist_tracks_url = self.artist_tracks_url(artist_url=self.download_options.url)
|
||||
tracks_entry_dicts = self.extract_info_via_info_json(
|
||||
only_info_json=True,
|
||||
log_prefix_on_info_json_dl="Downloading metadata for",
|
||||
url=artist_tracks_url,
|
||||
)
|
||||
|
||||
# Then, get all singles
|
||||
tracks = self._get_singles(entry_dicts=tracks_entry_dicts, albums=albums)
|
||||
for track in tracks:
|
||||
# Filter any premiere tracks if specified
|
||||
if self.download_options.skip_premiere_tracks and track.is_premiere():
|
||||
continue
|
||||
|
||||
download_logger.info("Downloading single track %s", track.title)
|
||||
if not self.is_dry_run:
|
||||
_ = self.extract_info_with_retry(
|
||||
is_downloaded_fn=track.is_downloaded,
|
||||
url=track.webpage_url,
|
||||
ytdl_options_overrides={"writeinfojson": False},
|
||||
)
|
||||
|
||||
yield track
|
||||
|
||||
def download(self) -> Generator[SoundcloudTrack, None, None]:
|
||||
"""
|
||||
Soundcloud subscription to download albums and tracks as singles.
|
||||
"""
|
||||
albums = self._get_albums()
|
||||
for album_track in self._get_album_tracks(albums=albums):
|
||||
yield album_track
|
||||
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 single_track in self._get_single_tracks(albums=albums):
|
||||
yield single_track
|
||||
for entry in downloader.download():
|
||||
yield entry
|
||||
|
|
|
|||
|
|
@ -28,10 +28,32 @@ class EntryParent(BaseEntry):
|
|||
if not child_entries:
|
||||
return {}
|
||||
|
||||
return {"playlist_max_upload_year": max(entry.upload_year for entry in child_entries)}
|
||||
return {
|
||||
"playlist_max_upload_year": max(
|
||||
entry.upload_year for entry in child_entries if isinstance(entry, Entry)
|
||||
)
|
||||
}
|
||||
|
||||
# pylint: enable=no-self-use
|
||||
|
||||
def read_nested_children_from_entry_dicts(self, entry_dicts: List[Dict]):
|
||||
child_entries: List["EntryParent"] = []
|
||||
|
||||
for entry_dict in entry_dicts:
|
||||
if entry_dict.get("playlist_id") == self.uid:
|
||||
child_entries.append(
|
||||
self.__class__(
|
||||
entry_dict=entry_dict,
|
||||
working_directory=self.working_directory(),
|
||||
entry_parent=self,
|
||||
)
|
||||
)
|
||||
|
||||
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":
|
||||
|
|
@ -136,11 +158,4 @@ class EntryParent(BaseEntry):
|
|||
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
|
||||
return Entry(entry_dict=self._kwargs, working_directory=self.working_directory())
|
||||
|
|
|
|||
|
|
@ -1,3 +1,5 @@
|
|||
from yt_dlp.utils import sanitize_filename
|
||||
|
||||
from ytdl_sub.entries.base_entry import BaseEntry
|
||||
from ytdl_sub.entries.base_entry import BaseEntryVariables
|
||||
|
||||
|
|
@ -14,6 +16,72 @@ _days_in_month = [0, 31, 28, 31, 30, 31, 30, 31, 31, 30, 31, 30, 31]
|
|||
|
||||
|
||||
class EntryVariables(BaseEntryVariables):
|
||||
@property
|
||||
def playlist(self: BaseEntry) -> str:
|
||||
"""
|
||||
Returns
|
||||
-------
|
||||
str
|
||||
Name of its parent playlist/channel if it exists, otherwise returns its title.
|
||||
"""
|
||||
return self.kwargs_get("playlist", self.title)
|
||||
|
||||
@property
|
||||
def playlist_sanitized(self) -> str:
|
||||
"""
|
||||
Returns
|
||||
-------
|
||||
str
|
||||
The playlist name, sanitized
|
||||
"""
|
||||
return sanitize_filename(self.playlist)
|
||||
|
||||
@property
|
||||
def playlist_index(self: BaseEntry) -> int:
|
||||
"""
|
||||
Returns
|
||||
-------
|
||||
int
|
||||
Playlist index if it exists, otherwise returns ``1``.
|
||||
|
||||
Note that for channels/playlists, an index of 1 implies it's the most recent
|
||||
uploaded entry. It is recommended to not use this unless you know the channel/playlist
|
||||
will never add new content, i.e. a music album.
|
||||
"""
|
||||
return self.kwargs_get("playlist_index", 1)
|
||||
|
||||
@property
|
||||
def playlist_index_padded(self) -> str:
|
||||
"""
|
||||
Returns
|
||||
-------
|
||||
str
|
||||
playlist_index padded two digits
|
||||
"""
|
||||
return _pad(self.playlist_index, width=2)
|
||||
|
||||
@property
|
||||
def playlist_count(self: BaseEntry) -> int:
|
||||
"""
|
||||
Returns
|
||||
-------
|
||||
int
|
||||
Playlist count if it exists, otherwise returns ``1``.
|
||||
"""
|
||||
return self.kwargs_get("playlist_count", 1)
|
||||
|
||||
@property
|
||||
def playlist_max_upload_year(self) -> int:
|
||||
"""
|
||||
Returns
|
||||
-------
|
||||
int
|
||||
Max upload_year for all entries in this entry's playlist if it exists, otherwise returns
|
||||
``upload_year``
|
||||
"""
|
||||
# override in EntryParent
|
||||
return self.upload_year
|
||||
|
||||
@property
|
||||
def ext(self: BaseEntry) -> str:
|
||||
"""
|
||||
|
|
|
|||
|
|
@ -598,7 +598,7 @@ class DownloadArchiver:
|
|||
"""
|
||||
|
||||
def __init__(self, enhanced_download_archive: EnhancedDownloadArchive):
|
||||
self.__enhanced_download_archive = enhanced_download_archive
|
||||
self._enhanced_download_archive = enhanced_download_archive
|
||||
|
||||
@property
|
||||
def working_directory(self) -> str:
|
||||
|
|
@ -607,7 +607,7 @@ class DownloadArchiver:
|
|||
-------
|
||||
Path to the working directory
|
||||
"""
|
||||
return self.__enhanced_download_archive.working_directory
|
||||
return self._enhanced_download_archive.working_directory
|
||||
|
||||
@property
|
||||
def is_dry_run(self) -> bool:
|
||||
|
|
@ -616,7 +616,7 @@ class DownloadArchiver:
|
|||
-------
|
||||
True if this session is a dry-run. False otherwise.
|
||||
"""
|
||||
return self.__enhanced_download_archive.is_dry_run
|
||||
return self._enhanced_download_archive.is_dry_run
|
||||
|
||||
def save_file(
|
||||
self,
|
||||
|
|
@ -640,7 +640,7 @@ class DownloadArchiver:
|
|||
entry
|
||||
Optional. Entry that the file belongs to
|
||||
"""
|
||||
self.__enhanced_download_archive.save_file_to_output_directory(
|
||||
self._enhanced_download_archive.save_file_to_output_directory(
|
||||
file_name=file_name,
|
||||
file_metadata=file_metadata,
|
||||
output_file_name=output_file_name,
|
||||
|
|
|
|||
Loading…
Reference in a new issue