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

finally some proper validation

Browse source
This commit is contained in:
Jesse Bannon 2023-12-11 23:06:29 -08:00
parent 347f3ac375
commit 5d1ff812f0
9 changed files with 146 additions and 134 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
src/ytdl_sub/config/overrides.py
View file

@ -67,10 +67,11 @@ class Overrides(DictFormatterValidator, Scriptable):
)
return ScriptUtils.add_sanitized_variables(initial_variables)
def initialize_script(self, unresolved_variables: Dict[str, str]) -> None:
def initialize_script(self, unresolved_variables: Dict[str, str]) -> "Overrides":
self.script.add(self.initial_variables(unresolved_variables=unresolved_variables))
self.unresolvable.update(set(unresolved_variables.keys()))
self.update_script()
return self
@property
def subscription_name(self) -> str:

5
src/ytdl_sub/config/plugin.py
View file

@ -1,5 +1,6 @@
from abc import ABC
from abc import abstractmethod
from enum import Enum
from typing import Dict
from typing import Generic
from typing import List
@ -20,6 +21,8 @@ class PluginPriority:
Defines priority for plugins, 0 is highest priority
"""
MODIFY_ENTRY_SPLIT = 9
# If modify_entry priority is >= to this value, run after split
MODIFY_ENTRY_AFTER_SPLIT = 10
@ -144,6 +147,8 @@ class SplitPlugin(Plugin[TOptionsValidator], Generic[TOptionsValidator], ABC):
Plugin that splits entries into zero or more entries
"""
priority = PluginPriority(modify_entry_metadata=PluginPriority.MODIFY_ENTRY_SPLIT)
@abstractmethod
def split(self, entry: Entry) -> List[Tuple[Entry, FileMetadata]]:
"""

116
src/ytdl_sub/config/preset.py
View file

@ -18,6 +18,7 @@ from ytdl_sub.config.plugin import Plugin
from ytdl_sub.config.plugin_mapping import PluginMapping
from ytdl_sub.config.preset_options import OptionsValidator
from ytdl_sub.config.preset_options import OutputOptions
from ytdl_sub.config.preset_options import PluginOperation
from ytdl_sub.config.preset_options import TOptionsValidator
from ytdl_sub.config.preset_options import YTDLOptions
from ytdl_sub.downloaders.url.validators import MultiUrlValidator
@ -26,13 +27,11 @@ from ytdl_sub.entries.script.variable_scripts import VARIABLE_SCRIPTS
from ytdl_sub.prebuilt_presets import PREBUILT_PRESET_NAMES
from ytdl_sub.prebuilt_presets import PUBLISHED_PRESET_NAMES
from ytdl_sub.script.script import Script
from ytdl_sub.script.script import ScriptBuilder
from ytdl_sub.script.utils.exceptions import VariableDoesNotExist
from ytdl_sub.utils.exceptions import StringFormattingVariableNotFoundException
from ytdl_sub.utils.exceptions import ValidationException
from ytdl_sub.utils.logger import Logger
from ytdl_sub.utils.script import ScriptUtils
from ytdl_sub.utils.scriptable import Scriptable
from ytdl_sub.utils.yaml import dump_yaml
from ytdl_sub.validators.strict_dict_validator import StrictDictValidator
from ytdl_sub.validators.string_formatter_validators import DictFormatterValidator
@ -168,35 +167,19 @@ class Preset(_PresetShell):
return list(VARIABLE_SCRIPTS.keys())
@property
def _added_variables(self) -> Dict[str, str]:
added_variables: Dict[str, str] = {
var_name: "dummy_string"
for var_name in self.downloader_options.added_source_variables()
}
def _added_variables(self) -> Set[str]:
added_variables: Set[str] = set()
options: List[OptionsValidator] = self.plugins.plugin_options
options.append(self.downloader_options)
for plugin_options in options:
for plugin_added_variables in plugin_options.added_source_variables(
unresolved_variables=set()
).values():
added_variables |= set(plugin_added_variables)
for plugin_options in self.plugins.plugin_options:
for source_var in plugin_options.added_source_variables():
added_variables[source_var] = "dummy_string"
return added_variables
@functools.cached_property
def _script_builder(self) -> ScriptBuilder:
# Set the formatter variables to be the overrides
script = ScriptBuilder(
self.overrides.initial_variables(unresolved_variables=self._added_variables)
)
script.add_resolved(ScriptUtils.add_dummy_variables(self._source_variables))
return script
@functools.cached_property
def _script(self) -> Script:
"""
Contains actualized script which should hold all Override variables
"""
return self._script_builder.add_resolved(
ScriptUtils.add_dummy_variables(self._added_variables)
).partial_build()
def __validate_and_get_plugins(self) -> PresetPlugins:
preset_plugins = PresetPlugins()
@ -211,21 +194,43 @@ class Preset(_PresetShell):
return preset_plugins
def __validate_added_variables(self):
script_builder = copy.deepcopy(self._script_builder)
self.downloader_options.validate_with_variables(script=script_builder.partial_build())
script_builder.add_resolved(
ScriptUtils.add_dummy_variables(self.downloader_options.added_source_variables())
def _validate_added_variables(self) -> Script:
"""
Validate variables resolve as plugins are executed, and return
a mock script which contains actualized added variables from the plugins
"""
script = copy.deepcopy(self.overrides.script).add(
ScriptUtils.add_dummy_variables(self._source_variables)
)
unresolved_variables = copy.deepcopy(self._added_variables)
added_variables: Set[str] = self.downloader_options.added_source_variables(
unresolved_variables
).get(PluginOperation.DOWNLOADER, set())
script.add(ScriptUtils.add_dummy_variables(added_variables))
unresolved_variables -= added_variables
for _, plugin_options in sorted(
self.plugins.zipped(), key=lambda pl: pl[0].priority.modify_entry_metadata
):
added_variables = plugin_options.added_source_variables(
unresolved_variables=unresolved_variables
).get(PluginOperation.MODIFY_ENTRY_METADATA, set())
script.add(ScriptUtils.add_dummy_variables(added_variables))
unresolved_variables -= added_variables
for _, plugin_options in sorted(
self.plugins.zipped(), key=lambda pl: pl[0].priority.modify_entry
):
# Validate current plugin using source + added plugin variables
plugin_options.validate_with_variables(script=script_builder.partial_build())
script_builder.add_resolved(
ScriptUtils.add_dummy_variables(self.downloader_options.added_source_variables())
)
added_variables = plugin_options.added_source_variables(
unresolved_variables=unresolved_variables
).get(PluginOperation.MODIFY_ENTRY, set())
script.add(ScriptUtils.add_dummy_variables(added_variables))
unresolved_variables -= added_variables
assert not unresolved_variables
_ = script.resolve(update=True)
return script
@functools.cache
def _get_unresolvable_variables(
@ -233,7 +238,7 @@ class Preset(_PresetShell):
formatter_validator: Union[StringFormatterValidator, OverridesStringFormatterValidator],
) -> Optional[Set[str]]:
unresolvable = (
set([VARIABLES.entry_metadata.variable_name] + list(self._added_variables.keys()))
self._added_variables.union([VARIABLES.entry_metadata.variable_name])
if isinstance(formatter_validator, OverridesStringFormatterValidator)
else None
)
@ -241,10 +246,11 @@ class Preset(_PresetShell):
def __validate_override_string_formatter_validator(
self,
mock_script: Script,
formatter_validator: Union[StringFormatterValidator, OverridesStringFormatterValidator],
) -> None:
try:
self._script.resolve_once(
mock_script.resolve_once(
{"tmp_var": formatter_validator.format_string},
unresolvable=self._get_unresolvable_variables(formatter_validator),
)
@ -253,6 +259,7 @@ class Preset(_PresetShell):
def __recursive_preset_validate(
self,
mock_script: Script,
validator: Optional[Validator] = None,
) -> None:
"""
@ -267,16 +274,20 @@ class Preset(_PresetShell):
# Usage of protected variables in other validators is fine. The reason to keep
# them protected is for readability when using them in subscriptions.
for validator_value in validator._validator_dict.values():
self.__recursive_preset_validate(validator_value)
self.__recursive_preset_validate(mock_script=mock_script, validator=validator_value)
# pylint: enable=protected-access
elif isinstance(validator, ListValidator):
for list_value in validator.list:
self.__recursive_preset_validate(list_value)
self.__recursive_preset_validate(mock_script=mock_script, validator=list_value)
elif isinstance(validator, (StringFormatterValidator, OverridesStringFormatterValidator)):
self.__validate_override_string_formatter_validator(validator)
self.__validate_override_string_formatter_validator(
mock_script=mock_script, formatter_validator=validator
)
elif isinstance(validator, (DictFormatterValidator, OverridesDictFormatterValidator)):
for validator_value in validator.dict.values():
self.__validate_override_string_formatter_validator(validator_value)
self.__validate_override_string_formatter_validator(
mock_script=mock_script, formatter_validator=validator_value
)
def _get_presets_to_merge(
self, parent_presets: str | List[str], seen_presets: List[str], config: ConfigValidator
@ -353,21 +364,22 @@ class Preset(_PresetShell):
key="ytdl_options", validator=YTDLOptions, default={}
)
self.overrides = self._validate_key(key="overrides", validator=Overrides, default={})
self.plugins: PresetPlugins = self.__validate_and_get_plugins()
self.__validate_added_variables()
# After all options are initialized, perform a recursive post-validate that requires
# values from multiple validators
self.__recursive_preset_validate()
self.overrides.initialize_script(
self.overrides = self._validate_key(
key="overrides", validator=Overrides, default={}
).initialize_script(
unresolved_variables={
var_name: f"{{%throw('Plugin variable {var_name} has not been created yet')}}"
for var_name in self._added_variables
}
)
mock_script = self._validate_added_variables()
# After all options are initialized, perform a recursive post-validate that requires
# values from multiple validators
self.__recursive_preset_validate(mock_script=mock_script)
@property
def name(self) -> str:
"""

28
src/ytdl_sub/config/preset_options.py
View file

@ -1,12 +1,13 @@
from abc import ABC
from enum import Enum
from typing import Any
from typing import Dict
from typing import List
from typing import Optional
from typing import Set
from typing import TypeVar
from ytdl_sub.config.defaults import DEFAULT_DOWNLOAD_ARCHIVE_NAME
from ytdl_sub.script.script import Script
from ytdl_sub.script.script import ScriptBuilder
from ytdl_sub.utils.exceptions import ValidationException
from ytdl_sub.validators.file_path_validators import OverridesStringFormatterFilePathValidator
from ytdl_sub.validators.file_path_validators import StringFormatterFileNameValidator
@ -20,6 +21,12 @@ from ytdl_sub.validators.validators import LiteralDictValidator
from ytdl_sub.validators.validators import Validator
class PluginOperation(Enum):
DOWNLOADER = -1
MODIFY_ENTRY_METADATA = 0
MODIFY_ENTRY = 1
# pylint: disable=no-self-use
# pylint: disable=unused-argument
class OptionsValidator(Validator, ABC):
@ -44,7 +51,9 @@ class OptionsValidator(Validator, ABC):
"""
return self._validation_exception(error_message=error_message)
def added_source_variables(self) -> List[str]:
def added_source_variables(
self, unresolved_variables: Set[str]
) -> Dict[PluginOperation, Set[str]]:
"""
If the plugin adds source variables, list them here.
@ -52,18 +61,7 @@ class OptionsValidator(Validator, ABC):
-------
List of added source variables this plugin creates
"""
return []
def validate_with_variables(self, script: Script) -> None:
"""
Optional validation after init with the session's source and override variables.
Parameters
----------
script
Script containing all current variables
"""
return None
return {}
TOptionsValidator = TypeVar("TOptionsValidator", bound=OptionsValidator)

29
src/ytdl_sub/downloaders/url/validators.py
View file

@ -3,8 +3,10 @@ from typing import Any
from typing import Dict
from typing import List
from typing import Optional
from typing import Set
from ytdl_sub.config.preset_options import OptionsValidator
from ytdl_sub.config.preset_options import PluginOperation
from ytdl_sub.script.script import Script
from ytdl_sub.script.script import ScriptBuilder
from ytdl_sub.validators.strict_dict_validator import StrictDictValidator
@ -245,31 +247,12 @@ class MultiUrlValidator(OptionsValidator):
# keep for readthedocs documentation
return self._urls.list[0].variables
def added_source_variables(self) -> List[str]:
def added_source_variables(
self, unresolved_variables: Set[str]
) -> Dict[PluginOperation, Set[str]]:
"""
Returns
-------
List of variables added. The first collection url always contains all the variables.
"""
return list(self._urls.list[0].variables.keys)
def validate_with_variables(self, script: Script) -> None:
"""
Ensures new variables added are not existing variables
"""
# Apply formatting to each new source variable, ensure it resolves
for collection_url in self.urls.list:
script.resolve_once(collection_url.variables.dict_with_format_strings)
# Ensure at least URL is non-empty
has_non_empty_url = False
url_variables = {
f"tmp_var_url_{idx}": url_validator.url.format_string
for idx, url_validator in enumerate(self.urls.list)
}
output = script.resolve_once(url_variables)
for out in output.values():
has_non_empty_url |= bool(str(out))
if not output or not has_non_empty_url:
raise self._validation_exception("Must contain at least one url that is non-empty")
return {PluginOperation.DOWNLOADER: set(self._urls.list[0].variables.keys)}

60
src/ytdl_sub/plugins/regex.py
View file

@ -2,15 +2,18 @@ from typing import Any
from typing import Dict
from typing import List
from typing import Optional
from typing import Set
from yt_dlp.utils import sanitize_filename
from ytdl_sub.config.plugin import Plugin
from ytdl_sub.config.plugin import PluginPriority
from ytdl_sub.config.preset_options import OptionsDictValidator
from ytdl_sub.config.preset_options import PluginOperation
from ytdl_sub.entries.entry import Entry
from ytdl_sub.entries.script.variable_scripts import VARIABLE_SCRIPTS
from ytdl_sub.entries.variables.kwargs import YTDL_SUB_REGEX_SOURCE_VARS
from ytdl_sub.script.parser import parse
from ytdl_sub.script.script import Script
from ytdl_sub.utils.exceptions import RegexNoMatchException
from ytdl_sub.utils.exceptions import StringFormattingVariableNotFoundException
@ -211,27 +214,6 @@ class RegexOptions(OptionsDictValidator):
"""
return self._skip_if_match_fails
def validate_with_variables(self, script: Script) -> None:
for key, regex_options in self.source_variable_capture_dict.items():
# Ensure each variable getting captured is a source variable
if key not in script._variables:
raise self._validation_exception(
f"cannot regex capture '{key}' because it is not a source or override variable"
)
# Ensure the capture group names are not existing source/override variables
for capture_group_name in regex_options.capture_group_names:
if capture_group_name in VARIABLE_SCRIPTS:
raise self._validation_exception(
f"'{capture_group_name}' cannot be used as a capture group name because it "
f"is a source variable"
)
if capture_group_name in script._variables:
raise self._validation_exception(
f"'{capture_group_name}' cannot be used as a capture group name because it "
f"is an override variable"
)
@property
def source_variable_capture_dict(self) -> Dict[str, VariableRegex]:
"""
@ -241,19 +223,39 @@ class RegexOptions(OptionsDictValidator):
"""
return self._from.variable_capture_dict
def added_source_variables(self) -> List[str]:
def _is_evaluatable_at_metadata_time(
self, unresolved_variables: Set[str], input_variable_name: str, regex_options: VariableRegex
) -> bool:
if input_variable_name in unresolved_variables:
return False
for capture_group_default in regex_options.capture_group_defaults:
parsed_default = parse(capture_group_default.format_string)
if parsed_default.variables.issubset(unresolved_variables):
return False
return True
def added_source_variables(
self, unresolved_variables: Set[str]
) -> Dict[PluginOperation, Set[str]]:
"""
Returns
-------
List of new source variables created via regex capture
"""
added_source_vars: List[str] = []
for regex_options in self.source_variable_capture_dict.values():
added_source_vars.extend(regex_options.capture_group_names)
added_source_vars.extend(
f"{capture_group_name}_sanitized"
for capture_group_name in regex_options.capture_group_names
)
added_source_vars: Dict[PluginOperation, Set[str]] = {
PluginOperation.MODIFY_ENTRY_METADATA: set(),
PluginOperation.MODIFY_ENTRY: set(),
}
for input_variable_name, regex_options in self.source_variable_capture_dict.items():
key = PluginOperation.MODIFY_ENTRY
if self._is_evaluatable_at_metadata_time(
unresolved_variables=unresolved_variables,
input_variable_name=input_variable_name,
regex_options=regex_options,
):
key = PluginOperation.MODIFY_ENTRY_METADATA
added_source_vars[key] |= set(regex_options.capture_group_names)
return added_source_vars

23
src/ytdl_sub/plugins/split_by_chapters.py
View file

@ -1,14 +1,17 @@
import copy
from pathlib import Path
from typing import Any
from typing import Dict
from typing import List
from typing import Optional
from typing import Set
from typing import Tuple
from yt_dlp.utils import sanitize_filename
from ytdl_sub.config.plugin import SplitPlugin
from ytdl_sub.config.preset_options import OptionsDictValidator
from ytdl_sub.config.preset_options import PluginOperation
from ytdl_sub.entries.entry import Entry
from ytdl_sub.entries.variables.kwargs import CHAPTERS
from ytdl_sub.entries.variables.kwargs import SPLIT_BY_CHAPTERS_PARENT_ENTRY
@ -81,14 +84,18 @@ class SplitByChaptersOptions(OptionsDictValidator):
key="when_no_chapters", validator=WhenNoChaptersValidator
).value
def added_source_variables(self) -> List[str]:
return [
"chapter_title",
"chapter_title_sanitized",
"chapter_index",
"chapter_index_padded",
"chapter_count",
]
def added_source_variables(
self, unresolved_variables: Set[str]
) -> Dict[PluginOperation, Set[str]]:
return {
PluginOperation.MODIFY_ENTRY: {
"chapter_title",
"chapter_title_sanitized",
"chapter_index",
"chapter_index_padded",
"chapter_count",
}
}
@property
def when_no_chapters(self) -> str:

8
src/ytdl_sub/plugins/subtitles.py
View file

@ -2,9 +2,11 @@ from pathlib import Path
from typing import Dict
from typing import List
from typing import Optional
from typing import Set
from ytdl_sub.config.plugin import Plugin
from ytdl_sub.config.preset_options import OptionsDictValidator
from ytdl_sub.config.preset_options import PluginOperation
from ytdl_sub.downloaders.ytdl_options_builder import YTDLOptionsBuilder
from ytdl_sub.entries.entry import Entry
from ytdl_sub.utils.file_handler import FileHandler
@ -113,13 +115,15 @@ class SubtitleOptions(OptionsDictValidator):
"""
return self._allow_auto_generated_subtitles
def added_source_variables(self) -> List[str]:
def added_source_variables(
self, unresolved_variables: Set[str]
) -> Dict[PluginOperation, Set[str]]:
"""
Returns
-------
List of new source variables created by using the subtitles plugin
"""
return ["lang", "subtitles_ext"]
return {PluginOperation.MODIFY_ENTRY: {"lang", "subtitles_ext"}}
class SubtitlesPlugin(Plugin[SubtitleOptions]):

8
src/ytdl_sub/utils/script.py
View file

@ -9,11 +9,11 @@ from ytdl_sub.script.types.resolvable import String
class ScriptUtils:
@classmethod
def add_dummy_variables(cls, variables: Iterable[str]) -> Dict[str, Resolvable]:
dummy_variables: Dict[str, Resolvable] = {}
def add_dummy_variables(cls, variables: Iterable[str]) -> Dict[str, str]:
dummy_variables: Dict[str, str] = {}
for var in variables:
dummy_variables[var] = String("dummy_string")
dummy_variables[f"{var}_sanitized"] = String("dummy_string")
dummy_variables[var] = ""
dummy_variables[f"{var}_sanitized"] = ""
return dummy_variables