entries

README.md

@@ -2,3 +2,8 @@
 Automate downloading and tagging files using YoutubeDL.
 
 README is WIP but the package is usable. See the provided `subscriptions.yaml`, install the `requirements.txt`, and run `PYTHONPATH=. python ytdl_subscribe main -h` to get started. 
+
+
+## TODO
+- Add tests with coverage
+- Test output paths work before downloading

requirements.txt

@@ -1,4 +1,4 @@
-black
+black==21.11b1
 argparse==1.4.0
 dicttoxml==1.7.4
 mergedeep==1.3.4
@@ -8,3 +8,4 @@ Pillow==8.3.2
 pyYAML==5.4.1
 sanitize-filename==1.2.0
 yt-dlp==2021.9.2
+pytest==7.1.1

subscriptions.yaml

@@ -12,14 +12,14 @@ presets:
       audioformat: "mp3"  # convert to mp3
       noplaylist: True
     post_process:
-      file_name: "[{album_year}] {sanitized_album}/{tracknumberpadded} - {sanitized_title}.{ext}"
+      file_name: "[{album_year}] {sanitized_album}/{track_number_padded} - {sanitized_title}.{ext}"
       thumbnail_name: "[{album_year}] {sanitized_album}/folder.{thumbnail_ext}"
       tagging:
         artist: "{artist}"
         albumartist: "{artist}"
         title: "{title}"
         album: "{album}"
-        tracknumber: "{tracknumber}"
+        tracknumber: "{track_number}"
         year: "{album_year}"
         genre: "{genre}"
 
@@ -62,7 +62,7 @@ subscriptions:
     preset: "soundcloud_with_id3_tags"
     soundcloud:
       username: bl00dwave
-    output_path: "/mnt/nas/Downloads/ytdl/music/bl00dwave"
+    output_path: "/tmp/bl00dwave"
     overrides:
       artist: "bl00dwave"
       genre: "Synthwave / Electronic"

ytdl_subscribe/entries/entry.py

@@ -0,0 +1,67 @@
+from string import Formatter
+from typing import Any, Dict, Optional
+
+from sanitize_filename import sanitize
+
+
+class Entry(object):
+    """
+    Entry object to represent a single media object returned from yt-dlp.
+    """
+
+    def __init__(self, **kwargs):
+        self._kwargs = kwargs
+
+    def kwargs(self, key) -> Any:
+        if key not in self._kwargs:
+            raise KeyError(
+                f"Expected '{key}' in {self.__class__.__name__} but does not exist."
+            )
+        return self._kwargs[key]
+
+    @property
+    def uid(self) -> str:
+        return self.kwargs("id")
+
+    @property
+    def title(self) -> str:
+        return self.kwargs("title")
+
+    @property
+    def sanitized_title(self) -> str:
+        return sanitize(self.title)
+
+    @property
+    def ext(self) -> str:
+        return self.kwargs("ext")
+
+    @property
+    def download_file_name(self) -> str:
+        return f"{self.uid}.{self.ext}"
+
+    def file_path(self, relative_directory: str):
+        return f"{relative_directory}/{self.download_file_name}"
+
+    def to_dict(self) -> Dict:
+        return {
+            "uid": self.uid,
+            "title": self.title,
+            "sanitized_title": self.sanitized_title,
+            "ext": self.ext,
+        }
+
+    def apply_formatter(self, format_string: str, overrides: Optional[Dict]) -> str:
+        entry_dict = self.to_dict()
+        if overrides:
+            entry_dict = dict(entry_dict, **overrides)
+
+        field_names = [
+            fname for _, fname, _, _ in Formatter().parse(format_string) if fname
+        ]
+        for fname in field_names:
+            if fname not in entry_dict:
+                raise KeyError(
+                    f"Format variable '{fname}' does not exist for {self.__class__.__name__}."
+                )
+
+        return format_string.format(**entry_dict)

ytdl_subscribe/entries/soundcloud.py

@@ -0,0 +1,106 @@
+from typing import List, Type, Dict
+
+from sanitize_filename import sanitize
+
+from ytdl_subscribe.entries.entry import Entry
+
+
+class SoundcloudTrack(Entry):
+    @property
+    def upload_date(self) -> str:
+        return self.kwargs("upload_date")
+
+    @property
+    def upload_year(self) -> int:
+        return int(self.upload_date[:4])
+
+    @property
+    def thumbnail(self) -> str:
+        return self.kwargs("thumbnail")
+
+    @property
+    def thumbnail_ext(self) -> str:
+        return self.thumbnail.split(".")[-1]
+
+    @property
+    def track_number(self) -> int:
+        return 1
+
+    @property
+    def track_number_padded(self) -> str:
+        return f"{self.track_number:02d}"
+
+    @property
+    def album(self) -> str:
+        return self.title
+
+    @property
+    def sanitized_album(self) -> str:
+        return sanitize(self.album)
+
+    @property
+    def album_year(self) -> int:
+        return self.upload_year
+
+    @property
+    def is_premiere(self) -> bool:
+        return "/preview/" in self.kwargs("url")
+
+    def to_dict(self) -> Dict:
+        return dict(
+            super(SoundcloudTrack, self).to_dict(),
+            **{
+                "upload_date": self.upload_date,
+                "upload_year": self.upload_year,
+                "thumbnail": self.thumbnail,
+                "thumbnail_ext": self.thumbnail_ext,
+                "track_number": self.track_number,
+                "track_number_padded": self.track_number_padded,
+                "album": self.album,
+                "sanitized_album": self.sanitized_album,
+                "album_year": self.album_year,
+                "is_premiere": self.is_premiere,
+            },
+        )
+
+
+class SoundcloudAlbumTrack(SoundcloudTrack):
+    def __init__(self, album, track_number, **kwargs):
+        super(SoundcloudAlbumTrack, self).__init__(**kwargs)
+        self._album = album
+        self._track_number = track_number
+
+    @property
+    def track_number(self) -> int:
+        return self._track_number
+
+    @property
+    def album(self) -> str:
+        return self._album.title
+
+    @property
+    def album_year(self) -> int:
+        return self._album.album_year
+
+class SoundcloudAlbum(Entry):
+    def __init__(self, skip_premiere_tracks: bool, **kwargs):
+        super(SoundcloudAlbum, self).__init__(**kwargs)
+        self.skip_premiere_tracks = skip_premiere_tracks
+
+    @property
+    def album_tracks(self) -> List[SoundcloudAlbumTrack]:
+        tracks = [
+            SoundcloudAlbumTrack(album=self, track_number=i + 1, **entry)
+            for i, entry in enumerate(self.kwargs("entries"))
+        ]
+        if self.skip_premiere_tracks:
+            tracks = [t for t in tracks if not t.is_premiere]
+
+        return tracks
+
+    @property
+    def album_year(self) -> int:
+        return min([track.upload_year for track in self.album_tracks])
+
+    def contains(self, track: SoundcloudTrack) -> bool:
+        return any([track.uid == t.uid for t in self.album_tracks])

ytdl_subscribe/parse.py

@@ -1,3 +1,5 @@
+from typing import Optional
+
 import yaml
 
 from mergedeep import mergedeep
@@ -57,7 +59,7 @@ def parse_presets(yaml_dict):
     return presets
 
 
-def parse_subscriptions(yaml_dict, presets, subscriptions=None):
+def parse_subscriptions(yaml_dict: dict, presets: dict, subscriptions: Optional[list]):
     """
     Parses subscriptions from a subscription yaml dict
 
@@ -65,12 +67,12 @@ def parse_subscriptions(yaml_dict, presets, subscriptions=None):
     ----------
     yaml_dict: dict
     presets: dict
-    subscriptions: list of str or None
+    subscriptions: list[str] or None
         If present, only parse these subscriptions
 
     Returns
     -------
-    list of Subscription
+    list[Subscription]
     """
     parsed_subscriptions = []
 

ytdl_subscribe/subscriptions/soundcloud.py

@@ -1,46 +1,18 @@
+from typing import List, Type
+
 import yt_dlp as ytdl
-from sanitize_filename import sanitize
 
 from ytdl_subscribe import SubscriptionSource
+from ytdl_subscribe.entries.soundcloud import SoundcloudAlbum, SoundcloudTrack
 from ytdl_subscribe.subscriptions.subscription import Subscription
 
 
 class SoundcloudSubscription(Subscription):
     source = SubscriptionSource.SOUNDCLOUD
 
-    def is_entry_skippable(self, entry):
-        return self.options["skip_premiere_tracks"] and "/preview/" in entry["url"]
-
-    def parse_entry(self, entry):
-        entry = super(SoundcloudSubscription, self).parse_entry(entry)
-        entry["upload_year"] = entry["upload_date"][:4]
-
-        # Add thumbnail ext value
-        entry["thumbnail_ext"] = entry["thumbnail"].split(".")[-1]
-
-        # If the entry does not have album fields, set them to be the track fields
-        if "album" not in entry:
-            entry["album"] = entry["title"]
-            entry["sanitized_album"] = entry["sanitized_title"]
-            entry["album_year"] = entry["upload_year"]
-            entry["tracknumber"] = 1
-            entry["tracknumberpadded"] = f"{1:02d}"
-
-        return entry
-
-    def parse_album_entry(self, album_entry):
-        album_year = min([int(e["upload_date"][:4]) for e in album_entry["entries"]])
-        for track_number, e in enumerate(album_entry["entries"], start=1):
-            e["album"] = album_entry["title"]
-            e["sanitized_album"] = sanitize(album_entry["title"])
-            e["album_year"] = album_year
-            e["tracknumber"] = track_number
-            e["tracknumberpadded"] = f"{track_number:02d}"
-        return album_entry
-
     def extract_info(self):
         base_url = f"https://soundcloud.com/{self.options['username']}"
-        tracks = []
+        tracks: List[Type[SoundcloudTrack]] = []
 
         if self.options.get("download_strategy") == "albums_then_tracks":
             # Get the album info first. This tells us which track ids belong
@@ -49,23 +21,24 @@ class SoundcloudSubscription(Subscription):
                 info = ytd.extract_info(base_url + "/albums")
 
             # For each album, parse each entry in the album
-            album_entries = [self.parse_album_entry(a) for a in info["entries"]]
-            for album_entry in album_entries:
-                tracks += [
-                    self.parse_entry(e)
-                    for e in album_entry["entries"]
-                    if not self.is_entry_skippable(e)
-                ]
+            albums = [
+                SoundcloudAlbum(
+                    skip_premiere_tracks=self.options["skip_premiere_tracks"], **e
+                )
+                for e in info["entries"]
+            ]
+            for album in albums:
+                tracks += album.album_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
-            album_track_ids = [t["id"] for t in tracks]
+            single_tracks = [SoundcloudTrack(**e) for e in info["entries"]]
             tracks += [
-                self.parse_entry(e)
-                for e in info["entries"]
-                if e["id"] not in album_track_ids and not self.is_entry_skippable(e)
+                track
+                for track in single_tracks
+                if not any([album.contains(track) for album in albums])
             ]
 
             for e in tracks:

ytdl_subscribe/subscriptions/subscription.py

@@ -1,5 +1,7 @@
 import os
+from copy import deepcopy
 from shutil import copyfile
+from typing import Type
 
 import dicttoxml
 import music_tag
@@ -7,6 +9,7 @@ from PIL import Image
 from sanitize_filename import sanitize
 
 from ytdl_subscribe import SubscriptionSource
+from ytdl_subscribe.entries.entry import Entry
 
 
 class Subscription(object):
@@ -36,6 +39,8 @@ class Subscription(object):
         self.ytdl_opts = ytdl_opts or dict()
         self.post_process = post_process
         self.overrides = overrides
+        for k, v in deepcopy(overrides).items():
+            self.overrides[f'sanitized_{k}'] = sanitize(v)
         self.output_path = output_path
 
         # Separate each subscription's working directory
@@ -45,30 +50,17 @@ class Subscription(object):
         self.ytdl_opts["outtmpl"] = self.WORKING_DIRECTORY + "/%(id)s.%(ext)s"
         self.ytdl_opts["writethumbnail"] = True
 
-    def format_entry(self, value, entry):
-        """
-        Parameters
-        ----------
-        value: str
-            String with format variables that should be populated
-        entry: dict
-            Entry used to populate any format variables in the value
-
-        Returns
-        -------
-        str
-        """
-        return value.format(**entry)
-
-    def format_filepath(self, filepath, entry, makedirs=False):
+    def format_filepath(
+        self, filepath_formatter: str, entry: Type[Entry], makedirs=False
+    ):
         """
         Convert a filepath value in the config to an actual filepath.
 
         Parameters
         ----------
-        filepath: str
+        filepath_formatter: str
             File path relative to the specified output path
-        entry: dict
+        entry: Type[Entry]
             Entry used to populate any format variables in the filepath
         makedirs: bool
             Whether to create all directories in the final filepath.
@@ -78,36 +70,25 @@ class Subscription(object):
         str
             Full filepath.
         """
-        file_name = self.format_entry(filepath, entry)
+        file_name = entry.apply_formatter(filepath_formatter, overrides=self.overrides)
         output_file_path = f"{self.output_path}/{file_name}"
         if makedirs:
             os.makedirs(os.path.dirname(output_file_path), exist_ok=True)
 
         return output_file_path
 
-    def parse_entry(self, entry):
-        # Add overrides to the entry
-        entry = dict(entry, **self.overrides)
-
-        # Add the file path to the entry, assert it exists
-        entry["file_path"] = f"{self.WORKING_DIRECTORY}/{entry['id']}.{entry['ext']}"
-
-        # Add sanitized values
-        entry["sanitized_artist"] = sanitize(entry["artist"])
-        entry["sanitized_title"] = sanitize(entry["title"])
-
-        return entry
-
-    def _post_process_tagging(self, entry):
-        t = music_tag.load_file(entry["file_path"])
+    def _post_process_tagging(self, entry: Type[Entry]):
+        t = music_tag.load_file(
+            entry.file_path(relative_directory=self.WORKING_DIRECTORY)
+        )
         for tag, tag_formatter in self.post_process["tagging"].items():
-            t[tag] = self.format_entry(tag_formatter, entry)
+            t[tag] = entry.apply_formatter(tag_formatter, overrides=self.overrides)
         t.save()
 
     def _post_process_nfo(self, entry):
         nfo = {}
         for tag, tag_formatter in self.post_process["nfo"].items():
-            nfo[tag] = self.format_entry(tag_formatter, entry)
+            nfo[tag] = entry.apply_formatter(tag_formatter, overrides=self.overrides)
 
         xml = dicttoxml.dicttoxml(
             obj=nfo,
@@ -127,7 +108,7 @@ class Subscription(object):
         """
         raise NotImplemented("Each source needs to implement how it extracts info")
 
-    def post_process_entry(self, entry):
+    def post_process_entry(self, entry: Type[Entry]):
         if "tagging" in self.post_process:
             self._post_process_tagging(entry)
 
@@ -135,7 +116,9 @@ class Subscription(object):
         output_file_path = self.format_filepath(
             self.post_process["file_name"], entry, makedirs=True
         )
-        copyfile(entry["file_path"], output_file_path)
+        copyfile(
+            entry.file_path(relative_directory=self.WORKING_DIRECTORY), output_file_path
+        )
 
         # Download the thumbnail if its present
         if "thumbnail_name" in self.post_process:
@@ -146,7 +129,7 @@ class Subscription(object):
             )
             if not os.path.isfile(thumbnail_dest_path):
                 thumbnail_file_path = (
-                    f"{self.WORKING_DIRECTORY}/{entry['id']}.{entry['thumbnail_ext']}"
+                    f"{self.WORKING_DIRECTORY}/{entry.uid}.{entry.thumbnail_ext}"
                 )
                 if os.path.isfile(thumbnail_file_path):
                     copyfile(thumbnail_file_path, thumbnail_dest_path)