From f52f829a2bc31558b9ffee15f2732453b3f9e2a5 Mon Sep 17 00:00:00 2001 From: Jesse Bannon Date: Sat, 30 Apr 2022 22:06:47 -0700 Subject: [PATCH] Use info.json files for all download strategies (#18) --- src/ytdl_sub/downloaders/downloader.py | 72 +++++------- .../downloaders/soundcloud_downloader.py | 104 +++++++++++------- .../downloaders/youtube_downloader.py | 36 ++++-- src/ytdl_sub/entries/base_entry.py | 13 --- src/ytdl_sub/entries/soundcloud.py | 79 ++++--------- src/ytdl_sub/entries/youtube.py | 19 ---- 6 files changed, 139 insertions(+), 184 deletions(-) diff --git a/src/ytdl_sub/downloaders/downloader.py b/src/ytdl_sub/downloaders/downloader.py index bb5c7148..1055ac6f 100644 --- a/src/ytdl_sub/downloaders/downloader.py +++ b/src/ytdl_sub/downloaders/downloader.py @@ -8,7 +8,6 @@ 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 @@ -132,14 +131,35 @@ class Downloader(Generic[DownloaderOptionsT, DownloaderEntryT], ABC): 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: + def _get_entry_dicts_from_info_json_files(self) -> List[Dict]: """ - Wrapper around yt_dlp.YoutubeDL.YoutubeDL.extract_info + Returns + ------- + List of all info.json files read as JSON dicts + """ + entry_dicts: List[Dict] = [] + info_json_paths = [ + Path(self.working_directory) / file_name + for file_name in os.listdir(self.working_directory) + if file_name.endswith(".info.json") + ] - All kwargs will passed to the extract_info function. This also enables ytdl to write - .info.json files for all media downloaded. + for info_json_path in info_json_paths: + with open(info_json_path, "r", encoding="utf-8") as file: + entry_dicts.append(json.load(file)) - Catches RejectedVideoReached and ExistingVideoReached exceptions. + return entry_dicts + + def extract_info_via_info_json( + self, ytdl_options_overrides: Optional[Dict] = None, **kwargs + ) -> List[Dict]: + """ + Wrapper around yt_dlp.YoutubeDL.YoutubeDL.extract_info with infojson enabled. Entry dicts + are extracted via reading all info.json files in the working directory rather than + from the output of extract_info. + + This allows us to catch RejectedVideoReached and ExistingVideoReached exceptions, and + simply ignore while still being able to read downloaded entry metadata. Parameters ---------- @@ -158,45 +178,7 @@ class Downloader(Generic[DownloaderOptionsT, DownloaderEntryT], ABC): 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 + return self._get_entry_dicts_from_info_json_files() @abc.abstractmethod def download(self) -> List[DownloaderEntryT]: diff --git a/src/ytdl_sub/downloaders/soundcloud_downloader.py b/src/ytdl_sub/downloaders/soundcloud_downloader.py index 7cc3a48f..8b0509d9 100644 --- a/src/ytdl_sub/downloaders/soundcloud_downloader.py +++ b/src/ytdl_sub/downloaders/soundcloud_downloader.py @@ -58,32 +58,6 @@ class SoundcloudDownloader( """Returns full artist url""" return f"https://soundcloud.com/{artist_name}" - def download_albums(self, artist_name: str) -> List[SoundcloudAlbum]: - """ - Given an artist name, download all of their albums - """ - artist_albums_url = f"{self.artist_url(artist_name)}/albums" - - info = self.extract_info(url=artist_albums_url) - albums = [ - SoundcloudAlbum(entry_dict=e, working_directory=self.working_directory) - for e in info["entries"] - ] - - return [album for album in albums if album.track_count > 0] - - def download_tracks(self, artist_name) -> List[SoundcloudTrack]: - """ - Given an artist name, download all of their tracks - """ - artist_tracks_url = f"{self.artist_url(artist_name)}/tracks" - - info = self.extract_info(url=artist_tracks_url) - return [ - SoundcloudTrack(entry_dict=e, working_directory=self.working_directory) - for e in info["entries"] - ] - ############################################################################### # Soundcloud albums and singles downloader + options @@ -102,27 +76,77 @@ class SoundcloudAlbumsAndSinglesDownloader( ): downloader_options_type = SoundcloudAlbumsAndSinglesDownloadOptions + def _get_albums(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 entry_dicts: + if entry_dict.get("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 entry_dicts: + album_id = entry_dict.get("playlist_id") + if entry_dict.get("extractor") == "soundcloud" and 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]) -> List[SoundcloudTrack]: + artist_id = None + tracks: List[SoundcloudTrack] = [] + + # First, get the artist entry. All single tracks that do not belong to an album will belong + # to the 'artist' playlist + for entry_dict in entry_dicts: + if entry_dict.get("extractor") == "soundcloud:user": + artist_id = entry_dict["id"] + + # Then, get all singles that belong to the 'artist' playlist + for entry_dict in entry_dicts: + if ( + entry_dict.get("extractor") == "soundcloud" + and entry_dict.get("playlist_id") == artist_id + ): + tracks.append( + SoundcloudTrack(entry_dict=entry_dict, working_directory=self.working_directory) + ) + + return tracks + def download(self) -> List[SoundcloudTrack]: """ Soundcloud subscription to download albums and tracks as singles. """ - tracks: List[SoundcloudTrack] = [] + artist_url = self.artist_url(artist_name=self.download_options.username.value) + entry_dicts = self.extract_info_via_info_json(url=artist_url) - # Get the album info first. This tells us which track ids belong - # to an album. Unfortunately we cannot use download_archive or info.json for this - albums: List[SoundcloudAlbum] = self.download_albums( - artist_name=self.download_options.username.value - ) + # Get all of the artist's albums + albums = self._get_albums(entry_dicts=entry_dicts) + # Then, get all singles + tracks = self._get_singles(entry_dicts=entry_dicts) + + # Append all album tracks as SoundcloudAlbumTrack classes to the singles for album in albums: - tracks += album.album_tracks( - skip_premiere_tracks=self.download_options.skip_premiere_tracks.value - ) + tracks += album.album_tracks() - # only add tracks that are not part of an album - single_tracks = self.download_tracks(artist_name=self.download_options.username.value) - tracks += [ - track for track in single_tracks if not any(album.contains(track) for album in albums) - ] + # Filter any premiere tracks if specified + if self.download_options.skip_premiere_tracks.value: + tracks = [track for track in tracks if not track.is_premiere()] return tracks diff --git a/src/ytdl_sub/downloaders/youtube_downloader.py b/src/ytdl_sub/downloaders/youtube_downloader.py index bffbb043..8f4ed770 100644 --- a/src/ytdl_sub/downloaders/youtube_downloader.py +++ b/src/ytdl_sub/downloaders/youtube_downloader.py @@ -15,7 +15,6 @@ 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 @@ -110,14 +109,18 @@ class YoutubePlaylistDownloader( """ playlist_id = self.download_options.playlist_id.value playlist_url = self.playlist_url(playlist_id=playlist_id) + playlist_videos: List[YoutubePlaylistVideo] = [] - self.extract_info_json(url=playlist_url) + entry_dicts = self.extract_info_via_info_json(url=playlist_url) + for entry_dict in entry_dicts: + if entry_dict.get("extractor") == "youtube": + playlist_videos.append( + YoutubePlaylistVideo( + entry_dict=entry_dict, working_directory=self.working_directory + ) + ) - _, videos = self.extract_from_info_json( - parent_prefix=playlist_id, parent_entry_type=YoutubePlaylist - ) - - return videos + return playlist_videos ############################################################################### @@ -178,6 +181,7 @@ class YoutubeChannelDownloader(YoutubeDownloader[YoutubeChannelDownloaderOptions Downloads all videos from a channel """ channel_url = self.channel_url(channel_id=self.channel_id) + channel_videos: List[YoutubeVideo] = [] ytdl_options_overrides = {} # If a date range is specified when download a YT channel, add it into the ytdl options @@ -185,13 +189,21 @@ class YoutubeChannelDownloader(YoutubeDownloader[YoutubeChannelDownloaderOptions if source_date_range: ytdl_options_overrides["daterange"] = source_date_range - 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 + entry_dicts = self.extract_info_via_info_json( + ytdl_options_overrides=ytdl_options_overrides, url=channel_url ) - return videos + for entry_dict in entry_dicts: + if entry_dict.get("extractor") == "youtube": + channel_videos.append( + YoutubeVideo(entry_dict=entry_dict, working_directory=self.working_directory) + ) + if entry_dict.get("extractor") == "youtube:tab": + self.channel = YoutubeChannel( + entry_dict=entry_dict, working_directory=self.working_directory + ) + + return channel_videos def _download_thumbnail( self, diff --git a/src/ytdl_sub/entries/base_entry.py b/src/ytdl_sub/entries/base_entry.py index a4078c2c..f31bbc6f 100644 --- a/src/ytdl_sub/entries/base_entry.py +++ b/src/ytdl_sub/entries/base_entry.py @@ -1,22 +1,9 @@ from abc import ABC -from dataclasses import dataclass from typing import Any from typing import Dict from typing import Optional -@dataclass -class PlaylistMetadata: - """ - Metadata for storing playlist information. - """ - - playlist_index: int - playlist_id: str - playlist_extractor: str - playlist_count: int - - class BaseEntry(ABC): """ Abstract entry object to represent anything download from ytdl (playlist metadata, media, etc). diff --git a/src/ytdl_sub/entries/soundcloud.py b/src/ytdl_sub/entries/soundcloud.py index 4278482b..e6c2353e 100644 --- a/src/ytdl_sub/entries/soundcloud.py +++ b/src/ytdl_sub/entries/soundcloud.py @@ -1,7 +1,7 @@ from typing import 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.soundcloud_variables import SoundcloudVariables @@ -29,32 +29,28 @@ class SoundcloudAlbumTrack(SoundcloudTrack): self, entry_dict: Dict, working_directory: str, - album: str, album_year: int, - playlist_metadata: PlaylistMetadata, ): """ 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 = album self._album_year = album_year - self._playlist_metadata = playlist_metadata @property def track_number(self) -> int: """Returns the entry's track number""" - return self._playlist_metadata.playlist_index + return self.kwargs("playlist_index") + + @property + def track_count(self) -> int: + """Returns the entry's total tracks in album""" + return self.kwargs("playlist_count") @property def album(self) -> str: """Returns the entry's album name, fetched from its internal album""" - return self._album - - @property - def track_count(self) -> int: - """Returns the entry's total tracks in album for singles this is 1""" - return self._playlist_metadata.playlist_count + return self.kwargs("playlist") @property def album_year(self) -> int: @@ -64,22 +60,16 @@ class SoundcloudAlbumTrack(SoundcloudTrack): @classmethod def from_soundcloud_track( cls, - album: str, album_year: int, soundcloud_track: SoundcloudTrack, - playlist_metadata: PlaylistMetadata, ) -> "SoundcloudAlbumTrack": """ Parameters ---------- - album: - Album name album_year: Album year soundcloud_track: Track to convert to an album track - playlist_metadata: - Metadata for playlist ordering Returns ------- @@ -88,9 +78,7 @@ class SoundcloudAlbumTrack(SoundcloudTrack): return SoundcloudAlbumTrack( entry_dict=soundcloud_track._kwargs, # pylint: disable=protected-access working_directory=soundcloud_track.working_directory(), - album=album, album_year=album_year, - playlist_metadata=playlist_metadata, ) @@ -99,43 +87,32 @@ class SoundcloudAlbum(Entry): Entry object to represent a Soundcloud album. """ - @property - def _single_tracks(self) -> List[SoundcloudTrack]: + def __init__(self, entry_dict: Dict, working_directory: Optional[str] = None): """ - Returns all tracks in the album represented by singles. Use this to fetch any - data needed from the tracks before representing it as an album track. - """ - return [ - SoundcloudTrack(entry_dict=entry, working_directory=self._working_directory) - for entry in self.kwargs("entries") - ] + Initialize the entry using ytdl metadata - def album_tracks(self, skip_premiere_tracks: bool = True) -> List[SoundcloudAlbumTrack]: + 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. """ - tracks = [ - track - for track in self._single_tracks - if not (skip_premiere_tracks and track.is_premiere()) - ] - album_tracks = [ SoundcloudAlbumTrack.from_soundcloud_track( - album=self.title, - album_year=self.album_year, - soundcloud_track=track, - playlist_metadata=PlaylistMetadata( - playlist_id=self.uid, - playlist_extractor=self.extractor, - playlist_index=track.kwargs("playlist_index"), - playlist_count=self.track_count, - ), + album_year=self.album_year, soundcloud_track=track ) - for track in tracks + for track in self.tracks ] return album_tracks @@ -147,7 +124,7 @@ class SoundcloudAlbum(Entry): ------- The album's year, computed by the max upload year amongst all album tracks """ - return max(track.upload_year for track in self._single_tracks) + return max(track.upload_year for track in self.tracks) @property def track_count(self) -> int: @@ -157,11 +134,3 @@ class SoundcloudAlbum(Entry): Number of tracks in the album (technically a playlist) """ return self.kwargs("playlist_count") - - def contains(self, track: SoundcloudTrack) -> bool: - """ - Returns - ------- - True if this album contains this track. False otherwise. - """ - return any(track.uid == t.uid for t in self._single_tracks) diff --git a/src/ytdl_sub/entries/youtube.py b/src/ytdl_sub/entries/youtube.py index e9e74b2c..8f6b65d9 100644 --- a/src/ytdl_sub/entries/youtube.py +++ b/src/ytdl_sub/entries/youtube.py @@ -1,6 +1,5 @@ import os.path from pathlib import Path -from typing import List from typing import Optional from ytdl_sub.entries.entry import Entry @@ -55,25 +54,7 @@ class YoutubePlaylistVideo(YoutubeVideo): return self.kwargs("playlist_count") -class YoutubePlaylist(Entry): - """ - Class placeholder for youtube playlists - """ - - class YoutubeChannel(Entry): - def videos(self) -> List[YoutubeVideo]: - """ - Returns - ------- - All videos in the playlist represented as YoutubePlaylistVideos. This updates - playlist-specific fields like playlist_index and playlist_size with its actual value. - """ - return [ - 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