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

Use info.json files for all download strategies (#18)

Browse source
This commit is contained in:
Jesse Bannon 2022-04-30 22:06:47 -07:00 committed by GitHub
parent 9734d391a6
commit f52f829a2b
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
6 changed files with 139 additions and 184 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -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]:

104
src/ytdl_sub/downloaders/soundcloud_downloader.py
View file

@ -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

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

@ -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,

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

@ -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).

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

@ -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)

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

@ -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