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

make download strategy a psuedo validator

Browse source
This commit is contained in:
jbannon 2022-04-09 06:01:01 +00:00
parent 4b05a13714
commit 2f3a647775
10 changed files with 125 additions and 128 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

11
ytdl_subscribe/subscriptions/soundcloud.py
View file

@ -7,13 +7,10 @@ from ytdl_subscribe.subscriptions.subscription import Subscription
from ytdl_subscribe.validators.config.source_options.soundcloud_validators import (
SoundcloudAlbumsAndSinglesDownloadValidator,
)
from ytdl_subscribe.validators.config.source_options.soundcloud_validators import (
SoundcloudSourceValidator,
)
class SoundcloudAlbumsAndSinglesSubscription(
Subscription[SoundcloudSourceValidator, SoundcloudAlbumsAndSinglesDownloadValidator]
Subscription[SoundcloudAlbumsAndSinglesDownloadValidator]
):
def extract_info(self):
tracks: List[SoundcloudTrack] = []
@ -22,7 +19,7 @@ class SoundcloudAlbumsAndSinglesSubscription(
# 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] = downloader.download_albums(
artist_name=self.download_strategy_options.username.value
artist_name=self.source_options.username.value
)
for album in albums:
@ -31,9 +28,7 @@ class SoundcloudAlbumsAndSinglesSubscription(
)
# only add tracks that are not part of an album
single_tracks = downloader.download_tracks(
artist_name=self.download_strategy_options.username.value
)
single_tracks = downloader.download_tracks(artist_name=self.source_options.username.value)
tracks += [
track for track in single_tracks if not any(album.contains(track) for album in albums)
]

13
ytdl_subscribe/subscriptions/subscription.py
View file

@ -24,17 +24,13 @@ from ytdl_subscribe.validators.config.output_options.output_options_validator im
)
from ytdl_subscribe.validators.config.overrides.overrides_validator import OverridesValidator
from ytdl_subscribe.validators.config.preset_validator import PresetValidator
from ytdl_subscribe.validators.config.source_options.source_validator import (
DownloadStrategyValidator,
)
from ytdl_subscribe.validators.config.source_options.source_validator import SourceValidator
from ytdl_subscribe.validators.config.source_options.source_validators import SourceValidator
T = TypeVar("T", bound=SourceValidator)
U = TypeVar("U", bound=Downloader)
V = TypeVar("V", bound=DownloadStrategyValidator)
class Subscription(Generic[T, V], ABC):
class Subscription(Generic[T], ABC):
def __init__(
self,
name: str,
@ -65,11 +61,6 @@ class Subscription(Generic[T, V], ABC):
ytdl_options=self.__preset_options.ytdl_options.dict,
)
@property
def download_strategy_options(self) -> V:
"""Returns the download strategy options defined for this subscription"""
return self.source_options.download_strategy
@property
def output_options(self) -> OutputOptionsValidator:
"""Returns the output options defined for this subscription"""

9
ytdl_subscribe/subscriptions/youtube.py
View file

@ -3,17 +3,12 @@ from ytdl_subscribe.subscriptions.subscription import Subscription
from ytdl_subscribe.validators.config.source_options.youtube_validators import (
YoutubePlaylistDownloadValidator,
)
from ytdl_subscribe.validators.config.source_options.youtube_validators import (
YoutubeSourceValidator,
)
class YoutubePlaylistSubscription(
Subscription[YoutubeSourceValidator, YoutubePlaylistDownloadValidator]
):
class YoutubePlaylistSubscription(Subscription[YoutubePlaylistDownloadValidator]):
def extract_info(self):
entries = self.get_downloader(YoutubeDownloader).download_playlist(
playlist_id=self.download_strategy_options.playlist_id.value
playlist_id=self.source_options.playlist_id.value
)
for entry in entries:

30
ytdl_subscribe/validators/config/preset_validator.py
View file

@ -12,21 +12,24 @@ from ytdl_subscribe.validators.config.output_options.output_options_validator im
OutputOptionsValidator,
)
from ytdl_subscribe.validators.config.overrides.overrides_validator import OverridesValidator
from ytdl_subscribe.validators.config.source_options.soundcloud_validators import (
SoundcloudSourceValidator,
from ytdl_subscribe.validators.config.source_options.download_strategy_validators import (
DownloadStrategyValidator,
)
from ytdl_subscribe.validators.config.source_options.source_validator import SourceValidator
from ytdl_subscribe.validators.config.source_options.youtube_validators import (
YoutubeSourceValidator,
from ytdl_subscribe.validators.config.source_options.download_strategy_validators import (
SoundcloudDownloadStrategyValidator,
)
from ytdl_subscribe.validators.config.source_options.download_strategy_validators import (
YoutubeDownloadStrategyValidator,
)
from ytdl_subscribe.validators.config.source_options.source_validators import SourceValidator
from ytdl_subscribe.validators.config.ytdl_options.ytdl_options_validator import (
YTDLOptionsValidator,
)
from ytdl_subscribe.validators.exceptions import ValidationException
PRESET_SOURCE_VALIDATOR_MAPPING: Dict[str, Type[SourceValidator]] = {
"soundcloud": SoundcloudSourceValidator,
"youtube": YoutubeSourceValidator,
PRESET_SOURCE_VALIDATOR_MAPPING: Dict[str, Type[DownloadStrategyValidator]] = {
"soundcloud": SoundcloudDownloadStrategyValidator,
"youtube": YoutubeDownloadStrategyValidator,
}
PRESET_REQUIRED_KEYS = {"output_options"}
@ -47,28 +50,29 @@ class PresetValidator(StrictDictValidator):
return sorted(list(PRESET_SOURCE_VALIDATOR_MAPPING.keys()))
def __validate_and_get_subscription_source(self) -> SourceValidator:
subscription_source: Optional[SourceValidator] = None
download_strategy_validator: Optional[DownloadStrategyValidator] = None
for key in self._keys:
if key in self.__available_sources and subscription_source:
# Ensure there are not multiple sources, i.e. youtube and soundcloud
if key in self.__available_sources and download_strategy_validator:
raise ValidationException(
f"'{self._name}' can only have one of the following sources: "
f"{', '.join(self.__available_sources)}"
)
if key in PRESET_SOURCE_VALIDATOR_MAPPING:
subscription_source = self._validate_key(
download_strategy_validator = self._validate_key(
key=key, validator=PRESET_SOURCE_VALIDATOR_MAPPING[key]
)
# If subscription source was not set, error
if not subscription_source:
if not download_strategy_validator:
raise ValidationException(
f"'{self._name} must have one of the following sources: "
f"{', '.join(self.__available_sources)}"
)
return subscription_source
return download_strategy_validator.source_validator
def __init__(self, name: str, value: Any):
super().__init__(name=name, value=value)

64
ytdl_subscribe/validators/config/source_options/download_strategy_validators.py Normal file
View file

@ -0,0 +1,64 @@
import copy
from abc import ABC
from typing import Any
from typing import Dict
from typing import Type
from ytdl_subscribe.validators.base.strict_dict_validator import StrictDictValidator
from ytdl_subscribe.validators.base.validators import StringValidator
from ytdl_subscribe.validators.config.source_options.soundcloud_validators import (
SoundcloudAlbumsAndSinglesDownloadValidator,
)
from ytdl_subscribe.validators.config.source_options.source_validators import SourceValidator
from ytdl_subscribe.validators.config.source_options.youtube_validators import (
YoutubePlaylistDownloadValidator,
)
class DownloadStrategyValidator(StrictDictValidator, ABC):
"""
Validates the download strategy of a source. Does not validate the source options.
"""
# All media sources must define a download strategy
_required_keys = {"download_strategy"}
# Extra fields will be strict-validated using other StictDictValidators
_allow_extra_keys = True
_download_strategy_to_source_mapping: Dict[str, Type[SourceValidator]] = {}
def __init__(self, name: str, value: Any):
super().__init__(name=name, value=value)
download_strategy_name = self._validate_key(
key="download_strategy",
validator=StringValidator,
).value
if download_strategy_name not in self._possible_download_strategies:
raise self._validation_exception(
f"download_strategy must be one of the following: "
f"{', '.join(self._possible_download_strategies)}"
)
# Remove the 'download_strategy' key before passing the other keys to the actual
# source validator
source_validator_dict = copy.deepcopy(self._dict)
del source_validator_dict["download_strategy"]
source_validator_class = self._download_strategy_to_source_mapping[download_strategy_name]
self.source_validator = source_validator_class(name=self._name, value=source_validator_dict)
@property
def _possible_download_strategies(self):
return sorted(list(self._download_strategy_to_source_mapping.keys()))
class SoundcloudDownloadStrategyValidator(DownloadStrategyValidator):
_download_strategy_to_source_mapping = {
"albums_and_singles": SoundcloudAlbumsAndSinglesDownloadValidator
}
class YoutubeDownloadStrategyValidator(DownloadStrategyValidator):
_download_strategy_to_source_mapping = {"playlist": YoutubePlaylistDownloadValidator}

22
ytdl_subscribe/validators/config/source_options/soundcloud_validators.py
View file

@ -1,28 +1,12 @@
from ytdl_subscribe.validators.base.validators import BoolValidator
from ytdl_subscribe.validators.base.validators import StringValidator
from ytdl_subscribe.validators.config.source_options.source_validator import (
DownloadStrategyValidator,
from ytdl_subscribe.validators.config.source_options.source_validators import (
SoundcloudSourceValidator,
)
from ytdl_subscribe.validators.config.source_options.source_validator import SourceValidator
class SoundcloudAlbumsAndSinglesDownloadValidator(DownloadStrategyValidator):
class SoundcloudAlbumsAndSinglesDownloadValidator(SoundcloudSourceValidator):
_required_keys = {"username"}
def __init__(self, name, value):
super().__init__(name, value)
self.username = self._validate_key(key="username", validator=StringValidator)
class SoundcloudSourceValidator(SourceValidator):
_optional_keys = {"skip_premiere_tracks"}
_download_strategy_validator_mapping = {
"albums_and_singles": SoundcloudAlbumsAndSinglesDownloadValidator
}
def __init__(self, name: str, value: dict):
super().__init__(name=name, value=value)
self.skip_premiere_tracks = self._validate_key(
"skip_premiere_tracks", BoolValidator, default=True
)

48
ytdl_subscribe/validators/config/source_options/source_validator.py
View file

@ -1,48 +0,0 @@
import copy
from typing import Any
from typing import Dict
from typing import Type
from ytdl_subscribe.validators.base.strict_dict_validator import StrictDictValidator
from ytdl_subscribe.validators.base.validators import StringValidator
class DownloadStrategyValidator(StrictDictValidator):
pass
class SourceValidator(StrictDictValidator):
# All media sources must define a download strategy
_required_keys = {"download_strategy"}
# Extra fields will be strict-validated using other StictDictValidators
_allow_extra_keys = True
_download_strategy_validator_mapping: Dict[str, Type[DownloadStrategyValidator]] = {}
def __init__(self, name: str, value: Any):
super().__init__(name=name, value=value)
download_strategy_name = self._validate_key(
key="download_strategy",
validator=StringValidator,
).value
if download_strategy_name not in self._possible_download_strategies:
raise self._validation_exception(
f"download_strategy must be one of the following: "
f"{', '.join(self._possible_download_strategies)}"
)
# Remove all non-download strategy keys before passing the dict to the validator
download_strategy_dict = copy.deepcopy(self._dict)
for key_to_delete in self._allowed_keys:
del download_strategy_dict[key_to_delete]
download_strategy_class = self._download_strategy_validator_mapping[download_strategy_name]
self.download_strategy = download_strategy_class(
name=self._name, value=download_strategy_dict
)
@property
def _possible_download_strategies(self):
return sorted(list(self._download_strategy_validator_mapping.keys()))

30
ytdl_subscribe/validators/config/source_options/source_validators.py Normal file
View file

@ -0,0 +1,30 @@
from abc import ABC
from ytdl_subscribe.validators.base.strict_dict_validator import StrictDictValidator
from ytdl_subscribe.validators.base.validators import BoolValidator
class SourceValidator(StrictDictValidator):
"""
Abstract class for any source validator
"""
class SoundcloudSourceValidator(SourceValidator, ABC):
"""
Abstract source validator for all soundcloud sources.
"""
_optional_keys = {"skip_premiere_tracks"}
def __init__(self, name: str, value: dict):
super().__init__(name=name, value=value)
self.skip_premiere_tracks = self._validate_key(
"skip_premiere_tracks", BoolValidator, default=True
)
class YoutubeSourceValidator(SourceValidator, ABC):
"""
Abstract source validator for all soundcloud sources.
"""

16
ytdl_subscribe/validators/config/source_options/youtube_validators.py
View file

@ -1,22 +1,10 @@
from typing import Any
from ytdl_subscribe.validators.base.validators import StringValidator
from ytdl_subscribe.validators.config.source_options.source_validator import (
DownloadStrategyValidator,
)
from ytdl_subscribe.validators.config.source_options.source_validator import SourceValidator
from ytdl_subscribe.validators.config.source_options.source_validators import YoutubeSourceValidator
class YoutubePlaylistDownloadValidator(DownloadStrategyValidator):
class YoutubePlaylistDownloadValidator(YoutubeSourceValidator):
_required_keys = {"playlist_id"}
def __init__(self, name, value):
super().__init__(name, value)
self.playlist_id = self._validate_key("playlist_id", StringValidator)
class YoutubeSourceValidator(SourceValidator):
_download_strategy_validator_mapping = {"playlist": YoutubePlaylistDownloadValidator}
def __init__(self, name: str, value: Any):
super().__init__(name=name, value=value)

10
ytdl_subscribe/validators/config/subscription_validator.py
View file

@ -65,15 +65,9 @@ class SubscriptionValidator(StrictDictValidator):
)
def to_subscription(self) -> Subscription:
if isinstance(
self.preset.subscription_source.download_strategy,
SoundcloudAlbumsAndSinglesDownloadValidator,
):
if isinstance(self.preset.subscription_source, SoundcloudAlbumsAndSinglesDownloadValidator):
subscription_class = SoundcloudAlbumsAndSinglesSubscription
elif isinstance(
self.preset.subscription_source.download_strategy,
YoutubePlaylistDownloadValidator,
):
elif isinstance(self.preset.subscription_source, YoutubePlaylistDownloadValidator):
subscription_class = YoutubePlaylistSubscription
else:
raise ValueError("subscription source class not found")