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

YT playlist simplified

Browse source
This commit is contained in:
jbannon 2022-05-01 02:11:54 +00:00
parent c12504d257
commit 10e2691b3b
4 changed files with 212 additions and 171 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -1,4 +1,6 @@
import abc
import json
import os
from abc import ABC
from contextlib import contextmanager
from pathlib import Path
@ -6,12 +8,16 @@ from typing import Dict
from typing import Generic
from typing import List
from typing import Optional
from typing import Tuple
from typing import Type
from typing import TypeVar
import yt_dlp as ytdl
from yt_dlp.utils import ExistingVideoReached
from yt_dlp.utils import RejectedVideoReached
from ytdl_sub.config.preset_options import Overrides
from ytdl_sub.entries.base_entry import BaseEntry
from ytdl_sub.entries.entry import Entry
from ytdl_sub.validators.strict_dict_validator import StrictDictValidator
@ -24,6 +30,7 @@ class DownloaderValidator(StrictDictValidator, ABC):
DownloaderOptionsT = TypeVar("DownloaderOptionsT", bound=DownloaderValidator)
DownloaderEntryT = TypeVar("DownloaderEntryT", bound=Entry)
DownloaderParentEntryT = TypeVar("DownloaderParentEntryT", bound=BaseEntry)
class Downloader(Generic[DownloaderOptionsT, DownloaderEntryT], ABC):
@ -32,8 +39,8 @@ class Downloader(Generic[DownloaderOptionsT, DownloaderEntryT], ABC):
and should translate that to list of Entry objects.
"""
downloader_options_type: Type[DownloaderOptionsT] = NotImplemented
downloader_entry_type: Type[DownloaderEntryT] = NotImplemented
downloader_options_type: Type[DownloaderValidator] = DownloaderValidator
downloader_entry_type: Type[Entry] = Entry
@classmethod
def ytdl_option_overrides(cls) -> Dict:
@ -113,10 +120,84 @@ class Downloader(Generic[DownloaderOptionsT, DownloaderEntryT], ABC):
"""
Wrapper around yt_dlp.YoutubeDL.YoutubeDL.extract_info
All kwargs will passed to the extract_info function.
Parameters
----------
ytdl_options_overrides
Optional. Dict containing ytdl args to override other predefined ytdl args
**kwargs
arguments passed directory to YoutubeDL extract_info
"""
with self.ytdl_downloader(ytdl_options_overrides) as ytdl_downloader:
return ytdl_downloader.extract_info(**kwargs)
def extract_info_json(self, ytdl_options_overrides: Optional[Dict] = None, **kwargs) -> None:
"""
Wrapper around yt_dlp.YoutubeDL.YoutubeDL.extract_info
All kwargs will passed to the extract_info function. This also enables ytdl to write
.info.json files for all media downloaded.
Catches RejectedVideoReached and ExistingVideoReached exceptions.
Parameters
----------
ytdl_options_overrides
Optional. Dict containing ytdl args to override other predefined ytdl args
**kwargs
arguments passed directory to YoutubeDL extract_info
"""
if ytdl_options_overrides is None:
ytdl_options_overrides = {}
ytdl_options_overrides = dict(ytdl_options_overrides, **{"writeinfojson": True})
try:
_ = self.extract_info(ytdl_options_overrides=ytdl_options_overrides, **kwargs)
except (RejectedVideoReached, ExistingVideoReached):
pass
def extract_from_info_json(
self, parent_prefix: str, parent_entry_type: Type[DownloaderParentEntryT]
) -> Tuple[DownloaderParentEntryT, List[DownloaderEntryT]]:
"""
Reads all .info.json files in the working directory, and casts them to the
parent_entry_type (i.e. YoutubeChannel) and downloader_entry_type (i.e. YoutubeVideo)
Parameters
----------
parent_prefix
info.json file name prefix to indicate its the parent (i.e. the YT channel id)
parent_entry_type
Class type for the parent entry
Returns
-------
Tuple containing parent, list of videos belong to the parent
"""
# Load the entries from info.json
parent_entry: Optional[DownloaderParentEntryT] = None
entries: List[DownloaderEntryT] = []
for file_name in os.listdir(self.working_directory):
if file_name.endswith(".info.json"):
with open(Path(self.working_directory) / file_name, "r", encoding="utf-8") as file:
if file_name.startswith(parent_prefix):
parent_entry = parent_entry_type(
entry_dict=json.load(file), working_directory=self.working_directory
)
else:
entries.append(
self.downloader_entry_type(
entry_dict=json.load(file), working_directory=self.working_directory
)
)
return parent_entry, entries
@abc.abstractmethod
def download(self) -> List[DownloaderEntryT]:
"""The function to perform the download of all media entries"""

159
src/ytdl_sub/downloaders/youtube_downloader.py
View file

@ -1,5 +1,3 @@
import json
import os
from abc import ABC
from pathlib import Path
from typing import Dict
@ -11,11 +9,14 @@ from urllib.request import urlopen
from PIL.Image import Image
from PIL.Image import open as pil_open
from yt_dlp.utils import RejectedVideoReached
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.entries.youtube import YoutubeChannel
from ytdl_sub.entries.youtube import YoutubePlaylist
from ytdl_sub.entries.youtube import YoutubePlaylistVideo
from ytdl_sub.entries.youtube import YoutubeVideo
from ytdl_sub.utils.logger import Logger
from ytdl_sub.validators.date_range_validator import DateRangeValidator
@ -35,56 +36,20 @@ class YoutubeDownloaderOptions(DownloaderValidator, ABC):
YoutubeDownloaderOptionsT = TypeVar("YoutubeDownloaderOptionsT", bound=YoutubeDownloaderOptions)
YoutubeVideoT = TypeVar("YoutubeVideoT", bound=YoutubeVideo)
class YoutubeDownloader(
Downloader[YoutubeDownloaderOptionsT, YoutubeVideo], Generic[YoutubeDownloaderOptionsT], ABC
Downloader[YoutubeDownloaderOptionsT, YoutubeVideoT],
Generic[YoutubeDownloaderOptionsT, YoutubeVideoT],
ABC,
):
"""
Class that handles downloading youtube entries via ytdl and converting them into
YoutubeVideo objects
YoutubeVideo like objects. Reserved for any future logic that is shared amongst all YT
downloaders.
"""
downloader_entry_type = YoutubeVideo
def _download_using_metadata(
self,
url: str,
ignore_prefix: str,
ytdl_options_overrides: Optional[Dict] = None,
) -> List[YoutubeVideo]:
"""
Do not get entries from the extract info, let it write to the info.json file and load
that instead. This is because if the video is already downloaded in a playlist, it will
not fetch the metadata (maybe there is a way??)
"""
entries: List[YoutubeVideo] = []
ytdl_overrides = {
"writeinfojson": True,
}
if ytdl_options_overrides:
ytdl_overrides = dict(ytdl_overrides, **ytdl_options_overrides)
try:
_ = self.extract_info(ytdl_options_overrides=ytdl_overrides, url=url)
except RejectedVideoReached:
pass
# Load the entries from info.json, ignore the playlist entry
for file_name in os.listdir(self.working_directory):
if file_name.startswith(ignore_prefix) or not file_name.endswith(".info.json"):
continue
with open(Path(self.working_directory) / file_name, "r", encoding="utf-8") as file:
entries.append(
YoutubeVideo(
entry_dict=json.load(file), working_directory=self.working_directory
)
)
return entries
###############################################################################
# Youtube single video downloader + options
@ -98,8 +63,9 @@ class YoutubeVideoDownloaderOptions(YoutubeDownloaderOptions):
self.video_id = self._validate_key("video_id", StringValidator)
class YoutubeVideoDownloader(YoutubeDownloader[YoutubeVideoDownloaderOptions]):
class YoutubeVideoDownloader(YoutubeDownloader[YoutubeVideoDownloaderOptions, YoutubeVideo]):
downloader_options_type = YoutubeVideoDownloaderOptions
downloader_entry_type = YoutubeVideo
@classmethod
def video_url(cls, video_id: str) -> str:
@ -111,8 +77,8 @@ class YoutubeVideoDownloader(YoutubeDownloader[YoutubeVideoDownloaderOptions]):
video_id = self.download_options.video_id.value
video_url = self.video_url(video_id=video_id)
entry = self.extract_info(url=video_url)
return [YoutubeVideo(entry_dict=entry, working_directory=self.working_directory)]
entry_dict = self.extract_info(url=video_url)
return [YoutubeVideo(entry_dict=entry_dict, working_directory=self.working_directory)]
###############################################################################
@ -127,22 +93,31 @@ class YoutubePlaylistDownloaderOptions(YoutubeDownloaderOptions):
self.playlist_id = self._validate_key("playlist_id", StringValidator)
class YoutubePlaylistDownloader(YoutubeDownloader[YoutubePlaylistDownloaderOptions]):
class YoutubePlaylistDownloader(
YoutubeDownloader[YoutubePlaylistDownloaderOptions, YoutubePlaylistVideo]
):
downloader_options_type = YoutubePlaylistDownloaderOptions
downloader_entry_type = YoutubePlaylistVideo
@classmethod
def playlist_url(cls, playlist_id: str) -> str:
"""Returns full playlist url"""
return f"https://youtube.com/playlist?list={playlist_id}"
def download(self) -> List[YoutubeVideo]:
def download(self) -> List[YoutubePlaylistVideo]:
"""
Downloads all videos in a Youtube playlist
"""
playlist_id = self.download_options.playlist_id.value
playlist_url = self.playlist_url(playlist_id=playlist_id)
return self._download_using_metadata(url=playlist_url, ignore_prefix=playlist_id)
self.extract_info_json(url=playlist_url)
_, videos = self.extract_from_info_json(
parent_prefix=playlist_id, parent_entry_type=YoutubePlaylist
)
return videos
###############################################################################
@ -165,8 +140,24 @@ class YoutubeChannelDownloaderOptions(YoutubeDownloaderOptions, DateRangeValidat
)
class YoutubeChannelDownloader(YoutubeDownloader[YoutubeChannelDownloaderOptions]):
class YoutubeChannelDownloader(YoutubeDownloader[YoutubeChannelDownloaderOptions, YoutubeVideo]):
downloader_options_type = YoutubeChannelDownloaderOptions
downloader_entry_type = YoutubeVideo
def __init__(
self,
working_directory: str,
download_options: DownloaderOptionsT,
ytdl_options: Optional[Dict] = None,
download_archive_file_name: Optional[str] = None,
):
super().__init__(
working_directory=working_directory,
download_options=download_options,
ytdl_options=ytdl_options,
download_archive_file_name=download_archive_file_name,
)
self.channel: Optional[YoutubeChannel] = None
@classmethod
def channel_url(cls, channel_id: str) -> str:
@ -187,46 +178,38 @@ class YoutubeChannelDownloader(YoutubeDownloader[YoutubeChannelDownloaderOptions
Downloads all videos from a channel
"""
channel_url = self.channel_url(channel_id=self.channel_id)
ytdl_options_overrides = {}
# If a date range is specified when download a YT channel, add it into the ytdl options
ytdl_options_overrides = {}
source_date_range = self.download_options.get_date_range()
if source_date_range:
ytdl_options_overrides["daterange"] = source_date_range
return self._download_using_metadata(
url=channel_url,
ignore_prefix=self.channel_id,
ytdl_options_overrides=ytdl_options_overrides,
self.extract_info_json(ytdl_options_overrides=ytdl_options_overrides, url=channel_url)
self.channel, videos = self.extract_from_info_json(
parent_prefix=self.channel_id, parent_entry_type=YoutubeChannel
)
@classmethod
def __download_thumbnail(
cls,
entry_dict: dict,
thumbnail_id: str,
return videos
def _download_thumbnail(
self,
thumbnail_url: str,
output_thumbnail_path: str,
):
"""
Downloads a specific thumbnail from a YTDL entry's thumbnail list
Downloads a thumbnail and stores it in the output directory
Parameters
----------
entry_dict:
YTDL entry dict
thumbnail_id:
Id of the thumbnail defined in the YTDL thumnail
thumbnail_url:
Url of the thumbnail
output_thumbnail_path:
Path to store the thumbnail after downloading
"""
thumbnail_url = None
for thumbnail in entry_dict.get("thumbnails", []):
if thumbnail["id"] == thumbnail_id:
thumbnail_url = thumbnail["url"]
break
if not thumbnail_url:
logger.warning("Could not find a thumbnail for %s", entry_dict.get("id"))
logger.warning("Could not find a thumbnail for %s", self.channel.uid)
return
with urlopen(thumbnail_url) as file:
@ -245,22 +228,14 @@ class YoutubeChannelDownloader(YoutubeDownloader[YoutubeChannelDownloaderOptions
output_directory
Output directory path
"""
channel_json_file_path = Path(self.working_directory) / f"{self.channel_id}.info.json"
with open(channel_json_file_path, "r", encoding="utf-8") as channel_json:
channel_entry = json.load(channel_json)
avatar_thumbnail_name = overrides.apply_formatter(self.download_options.channel_avatar_path)
self._download_thumbnail(
thumbnail_url=self.channel.avatar_thumbnail_url(),
output_thumbnail_path=str(Path(output_directory) / avatar_thumbnail_name),
)
if self.download_options.channel_avatar_path:
thumbnail_name = overrides.apply_formatter(self.download_options.channel_avatar_path)
self.__download_thumbnail(
entry_dict=channel_entry,
thumbnail_id="avatar_uncropped",
output_thumbnail_path=str(Path(output_directory) / thumbnail_name),
)
if self.download_options.channel_banner_path:
thumbnail_name = overrides.apply_formatter(self.download_options.channel_banner_path)
self.__download_thumbnail(
entry_dict=channel_entry,
thumbnail_id="banner_uncropped",
output_thumbnail_path=str(Path(output_directory) / thumbnail_name),
)
banner_thumbnail_name = overrides.apply_formatter(self.download_options.channel_banner_path)
self._download_thumbnail(
thumbnail_url=self.channel.banner_thumbnail_url(),
output_thumbnail_path=str(Path(output_directory) / banner_thumbnail_name),
)

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

@ -1,8 +1,8 @@
import os.path
from pathlib import Path
from typing import List, Dict
from typing import List
from typing import Optional
from ytdl_sub.entries.base_entry import PlaylistMetadata
from ytdl_sub.entries.entry import Entry
from ytdl_sub.entries.variables.youtube_variables import YoutubeVideoVariables
@ -36,42 +36,6 @@ class YoutubeVideo(YoutubeVideoVariables, Entry):
class YoutubePlaylistVideo(YoutubeVideo):
def __init__(
self,
entry_dict: Dict,
working_directory: str,
playlist_metadata: PlaylistMetadata,
):
"""
Initialize the playlist video with playlist metadata
"""
super().__init__(entry_dict=entry_dict, working_directory=working_directory)
self._playlist_metadata = playlist_metadata
@classmethod
def from_youtube_video(
cls,
youtube_video: YoutubeVideo,
playlist_metadata: PlaylistMetadata,
) -> "YoutubePlaylistVideo":
"""
Parameters
----------
youtube_video:
Video to convert to an playlist video
playlist_metadata:
Metadata for playlist ordering
Returns
-------
YoutubeVideo converted to a YoutubePlaylistVideo
"""
return YoutubePlaylistVideo(
entry_dict=youtube_video._kwargs, # pylint: disable=protected-access
working_directory=youtube_video.working_directory(),
playlist_metadata=playlist_metadata,
)
@property
def playlist_index(self) -> int:
"""
@ -79,7 +43,7 @@ class YoutubePlaylistVideo(YoutubeVideo):
-------
The playlist index
"""
return self._playlist_metadata.playlist_index
return self.kwargs("playlist_index")
@property
def playlist_size(self) -> int:
@ -88,22 +52,17 @@ class YoutubePlaylistVideo(YoutubeVideo):
-------
The size of the playlist
"""
return self._playlist_metadata.playlist_count
return self.kwargs("playlist_count")
class YoutubePlaylist(Entry):
@property
def _videos(self) -> List[YoutubeVideo]:
"""
Returns all videos in the playlist represented by non-playlist Videos. Use this to fetch any
data needed from the videos before representing it as a playlist video.
"""
return [
YoutubeVideo(entry_dict=entry, working_directory=self._working_directory)
for entry in self.kwargs("entries")
]
"""
Class placeholder for youtube playlists
"""
def playlist_videos(self) -> List[YoutubePlaylistVideo]:
class YoutubeChannel(Entry):
def videos(self) -> List[YoutubeVideo]:
"""
Returns
-------
@ -111,14 +70,40 @@ class YoutubePlaylist(Entry):
playlist-specific fields like playlist_index and playlist_size with its actual value.
"""
return [
YoutubePlaylistVideo.from_youtube_video(
youtube_video=video,
playlist_metadata=PlaylistMetadata(
playlist_id=self.uid,
playlist_extractor=self.extractor,
playlist_index=video.kwargs("playlist_index"),
playlist_count=self.kwargs('playlist_count'),
),
)
for video in self._videos
YoutubeVideo(entry_dict=entry, working_directory=self._working_directory)
for entry in self.kwargs("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 channel's thumbnail
Returns
-------
Desired thumbnail url if it exists. None if it does not.
"""
for thumbnail in self.kwargs("thumbnails"):
if thumbnail["id"] == thumbnail_id:
return thumbnail["url"]
return None
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")

32
subscriptions.yaml
View file

@ -18,23 +18,23 @@
# artist: "DeLorra"
# genre: "Synthwave / Electronic"
#
bl00dwave:
preset: "soundcloud_with_id3_tags"
soundcloud:
username: bl00dwave
overrides:
artist: "bl00dwave"
genre: "Synthwave / Electronic"
#
#tom_petty:
# preset: "music_videos"
# youtube:
# download_strategy: "playlist"
# playlist_id: PLoopXDarluPBnuxs4PTC55Sc_2ShAXC0i
# output_options:
# output_directory: "/tmp/Tom Petty"
#bl00dwave:
# preset: "soundcloud_with_id3_tags"
# soundcloud:
# username: bl00dwave
# overrides:
# artist: "Tom Petty and the Heartbreakers"
# artist: "bl00dwave"
# genre: "Synthwave / Electronic"
#
tom_petty:
preset: "music_videos"
youtube:
download_strategy: "playlist"
playlist_id: PLoopXDarluPBnuxs4PTC55Sc_2ShAXC0i
output_options:
output_directory: "/tmp/Tom Petty"
overrides:
artist: "Tom Petty and the Heartbreakers"
#
#rammstein:
# preset: "music_videos"