diff --git a/subscriptions.yaml b/subscriptions.yaml index 3729bdde..e9662ffd 100644 --- a/subscriptions.yaml +++ b/subscriptions.yaml @@ -7,10 +7,7 @@ presets: download_strategy: "albums_then_tracks" skip_premiere_tracks: True ytdl_opts: - format: 'bestaudio/best' - extractaudio: True # only keep the audio - audioformat: "mp3" # convert to mp3 - noplaylist: True + format: 'bestaudio[ext=mp3]' post_process: file_name: "[{album_year}] {sanitized_album}/{track_number_padded} - {sanitized_title}.{ext}" thumbnail_name: "[{album_year}] {sanitized_album}/folder.{thumbnail_ext}" @@ -30,14 +27,14 @@ presets: format: 'best' ignoreerrors: True post_process: - file_name: "{sanitized_artist} - {sanitized_track}.{ext}" + file_name: "{sanitized_artist} - {sanitized_title}.{ext}" convert_thumbnail: "jpg" - thumbnail_name: "{sanitized_artist} - {sanitized_track}.jpg" - nfo_name: "{sanitized_artist} - {sanitized_track}.nfo" + thumbnail_name: "{sanitized_artist} - {sanitized_title}.jpg" + nfo_name: "{sanitized_artist} - {sanitized_title}.nfo" nfo_root: "musicvideo" nfo: artist: "{artist}" - title: "{sanitized_track}" + title: "{title}" album: "Music Videos" year: "{upload_year}" diff --git a/ytdl_subscribe/downloaders/__init__.py b/ytdl_subscribe/downloaders/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/ytdl_subscribe/downloaders/downloader.py b/ytdl_subscribe/downloaders/downloader.py new file mode 100644 index 00000000..3f4bc280 --- /dev/null +++ b/ytdl_subscribe/downloaders/downloader.py @@ -0,0 +1,83 @@ +import tempfile +from contextlib import contextmanager +from pathlib import Path +from typing import Dict +from typing import Optional + +import yt_dlp as ytdl + + +class Downloader: + """ + Class that interacts with ytdl to perform the download of metadata and content, and should translate that to + list of Entry objects. + """ + + @classmethod + def ytdl_option_overrides(cls) -> Dict: + """Global overrides that even overwrite user input""" + return {"writethumbnail": True, "noplaylist": True} + + @classmethod + def ytdl_option_defaults(cls) -> Dict: + """Downloader defaults that can be overwritten from user input""" + return {} + + @classmethod + def _configure_ytdl_options( + cls, ytdl_options: Optional[Dict], working_directory: str + ) -> Dict: + """Configure the ytdl options for the downloader""" + if ytdl_options is None: + ytdl_options = dict() + + # Overwrite defaults with input + ytdl_options = dict(cls.ytdl_option_defaults(), **ytdl_options) + + # Overwrite defaults + input with global options + ytdl_options = dict(ytdl_options, **cls.ytdl_option_overrides()) + + # Finally overwrite the output location with the specified working directory + ytdl_options["outtmpl"] = str(Path(working_directory) / "%(id)s.%(ext)s") + return ytdl_options + + def __init__( + self, + output_directory: str, + working_directory: Optional[str] = None, + ytdl_options: Optional[Dict] = None, + ): + self.output_path = output_directory + + self.working_directory = working_directory + if self.working_directory is None: + self.working_directory = tempfile.TemporaryDirectory().name + + self.ytdl_options = Downloader._configure_ytdl_options( + ytdl_options=ytdl_options, + working_directory=self.working_directory, + ) + + @contextmanager + def ytdl_downloader( + self, ytdl_options_overrides: Optional[Dict] = None + ) -> ytdl.YoutubeDL: + """ + Context manager to interact with yt_dlp. + """ + ytdl_options = self.ytdl_options + if ytdl_options_overrides is not None: + ytdl_options = dict(ytdl_options, **ytdl_options_overrides) + + with ytdl.YoutubeDL(ytdl_options) as ytdl_downloader: + yield ytdl_downloader + + def extract_info( + self, ytdl_options_overrides: Optional[Dict] = None, **kwargs + ) -> Dict: + """ + Wrapper around yt_dlp.YoutubeDL.YoutubeDL.extract_info + All kwargs will passed to the extract_info function. + """ + with self.ytdl_downloader(ytdl_options_overrides) as ytdl_downloader: + return ytdl_downloader.extract_info(**kwargs) diff --git a/ytdl_subscribe/downloaders/soundcloud_downloader.py b/ytdl_subscribe/downloaders/soundcloud_downloader.py new file mode 100644 index 00000000..18d64d60 --- /dev/null +++ b/ytdl_subscribe/downloaders/soundcloud_downloader.py @@ -0,0 +1,30 @@ +from typing import Dict +from typing import List + +from ytdl_subscribe.downloaders.downloader import Downloader +from ytdl_subscribe.entries.soundcloud import SoundcloudAlbum +from ytdl_subscribe.entries.soundcloud import SoundcloudTrack + + +class SoundcloudDownloader(Downloader): + @classmethod + def ytdl_option_defaults(cls) -> Dict: + return { + "format": "bestaudio[ext=mp3]", + } + + @classmethod + def artist_url(cls, artist_name: str) -> str: + return f"https://soundcloud.com/{artist_name}" + + def download_albums(self, artist_name) -> List[SoundcloudAlbum]: + artist_albums_url = f"{self.artist_url(artist_name)}/albums" + + info = self.extract_info(url=artist_albums_url) + return [SoundcloudAlbum(**e) for e in info["entries"]] + + def download_tracks(self, artist_name) -> List[SoundcloudTrack]: + artist_tracks_url = f"{self.artist_url(artist_name)}/tracks" + + info = self.extract_info(url=artist_tracks_url) + return [SoundcloudTrack(**e) for e in info["entries"]] diff --git a/ytdl_subscribe/downloaders/youtube_downloader.py b/ytdl_subscribe/downloaders/youtube_downloader.py new file mode 100644 index 00000000..d93ea4b8 --- /dev/null +++ b/ytdl_subscribe/downloaders/youtube_downloader.py @@ -0,0 +1,48 @@ +import json +import os +from pathlib import Path +from typing import List + +from ytdl_subscribe.downloaders.downloader import Downloader +from ytdl_subscribe.entries.youtube import YoutubeVideo + + +class YoutubeDownloader(Downloader): + @classmethod + def playlist_url(cls, playlist_id: str) -> str: + return f"https://youtube.com/playlist?list={playlist_id}" + + def _download_metadata(self, url: str) -> None: + """ + 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??) + """ + ytdl_metadata_override = { + "download_archive": str( + Path(self.working_directory) / "ytdl-download-archive.txt" + ), + "writeinfojson": True, + } + _ = self.extract_info(ytdl_options_overrides=ytdl_metadata_override, url=url) + + def download_playlist(self, playlist_id: str) -> List[YoutubeVideo]: + """ + Downloads all videos in a playlist + """ + playlist_url = self.playlist_url(playlist_id=playlist_id) + + self._download_metadata(url=playlist_url) + + # Load the entries from info.json, ignore the playlist entry + entries: List[YoutubeVideo] = [] + + # Load the entries from info.json, ignore the playlist entry + for file_name in os.listdir(self.working_directory): + if file_name.endswith(".info.json") and not file_name.startswith( + playlist_id + ): + with open(Path(self.working_directory) / file_name, "r") as f: + entries.append(YoutubeVideo(**json.load(f))) + + return entries diff --git a/ytdl_subscribe/entries/soundcloud.py b/ytdl_subscribe/entries/soundcloud.py index 476cce7f..9cf5dfb6 100644 --- a/ytdl_subscribe/entries/soundcloud.py +++ b/ytdl_subscribe/entries/soundcloud.py @@ -95,10 +95,6 @@ class SoundcloudAlbum(Entry): Entry object to represent a Soundcloud album. """ - def __init__(self, skip_premiere_tracks: bool, **kwargs): - super().__init__(**kwargs) - self.skip_premiere_tracks = skip_premiere_tracks - @property def _single_tracks(self) -> List[SoundcloudTrack]: """ @@ -107,8 +103,9 @@ class SoundcloudAlbum(Entry): """ return [SoundcloudTrack(**entry) for entry in self.kwargs("entries")] - @property - def album_tracks(self) -> List[SoundcloudAlbumTrack]: + def album_tracks( + self, skip_premiere_tracks: bool = True + ) -> 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. @@ -122,7 +119,7 @@ class SoundcloudAlbum(Entry): ) for i, entry in enumerate(self.kwargs("entries")) ] - if self.skip_premiere_tracks: + if skip_premiere_tracks: album_tracks = [t for t in album_tracks if not t.is_premiere] return album_tracks diff --git a/ytdl_subscribe/subscriptions/soundcloud.py b/ytdl_subscribe/subscriptions/soundcloud.py index 42eebb41..7212e985 100644 --- a/ytdl_subscribe/subscriptions/soundcloud.py +++ b/ytdl_subscribe/subscriptions/soundcloud.py @@ -1,7 +1,6 @@ from typing import List -import yt_dlp as ytdl - +from ytdl_subscribe.downloaders.soundcloud_downloader import SoundcloudDownloader from ytdl_subscribe.entries.soundcloud import SoundcloudAlbum from ytdl_subscribe.entries.soundcloud import SoundcloudTrack from ytdl_subscribe.enums import SubscriptionSource @@ -12,34 +11,33 @@ class SoundcloudSubscription(Subscription): source = SubscriptionSource.SOUNDCLOUD def extract_info(self): - base_url = f"https://soundcloud.com/{self.options['username']}" tracks: List[SoundcloudTrack] = [] + soundcloud_downloader = SoundcloudDownloader( + output_directory=self.output_path, + working_directory=self.WORKING_DIRECTORY, + ytdl_options=self.ytdl_opts, + ) if self.options.get("download_strategy") == "albums_then_tracks": # 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 - with ytdl.YoutubeDL(self.ytdl_opts) as ytd: - info = ytd.extract_info(base_url + "/albums") + albums: List[SoundcloudAlbum] = soundcloud_downloader.download_albums( + artist_name=self.options["username"] + ) - # For each album, parse each entry in the album - albums = [ - SoundcloudAlbum( - skip_premiere_tracks=self.options["skip_premiere_tracks"], **e - ) - for e in info["entries"] - ] for album in albums: - tracks += album.album_tracks + tracks += album.album_tracks( + skip_premiere_tracks=self.options["skip_premiere_tracks"] + ) - with ytdl.YoutubeDL(self.ytdl_opts) as ytd: - info = ytd.extract_info(base_url + "/tracks") - - # Skip parsing entries that have already been parsed when parsing albums - single_tracks = [SoundcloudTrack(**e) for e in info["entries"]] + # only add tracks that are not part of an album + single_tracks = soundcloud_downloader.download_tracks( + artist_name=self.options["username"] + ) tracks += [ track for track in single_tracks - if not any([album.contains(track) for album in albums]) + if not any(album.contains(track) for album in albums) ] for e in tracks: diff --git a/ytdl_subscribe/subscriptions/youtube.py b/ytdl_subscribe/subscriptions/youtube.py index 55b093b5..e4dff3f2 100644 --- a/ytdl_subscribe/subscriptions/youtube.py +++ b/ytdl_subscribe/subscriptions/youtube.py @@ -1,10 +1,4 @@ -import json -import os -from typing import List - -import yt_dlp as ytdl - -from ytdl_subscribe.entries.youtube import YoutubeVideo +from ytdl_subscribe.downloaders.youtube_downloader import YoutubeDownloader from ytdl_subscribe.enums import SubscriptionSource from ytdl_subscribe.subscriptions.subscription import Subscription @@ -13,27 +7,15 @@ class YoutubeSubscription(Subscription): source = SubscriptionSource.YOUTUBE def extract_info(self): - playlist_id = self.options["playlist_id"] - url = f"https://youtube.com/playlist?list={playlist_id}" - track_ytdl_opts = { - "download_archive": self.WORKING_DIRECTORY + "/ytdl-download-archive.txt", - "writeinfojson": True, - } + youtube_downloader = YoutubeDownloader( + output_directory=self.output_path, + working_directory=self.WORKING_DIRECTORY, + ytdl_options=self.ytdl_opts, + ) - # 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, it will - # not fetch the metadata (maybe there is a way??) - with ytdl.YoutubeDL(dict(self.ytdl_opts, **track_ytdl_opts)) as ytd: - _ = ytd.extract_info(url) - - # Load the entries from info.json, ignore the playlist entry - entries: List[YoutubeVideo] = [] - for file_name in os.listdir(self.WORKING_DIRECTORY): - if file_name.endswith(".info.json") and not file_name.startswith( - playlist_id - ): - with open(self.WORKING_DIRECTORY + "/" + file_name, "r") as f: - entries.append(YoutubeVideo(**json.load(f))) + entries = youtube_downloader.download_playlist( + playlist_id=self.options["playlist_id"] + ) for e in entries: self.post_process_entry(e)