From 6662bc511259f926823bc4ef678653c29989930a Mon Sep 17 00:00:00 2001 From: Jesse Bannon Date: Wed, 30 Mar 2022 07:41:12 +0000 Subject: [PATCH] entries --- README.md | 5 + requirements.txt | 3 +- subscriptions.yaml | 6 +- tests/__init__.py | 0 ytdl_subscribe/entries/__init__.py | 0 ytdl_subscribe/entries/entry.py | 67 ++++++++++++ ytdl_subscribe/entries/soundcloud.py | 106 +++++++++++++++++++ ytdl_subscribe/parse.py | 8 +- ytdl_subscribe/subscriptions/soundcloud.py | 59 +++-------- ytdl_subscribe/subscriptions/subscription.py | 61 ++++------- 10 files changed, 226 insertions(+), 89 deletions(-) create mode 100644 tests/__init__.py create mode 100644 ytdl_subscribe/entries/__init__.py create mode 100644 ytdl_subscribe/entries/entry.py create mode 100644 ytdl_subscribe/entries/soundcloud.py diff --git a/README.md b/README.md index b7492a25..fbb4f3ae 100644 --- a/README.md +++ b/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 diff --git a/requirements.txt b/requirements.txt index 581a17f5..cd1dd136 100644 --- a/requirements.txt +++ b/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 diff --git a/subscriptions.yaml b/subscriptions.yaml index 1b5c16f3..3729bdde 100644 --- a/subscriptions.yaml +++ b/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" diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/ytdl_subscribe/entries/__init__.py b/ytdl_subscribe/entries/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/ytdl_subscribe/entries/entry.py b/ytdl_subscribe/entries/entry.py new file mode 100644 index 00000000..aab67d3d --- /dev/null +++ b/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) diff --git a/ytdl_subscribe/entries/soundcloud.py b/ytdl_subscribe/entries/soundcloud.py new file mode 100644 index 00000000..001005ef --- /dev/null +++ b/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]) diff --git a/ytdl_subscribe/parse.py b/ytdl_subscribe/parse.py index 1ae98410..79e40216 100644 --- a/ytdl_subscribe/parse.py +++ b/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 = [] diff --git a/ytdl_subscribe/subscriptions/soundcloud.py b/ytdl_subscribe/subscriptions/soundcloud.py index 27174e4f..8e9ab9c5 100644 --- a/ytdl_subscribe/subscriptions/soundcloud.py +++ b/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: diff --git a/ytdl_subscribe/subscriptions/subscription.py b/ytdl_subscribe/subscriptions/subscription.py index a62affc2..c7fdc5fc 100644 --- a/ytdl_subscribe/subscriptions/subscription.py +++ b/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)