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

[BACKEND] EntryParent class, use it to make downloaders more generic (#229)

Browse source
This commit is contained in:
Jesse Bannon 2022-09-10 22:50:27 -07:00 committed by GitHub
parent c7e9380fe0
commit c9f0300b39
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
9 changed files with 271 additions and 296 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -5,7 +5,7 @@ from typing import List
from ytdl_sub.downloaders.downloader import download_logger
from ytdl_sub.downloaders.soundcloud.abc import SoundcloudDownloader
from ytdl_sub.downloaders.soundcloud.abc import SoundcloudDownloaderOptions
from ytdl_sub.entries.soundcloud import SoundcloudAlbum
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
@ -72,37 +72,8 @@ class SoundcloudAlbumsAndSinglesDownloader(
},
)
def _get_albums_from_entry_dicts(self, entry_dicts: List[Dict]) -> List[SoundcloudAlbum]:
"""
Parameters
----------
entry_dicts
Entry dicts from extracting info jsons
Returns
-------
Dict containing album_id: album class
"""
albums: Dict[str, SoundcloudAlbum] = {}
# First, get the albums themselves
for entry_dict in self._filter_entry_dicts(entry_dicts, extractor="soundcloud:set"):
albums[entry_dict["id"]] = SoundcloudAlbum(
entry_dict=entry_dict, working_directory=self.working_directory
)
# Then, get all tracks that belong to the album
for entry_dict in self._filter_entry_dicts(entry_dicts):
album_id = entry_dict.get("playlist_id")
if album_id in albums:
albums[album_id].tracks.append(
SoundcloudTrack(entry_dict=entry_dict, working_directory=self.working_directory)
)
return list(albums.values())
def _get_singles(
self, entry_dicts: List[Dict], albums: List[SoundcloudAlbum]
self, entry_dicts: List[Dict], albums: List[EntryParent]
) -> List[SoundcloudTrack]:
tracks: List[SoundcloudTrack] = []
@ -115,26 +86,35 @@ class SoundcloudAlbumsAndSinglesDownloader(
return tracks
def _get_albums(self) -> List[SoundcloudAlbum]:
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)
album_entry_dicts = self.extract_info_via_info_json(
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 = self._get_albums_from_entry_dicts(entry_dicts=album_entry_dicts)
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[SoundcloudAlbum]
self, albums: List[EntryParent]
) -> Generator[SoundcloudAlbumTrack, None, None]:
for album in albums:
if len(album.album_tracks()) > 0:
if album.child_count > 0:
download_logger.info("Downloading album %s", album.title)
for track in album.album_tracks():
for track in album.child_entries:
if self.download_options.skip_premiere_tracks and track.is_premiere():
continue
@ -157,7 +137,7 @@ class SoundcloudAlbumsAndSinglesDownloader(
yield track
def _get_single_tracks(
self, albums: List[SoundcloudAlbum]
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(

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

@ -10,7 +10,7 @@ from ytdl_sub.downloaders.downloader import download_logger
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.youtube import YoutubeChannel
from ytdl_sub.entries.entry_parent import EntryParent
from ytdl_sub.entries.youtube import YoutubeVideo
from ytdl_sub.utils.datetime import to_date_range_hack
from ytdl_sub.utils.thumbnail import convert_url_thumbnail
@ -156,19 +156,7 @@ class YoutubeChannelDownloader(YoutubeDownloader[YoutubeChannelDownloaderOptions
overrides=overrides,
)
self.channel: Optional[YoutubeChannel] = None
def _get_channel(self, entry_dicts: List[Dict]) -> YoutubeChannel:
return YoutubeChannel(
entry_dict=self._filter_entry_dicts(entry_dicts, extractor="youtube:tab")[0],
working_directory=self.working_directory,
)
def _get_channel_videos(self, entry_dicts: List[Dict]) -> List[YoutubeVideo]:
return [
YoutubeVideo(entry_dict=entry_dict, working_directory=self.working_directory)
for entry_dict in self._filter_entry_dicts(entry_dicts, sort_by="playlist_index")
]
self.channel: Optional[EntryParent] = None
def download(self) -> Generator[YoutubeVideo, None, None]:
"""
@ -192,8 +180,12 @@ class YoutubeChannelDownloader(YoutubeDownloader[YoutubeChannelDownloaderOptions
url=self.download_options.channel_url,
)
self.channel = self._get_channel(entry_dicts=entry_dicts)
entries_to_download = self._get_channel_videos(entry_dicts=entry_dicts)
self.channel = EntryParent.from_entry_dicts_with_children(
entry_dicts=entry_dicts,
working_directory=self.working_directory,
child_class=YoutubeVideo,
extractor="youtube:tab",
)
self.overrides.add_override_variables(
variables_to_add={
@ -206,8 +198,8 @@ class YoutubeChannelDownloader(YoutubeDownloader[YoutubeChannelDownloaderOptions
# Iterate in descending order to process older videos first. In case an error occurs and a
# 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(entries_to_download), start=1):
download_logger.info("Downloading %d/%d %s", idx, len(entries_to_download), video.title)
for idx, video in enumerate(reversed(self.channel.child_entries), start=1):
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,
# channels do not download subtitles or subtitle metadata
@ -260,7 +252,7 @@ class YoutubeChannelDownloader(YoutubeDownloader[YoutubeChannelDownloaderOptions
self.download_options.channel_avatar_path
)
if self._download_thumbnail(
thumbnail_url=self.channel.avatar_thumbnail_url(),
thumbnail_url=self.channel.get_thumbnail_url("avatar_uncropped"),
output_thumbnail_path=str(Path(self.working_directory) / avatar_thumbnail_name),
):
self.save_file(file_name=avatar_thumbnail_name)
@ -272,7 +264,7 @@ class YoutubeChannelDownloader(YoutubeDownloader[YoutubeChannelDownloaderOptions
self.download_options.channel_banner_path
)
if self._download_thumbnail(
thumbnail_url=self.channel.banner_thumbnail_url(),
thumbnail_url=self.channel.get_thumbnail_url("banner_uncropped"),
output_thumbnail_path=str(Path(self.working_directory) / banner_thumbnail_name),
):
self.save_file(file_name=banner_thumbnail_name)

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

@ -5,6 +5,7 @@ from typing import List
from ytdl_sub.downloaders.downloader import download_logger
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
@ -94,24 +95,24 @@ class YoutubePlaylistDownloader(
url=self.download_options.playlist_url,
)
playlist = self._filter_entry_dicts(entry_dicts, extractor="youtube:tab")[0]
playlist_videos = self._filter_entry_dicts(entry_dicts, sort_by="playlist_index")
playlist: EntryParent = EntryParent.from_entry_dicts_with_children(
entry_dicts=entry_dicts,
working_directory=self.working_directory,
child_class=YoutubePlaylistVideo,
extractor="youtube:tab",
)
self.overrides.add_override_variables(
variables_to_add={
"source_title": playlist["title"],
"source_uploader": playlist.get("uploader", "__failed_to_scrape__"),
"source_description": playlist.get("description", ""),
"source_title": playlist.title,
"source_uploader": playlist.kwargs_get("uploader", "__failed_to_scrape__"),
"source_description": playlist.kwargs_get("description", ""),
}
)
# Iterate in reverse order to process older videos first. In case an error occurs and a
# 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, entry_dict in enumerate(reversed(playlist_videos), start=1):
video = YoutubePlaylistVideo(
entry_dict=entry_dict, working_directory=self.working_directory
)
for idx, video in enumerate(reversed(playlist.child_entries), start=1):
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,

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

@ -5,8 +5,85 @@ from typing import List
from typing import Optional
from typing import final
from yt_dlp.utils import sanitize_filename
class BaseEntry(ABC):
# pylint: disable=no-member
class BaseEntryVariables:
"""
Source variables are ``{variables}`` that contain metadata from downloaded media.
These variables can be used with fields that expect
:class:`~ytdl_sub.validators.string_formatter_validators.StringFormatterValidator`,
but not
:class:`~ytdl_sub.validators.string_formatter_validators.OverridesStringFormatterValidator`.
"""
@property
def uid(self: "BaseEntry") -> str:
"""
Returns
-------
str
The entry's unique ID
"""
return self.kwargs("id")
@property
def extractor(self: "BaseEntry") -> str:
"""
Returns
-------
str
The ytdl extractor name
"""
return self.kwargs("extractor")
@property
def title(self: "BaseEntry") -> str:
"""
Returns
-------
str
The title of the entry
"""
return self.kwargs("title")
@property
def title_sanitized(self) -> str:
"""
Returns
-------
str
The sanitized title of the entry, which is safe to use for Unix and Windows file names.
"""
return sanitize_filename(self.title)
@property
def webpage_url(self: "BaseEntry") -> str:
"""
Returns
-------
str
The url to the webpage.
"""
return self.kwargs("webpage_url")
@property
def info_json_ext(self) -> str:
"""
Returns
-------
str
The "info.json" extension
"""
return "info.json"
# pylint: enable=no-member
class BaseEntry(BaseEntryVariables, ABC):
"""
Abstract entry object to represent anything download from ytdl (playlist metadata, media, etc).
"""
@ -28,7 +105,7 @@ class BaseEntry(ABC):
self._working_directory = working_directory
self._kwargs = entry_dict
self._additional_variables: Dict[str, str] = {}
self._additional_variables: Dict[str, str | int] = {}
def kwargs_contains(self, key: str) -> bool:
"""Returns whether internal kwargs contains the specified key"""
@ -110,3 +187,19 @@ class BaseEntry(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)

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

@ -6,7 +6,6 @@ from typing import Optional
from typing import final
from ytdl_sub.entries.base_entry import BaseEntry
from ytdl_sub.entries.variables.entry_variables import BaseEntryVariables
from ytdl_sub.entries.variables.entry_variables import EntryVariables
from ytdl_sub.validators.audo_codec_validator import AUDIO_CODEC_EXTS
@ -103,27 +102,3 @@ class Entry(EntryVariables, BaseEntry):
break
return file_exists
class ParentEntry(BaseEntryVariables, BaseEntry):
"""
Entry to represent parent entry objects like a Channel or Playlist
"""
def _get_thumbnail_url(self, thumbnail_id: str) -> Optional[str]:
"""
Downloads a specific thumbnail from a YTDL entry's thumbnail list
Parameters
----------
thumbnail_id:
Id of the thumbnail defined in the parent's thumbnail
Returns
-------
Desired thumbnail url if it exists. None if it does not.
"""
for thumbnail in self.kwargs_get("thumbnails", []):
if thumbnail["id"] == thumbnail_id:
return thumbnail["url"]
return None

132
src/ytdl_sub/entries/entry_parent.py Normal file
View file

@ -0,0 +1,132 @@
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
TChildEntry = TypeVar("TChildEntry", bound=BaseEntry)
class EntryParent(BaseEntry):
def __init__(self, entry_dict: Dict, working_directory: str):
super().__init__(entry_dict=entry_dict, working_directory=working_directory)
self._child_entries: List[TChildEntry] = []
# pylint: disable=no-self-use
def _get_children_entry_variables_to_add(
self, child_entries: List[TChildEntry]
) -> Dict[str, str | int]:
"""
Adds source variables to the child entry derived from the parent entry.
"""
if not child_entries:
return {}
return {"playlist_max_upload_year": max(entry.upload_year for entry in child_entries)}
# pylint: enable=no-self-use
def read_children_from_entry_dicts(
self, entry_dicts: List[Dict], child_class: Type[TChildEntry]
) -> "EntryParent":
"""
Parameters
----------
entry_dicts
Entry dicts to look for children from
child_class
The class to convert the entry dict to
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())
)
child_variables_to_add = self._get_children_entry_variables_to_add(child_entries)
for child_entry in child_entries:
child_entry.add_variables(variables_to_add=child_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
-------
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
Parameters
----------
thumbnail_id:
Id of the thumbnail defined in the parent's thumbnail
Returns
-------
Desired thumbnail url if it exists. None if it does not.
"""
for thumbnail in self.kwargs_get("thumbnails", []):
if thumbnail["id"] == thumbnail_id:
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":
"""
Load the parent entry and its children
"""
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

110
src/ytdl_sub/entries/soundcloud.py
View file

@ -1,9 +1,4 @@
from typing import Dict
from typing import List
from typing import Optional
from ytdl_sub.entries.entry import Entry
from ytdl_sub.entries.entry import ParentEntry
from ytdl_sub.entries.variables.soundcloud_variables import SoundcloudVariables
@ -28,18 +23,6 @@ class SoundcloudAlbumTrack(SoundcloudTrack):
Entry object to represent a Soundcloud track yt-dlp that belongs to an album.
"""
def __init__(
self,
entry_dict: Dict,
working_directory: str,
album_year: int,
):
"""
Initialize the album track using album metadata and ytdl metadata for the specific track.
"""
super().__init__(entry_dict=entry_dict, working_directory=working_directory)
self._album_year = album_year
@property
def track_number(self) -> int:
"""Returns the entry's track number"""
@ -58,94 +41,5 @@ class SoundcloudAlbumTrack(SoundcloudTrack):
@property
def album_year(self) -> int:
"""Returns the entry's album year, fetched from its internal album"""
return self._album_year
@classmethod
def from_soundcloud_track(
cls,
album_year: int,
soundcloud_track: SoundcloudTrack,
) -> "SoundcloudAlbumTrack":
"""
Parameters
----------
album_year:
Album year
soundcloud_track:
Track to convert to an album track
Returns
-------
SoundcloudTrack converted to a SoundcloudAlbumTrack
"""
return SoundcloudAlbumTrack(
entry_dict=soundcloud_track._kwargs, # pylint: disable=protected-access
working_directory=soundcloud_track.working_directory(),
album_year=album_year,
)
class SoundcloudAlbum(ParentEntry):
"""
Entry object to represent a Soundcloud album.
"""
entry_extractor = "soundcloud:set"
def __init__(self, entry_dict: Dict, working_directory: Optional[str] = None):
"""
Initialize the entry using ytdl metadata
Parameters
----------
entry_dict
Entry metadata
working_directory
Optional. Directory that the entry is downloaded to
"""
super().__init__(entry_dict=entry_dict, working_directory=working_directory)
self.tracks: List[SoundcloudTrack] = []
def album_tracks(self) -> List[SoundcloudAlbumTrack]:
"""
Returns
-------
All tracks in the album represented as album-tracks. They will share the
same album name, have ordered track numbers, and a shared album year.
"""
album_tracks = [
SoundcloudAlbumTrack.from_soundcloud_track(
album_year=self.album_year, soundcloud_track=track
)
for track in self.tracks
]
return sorted(album_tracks, key=lambda track: track.track_number)
@property
def album_year(self) -> int:
"""
Returns
-------
The album's year, computed by the max upload year amongst all album tracks
"""
return max(track.upload_year for track in self.tracks)
@property
def track_count(self) -> int:
"""
Returns
-------
Number of tracks in the album (technically a playlist)
"""
return self.kwargs("playlist_count")
def __contains__(self, item):
"""
Returns
-------
True if the the item (entry_dict) has the same id as one of the tracks. False otherwise.
"""
if isinstance(item, dict):
return any(item.get("id") == track.uid for track in self.tracks)
return False
# added from parent entry
return self._additional_variables["playlist_max_upload_year"]

73
src/ytdl_sub/entries/variables/entry_variables.py
View file

@ -1,6 +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
# This file contains mixins to a BaseEntry subclass. Ignore pylint's "no kwargs member" suggestion
# pylint: disable=no-member
@ -14,76 +13,6 @@ def _pad(num: int, width: int = 2):
_days_in_month = [0, 31, 28, 31, 30, 31, 30, 31, 31, 30, 31, 30, 31]
class BaseEntryVariables:
"""
Source variables are ``{variables}`` that contain metadata from downloaded media.
These variables can be used with fields that expect
:class:`~ytdl_sub.validators.string_formatter_validators.StringFormatterValidator`,
but not
:class:`~ytdl_sub.validators.string_formatter_validators.OverridesStringFormatterValidator`.
"""
@property
def uid(self: BaseEntry) -> str:
"""
Returns
-------
str
The entry's unique ID
"""
return self.kwargs("id")
@property
def extractor(self: BaseEntry) -> str:
"""
Returns
-------
str
The ytdl extractor name
"""
return self.kwargs("extractor")
@property
def title(self: BaseEntry) -> str:
"""
Returns
-------
str
The title of the entry
"""
return self.kwargs("title")
@property
def title_sanitized(self) -> str:
"""
Returns
-------
str
The sanitized title of the entry, which is safe to use for Unix and Windows file names.
"""
return sanitize_filename(self.title)
@property
def webpage_url(self: BaseEntry) -> str:
"""
Returns
-------
str
The url to the webpage.
"""
return self.kwargs("webpage_url")
@property
def info_json_ext(self) -> str:
"""
Returns
-------
str
The "info.json" extension
"""
return "info.json"
class EntryVariables(BaseEntryVariables):
@property
def ext(self: BaseEntry) -> str:

21
src/ytdl_sub/entries/youtube.py
View file

@ -2,7 +2,6 @@ import os.path
from pathlib import Path
from ytdl_sub.entries.entry import Entry
from ytdl_sub.entries.entry import ParentEntry
from ytdl_sub.entries.variables.youtube_variables import YoutubeVideoVariables
@ -44,23 +43,3 @@ class YoutubePlaylistVideo(YoutubeVideo):
The size of the playlist
"""
return self.kwargs("playlist_count")
class YoutubeChannel(ParentEntry):
entry_extractor = "youtube:tab"
def avatar_thumbnail_url(self) -> str:
"""
Returns
-------
The channel's uncropped avatar image url
"""
return self._get_thumbnail_url(thumbnail_id="avatar_uncropped")
def banner_thumbnail_url(self) -> str:
"""
Returns
-------
The channel's uncropped banner image url
"""
return self._get_thumbnail_url(thumbnail_id="banner_uncropped")