From 3a22a9bf10b12b2ea9c5aa6f6af4c9f1585e1af5 Mon Sep 17 00:00:00 2001 From: jbannon Date: Mon, 4 Apr 2022 22:30:48 +0000 Subject: [PATCH] refactor subscriptions to use validators --- ytdl_subscribe/entries/entry.py | 9 + ytdl_subscribe/subscriptions/soundcloud.py | 73 ++++--- ytdl_subscribe/subscriptions/subscription.py | 193 ++++++++++-------- ytdl_subscribe/subscriptions/youtube.py | 25 ++- .../validators/config/preset_validator.py | 48 +++-- .../config/sources/soundcloud_validators.py | 6 +- .../config/subscription_validator.py | 40 +++- 7 files changed, 249 insertions(+), 145 deletions(-) diff --git a/ytdl_subscribe/entries/entry.py b/ytdl_subscribe/entries/entry.py index 4cb19f00..1f8a5651 100644 --- a/ytdl_subscribe/entries/entry.py +++ b/ytdl_subscribe/entries/entry.py @@ -78,10 +78,19 @@ class Entry: """Returns the entry's file name when downloaded locally""" return f"{self.uid}.{self.ext}" + @property + def download_thumbnail_name(self) -> str: + """Returns the thumbnail's file name when downloaded locally TODO: unit test this""" + return f"{self.uid}.{self.thumbnail_ext}" + def file_path(self, relative_directory: str): """Returns the entry's file path with respect to the relative directory""" return str(Path(relative_directory) / self.download_file_name) + def thumbnail_path(self, relative_directory: str): + """Returns the entry's thumbnail path with respect to the relative directory""" + return str(Path(relative_directory) / self.download_thumbnail_name) + def to_dict(self) -> Dict: """Returns the entry's values as a dictionary""" return { diff --git a/ytdl_subscribe/subscriptions/soundcloud.py b/ytdl_subscribe/subscriptions/soundcloud.py index 3e7d8268..9e9eb955 100644 --- a/ytdl_subscribe/subscriptions/soundcloud.py +++ b/ytdl_subscribe/subscriptions/soundcloud.py @@ -4,40 +4,57 @@ from ytdl_subscribe.downloaders.soundcloud_downloader import SoundcloudDownloade from ytdl_subscribe.entries.soundcloud import SoundcloudAlbum from ytdl_subscribe.entries.soundcloud import SoundcloudTrack from ytdl_subscribe.subscriptions.subscription import Subscription +from ytdl_subscribe.validators.config.sources.soundcloud_validators import ( + SoundcloudAlbumsAndSinglesDownloadValidator, +) +from ytdl_subscribe.validators.config.sources.soundcloud_validators import ( + SoundcloudSourceValidator, +) -class SoundcloudSubscription(Subscription): - def extract_info(self): +class SoundcloudAlbumsAndSinglesSubscription(Subscription): + source_validator_type = SoundcloudSourceValidator + download_strategy_type = SoundcloudAlbumsAndSinglesDownloadValidator + + @property + def source_options(self) -> SoundcloudSourceValidator: + return super().source_options + + @property + def download_strategy_options(self) -> SoundcloudAlbumsAndSinglesDownloadValidator: + return super().download_strategy_options + + def extract_info( + self, + ): + tracks: List[SoundcloudTrack] = [] soundcloud_downloader = SoundcloudDownloader( - output_directory=self.output_path, - working_directory=self.WORKING_DIRECTORY, - ytdl_options=self.ytdl_opts, + output_directory=self.output_options.output_directory.value, + working_directory=self.config_options.working_directory.value, + ytdl_options=self.ytdl_options.dict, ) - if self.options.get("download_strategy") == "albums_and_singles": - # 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] = soundcloud_downloader.download_albums( - artist_name=self.options["username"] + # 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] = soundcloud_downloader.download_albums( + artist_name=self.download_strategy_options.username.value + ) + + for album in albums: + tracks += album.album_tracks( + skip_premiere_tracks=self.source_options.skip_premiere_tracks.value ) - for album in albums: - tracks += album.album_tracks( - skip_premiere_tracks=self.options["skip_premiere_tracks"] - ) + # only add tracks that are not part of an album + single_tracks = soundcloud_downloader.download_tracks( + artist_name=self.download_strategy_options.username.value + ) + tracks += [ + track + for track in single_tracks + if not any(album.contains(track) for album in albums) + ] - # 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) - ] - - for e in tracks: - self.post_process_entry(e) - else: - raise ValueError("Invalid download_strategy field for Soundcloud") + for e in tracks: + self.post_process_entry(e) diff --git a/ytdl_subscribe/subscriptions/subscription.py b/ytdl_subscribe/subscriptions/subscription.py index 1f0afb6a..a1b5efe6 100644 --- a/ytdl_subscribe/subscriptions/subscription.py +++ b/ytdl_subscribe/subscriptions/subscription.py @@ -1,99 +1,115 @@ import os -from copy import deepcopy from shutil import copyfile +from typing import Type +from typing import TypeVar import dicttoxml import music_tag from PIL import Image -from sanitize_filename import sanitize from ytdl_subscribe.entries.entry import Entry +from ytdl_subscribe.validators.config.config_validator import ConfigValidator +from ytdl_subscribe.validators.config.metadata_options.metadata_options_validator import ( + MetadataOptionsValidator, +) +from ytdl_subscribe.validators.config.output_options.output_options_validator import ( + OutputOptionsValidator, +) +from ytdl_subscribe.validators.config.preset_validator import OverridesValidator +from ytdl_subscribe.validators.config.preset_validator import YTDLOptionsValidator +from ytdl_subscribe.validators.config.sources.source_validator import ( + DownloadStrategyValidator, +) +from ytdl_subscribe.validators.config.sources.source_validator import SourceValidator + +SOURCE_T = TypeVar("SOURCE_T", bound=SourceValidator) +DOWNLOAD_STRATEGY_T = TypeVar("DOWNLOAD_STRATEGY_T", bound=DownloadStrategyValidator) class Subscription(object): - WORKING_DIRECTORY = "" + source_validator_type: Type[SOURCE_T] + download_strategy_type: Type[DOWNLOAD_STRATEGY_T] - def __init__(self, name, options, ytdl_opts, post_process, overrides, output_path): + def __init__( + self, + name: str, + config_options: ConfigValidator, + source_options: SourceValidator, + output_options: OutputOptionsValidator, + metadata_options: MetadataOptionsValidator, + ytdl_options: YTDLOptionsValidator, + overrides: OverridesValidator, + ): """ Parameters ---------- name: str Name of the subscription - options: dict - Dictionary of ytdl options, specific to the source type - ytdl_opts: dict - Dictionary of options passed directly to ytdl. - See `youtube_dl.YoutubeDL.YoutubeDL` for options. - post_process: dict - Dictionary of ytdl-subscribe post processing options - overrides: dict - Dictionary that overrides every ytdl entry. Be careful what you override! - output_path: str - Base path to save files to + config_options: ConfigValidator + source_options: SourceValidator + output_options: OutputOptionsValidator + metadata_options: MetadataOptionsValidator + ytdl_options: YTDLOptionsValidator + overrides: OverridesValidator """ self.name = name - self.options = options - self.ytdl_opts = ytdl_opts or dict() - self.post_process = post_process + self.config_options = config_options + + self.__source_options = source_options + self.__download_strategy_options = source_options.download_strategy + + if not isinstance(self.__source_options, self.source_validator_type): + raise ValueError("Source options does not match the expected type") + + if not isinstance( + self.__download_strategy_options, self.download_strategy_type + ): + raise ValueError("Download strategy does not match the expected type") + + self.output_options = output_options + self.metadata_options = metadata_options + self.ytdl_options = ytdl_options 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 - self.WORKING_DIRECTORY += f"{'/' if self.WORKING_DIRECTORY else ''}{self.name}" + @property + def source_options(self) -> SOURCE_T: + return self.__source_options - # Always set outtmpl to the id and extension. Will be renamed using the subscription's - # output_path value - self.ytdl_opts["outtmpl"] = self.WORKING_DIRECTORY + "/%(id)s.%(ext)s" - self.ytdl_opts["writethumbnail"] = True - - def format_filepath(self, filepath_formatter: str, entry: Entry, makedirs=False): - """ - Convert a filepath value in the config to an actual filepath. - - Parameters - ---------- - filepath_formatter: str - File path relative to the specified output path - entry: Entry - Entry used to populate any format variables in the filepath - makedirs: bool - Whether to create all directories in the final filepath. - - Returns - ------- - str - Full filepath. - """ - 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 + @property + def download_strategy_options(self) -> DOWNLOAD_STRATEGY_T: + return self.__download_strategy_options def _post_process_tagging(self, entry: Entry): + id3_options = self.metadata_options.id3 t = music_tag.load_file( - entry.file_path(relative_directory=self.WORKING_DIRECTORY) + entry.file_path( + relative_directory=self.config_options.working_directory.value + ) ) - for tag, tag_formatter in self.post_process["tagging"].items(): - t[tag] = entry.apply_formatter(tag_formatter, overrides=self.overrides) + for tag, tag_formatter in id3_options.tags.dict.items(): + t[tag] = entry.apply_formatter( + format_string=tag_formatter, overrides=self.overrides.dict + ) t.save() def _post_process_nfo(self, entry): nfo = {} - for tag, tag_formatter in self.post_process["nfo"].items(): - nfo[tag] = entry.apply_formatter(tag_formatter, overrides=self.overrides) + nfo_options = self.metadata_options.nfo + + for tag, tag_formatter in nfo_options.tags.dict.items(): + nfo[tag] = entry.apply_formatter( + format_string=tag_formatter, overrides=self.overrides + ) xml = dicttoxml.dicttoxml( obj=nfo, - root="nfo_root" in self.post_process, - custom_root=self.post_process.get("nfo_root"), + root=True, # We assume all NFOs have a root. Maybe we should not? + custom_root=nfo_options.nfo_root.value, attr_type=False, ) - nfo_file_path = self.format_filepath( - self.post_process["nfo_name"], entry, makedirs=True + nfo_file_path = entry.apply_formatter( + format_string=nfo_options.nfo_name.format_string, + overrides=self.overrides.dict, ) with open(nfo_file_path, "wb") as f: f.write(xml) @@ -105,37 +121,46 @@ class Subscription(object): raise NotImplemented("Each source needs to implement how it extracts info") def post_process_entry(self, entry: Entry): - if "tagging" in self.post_process: + if self.metadata_options.id3: self._post_process_tagging(entry) # Move the file after all direct file modifications are complete - output_file_path = self.format_filepath( - self.post_process["file_name"], entry, makedirs=True + entry_source_file_path = entry.file_path( + relative_directory=self.config_options.working_directory.value ) - copyfile( - entry.file_path(relative_directory=self.WORKING_DIRECTORY), output_file_path + + entry_destination_file_path = entry.apply_formatter( + format_string=self.output_options.file_name.format_string, + overrides=self.overrides.dict, ) + os.makedirs(os.path.dirname(entry_destination_file_path), exist_ok=True) + copyfile(entry_source_file_path, entry_destination_file_path) + # Download the thumbnail if its present - if "thumbnail_name" in self.post_process: - thumbnail_dest_path = self.format_filepath( - self.post_process["thumbnail_name"], - entry, - makedirs=True, + if self.output_options.thumbnail_name: + thumbnail_source_path = entry.thumbnail_path( + relative_directory=self.config_options.working_directory.value ) - if not os.path.isfile(thumbnail_dest_path): - thumbnail_file_path = ( - f"{self.WORKING_DIRECTORY}/{entry.uid}.{entry.thumbnail_ext}" + + thumbnail_destination_path = entry.apply_formatter( + format_string=self.output_options.thumbnail_name.format_string, + overrides=self.overrides.dict, + ) + + os.makedirs(os.path.dirname(entry_destination_file_path), exist_ok=True) + + # If the thumbnail is to be converted, then save the converted thumbnail to the + # output filepath + if self.output_options.convert_thumbnail: + im = Image.open(thumbnail_source_path).convert("RGB") + im.save( + fp=thumbnail_destination_path, + format=self.output_options.convert_thumbnail.value, ) - if os.path.isfile(thumbnail_file_path): - copyfile(thumbnail_file_path, thumbnail_dest_path) + # Otherwise, just copy the downloaded thumbnail + else: + copyfile(thumbnail_source_path, thumbnail_destination_path) - if "convert_thumbnail" in self.post_process: - # TODO: Clean with yaml definitions - if self.post_process["convert_thumbnail"] == "jpg": - self.post_process["convert_thumbnail"] = "jpeg" - im = Image.open(thumbnail_dest_path).convert("RGB") - im.save(thumbnail_dest_path, self.post_process["convert_thumbnail"]) - - if "nfo" in self.post_process: + if self.metadata_options.nfo: self._post_process_nfo(entry) diff --git a/ytdl_subscribe/subscriptions/youtube.py b/ytdl_subscribe/subscriptions/youtube.py index b83e1b9b..2136c705 100644 --- a/ytdl_subscribe/subscriptions/youtube.py +++ b/ytdl_subscribe/subscriptions/youtube.py @@ -1,17 +1,34 @@ from ytdl_subscribe.downloaders.youtube_downloader import YoutubeDownloader from ytdl_subscribe.subscriptions.subscription import Subscription +from ytdl_subscribe.validators.config.sources.youtube_validators import ( + YoutubePlaylistDownloadValidator, +) +from ytdl_subscribe.validators.config.sources.youtube_validators import ( + YoutubeSourceValidator, +) class YoutubeSubscription(Subscription): + source_validator_type = YoutubeSourceValidator + download_strategy_type = YoutubePlaylistDownloadValidator + + @property + def source_options(self) -> YoutubeSourceValidator: + return super().source_options + + @property + def download_strategy_options(self) -> YoutubePlaylistDownloadValidator: + return super().download_strategy_options + def extract_info(self): youtube_downloader = YoutubeDownloader( - output_directory=self.output_path, - working_directory=self.WORKING_DIRECTORY, - ytdl_options=self.ytdl_opts, + output_directory=self.output_options.output_directory.value, + working_directory=self.config_options.working_directory.value, + ytdl_options=self.ytdl_options.dict, ) entries = youtube_downloader.download_playlist( - playlist_id=self.options["playlist_id"] + playlist_id=self.download_strategy_options.playlist_id.value ) for e in entries: diff --git a/ytdl_subscribe/validators/config/preset_validator.py b/ytdl_subscribe/validators/config/preset_validator.py index 46ffb596..2bd2c191 100644 --- a/ytdl_subscribe/validators/config/preset_validator.py +++ b/ytdl_subscribe/validators/config/preset_validator.py @@ -1,6 +1,11 @@ +import copy from typing import Any +from typing import Dict from typing import Optional from typing import Tuple +from typing import Type + +import sanitize_filename from ytdl_subscribe.utils.enums import SubscriptionSourceName from ytdl_subscribe.validators.base.strict_dict_validator import StrictDictValidator @@ -24,6 +29,23 @@ from ytdl_subscribe.validators.config.sources.youtube_validators import ( from ytdl_subscribe.validators.exceptions import ValidationException +class YTDLOptionsValidator(DictValidator): + pass + + +class OverridesValidator(DictFormatterValidator): + @property + def dict(self) -> dict: + """For overrides, create sanitized versions of each entry for convenience""" + output_dict = copy.deepcopy(super().dict) + for key in self.keys: + output_dict[f"sanitized_{key}"] = sanitize_filename.sanitize( + output_dict[key] + ) + + return output_dict + + class PresetValidator(StrictDictValidator): required_keys = {"output_options, metadata_options"} optional_keys = { @@ -32,6 +54,11 @@ class PresetValidator(StrictDictValidator): *SubscriptionSourceName.all(), } + subscription_source_validator_mapping: Dict[str, Type[SourceValidator]] = { + SubscriptionSourceName.SOUNDCLOUD: SoundcloudSourceValidator, + SubscriptionSourceName.YOUTUBE: YoutubeSourceValidator, + } + def __validate_and_get_subscription_source(self) -> Tuple[str, SourceValidator]: subscription_source: Optional[SourceValidator] = None subscription_source_name: Optional[str] = None @@ -42,17 +69,10 @@ class PresetValidator(StrictDictValidator): f"'{self.name}' can only have one of the following sources: {SubscriptionSourceName.pretty_all()}" ) - if key == SubscriptionSourceName.SOUNDCLOUD: - subscription_source_name = SubscriptionSourceName.SOUNDCLOUD + if key in self.subscription_source_validator_mapping: + subscription_source_name = key subscription_source = self.validate_key( - key=key, - validator=SoundcloudSourceValidator, - ) - elif key == SubscriptionSourceName.YOUTUBE: - subscription_source_name = SubscriptionSourceName.YOUTUBE - subscription_source = self.validate_key( - key=key, - validator=YoutubeSourceValidator, + key=key, validator=self.subscription_source_validator_mapping[key] ) # If subscription source was not set, error @@ -78,10 +98,10 @@ class PresetValidator(StrictDictValidator): key="metadata_options", validator=MetadataOptionsValidator ) - self.ytdl_options = self.validate_key_if_present( - key="ytdl_options", validator=DictValidator + self.ytdl_options = self.validate_key( + key="ytdl_options", validator=YTDLOptionsValidator, default={} ) - self.overrides = self.validate_key_if_present( - key="overrides", validator=DictFormatterValidator + self.overrides = self.validate_key( + key="overrides", validator=OverridesValidator, default={} ) diff --git a/ytdl_subscribe/validators/config/sources/soundcloud_validators.py b/ytdl_subscribe/validators/config/sources/soundcloud_validators.py index 806f95cd..97d71cf9 100644 --- a/ytdl_subscribe/validators/config/sources/soundcloud_validators.py +++ b/ytdl_subscribe/validators/config/sources/soundcloud_validators.py @@ -11,9 +11,7 @@ class SoundcloudAlbumsAndSinglesDownloadValidator(DownloadStrategyValidator): def __init__(self, name, value): super().__init__(name, value) - self.username = self.validate_key( - key="username", validator=StringValidator - ).value + self.username = self.validate_key(key="username", validator=StringValidator) class SoundcloudSourceValidator(SourceValidator): @@ -27,4 +25,4 @@ class SoundcloudSourceValidator(SourceValidator): super().__init__(name=name, value=value) self.skip_premiere_tracks = self.validate_key( "skip_premiere_tracks", BoolValidator, default=True - ).value + ) diff --git a/ytdl_subscribe/validators/config/subscription_validator.py b/ytdl_subscribe/validators/config/subscription_validator.py index fbe35d9e..06ce4e1b 100644 --- a/ytdl_subscribe/validators/config/subscription_validator.py +++ b/ytdl_subscribe/validators/config/subscription_validator.py @@ -4,7 +4,9 @@ from typing import List import yaml from mergedeep import mergedeep -from ytdl_subscribe.subscriptions.soundcloud import SoundcloudSubscription +from ytdl_subscribe.subscriptions.soundcloud import ( + SoundcloudAlbumsAndSinglesSubscription, +) from ytdl_subscribe.subscriptions.subscription import Subscription from ytdl_subscribe.subscriptions.youtube import YoutubeSubscription from ytdl_subscribe.utils.enums import SubscriptionSourceName @@ -12,7 +14,14 @@ from ytdl_subscribe.validators.base.strict_dict_validator import StrictDictValid from ytdl_subscribe.validators.base.validators import DictValidator from ytdl_subscribe.validators.base.validators import StringValidator from ytdl_subscribe.validators.config.config_validator import ConfigValidator +from ytdl_subscribe.validators.config.preset_validator import OverridesValidator from ytdl_subscribe.validators.config.preset_validator import PresetValidator +from ytdl_subscribe.validators.config.sources.soundcloud_validators import ( + SoundcloudSourceValidator, +) +from ytdl_subscribe.validators.config.sources.youtube_validators import ( + YoutubePlaylistDownloadValidator, +) class SubscriptionValidator(StrictDictValidator): @@ -26,9 +35,11 @@ class SubscriptionValidator(StrictDictValidator): def __init__(self, config: ConfigValidator, name: str, value: Any): super().__init__(name, value) self.config = config - self.overrides = self.validate_key( + + # Ensure the overrides defined here are valid + _ = self.validate_key( key="overrides", - validator=DictValidator, + validator=OverridesValidator, default={}, ) @@ -58,20 +69,27 @@ class SubscriptionValidator(StrictDictValidator): ) def to_subscription(self) -> Subscription: - if self.preset.subscription_source_name == SubscriptionSourceName.SOUNDCLOUD: - subscription_class = SoundcloudSubscription - elif self.preset.subscription_source_name == SubscriptionSourceName.YOUTUBE: + if isinstance( + self.preset.subscription_source.download_strategy, + SoundcloudAlbumsAndSinglesSubscription, + ): + subscription_class = SoundcloudAlbumsAndSinglesSubscription + elif isinstance( + self.preset.subscription_source.download_strategy, + YoutubePlaylistDownloadValidator, + ): subscription_class = YoutubeSubscription else: raise ValueError("subscription source class not found") return subscription_class( name=self.name, - options=self.preset.dict.get(self.preset.subscription_source_name), - ytdl_opts=self.preset.dict.get("ytdl_options"), - post_process=self.preset.dict.get("post_process"), - overrides=self.preset.dict.get("overrides"), - output_path=self.preset.dict.get("output_path"), + config_options=self.config, + source_options=self.preset.subscription_source, + output_options=self.preset.output_options, + metadata_options=self.preset.metadata_options, + ytdl_options=self.preset.ytdl_options, + overrides=self.preset.overrides, ) @classmethod