This commit is contained in:
jbannon 2022-04-29 05:09:19 +00:00
parent 1c172f1c82
commit 1326fbf07f
13 changed files with 218 additions and 17 deletions

View file

@ -1,5 +1,9 @@
[MASTER] [MASTER]
disable= disable=
C0115, # Missing class docstring
C0114, # missing-module-docstring C0114, # missing-module-docstring
R0903, # too-few-public-methods R0903, # too-few-public-methods
R0801, # similar lines R0801, # similar lines
W0511, # TODO
load-plugins=pylint.extensions.docparams

View file

@ -31,11 +31,31 @@ class ConfigFile(StrictDictValidator):
self.presets = self._validate_key("presets", LiteralDictValidator) self.presets = self._validate_key("presets", LiteralDictValidator)
@classmethod @classmethod
def from_dict(cls, config_dict) -> "ConfigFile": def from_dict(cls, config_dict: dict) -> "ConfigFile":
"""
Parameters
----------
config_dict:
The config in dictionary format
Returns
-------
Config file validator
"""
return ConfigFile(name="", value=config_dict) return ConfigFile(name="", value=config_dict)
@classmethod @classmethod
def from_file_path(cls, config_path) -> "ConfigFile": def from_file_path(cls, config_path: str) -> "ConfigFile":
"""
Parameters
----------
config_path:
Path to the config yaml
Returns
-------
Config file validator
"""
# TODO: Create separate yaml file loader class # TODO: Create separate yaml file loader class
with open(config_path, "r", encoding="utf-8") as file: with open(config_path, "r", encoding="utf-8") as file:
config_dict = yaml.safe_load(file) config_dict = yaml.safe_load(file)

View file

@ -50,6 +50,16 @@ class DownloadStrategyValidator(StrictDictValidator):
).value ).value
def get(self, downloader_source: str) -> Type[Downloader]: def get(self, downloader_source: str) -> Type[Downloader]:
"""
Parameters
----------
downloader_source:
Name of the download source
Returns
-------
The downloader class
"""
try: try:
return DownloadStrategyMapping.get( return DownloadStrategyMapping.get(
source=downloader_source, download_strategy=self.download_strategy_name source=downloader_source, download_strategy=self.download_strategy_name

View file

@ -79,6 +79,18 @@ class DownloadStrategyMapping:
@classmethod @classmethod
def get(cls, source: str, download_strategy: str) -> Type[Downloader]: def get(cls, source: str, download_strategy: str) -> Type[Downloader]:
"""
Parameters
----------
source:
The source (i.e. 'youtube', 'soundcloud') of the downloader
download_strategy:
The download strategy name
Returns
-------
The downloader class
"""
cls._validate_is_download_strategy(source, download_strategy) cls._validate_is_download_strategy(source, download_strategy)
return cls._MAPPING[source][download_strategy] return cls._MAPPING[source][download_strategy]
@ -106,6 +118,21 @@ class PluginMapping:
@classmethod @classmethod
def get(cls, plugin: str) -> Type[Plugin]: def get(cls, plugin: str) -> Type[Plugin]:
"""
Parameters
----------
plugin
Name of the plugin
Returns
-------
The plugin class
Raises
-------
ValueError
Raised if the plugin does not exist
"""
if plugin not in cls.plugins(): if plugin not in cls.plugins():
raise ValueError( raise ValueError(
f"Tried to use plugin '{plugin}' that does not exist. Available plugins: " f"Tried to use plugin '{plugin}' that does not exist. Available plugins: "

View file

@ -58,6 +58,11 @@ class SubscriptionValidator(StrictDictValidator):
) )
def to_subscription(self) -> Subscription: def to_subscription(self) -> Subscription:
"""
Returns
-------
The subscription after the config and preset have been validated
"""
return Subscription( return Subscription(
name=self._name, name=self._name,
config_options=self.config.config_options, config_options=self.config.config_options,
@ -66,14 +71,40 @@ class SubscriptionValidator(StrictDictValidator):
@classmethod @classmethod
def from_dict( def from_dict(
cls, config: ConfigFile, subscription_name, subscription_dict: Dict cls, config: ConfigFile, subscription_name: str, subscription_dict: Dict
) -> "SubscriptionValidator": ) -> "SubscriptionValidator":
"""
Parameters
----------
config:
Validated instance of the config
subscription_name:
Name of the subscription
subscription_dict:
The subscription config in dict format
Returns
-------
The Subscription validator
"""
return SubscriptionValidator(config=config, name=subscription_name, value=subscription_dict) return SubscriptionValidator(config=config, name=subscription_name, value=subscription_dict)
@classmethod @classmethod
def from_file_path( def from_file_path(
cls, config: ConfigFile, subscription_path: str cls, config: ConfigFile, subscription_path: str
) -> List["SubscriptionValidator"]: ) -> List["SubscriptionValidator"]:
"""
Parameters
----------
config:
Validated instance of the config
subscription_path:
File path to the subscription yaml file
Returns
-------
List of subscription validators, for each one in the subscription yaml
"""
# TODO: Create separate yaml file loader class # TODO: Create separate yaml file loader class
with open(subscription_path, "r", encoding="utf-8") as file: with open(subscription_path, "r", encoding="utf-8") as file:
subscription_dict = yaml.safe_load(file) subscription_dict = yaml.safe_load(file)

View file

@ -120,6 +120,8 @@ class Downloader(Generic[DownloaderOptionsT, DownloaderEntryT], ABC):
Parameters Parameters
---------- ----------
output_directory overrides:
Subscription overrides
output_directory:
Output directory to potentially store extra files downloaded Output directory to potentially store extra files downloaded
""" """

View file

@ -197,14 +197,27 @@ class YoutubeChannelDownloader(YoutubeDownloader[YoutubeChannelDownloaderOptions
ytdl_options_overrides=ytdl_options_overrides, ytdl_options_overrides=ytdl_options_overrides,
) )
@classmethod
def __download_thumbnail( def __download_thumbnail(
self, cls,
channel_dict: dict, entry_dict: dict,
thumbnail_id: str, thumbnail_id: str,
output_thumbnail_path: str, output_thumbnail_path: str,
): ):
"""
Downloads a specific thumbnail from a YTDL entry's thumbnail list
Parameters
----------
entry_dict:
YTDL entry dict
thumbnail_id:
Id of the thumbnail defined in the YTDL thumnail
output_thumbnail_path:
Path to store the thumbnail after downloading
"""
thumbnail_url = None thumbnail_url = None
for thumbnail in channel_dict.get("thumbnails", []): for thumbnail in entry_dict.get("thumbnails", []):
if thumbnail["id"] == thumbnail_id: if thumbnail["id"] == thumbnail_id:
thumbnail_url = thumbnail["url"] thumbnail_url = thumbnail["url"]
break break
@ -230,13 +243,13 @@ class YoutubeChannelDownloader(YoutubeDownloader[YoutubeChannelDownloaderOptions
Output directory path Output directory path
""" """
channel_json_file_path = Path(self.working_directory) / f"{self.channel_id}.info.json" channel_json_file_path = Path(self.working_directory) / f"{self.channel_id}.info.json"
with open(channel_json_file_path) as channel_json: with open(channel_json_file_path, "r", encoding="utf-8") as channel_json:
channel_entry = json.load(channel_json) channel_entry = json.load(channel_json)
if self.download_options.channel_avatar_path: if self.download_options.channel_avatar_path:
thumbnail_name = overrides.apply_formatter(self.download_options.channel_avatar_path) thumbnail_name = overrides.apply_formatter(self.download_options.channel_avatar_path)
self.__download_thumbnail( self.__download_thumbnail(
channel_dict=channel_entry, entry_dict=channel_entry,
thumbnail_id="avatar_uncropped", thumbnail_id="avatar_uncropped",
output_thumbnail_path=str(Path(output_directory) / thumbnail_name), output_thumbnail_path=str(Path(output_directory) / thumbnail_name),
) )
@ -244,7 +257,7 @@ class YoutubeChannelDownloader(YoutubeDownloader[YoutubeChannelDownloaderOptions
if self.download_options.channel_banner_path: if self.download_options.channel_banner_path:
thumbnail_name = overrides.apply_formatter(self.download_options.channel_banner_path) thumbnail_name = overrides.apply_formatter(self.download_options.channel_banner_path)
self.__download_thumbnail( self.__download_thumbnail(
channel_dict=channel_entry, entry_dict=channel_entry,
thumbnail_id="banner_uncropped", thumbnail_id="banner_uncropped",
output_thumbnail_path=str(Path(output_directory) / thumbnail_name), output_thumbnail_path=str(Path(output_directory) / thumbnail_name),
) )

View file

@ -46,6 +46,16 @@ class BaseEntry(ABC):
return self._kwargs[key] return self._kwargs[key]
def working_directory(self) -> str: def working_directory(self) -> str:
"""
Returns
-------
The working directory
Raises
-------
ValueError
The working directory was never defined in the init
"""
if self._working_directory is None: if self._working_directory is None:
raise ValueError( raise ValueError(
"Entry working directory is not set when trying to access its download file path" "Entry working directory is not set when trying to access its download file path"

View file

@ -30,7 +30,8 @@ class NfoTagsPlugin(Plugin[NfoTagsOptions]):
Parameters Parameters
---------- ----------
entry: Entry to create an NFO file for entry:
Entry to create an NFO file for
""" """
nfo = {} nfo = {}

View file

@ -61,7 +61,8 @@ class Plugin(Generic[PluginOptionsT], ABC):
Parameters Parameters
---------- ----------
entry: Entry to post process entry:
Entry to post process
""" """
def post_process_subscription(self): def post_process_subscription(self):

View file

@ -60,48 +60,101 @@ class Subscription:
@property @property
def downloader_class(self) -> Type[Downloader]: def downloader_class(self) -> Type[Downloader]:
"""
Returns
-------
This subscription's downloader class
"""
return self.__preset_options.downloader return self.__preset_options.downloader
@property @property
def downloader_options(self) -> DownloaderValidator: def downloader_options(self) -> DownloaderValidator:
"""
Returns
-------
The download options for this subscription's downloader
"""
return self.__preset_options.downloader_options return self.__preset_options.downloader_options
@property @property
def plugins(self) -> List[Tuple[Type[Plugin], PluginOptions]]: def plugins(self) -> List[Tuple[Type[Plugin], PluginOptions]]:
"""
Returns
-------
List of tuples containing (plugin class, plugin options)
"""
return self.__preset_options.plugins return self.__preset_options.plugins
@property @property
def ytdl_options(self) -> YTDLOptions: def ytdl_options(self) -> YTDLOptions:
"""
Returns
-------
YTDL options for this subscription
"""
return self.__preset_options.ytdl_options return self.__preset_options.ytdl_options
@property @property
def output_options(self) -> OutputOptions: def output_options(self) -> OutputOptions:
"""Returns the output options defined for this subscription""" """
Returns
-------
The output options defined for this subscription
"""
return self.__preset_options.output_options return self.__preset_options.output_options
@property @property
def overrides(self) -> Overrides: def overrides(self) -> Overrides:
"""Returns the overrides defined for this subscription""" """
Returns
-------
The overrides defined for this subscription
"""
return self.__preset_options.overrides return self.__preset_options.overrides
@property @property
def working_directory(self) -> str: def working_directory(self) -> str:
"""Returns the directory that the downloader saves files to""" """
Returns
-------
The directory that the downloader saves files to
"""
return str(Path(self.__config_options.working_directory.value) / Path(self.name)) return str(Path(self.__config_options.working_directory.value) / Path(self.name))
@property @property
def output_directory(self) -> str: def output_directory(self) -> str:
""" """
:return: The formatted output directory Returns
-------
The formatted output directory
""" """
return self.overrides.apply_formatter(formatter=self.output_options.output_directory) return self.overrides.apply_formatter(formatter=self.output_options.output_directory)
def _copy_file_to_output_directory(self, source_file_path: str, output_file_name: str): def _copy_file_to_output_directory(self, source_file_path: str, output_file_name: str):
"""
Helper function to move a file from the working directory to the output directory.
Parameters
----------
source_file_path:
Path to the source file
output_file_name:
Desired output file name with the output directory as its relative directory
"""
destination_file_path = Path(self.output_directory) / Path(output_file_name) destination_file_path = Path(self.output_directory) / Path(output_file_name)
os.makedirs(os.path.dirname(destination_file_path), exist_ok=True) os.makedirs(os.path.dirname(destination_file_path), exist_ok=True)
copyfile(source_file_path, destination_file_path) copyfile(source_file_path, destination_file_path)
def _copy_entry_files_to_output_directory(self, entry: Entry): def _copy_entry_files_to_output_directory(self, entry: Entry):
"""
Helper function to move the media file and optionally thumbnail file to the output directory
for a single entry.
Parameters
----------
entry:
The entry with files to move
"""
# Move the file after all direct file modifications are complete # Move the file after all direct file modifications are complete
output_file_name = self.overrides.apply_formatter( output_file_name = self.overrides.apply_formatter(
formatter=self.output_options.file_name, entry=entry formatter=self.output_options.file_name, entry=entry
@ -125,6 +178,10 @@ class Subscription:
@contextlib.contextmanager @contextlib.contextmanager
def _prepare_working_directory(self): def _prepare_working_directory(self):
"""
Context manager to create all directories to the working directory. Deletes the entire
working directory when cleaning up.
"""
os.makedirs(self.working_directory, exist_ok=True) os.makedirs(self.working_directory, exist_ok=True)
try: try:
@ -134,6 +191,9 @@ class Subscription:
@contextlib.contextmanager @contextlib.contextmanager
def _maintain_archive_file(self): def _maintain_archive_file(self):
"""
Context manager to initialize the enhanced download archive
"""
if self.output_options.maintain_download_archive: if self.output_options.maintain_download_archive:
self._enhanced_download_archive.prepare_download_archive() self._enhanced_download_archive.prepare_download_archive()
@ -150,6 +210,11 @@ class Subscription:
self._enhanced_download_archive.save_download_archive() self._enhanced_download_archive.save_download_archive()
def _initialize_plugins(self) -> List[Plugin]: def _initialize_plugins(self) -> List[Plugin]:
"""
Returns
-------
List of plugins defined in the subscription, initialized and ready to use.
"""
plugins: List[Plugin] = [] plugins: List[Plugin] = []
for plugin_type, plugin_options in self.plugins: for plugin_type, plugin_options in self.plugins:
plugin = plugin_type( plugin = plugin_type(

View file

@ -15,6 +15,11 @@ class DateRangeValidator(StrictDictValidator):
self.after = self._validate_key_if_present("after", StringDatetimeValidator) self.after = self._validate_key_if_present("after", StringDatetimeValidator)
def get_date_range(self) -> Optional[DateRange]: def get_date_range(self) -> Optional[DateRange]:
"""
Returns
-------
Date range if the 'before' or 'after' is defined. None otherwise.
"""
if self.before or self.after: if self.before or self.after:
return DateRange( return DateRange(
start=self.after.datetime_str if self.after else None, start=self.after.datetime_str if self.after else None,

View file

@ -100,6 +100,18 @@ class StringFormatterValidator(Validator):
@final @final
def apply_formatter(self, variable_dict: Dict[str, str]) -> str: def apply_formatter(self, variable_dict: Dict[str, str]) -> str:
"""
Calls `format` on the format string using the variable_dict as input kwargs
Parameters
----------
variable_dict
kwargs to pass to the format string
Returns
-------
Format string formatted
"""
# Keep formatting the format string until no format_variables are present # Keep formatting the format string until no format_variables are present
formatter = self formatter = self
recursion_depth = 0 recursion_depth = 0