lots o cleaning

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}"
 

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)

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"]]

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

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

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:

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)