danvics/ytdl-sub
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

using validators

Browse source
This commit is contained in:
jbannon 2022-03-31 05:20:41 +00:00
parent 74169b9939
commit 329562bb0b
15 changed files with 220 additions and 227 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

22
ytdl_subscribe/main.py
View file

@ -1,14 +1,12 @@
import argparse
from typing import List
from ytdl_subscribe.parse import parse_presets
from ytdl_subscribe.parse import parse_subscriptions
from ytdl_subscribe.parse import parse_subscriptions_file
from ytdl_subscribe.parse import read_config_file
###################################################################################################
# GLOBAL PARSER
from ytdl_subscribe.validators.config.config import Config
from ytdl_subscribe.validators.config.config_validator import ConfigValidator
from ytdl_subscribe.validators.config.subscription_validator import (
SubscriptionValidator,
)
parser = argparse.ArgumentParser(
description="YoutubeDL-Subscribe: Download and organize your favorite media hassle-free."
@ -43,10 +41,20 @@ download_parser.add_argument(
if __name__ == "__main__":
args = parser.parse_args()
config: Config = Config.from_file_path(args.config)
config: ConfigValidator = ConfigValidator.from_file_path(args.config)
if args.subparser == "sub":
subscription_paths: List[str] = args.subscription_paths
subscriptions: List[SubscriptionValidator] = []
for subscription_path in subscription_paths:
subscriptions += SubscriptionValidator.from_file_path(
config=config, subscription_path=subscription_path
)
for subscription in subscriptions:
subscription.to_subscription().extract_info()
print("hi")
# for subscription_path in subscription_paths:
#
#

121
ytdl_subscribe/parse.py
View file

@ -1,121 +0,0 @@
from typing import Optional
import yaml
from mergedeep import mergedeep
from ytdl_subscribe.enums import SubscriptionSourceName
from ytdl_subscribe.enums import YAMLSection
from ytdl_subscribe.subscriptions.soundcloud import SoundcloudSubscription
from ytdl_subscribe.subscriptions.subscription import Subscription
from ytdl_subscribe.subscriptions.youtube import YoutubeSubscription
from ytdl_subscribe.validators.config.config import Config
def _set_config_variables(config):
Subscription.WORKING_DIRECTORY = config.get("working_directory", "")
def read_config_file(config_path: str) -> Config:
with open(config_path, "r", encoding="utf-8") as file:
config_dict = yaml.safe_load(file)
return Config(name="config", value=config_dict).validate()
def parse_subscriptions_file(subscription_yaml_path, set_config_variables=True):
"""
Reads and parses a subscription yaml from its path.
TODO: trafaret the dict for validation
Parameters
----------
subscription_yaml_path: str
File path
set_config_variables: bool
Whether to set global config variables
Returns
-------
dict
The subscription yaml after safe_load
"""
with open(subscription_yaml_path, "r") as f:
yaml_dict = yaml.safe_load(f)
config = yaml_dict[YAMLSection.CONFIG_KEY]
if set_config_variables:
_set_config_variables(config)
return yaml_dict
def parse_presets(yaml_dict):
"""
Parses presets from a subscription yaml dict
Parameters
----------
yaml_dict: dict
Subscription YAML dict
Returns
-------
dict
Presets
"""
presets = yaml_dict[YAMLSection.PRESET_KEY]
return presets
def parse_subscriptions(yaml_dict: dict, presets: dict, subscriptions: Optional[list]):
"""
Parses config from a subscription yaml dict
Parameters
----------
yaml_dict: dict
presets: dict
subscriptions: list[str] or None
If present, only parse these config
Returns
-------
list[Subscription]
"""
parsed_subscriptions = []
# Parse all config even if all are not used for validation's sake
for name, subscription in yaml_dict[YAMLSection.SUBSCRIPTIONS_KEY].items():
preset = {}
if subscription.get("preset") in presets:
preset = presets[subscription["preset"]]
subscription = mergedeep.merge({}, preset, subscription)
if SubscriptionSourceName.SOUNDCLOUD in subscription:
subscription_source = SubscriptionSourceName.SOUNDCLOUD
subscription_class = SoundcloudSubscription
elif SubscriptionSourceName.YOUTUBE in subscription:
subscription_source = SubscriptionSourceName.YOUTUBE
subscription_class = YoutubeSubscription
else:
raise ValueError("dne")
parsed_subscriptions.append(
subscription_class(
name=name,
options=subscription[subscription_source],
ytdl_opts=subscription["ytdl_opts"],
post_process=subscription["post_process"],
overrides=subscription["overrides"],
output_path=subscription["output_path"],
)
)
# Filter config if present
if subscriptions:
parsed_subscriptions = [
sub for sub in parsed_subscriptions if sub.name in subscriptions
]
return parsed_subscriptions

2
ytdl_subscribe/subscriptions/soundcloud.py
View file

@ -15,7 +15,7 @@ class SoundcloudSubscription(Subscription):
ytdl_options=self.ytdl_opts,
)
if self.options.get("download_strategy") == "albums_then_tracks":
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(

6
ytdl_subscribe/enums.py → ytdl_subscribe/utils/enums.py
View file

@ -14,9 +14,3 @@ class SubscriptionSourceName(object):
def pretty_all(cls) -> str:
"""Returns all subscription sources pretty printed"""
return ", ".join(cls.all())
class YAMLSection(object):
CONFIG_KEY = "config"
PRESET_KEY = "presets"
SUBSCRIPTIONS_KEY = "config"

0
ytdl_subscribe/validators/native/__init__.py → ytdl_subscribe/validators/base/__init__.py
View file

4
ytdl_subscribe/validators/native/validator.py → ytdl_subscribe/validators/base/base_validator.py
View file

@ -1,10 +1,10 @@
from typing import Any
class Validator:
class BaseValidator:
def __init__(self, name: str, value: Any):
self.name = name
self.value = value
def validate(self) -> "Validator":
def validate(self) -> "BaseValidator":
return self

4
ytdl_subscribe/validators/native/object_validator.py → ytdl_subscribe/validators/base/object_validator.py
View file

@ -2,11 +2,11 @@ from typing import Any
from typing import Optional
from typing import Set
from ytdl_subscribe.validators.base.base_validator import BaseValidator
from ytdl_subscribe.validators.exceptions import ValidationException
from ytdl_subscribe.validators.native.validator import Validator
class ObjectValidator(Validator):
class ObjectValidator(BaseValidator):
required_fields: Set[str] = []
optional_fields: Set[str] = []

16
ytdl_subscribe/validators/config/config.py → ytdl_subscribe/validators/config/config_validator.py
View file

@ -4,34 +4,34 @@ from typing import Optional
import yaml
from ytdl_subscribe.validators.config.preset import Preset
from ytdl_subscribe.validators.native.object_validator import ObjectValidator
from ytdl_subscribe.validators.base.object_validator import ObjectValidator
from ytdl_subscribe.validators.config.preset_validator import PresetValidator
class Config(ObjectValidator):
class ConfigValidator(ObjectValidator):
required_fields = {"working_directory", "presets"}
def __init__(self, name: str, value: Any):
super().__init__(name, value)
self.working_directory: Optional[str] = None
self.presets: Optional[Dict[str, Preset]] = None
self.presets: Optional[Dict[str, PresetValidator]] = None
def validate(self) -> "Config":
def validate(self) -> "ConfigValidator":
super().validate()
self.working_directory = self.get("working_directory")
self.presets = {}
for preset_name, preset_obj in self.get("presets").items():
self.presets[preset_name] = Preset(
self.presets[preset_name] = PresetValidator(
name=f"{self.name}.presets.{preset_name}", value=preset_obj
).validate()
return self
@classmethod
def from_file_path(cls, config_path) -> "Config":
def from_file_path(cls, config_path) -> "ConfigValidator":
# TODO: Create separate yaml file loader class
with open(config_path, "r", encoding="utf-8") as file:
config_dict = yaml.safe_load(file)
return Config(name="config", value=config_dict).validate()
return ConfigValidator(name="config", value=config_dict).validate()

31
ytdl_subscribe/validators/config/preset.py → ytdl_subscribe/validators/config/preset_validator.py
View file

@ -1,22 +1,29 @@
from typing import Any
from typing import Optional
from ytdl_subscribe.enums import SubscriptionSourceName
from ytdl_subscribe.validators.config.sources import PresetSource
from ytdl_subscribe.validators.config.sources import SoundcloudSource
from ytdl_subscribe.validators.config.sources import YoutubeSource
from ytdl_subscribe.utils.enums import SubscriptionSourceName
from ytdl_subscribe.validators.base.object_validator import ObjectValidator
from ytdl_subscribe.validators.config.sources.base_source_validator import (
BaseSourceValidator,
)
from ytdl_subscribe.validators.config.sources.soundcloud_source_validator import (
SoundcloudSourceValidator,
)
from ytdl_subscribe.validators.config.sources.youtube_source_validator import (
YoutubeSourceValidator,
)
from ytdl_subscribe.validators.exceptions import ValidationException
from ytdl_subscribe.validators.native.object_validator import ObjectValidator
class Preset(ObjectValidator):
class PresetValidator(ObjectValidator):
required_fields = {"post_process"}
optional_fields = {"ytdl_options", *SubscriptionSourceName.all()}
optional_fields = {"ytdl_options", "output_path", *SubscriptionSourceName.all()}
allow_extra_fields = False
def __init__(self, name: str, value: Any):
super().__init__(name=name, value=value)
self.subscription_source: Optional[PresetSource] = None
self.subscription_source: Optional[BaseSourceValidator] = None
self.subscription_source_name: Optional[str] = None
def _validate_subscription_source(self):
# Make sure we only have a single subscription source
@ -27,11 +34,13 @@ class Preset(ObjectValidator):
)
if object_key == SubscriptionSourceName.SOUNDCLOUD:
self.subscription_source = SoundcloudSource(
self.subscription_source_name = SubscriptionSourceName.SOUNDCLOUD
self.subscription_source = SoundcloudSourceValidator(
name=f"{self.name}{object_key}", value=object_value
).validate()
elif object_key == SubscriptionSourceName.YOUTUBE:
self.subscription_source = YoutubeSource(
self.subscription_source_name = SubscriptionSourceName.YOUTUBE
self.subscription_source = YoutubeSourceValidator(
name=f"{self.name}{object_key}", value=object_value
).validate()
@ -41,7 +50,7 @@ class Preset(ObjectValidator):
f"'{self.name} must have one of the following sources: {SubscriptionSourceName.pretty_all()}"
)
def validate(self) -> "Preset":
def validate(self) -> "PresetValidator":
super().validate()
self._validate_subscription_source()
return self

69
ytdl_subscribe/validators/config/sources.py
View file

@ -1,69 +0,0 @@
from typing import Any
from typing import Optional
from typing import Set
from ytdl_subscribe.validators.exceptions import ValidationException
from ytdl_subscribe.validators.native.object_validator import ObjectValidator
class PresetSource(ObjectValidator):
required_fields = {"download_strategy"}
download_strategies: Set[str] = {}
def __init__(self, name: str, value: Any):
super().__init__(name=name, value=value)
self.download_strategy: Optional[str] = None
def validate(self) -> "PresetSource":
super().validate()
if self.value["download_strategy"] not in self.download_strategies:
raise ValidationException(
f"'download_strategy' must be one of the following: {', '.join(self.download_strategies)}"
)
return self
class YoutubeSource(PresetSource):
optional_fields = {"playlist_id"}
download_strategies = {"playlist"}
def __init__(self, name: str, value: Any):
super().__init__(name=name, value=value)
self.playlist_id: Optional[str] = None
def validate(self) -> "YoutubeSource":
super().validate()
if self.download_strategy == "playlist":
self.playlist_id = self.value.get("playlist_id")
if not self.playlist_id or not isinstance(self.playlist_id, str):
raise ValidationException(
"Youtube download_strategy 'playlist' requires the field 'playlist_id' to be a string"
)
return self
class SoundcloudSource(PresetSource):
optional_fields = {"username", "skip_premiere_tracks"}
download_strategies = {"albums_and_singles"}
def __init__(self, name: str, value: Any):
super().__init__(name=name, value=value)
self.username: Optional[str] = None
self.skip_premiere_tracks: Optional[bool] = None
def validate(self) -> "SoundcloudSource":
super().validate()
self.skip_premiere_tracks = self.get("skip_premiere_tracks", True)
if self.download_strategy == "albums_and_singles":
self.username = self.value.get("username")
if not self.username or not isinstance(self.username, str):
raise ValidationException(
"Soundcloud download_strategy 'albums_and_singles' requires the field 'username'"
)
return self

0
ytdl_subscribe/validators/config/sources/__init__.py Normal file
View file

25
ytdl_subscribe/validators/config/sources/base_source_validator.py Normal file
View file

@ -0,0 +1,25 @@
from typing import Any
from typing import Optional
from typing import Set
from ytdl_subscribe.validators.base.object_validator import ObjectValidator
from ytdl_subscribe.validators.exceptions import ValidationException
class BaseSourceValidator(ObjectValidator):
required_fields = {"download_strategy"}
download_strategies: Set[str] = {}
def __init__(self, name: str, value: Any):
super().__init__(name=name, value=value)
self.download_strategy: Optional[str] = None
def validate(self) -> "BaseSourceValidator":
super().validate()
if self.value["download_strategy"] not in self.download_strategies:
raise ValidationException(
f"'download_strategy' must be one of the following: {', '.join(self.download_strategies)}"
)
return self

30
ytdl_subscribe/validators/config/sources/soundcloud_source_validator.py Normal file
View file

@ -0,0 +1,30 @@
from typing import Any
from typing import Optional
from ytdl_subscribe.validators.config.sources.base_source_validator import (
BaseSourceValidator,
)
from ytdl_subscribe.validators.exceptions import ValidationException
class SoundcloudSourceValidator(BaseSourceValidator):
optional_fields = {"username", "skip_premiere_tracks"}
download_strategies = {"albums_and_singles"}
def __init__(self, name: str, value: Any):
super().__init__(name=name, value=value)
self.username: Optional[str] = None
self.skip_premiere_tracks: Optional[bool] = None
def validate(self) -> "SoundcloudSourceValidator":
super().validate()
self.skip_premiere_tracks = self.get("skip_premiere_tracks", True)
if self.download_strategy == "albums_and_singles":
self.username = self.value.get("username")
if not self.username or not isinstance(self.username, str):
raise ValidationException(
"Soundcloud download_strategy 'albums_and_singles' requires the field 'username'"
)
return self

28
ytdl_subscribe/validators/config/sources/youtube_source_validator.py Normal file
View file

@ -0,0 +1,28 @@
from typing import Any
from typing import Optional
from ytdl_subscribe.validators.config.sources.base_source_validator import (
BaseSourceValidator,
)
from ytdl_subscribe.validators.exceptions import ValidationException
class YoutubeSourceValidator(BaseSourceValidator):
optional_fields = {"playlist_id"}
download_strategies = {"playlist"}
def __init__(self, name: str, value: Any):
super().__init__(name=name, value=value)
self.playlist_id: Optional[str] = None
def validate(self) -> "YoutubeSourceValidator":
super().validate()
if self.download_strategy == "playlist":
self.playlist_id = self.value.get("playlist_id")
if not self.playlist_id or not isinstance(self.playlist_id, str):
raise ValidationException(
"Youtube download_strategy 'playlist' requires the field 'playlist_id' to be a string"
)
return self

89
ytdl_subscribe/validators/config/subscription_validator.py Normal file
View file

@ -0,0 +1,89 @@
from typing import Any
from typing import Dict
from typing import List
from typing import Optional
import yaml
from mergedeep import mergedeep
from ytdl_subscribe.subscriptions.soundcloud import SoundcloudSubscription
from ytdl_subscribe.subscriptions.subscription import Subscription
from ytdl_subscribe.subscriptions.youtube import YoutubeSubscription
from ytdl_subscribe.utils.enums import SubscriptionSourceName
from ytdl_subscribe.validators.config.config_validator import ConfigValidator
from ytdl_subscribe.validators.config.preset_validator import PresetValidator
from ytdl_subscribe.validators.exceptions import ValidationException
class SubscriptionValidator(PresetValidator):
"""
A Subscription is a preset but overrides it with specific values
"""
required_fields = {"preset"}
optional_fields = {"overrides"}.union(
PresetValidator.required_fields, PresetValidator.optional_fields
)
def __init__(self, config: ConfigValidator, name: str, value: Any):
super().__init__(name, value)
self.config = config
self.overrides: Optional[Dict] = None
def validate(self) -> "SubscriptionValidator":
# If a preset is present, update the
preset_name = self.get("preset")
available_presets = sorted(list(self.config.presets.keys()))
if preset_name not in available_presets:
raise ValidationException(
f"'{self.name}.preset' does not exist in the provided config. "
f"Available presets: {', '.join(available_presets)}"
)
self.overrides = self.get("overrides", {})
# A little hacky, we will override the preset with the contents of this subscription, then validate it
self.value = mergedeep.merge(
self.config.presets[preset_name].value,
self.value,
strategy=mergedeep.Strategy.REPLACE,
)
_ = super().validate()
return self
def to_subscription(self) -> Subscription:
subscription_class = None
if self.subscription_source_name == SubscriptionSourceName.SOUNDCLOUD:
subscription_class = SoundcloudSubscription
elif self.subscription_source_name == SubscriptionSourceName.YOUTUBE:
subscription_class = YoutubeSubscription
else:
raise ValueError("subscription source class not found")
return subscription_class(
name=self.name,
options=self.get(self.subscription_source_name),
ytdl_opts=self.get("ytdl_options"),
post_process=self.get("post_process"),
overrides=self.get("overrides"),
output_path=self.get("output_path"),
)
@classmethod
def from_file_path(
cls, config: ConfigValidator, subscription_path: str
) -> List["SubscriptionValidator"]:
# TODO: Create separate yaml file loader class
with open(subscription_path, "r", encoding="utf-8") as file:
subscription_dict = yaml.safe_load(file)
subscriptions: List["SubscriptionValidator"] = []
for subscription_key, subscription_object in subscription_dict.items():
subscriptions.append(
SubscriptionValidator(
config=config, name=subscription_key, value=subscription_object
).validate()
)
return subscriptions