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

[BACKEND] Refactor validation, preparation for subscription dissect (#1398)

Browse source 
Completely rewrite how subscription validation is performed. Optimizes it quite a bit while also adding backend support for the upcoming `dissect` sub-command, where you can resolve any subscription into its 'raw' form for easier debugging when making scripting changes.
This commit is contained in:
Jesse Bannon 2025-12-30 17:03:59 -08:00 committed by GitHub
parent 344753cc63
commit 1abe2a44f5
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
17 changed files with 575 additions and 391 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
.gitignore vendored
View file

@ -146,8 +146,11 @@ docker/testing/volumes
.local/
.ytdl-sub-working-directory
.ytdl-sub-lock
ffmpeg.exe
ffprobe.exe
tools/docgen/out
prof/

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

@ -4,15 +4,16 @@ from typing import Iterable
from typing import Optional
from typing import Set
import mergedeep
from ytdl_sub.entries.entry import Entry
from ytdl_sub.entries.script.variable_definitions import VARIABLES
from ytdl_sub.entries.variables.override_variables import REQUIRED_OVERRIDE_VARIABLE_NAMES
from ytdl_sub.entries.variables.override_variables import OverrideHelpers
from ytdl_sub.script.parser import parse
from ytdl_sub.script.script import Script
from ytdl_sub.script.types.function import BuiltInFunction
from ytdl_sub.script.types.resolvable import Resolvable
from ytdl_sub.script.types.resolvable import String
from ytdl_sub.script.types.syntax_tree import SyntaxTree
from ytdl_sub.script.utils.exceptions import ScriptVariableNotResolved
from ytdl_sub.utils.exceptions import InvalidVariableNameException
from ytdl_sub.utils.exceptions import StringFormattingException
@ -134,29 +135,35 @@ class Overrides(UnstructuredDictFormatterValidator, Scriptable):
)
def initial_variables(
self, unresolved_variables: Optional[Dict[str, str]] = None
) -> Dict[str, str]:
self, unresolved_variables: Optional[Dict[str, SyntaxTree]] = None
) -> Dict[str, SyntaxTree]:
"""
Returns
-------
Variables and format strings for all Override variables + additional variables (Optional)
"""
initial_variables: Dict[str, str] = {}
mergedeep.merge(
initial_variables,
self.dict_with_format_strings,
unresolved_variables if unresolved_variables else {},
)
return ScriptUtils.add_sanitized_variables(initial_variables)
initial_variables: Dict[str, SyntaxTree] = self.dict_with_parsed_format_strings
if unresolved_variables:
initial_variables |= unresolved_variables
return ScriptUtils.add_sanitized_parsed_variables(initial_variables)
def initialize_script(self, unresolved_variables: Set[str]) -> "Overrides":
"""
Initialize the override script with any unresolved variables
"""
self.script.add(
self.script.add_parsed(
self.initial_variables(
unresolved_variables={
var_name: f"{{%throw('Plugin variable {var_name} has not been created yet')}}"
var_name: SyntaxTree(
ast=[
BuiltInFunction(
name="throw",
args=[
String(f"Plugin variable {var_name} has not been created yet")
],
)
]
)
for var_name in unresolved_variables
}
)

33
src/ytdl_sub/config/plugin/preset_plugins.py
View file

@ -1,5 +1,7 @@
from typing import Iterable
from typing import List
from typing import Optional
from typing import Set
from typing import Tuple
from typing import Type
@ -44,3 +46,34 @@ class PresetPlugins:
if plugin_type in plugin_option_types:
return self.plugin_options[plugin_option_types.index(plugin_type)]
return None
def get_added_and_modified_variables(
self, additional_options: List[OptionsValidator]
) -> Iterable[Tuple[OptionsValidator, Set[str], Set[str]]]:
"""
Iterates and returns the plugin options, added variables, modified variables
"""
for plugin_options in self.plugin_options + additional_options:
added_variables: Set[str] = set()
modified_variables: Set[str] = set()
for plugin_added_variables in plugin_options.added_variables(
unresolved_variables=set(),
).values():
added_variables |= set(plugin_added_variables)
for plugin_modified_variables in plugin_options.modified_variables().values():
modified_variables = plugin_modified_variables
yield plugin_options, added_variables, modified_variables
def get_all_variables(self, additional_options: List[OptionsValidator]) -> Set[str]:
"""
Returns set of all added and modified variables' names.
"""
all_variables: Set[str] = set()
for _, added, modified in self.get_added_and_modified_variables(additional_options):
all_variables.update(added)
all_variables.update(modified)
return all_variables

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

@ -2,6 +2,7 @@ import copy
from typing import Any
from typing import Dict
from typing import List
from typing import Set
from mergedeep import mergedeep
@ -11,7 +12,6 @@ from ytdl_sub.config.plugin.plugin_mapping import PluginMapping
from ytdl_sub.config.plugin.preset_plugins import PresetPlugins
from ytdl_sub.config.preset_options import OutputOptions
from ytdl_sub.config.preset_options import YTDLOptions
from ytdl_sub.config.validators.variable_validation import VariableValidation
from ytdl_sub.downloaders.url.validators import MultiUrlValidator
from ytdl_sub.prebuilt_presets import PREBUILT_PRESET_NAMES
from ytdl_sub.prebuilt_presets import PUBLISHED_PRESET_NAMES
@ -172,6 +172,37 @@ class Preset(_PresetShell):
mergedeep.merge({}, *reversed(presets_to_merge), strategy=mergedeep.Strategy.ADDITIVE)
)
def _initialize_overrides_script(self, overrides: Overrides) -> Overrides:
"""
Do some gymnastics to initialize the Overrides script.
"""
unresolved_variables: Set[str] = set()
for (
plugin_options,
added_variables,
modified_variables,
) in self.plugins.get_added_and_modified_variables(
additional_options=[self.downloader_options, self.output_options]
):
for added_variable in added_variables:
if not overrides.ensure_added_plugin_variable_valid(added_variable=added_variable):
# pylint: disable=protected-access
raise plugin_options._validation_exception(
f"Cannot use the variable name {added_variable} because it exists as a"
" built-in ytdl-sub variable name."
)
# pylint: enable=protected-access
# Set unresolved as variables that are added but do not exist as
# entry/override variables since they are created at run-time
unresolved_variables |= added_variables | modified_variables
# Initialize overrides with unresolved variables + modified variables to throw an error.
# For modified variables, this is to prevent a resolve(update=True) to setting any
# dependencies until it has been explicitly added
return overrides.initialize_script(unresolved_variables=unresolved_variables)
def __init__(self, config: ConfigValidator, name: str, value: Any):
super().__init__(name=name, value=value)
@ -192,16 +223,11 @@ class Preset(_PresetShell):
)
self.plugins: PresetPlugins = self._validate_and_get_plugins()
self.overrides = self._validate_key(key="overrides", validator=Overrides, default={})
self.overrides = self._initialize_overrides_script(
overrides=self._validate_key(key="overrides", validator=Overrides, default={})
)
self.overrides.ensure_variable_names_not_a_plugin(plugin_names=PRESET_KEYS)
VariableValidation(
downloader_options=self.downloader_options,
output_options=self.output_options,
plugins=self.plugins,
).initialize_preset_overrides(overrides=self.overrides).ensure_proper_usage()
@property
def name(self) -> str:
"""

170
src/ytdl_sub/config/validators/variable_validation.py
View file

@ -1,10 +1,4 @@
import copy
from typing import Dict
from typing import Iterable
from typing import List
from typing import Optional
from typing import Set
from typing import Tuple
from ytdl_sub.config.overrides import Overrides
from ytdl_sub.config.plugin.plugin_mapping import PluginMapping
@ -13,155 +7,27 @@ from ytdl_sub.config.plugin.preset_plugins import PresetPlugins
from ytdl_sub.config.preset_options import OutputOptions
from ytdl_sub.config.validators.options import OptionsValidator
from ytdl_sub.downloaders.url.validators import MultiUrlValidator
from ytdl_sub.entries.variables.override_variables import REQUIRED_OVERRIDE_VARIABLE_NAMES
from ytdl_sub.script.script import Script
from ytdl_sub.script.script import _is_function
from ytdl_sub.utils.scriptable import BASE_SCRIPT
from ytdl_sub.validators.string_formatter_validators import to_variable_dependency_format_string
from ytdl_sub.validators.string_formatter_validators import validate_formatters
# Entry variables to mock during validation
_DUMMY_ENTRY_VARIABLES: Dict[str, str] = {
name: to_variable_dependency_format_string(
# pylint: disable=protected-access
script=BASE_SCRIPT,
parsed_format_string=BASE_SCRIPT._variables[name],
# pylint: enable=protected-access
)
for name in BASE_SCRIPT.variable_names
}
def _add_dummy_variables(variables: Iterable[str]) -> Dict[str, str]:
dummy_variables: Dict[str, str] = {}
for var in variables:
dummy_variables[var] = ""
dummy_variables[f"{var}_sanitized"] = ""
return dummy_variables
def _add_dummy_overrides(overrides: Overrides) -> Dict[str, str]:
# Have the dummy override variable contain all variable deps that it uses in the string
dummy_overrides: Dict[str, str] = {}
for override_name in _override_variables(overrides):
if _is_function(override_name):
continue
# pylint: disable=protected-access
dummy_overrides[override_name] = to_variable_dependency_format_string(
script=overrides.script, parsed_format_string=overrides.script._variables[override_name]
)
# pylint: enable=protected-access
return dummy_overrides
def _get_added_and_modified_variables(
plugins: PresetPlugins, downloader_options: MultiUrlValidator, output_options: OutputOptions
) -> Iterable[Tuple[OptionsValidator, Set[str], Set[str]]]:
"""
Iterates and returns the plugin options, added variables, modified variables
"""
options: List[OptionsValidator] = plugins.plugin_options
options.append(downloader_options)
options.append(output_options)
for plugin_options in options:
added_variables: Set[str] = set()
modified_variables: Set[str] = set()
for plugin_added_variables in plugin_options.added_variables(
unresolved_variables=set(),
).values():
added_variables |= set(plugin_added_variables)
for plugin_modified_variables in plugin_options.modified_variables().values():
modified_variables = plugin_modified_variables
yield plugin_options, added_variables, modified_variables
def _override_variables(overrides: Overrides) -> Set[str]:
return set(list(overrides.initial_variables().keys()))
class VariableValidation:
def __init__(
self,
overrides: Overrides,
downloader_options: MultiUrlValidator,
output_options: OutputOptions,
plugins: PresetPlugins,
):
self.overrides = overrides
self.downloader_options = downloader_options
self.output_options = output_options
self.plugins = plugins
self.script: Optional[Script] = None
self.resolved_variables: Set[str] = set()
self.unresolved_variables: Set[str] = set()
def initialize_preset_overrides(self, overrides: Overrides) -> "VariableValidation":
"""
Do some gymnastics to initialize the Overrides script.
"""
override_variables = set(list(overrides.initial_variables().keys()))
# Set resolved variables as all entry + override variables
# at this point to generate every possible added/modified variable
self.resolved_variables = set(_DUMMY_ENTRY_VARIABLES.keys()) | override_variables
plugin_variables: Set[str] = set()
for (
plugin_options,
added_variables,
modified_variables,
) in _get_added_and_modified_variables(
plugins=self.plugins,
downloader_options=self.downloader_options,
output_options=self.output_options,
):
for added_variable in added_variables:
if not overrides.ensure_added_plugin_variable_valid(added_variable=added_variable):
# pylint: disable=protected-access
raise plugin_options._validation_exception(
f"Cannot use the variable name {added_variable} because it exists as a"
" built-in ytdl-sub variable name."
)
# pylint: enable=protected-access
# Set unresolved as variables that are added but do not exist as
# entry/override variables since they are created at run-time
self.unresolved_variables |= added_variables | modified_variables
plugin_variables |= added_variables | modified_variables
# Then update resolved variables to reflect that
self.resolved_variables -= self.unresolved_variables
# Initialize overrides with unresolved variables + modified variables to throw an error.
# For modified variables, this is to prevent a resolve(update=True) to setting any
# dependencies until it has been explicitly added
overrides = overrides.initialize_script(unresolved_variables=self.unresolved_variables)
# copy the script and mock entry variables
self.script = copy.deepcopy(overrides.script)
self.script.add(
variables=_add_dummy_overrides(overrides=overrides)
| _add_dummy_variables(variables=plugin_variables)
| _DUMMY_ENTRY_VARIABLES
self.script = self.overrides.script
self.unresolved_variables = self.plugins.get_all_variables(
additional_options=[self.output_options, self.downloader_options]
)
return self
def _update_script(self) -> None:
_ = self.script.resolve(unresolvable=self.unresolved_variables, update=True)
def _add_subscription_override_variables(self) -> None:
"""
Add dummy subscription variables for script validation
"""
self.resolved_variables |= REQUIRED_OVERRIDE_VARIABLE_NAMES
def _add_variables(self, plugin_op: PluginOperation, options: OptionsValidator) -> None:
"""
Add dummy variables for script validation
@ -171,19 +37,17 @@ class VariableValidation:
).get(plugin_op, set())
modified_variables = options.modified_variables().get(plugin_op, set())
resolved_variables = added_variables | modified_variables
self.unresolved_variables -= added_variables | modified_variables
self.resolved_variables |= resolved_variables
self.unresolved_variables -= resolved_variables
def ensure_proper_usage(self) -> None:
def ensure_proper_usage(self) -> Dict:
"""
Validate variables resolve as plugins are executed, and return
a mock script which contains actualized added variables from the plugins
"""
resolved_subscription: Dict = {}
self._add_variables(PluginOperation.DOWNLOADER, options=self.downloader_options)
self._add_subscription_override_variables()
# Always add output options first
self._add_variables(PluginOperation.MODIFY_ENTRY_METADATA, options=self.output_options)
@ -200,16 +64,28 @@ class VariableValidation:
self._add_variables(PluginOperation.MODIFY_ENTRY, options=plugin_options)
# Validate that any formatter in the plugin options can resolve
validate_formatters(
resolved_subscription |= validate_formatters(
script=self.script,
unresolved_variables=self.unresolved_variables,
validator=plugin_options,
)
validate_formatters(
resolved_subscription |= validate_formatters(
script=self.script,
unresolved_variables=self.unresolved_variables,
validator=self.output_options,
)
# TODO: make this a function
raw_download_output = validate_formatters(
script=self.script,
unresolved_variables=self.unresolved_variables,
validator=self.downloader_options.urls,
)
resolved_subscription["download"] = []
for url_output in raw_download_output["download"]:
if url_output["url"]:
resolved_subscription["download"].append(url_output)
assert not self.unresolved_variables
return resolved_subscription

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

@ -21,7 +21,7 @@ class UrlThumbnailValidator(StrictDictValidator):
def __init__(self, name, value):
super().__init__(name, value)
self._name = self._validate_key(key="name", validator=StringFormatterValidator)
self._thumb_name = self._validate_key(key="name", validator=StringFormatterValidator)
self._uid = self._validate_key(key="uid", validator=OverridesStringFormatterValidator)
@property
@ -29,7 +29,7 @@ class UrlThumbnailValidator(StrictDictValidator):
"""
File name for the thumbnail
"""
return self._name
return self._thumb_name
@property
def uid(self) -> OverridesStringFormatterValidator:

5
src/ytdl_sub/script/parser.py
View file

@ -31,6 +31,8 @@ from ytdl_sub.script.utils.exceptions import InvalidSyntaxException
from ytdl_sub.script.utils.exceptions import InvalidVariableName
from ytdl_sub.script.utils.exceptions import UserException
from ytdl_sub.script.utils.exceptions import VariableDoesNotExist
from ytdl_sub.script.utils.name_validation import is_function
from ytdl_sub.script.utils.name_validation import to_function_name
from ytdl_sub.script.utils.name_validation import validate_variable_name
# pylint: disable=invalid-name
@ -144,6 +146,9 @@ class _Parser:
):
self._text = text
self._name = name
if name and is_function(name):
self._name = to_function_name(name)
self._custom_function_names = custom_function_names
self._variable_names = variable_names
self._pos = 0

121
src/ytdl_sub/script/script.py
View file

@ -20,28 +20,13 @@ from ytdl_sub.script.utils.exceptions import IncompatibleFunctionArguments
from ytdl_sub.script.utils.exceptions import InvalidCustomFunctionArguments
from ytdl_sub.script.utils.exceptions import RuntimeException
from ytdl_sub.script.utils.exceptions import ScriptVariableNotResolved
from ytdl_sub.script.utils.name_validation import is_function
from ytdl_sub.script.utils.name_validation import to_function_definition_name
from ytdl_sub.script.utils.name_validation import to_function_name
from ytdl_sub.script.utils.name_validation import validate_variable_name
from ytdl_sub.script.utils.type_checking import FunctionSpec
def _is_function(override_name: str):
return override_name.startswith("%")
def _function_name(function_key: str) -> str:
"""
Drop the % in %custom_function
"""
return function_key[1:]
def _to_function_definition_name(function_key: str) -> str:
"""
Add % in %custom_function
"""
return f"%{function_key}"
class Script:
"""
Takes a dictionary of both
@ -241,23 +226,23 @@ class Script:
def __init__(self, script: Dict[str, str]):
function_names: Set[str] = {
_function_name(name) for name in script.keys() if _is_function(name)
to_function_name(name) for name in script.keys() if is_function(name)
}
variable_names: Set[str] = {
validate_variable_name(name) for name in script.keys() if not _is_function(name)
validate_variable_name(name) for name in script.keys() if not is_function(name)
}
self._functions: Dict[str, SyntaxTree] = {
# custom_function_name must be passed to properly type custom function
# arguments uniquely if they're nested (i.e. $0 to $custom_func___0)
_function_name(function_key): parse(
to_function_name(function_key): parse(
text=function_value,
name=_function_name(function_key),
name=to_function_name(function_key),
custom_function_names=function_names,
variable_names=variable_names,
)
for function_key, function_value in script.items()
if _is_function(function_key)
if is_function(function_key)
}
self._variables: Dict[str, SyntaxTree] = {
@ -268,7 +253,7 @@ class Script:
variable_names=variable_names,
)
for variable_key, variable_value in script.items()
if not _is_function(variable_key)
if not is_function(variable_key)
}
self._validate()
@ -485,12 +470,12 @@ class Script:
added_variables_to_validate: Set[str] = set()
functions_to_add = {
_function_name(name): definition
to_function_name(name): definition
for name, definition in variables.items()
if _is_function(name)
if is_function(name)
}
variables_to_add = {
name: definition for name, definition in variables.items() if not _is_function(name)
name: definition for name, definition in variables.items() if not is_function(name)
}
custom_function_names = set(self._functions.keys()) | functions_to_add.keys()
@ -520,6 +505,46 @@ class Script:
return self
def add_parsed(self, variables: Dict[str, SyntaxTree]) -> "Script":
"""
Adds already parsed, new variables to the script.
Parameters
----------
variables
Mapping containing variable name to definition.
Returns
-------
Script
self
"""
added_variables_to_validate: Set[str] = set()
functions_to_add = {
to_function_name(name): definition
for name, definition in variables.items()
if is_function(name)
}
variables_to_add = {
name: definition for name, definition in variables.items() if not is_function(name)
}
for definitions in [functions_to_add, variables_to_add]:
for name, parsed in definitions.items():
if parsed.maybe_resolvable is None:
added_variables_to_validate.add(name)
if name in functions_to_add:
self._functions[name] = parsed
else:
self._variables[name] = parsed
if added_variables_to_validate:
self._validate(added_variables=added_variables_to_validate)
return self
def resolve_once(
self,
variable_definitions: Dict[str, str],
@ -560,6 +585,46 @@ class Script:
for name in variable_definitions.keys():
self._variables.pop(name, None)
def resolve_once_parsed(
self,
variable_definitions: Dict[str, SyntaxTree],
resolved: Optional[Dict[str, Resolvable]] = None,
unresolvable: Optional[Set[str]] = None,
update: bool = False,
) -> Dict[str, Resolvable]:
"""
Given a new set of variable definitions, resolve them using the Script, but do not
add them to the Script itself.
Parameters
----------
variable_definitions
Variables to resolve, but not store in the Script
resolved
Optional. Pre-resolved variables that should be used instead of what is in the script.
unresolvable
Optional. Unresolvable variables that will be ignored in resolution, including all
variables with a dependency to them.
update
Whether to update the script's state with resolved variables. Defaults to False.
Returns
-------
Dict[str, Resolvable]
Dict containing the variable names to their resolved values.
"""
try:
self.add_parsed(variable_definitions)
return self._resolve(
pre_resolved=resolved,
unresolvable=unresolvable,
output_filter=set(list(variable_definitions.keys())),
update=update,
).output
finally:
for name in variable_definitions.keys():
self._variables.pop(name, None)
def get(self, variable_name: str) -> Resolvable:
"""
Parameters
@ -605,4 +670,4 @@ class Script:
Set[str]
Names of all functions within the Script.
"""
return set(_to_function_definition_name(name) for name in self._functions.keys())
return set(to_function_definition_name(name) for name in self._functions.keys())

21
src/ytdl_sub/script/utils/name_validation.py
View file

@ -58,3 +58,24 @@ def validate_custom_function_name(custom_function_name: str) -> None:
f"Custom function name '%{custom_function_name}' is invalid:"
" The name is used by a built-in function and cannot be overwritten."
)
def is_function(override_name: str):
"""
Whether the definition is a function or not.
"""
return override_name.startswith("%")
def to_function_name(function_key: str) -> str:
"""
Drop the % in %custom_function
"""
return function_key[1:]
def to_function_definition_name(function_key: str) -> str:
"""
Add % in %custom_function
"""
return f"%{function_key}"

24
src/ytdl_sub/subscriptions/base_subscription.py
View file

@ -8,12 +8,14 @@ from ytdl_sub.config.plugin.preset_plugins import PresetPlugins
from ytdl_sub.config.preset import Preset
from ytdl_sub.config.preset_options import OutputOptions
from ytdl_sub.config.preset_options import YTDLOptions
from ytdl_sub.config.validators.variable_validation import VariableValidation
from ytdl_sub.downloaders.url.validators import MultiUrlValidator
from ytdl_sub.entries.variables.override_variables import SubscriptionVariables
from ytdl_sub.utils.exceptions import SubscriptionPermissionError
from ytdl_sub.utils.file_handler import FileHandler
from ytdl_sub.utils.file_handler import FileHandlerTransactionLog
from ytdl_sub.utils.logger import Logger
from ytdl_sub.utils.yaml import dump_yaml
from ytdl_sub.ytdl_additions.enhanced_download_archive import EnhancedDownloadArchive
logger = Logger.get("subscription")
@ -76,6 +78,14 @@ class BaseSubscription(ABC):
}
)
# Validate after adding the subscription name
self._validated_dict = VariableValidation(
overrides=self.overrides,
downloader_options=self.downloader_options,
output_options=self.output_options,
plugins=self.plugins,
).ensure_proper_usage()
self._enhanced_download_archive: Optional[EnhancedDownloadArchive] = (
_initialize_download_archive(
output_options=self.output_options,
@ -88,9 +98,9 @@ class BaseSubscription(ABC):
# Add post-archive variables
self.overrides.add(
{
SubscriptionVariables.subscription_has_download_archive(): f"""{{
%bool({self.download_archive.num_entries > 0})
}}""",
SubscriptionVariables.subscription_has_download_archive(): (
f"{{%bool({self.download_archive.num_entries > 0})}}"
),
}
)
@ -245,3 +255,11 @@ class BaseSubscription(ABC):
Subscription in yaml format
"""
return self._preset_options.yaml
def resolved_yaml(self) -> str:
"""
Returns
-------
Human-readable, condensed YAML definition of the subscription.
"""
return dump_yaml(self._validated_dict)

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

@ -4,7 +4,6 @@ from typing import Any
from typing import Dict
from ytdl_sub.script.parser import parse
from ytdl_sub.script.script import _is_function
from ytdl_sub.script.types.array import UnresolvedArray
from ytdl_sub.script.types.function import BuiltInFunction
from ytdl_sub.script.types.function import Function
@ -15,8 +14,10 @@ from ytdl_sub.script.types.resolvable import Float
from ytdl_sub.script.types.resolvable import Integer
from ytdl_sub.script.types.resolvable import Lambda
from ytdl_sub.script.types.resolvable import String
from ytdl_sub.script.types.syntax_tree import SyntaxTree
from ytdl_sub.script.types.variable import Variable
from ytdl_sub.script.utils.exceptions import UNREACHABLE
from ytdl_sub.script.utils.name_validation import is_function
# pylint: disable=too-many-return-statements
@ -30,10 +31,26 @@ class ScriptUtils:
sanitized_variables = {
f"{name}_sanitized": f"{{%sanitize({name})}}"
for name in variables.keys()
if not _is_function(name)
if not is_function(name)
}
return dict(variables, **sanitized_variables)
@classmethod
def add_sanitized_parsed_variables(
cls, variables: Dict[str, SyntaxTree]
) -> Dict[str, SyntaxTree]:
"""
Helper to add sanitized variables to a Script
"""
sanitized_variables = {
f"{name}_sanitized": SyntaxTree(
ast=[BuiltInFunction(name="sanitize", args=[Variable(name)])]
)
for name in variables.keys()
if not is_function(name)
}
return variables | sanitized_variables
@classmethod
def to_script(cls, value: Any, sort_keys: bool = True) -> str:
"""

92
src/ytdl_sub/validators/string_formatter_validators.py
View file

@ -11,6 +11,7 @@ from ytdl_sub.script.types.syntax_tree import SyntaxTree
from ytdl_sub.script.utils.exceptions import RuntimeException
from ytdl_sub.script.utils.exceptions import ScriptVariableNotResolved
from ytdl_sub.script.utils.exceptions import UserException
from ytdl_sub.script.utils.exceptions import UserThrownRuntimeError
from ytdl_sub.utils.exceptions import StringFormattingVariableNotFoundException
from ytdl_sub.utils.script import ScriptUtils
from ytdl_sub.validators.validators import DictValidator
@ -19,6 +20,8 @@ from ytdl_sub.validators.validators import LiteralDictValidator
from ytdl_sub.validators.validators import StringValidator
from ytdl_sub.validators.validators import Validator
# pylint: disable=protected-access
class StringFormatterValidator(StringValidator):
"""
@ -51,7 +54,10 @@ class StringFormatterValidator(StringValidator):
def __init__(self, name, value: str):
super().__init__(name=name, value=value)
try:
_ = parse(str(value))
self._parsed = parse(
text=str(value),
name=self.leaf_name,
)
except UserException as exc:
raise self._validation_exception(exc) from exc
@ -65,6 +71,16 @@ class StringFormatterValidator(StringValidator):
"""
return self._value
@property
@final
def parsed(self) -> SyntaxTree:
"""
Returns
-------
The parsed format string.
"""
return self._parsed
def post_process(self, resolved: str) -> str:
"""
Returns
@ -167,9 +183,14 @@ class DictFormatterValidator(LiteralDictValidator):
@property
def dict_with_format_strings(self) -> Dict[str, str]:
"""Returns dict with the format strings themselves"""
"""Returns dict with the format strings themselves."""
return {key: string_formatter.format_string for key, string_formatter in self.dict.items()}
@property
def dict_with_parsed_format_strings(self) -> Dict[str, SyntaxTree]:
"""Returns dict with the parsed format strings."""
return {key: string_formatter.parsed for key, string_formatter in self.dict.items()}
class OverridesDictFormatterValidator(DictFormatterValidator):
"""
@ -199,10 +220,8 @@ def to_variable_dependency_format_string(script: Script, parsed_format_string: S
dummy_format_string = ""
for var in parsed_format_string.variables:
dummy_format_string += f"{{ {var.name} }}"
# pylint: disable=protected-access
for variable_dependency in script._variables[var.name].variables:
dummy_format_string += f"{{ {variable_dependency.name} }}"
# pylint: enable=protected-access
return dummy_format_string
@ -210,16 +229,15 @@ def _validate_formatter(
mock_script: Script,
unresolved_variables: Set[str],
formatter_validator: Union[StringFormatterValidator, OverridesStringFormatterValidator],
) -> None:
is_static_formatter = False
unresolvable = unresolved_variables
if isinstance(formatter_validator, OverridesStringFormatterValidator):
is_static_formatter = True
unresolvable = unresolved_variables.union({VARIABLES.entry_metadata.variable_name})
) -> str:
parsed = formatter_validator.parsed
if resolved := parsed.maybe_resolvable:
return resolved.native
is_static_formatter = isinstance(formatter_validator, OverridesStringFormatterValidator)
if is_static_formatter:
unresolved_variables = unresolved_variables.union({VARIABLES.entry_metadata.variable_name})
parsed = parse(
text=formatter_validator.format_string,
)
variable_names = {var.name for var in parsed.variables}
custom_function_names = {f"%{func.name}" for func in parsed.custom_functions}
@ -233,21 +251,20 @@ def _validate_formatter(
"contains the following custom functions that do not exist: "
f"{', '.join(sorted(custom_function_names - mock_script.function_names))}"
)
if unresolved := variable_names.intersection(unresolvable):
if unresolved := variable_names.intersection(unresolved_variables):
raise StringFormattingVariableNotFoundException(
"contains the following variables that are unresolved when executing this "
f"formatter: {', '.join(sorted(unresolved))}"
)
try:
mock_script.resolve_once(
{
"tmp_var": to_variable_dependency_format_string(
script=mock_script, parsed_format_string=parsed
)
},
unresolvable=unresolvable,
update=True,
)
if is_static_formatter:
return mock_script.resolve_once_parsed(
{"tmp_var": formatter_validator.parsed},
unresolvable=unresolved_variables,
update=True,
)["tmp_var"].native
return formatter_validator.format_string
except RuntimeException as exc:
if isinstance(exc, ScriptVariableNotResolved) and is_static_formatter:
raise StringFormattingVariableNotFoundException(
@ -255,45 +272,60 @@ def _validate_formatter(
"entry variables"
) from exc
raise StringFormattingVariableNotFoundException(exc) from exc
except UserThrownRuntimeError as exc:
# Errors are expected for non-static formatters due to missing entry
# data. Raise otherwise.
if not is_static_formatter:
return formatter_validator.format_string
raise exc
def validate_formatters(
script: Script,
unresolved_variables: Set[str],
validator: Validator,
) -> None:
) -> Dict:
"""
Ensure all OverridesStringFormatterValidator's only contain variables from the overrides
and resolve.
"""
resolved_dict: Dict = {}
if isinstance(validator, DictValidator):
# pylint: disable=protected-access
resolved_dict[validator.leaf_name] = {}
# 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():
validate_formatters(
resolved_dict[validator.leaf_name] |= validate_formatters(
script=script,
unresolved_variables=unresolved_variables,
validator=validator_value,
)
# pylint: enable=protected-access
elif isinstance(validator, ListValidator):
resolved_dict[validator.leaf_name] = []
for list_value in validator.list:
validate_formatters(
list_output = validate_formatters(
script=script,
unresolved_variables=unresolved_variables,
validator=list_value,
)
assert len(list_output) == 1
resolved_dict[validator.leaf_name].append(list(list_output.values())[0])
elif isinstance(validator, (StringFormatterValidator, OverridesStringFormatterValidator)):
_validate_formatter(
resolved_dict[validator.leaf_name] = _validate_formatter(
mock_script=script,
unresolved_variables=unresolved_variables,
formatter_validator=validator,
)
elif isinstance(validator, (DictFormatterValidator, OverridesDictFormatterValidator)):
resolved_dict[validator.leaf_name] = {}
for validator_value in validator.dict.values():
_validate_formatter(
resolved_dict[validator.leaf_name] |= _validate_formatter(
mock_script=script,
unresolved_variables=unresolved_variables,
formatter_validator=validator_value,
)
else:
resolved_dict[validator.leaf_name] = validator._value
return resolved_dict

16
src/ytdl_sub/validators/validators.py
View file

@ -95,21 +95,11 @@ class Validator(ABC):
@final
@property
def _root_name(self) -> str:
def leaf_name(self) -> str:
"""
Returns
-------
"first" from the first.element.of.the.name
"""
return self._name.split(".")[0]
@final
@property
def _leaf_name(self) -> str:
"""
Returns
-------
"first" from the first.element.of.the.name
"name" from the first.element.of.the.name
"""
return self._name.split(".")[-1]
@ -274,7 +264,7 @@ class DictValidator(Validator):
value=self._dict.get(key, default),
)
self.__validator_dict[validator_name] = validator_instance
self.__validator_dict[key] = validator_instance
return validator_instance
@final

142
tests/unit/config/test_preset.py
View file

@ -6,7 +6,6 @@ import pytest
from ytdl_sub.config.preset import Preset
from ytdl_sub.plugins.nfo_tags import NfoTagsOptions
from ytdl_sub.utils.exceptions import StringFormattingVariableNotFoundException
from ytdl_sub.utils.exceptions import ValidationException
@ -157,100 +156,6 @@ class TestPreset:
},
)
def test_preset_error__source_variable_does_not_exist(
self, config_file, output_options, youtube_video
):
with pytest.raises(
StringFormattingVariableNotFoundException,
match="contains the following variables that do not exist: dne_var",
):
_ = Preset(
config=config_file,
name="test",
value={
"download": youtube_video,
"output_options": {"output_directory": "dir", "file_name": "{dne_var}"},
},
)
def test_preset_error__override_variable_does_not_exist(
self, config_file, output_options, youtube_video
):
with pytest.raises(
StringFormattingVariableNotFoundException,
match="contains the following variables that do not exist: dne_var",
):
_ = Preset(
config=config_file,
name="test",
value={
"download": youtube_video,
"output_options": {"output_directory": "{dne_var}", "file_name": "file"},
},
)
def test_preset_error__dict_source_variable_does_not_exist(
self, config_file, output_options, youtube_video
):
with pytest.raises(
StringFormattingVariableNotFoundException,
match="contains the following variables that do not exist: dne_var",
):
_ = Preset(
config=config_file,
name="test",
value={
"download": youtube_video,
"output_options": {"output_directory": "dir", "file_name": "file"},
"nfo_tags": {
"nfo_name": "the nfo name",
"nfo_root": "the root",
"tags": {"tag_a": "{dne_var}"},
},
},
)
def test_preset_error__dict_override_variable_does_not_exist(
self, config_file, output_options, youtube_video
):
with pytest.raises(
StringFormattingVariableNotFoundException,
match="contains the following variables that do not exist: dne_var",
):
_ = Preset(
config=config_file,
name="test",
value={
"download": youtube_video,
"output_options": output_options,
"output_directory_nfo_tags": {
"nfo_name": "the nfo name",
"nfo_root": "the root",
"tags": {"tag_a": "{dne_var}"},
},
},
)
def test_preset_error__dict_override_variable_not_static(
self, config_file, output_options, youtube_video
):
with pytest.raises(
StringFormattingVariableNotFoundException,
match="static formatters must contain variables that "
"have no dependency to entry variables",
):
_ = Preset(
config=config_file,
name="test",
value={
"download": youtube_video,
"output_options": {
"output_directory": "{title}",
"file_name": "{uid}",
},
},
)
def test_preset_with_multi_url__contains_empty_url(self, config_file, output_options):
_ = Preset(
config=config_file,
@ -412,50 +317,3 @@ class TestPreset:
"overrides": {name: "ack"},
},
)
def test_preset_error_added_url_variable_cannot_resolve(self, config_file, output_options):
with pytest.raises(
ValidationException,
match=re.escape(
"variable the_bad_one cannot use the variables subtitles_ext because it "
"depends on other variables that are computed later in execution"
),
):
_ = Preset(
config=config_file,
name="test",
value={
"download": {
"url": "youtube.com/watch?v=123abc",
"variables": {"the_bad_one": "{subtitles_ext}"},
},
"subtitles": {
"embed_subtitles": True,
},
"output_options": {"output_directory": "dir", "file_name": "acjk"},
},
)
def test_preset_error_override_name_conflicts_with_plugin(self, config_file, output_options):
with pytest.raises(
ValidationException,
match=re.escape(
"Override variable with name throttle_protection cannot be used since it is the "
"name of a plugin. Perhaps you meant to define it as a plugin? If so, indent it "
"left to make it at the same level as overrides."
),
):
_ = Preset(
config=config_file,
name="test",
value={
"download": {
"url": "youtube.com/watch?v=123abc",
},
"subtitles": {
"embed_subtitles": True,
},
"output_options": {"output_directory": "dir", "file_name": "acjk"},
"overrides": {"throttle_protection": "nope"},
},
)

61
tests/unit/config/test_subscription.py
View file

@ -6,6 +6,7 @@ from typing import Dict
from unittest.mock import patch
import pytest
import yaml
from ytdl_sub.config.config_file import ConfigFile
from ytdl_sub.plugins.nfo_tags import NfoTagsOptions
@ -540,9 +541,67 @@ def test_music_video_subscriptions(default_config: ConfigFile, music_video_subsc
assert gnr.get("url2").native == "https://www.youtube.com/watch?v=OldpIhHPsbs"
def test_default_docker_config_and_subscriptions(docker_default_subscription_path: Path):
def test_default_docker_config_and_subscriptions(
docker_default_subscription_path: Path, output_directory: str
):
default_config = ConfigFile.from_file_path("docker/root/defaults/config.yaml")
default_subs = Subscription.from_file_path(
config=default_config, subscription_path=docker_default_subscription_path
)
assert len(default_subs) == 1
resolved_yaml_as_json = yaml.safe_load(default_subs[0].resolved_yaml())
# Since this creates random values, ignore it for this test
assert "throttle_protection" in resolved_yaml_as_json
del resolved_yaml_as_json["throttle_protection"]
assert resolved_yaml_as_json == {
"chapters": {
"allow_chapters_from_comments": False,
"embed_chapters": True,
"enable": "True",
"force_key_frames": False,
},
"date_range": {"breaks": "True", "enable": "True", "type": "upload_date"},
"download": [
{
"download_reverse": "True",
"include_sibling_metadata": False,
"playlist_thumbnails": [
{"name": "{avatar_uncropped_thumbnail_file_name}", "uid": "avatar_uncropped"},
{"name": "{banner_uncropped_thumbnail_file_name}", "uid": "banner_uncropped"},
],
"source_thumbnails": [
{"name": "{avatar_uncropped_thumbnail_file_name}", "uid": "avatar_uncropped"},
{"name": "{banner_uncropped_thumbnail_file_name}", "uid": "banner_uncropped"},
],
"url": "https://www.youtube.com/@novapbs",
"variables": {},
"webpage_url": "{modified_webpage_url}",
"ytdl_options": {},
}
],
"file_convert": {"convert_to": "mp4", "convert_with": "yt-dlp", "enable": "True"},
"format": "(bv*[ext=mp4][vcodec~='^((he|a)vc|h26[45])']+ba[ext=m4a]) / (bv[ext=mp4]*+ba[ext=m4a]/b)",
"output_options": {
"download_archive_name": ".ytdl-sub-NOVA PBS-download-archive.json",
"file_name": "{episode_file_path}.{ext}",
"info_json_name": "{episode_file_path}.{info_json_ext}",
"keep_files_date_eval": "{episode_date_standardized}",
"maintain_download_archive": True,
"output_directory": f"{output_directory}/NOVA PBS",
"preserve_mtime": False,
"thumbnail_name": "{thumbnail_file_name}",
},
"video_tags": {
"contentRating": "{episode_content_rating}",
"date": "{episode_date_standardized}",
"episode_id": "{episode_number}",
"genre": "{tv_show_genre}",
"show": "{tv_show_name}",
"synopsis": "{episode_plot}",
"title": "{episode_title}",
"year": "{episode_year}",
},
}

172
tests/unit/config/test_subscription_validation.py Normal file
View file

@ -0,0 +1,172 @@
import re
import pytest
from ytdl_sub.config.preset import Preset
from ytdl_sub.subscriptions.subscription import Subscription
from ytdl_sub.utils.exceptions import StringFormattingVariableNotFoundException
from ytdl_sub.utils.exceptions import ValidationException
class TestSubscriptionValidation:
def test_preset_error__source_variable_does_not_exist(
self, config_file, output_options, youtube_video
):
with pytest.raises(
StringFormattingVariableNotFoundException,
match="contains the following variables that do not exist: dne_var",
):
_ = Subscription.from_preset(
preset=Preset(
config=config_file,
name="test",
value={
"download": youtube_video,
"output_options": {"output_directory": "dir", "file_name": "{dne_var}"},
},
),
config=config_file,
)
def test_preset_error__override_variable_does_not_exist(
self, config_file, output_options, youtube_video
):
with pytest.raises(
StringFormattingVariableNotFoundException,
match="contains the following variables that do not exist: dne_var",
):
_ = Subscription.from_preset(
preset=Preset(
config=config_file,
name="test",
value={
"download": youtube_video,
"output_options": {"output_directory": "{dne_var}", "file_name": "file"},
},
),
config=config_file,
)
def test_preset_error__dict_source_variable_does_not_exist(
self, config_file, output_options, youtube_video
):
with pytest.raises(
StringFormattingVariableNotFoundException,
match="contains the following variables that do not exist: dne_var",
):
_ = Subscription.from_preset(
preset=Preset(
config=config_file,
name="test",
value={
"download": youtube_video,
"output_options": {"output_directory": "dir", "file_name": "file"},
"nfo_tags": {
"nfo_name": "the nfo name",
"nfo_root": "the root",
"tags": {"tag_a": "{dne_var}"},
},
},
),
config=config_file,
)
def test_preset_error__dict_override_variable_does_not_exist(
self, config_file, output_options, youtube_video
):
with pytest.raises(
StringFormattingVariableNotFoundException,
match="contains the following variables that do not exist: dne_var",
):
_ = Subscription.from_preset(
preset=Preset(
config=config_file,
name="test",
value={
"download": youtube_video,
"output_options": output_options,
"output_directory_nfo_tags": {
"nfo_name": "the nfo name",
"nfo_root": "the root",
"tags": {"tag_a": "{dne_var}"},
},
},
),
config=config_file,
)
def test_preset_error__dict_override_variable_not_static(
self, config_file, output_options, youtube_video
):
with pytest.raises(
StringFormattingVariableNotFoundException,
match="static formatters must contain variables that "
"have no dependency to entry variables",
):
_ = Subscription.from_preset(
preset=Preset(
config=config_file,
name="test",
value={
"download": youtube_video,
"output_options": {
"output_directory": "{title}",
"file_name": "{uid}",
},
},
),
config=config_file,
)
def test_preset_error_added_url_variable_cannot_resolve(self, config_file, output_options):
with pytest.raises(
ValidationException,
match=re.escape(
"variable the_bad_one cannot use the variables subtitles_ext because it "
"depends on other variables that are computed later in execution"
),
):
_ = Subscription.from_preset(
preset=Preset(
config=config_file,
name="test",
value={
"download": {
"url": "youtube.com/watch?v=123abc",
"variables": {"the_bad_one": "{subtitles_ext}"},
},
"subtitles": {
"embed_subtitles": True,
},
"output_options": {"output_directory": "dir", "file_name": "acjk"},
},
),
config=config_file,
)
def test_preset_error_override_name_conflicts_with_plugin(self, config_file, output_options):
with pytest.raises(
ValidationException,
match=re.escape(
"Override variable with name throttle_protection cannot be used since it is the "
"name of a plugin. Perhaps you meant to define it as a plugin? If so, indent it "
"left to make it at the same level as overrides."
),
):
_ = Subscription.from_preset(
preset=Preset(
config=config_file,
name="test",
value={
"download": {
"url": "youtube.com/watch?v=123abc",
},
"subtitles": {
"embed_subtitles": True,
},
"output_options": {"output_directory": "dir", "file_name": "acjk"},
"overrides": {"throttle_protection": "nope"},
},
),
config=config_file,
)

2
tools/docgen/plugins.py
View file

@ -37,10 +37,12 @@ def should_filter_property(property_name: str) -> bool:
"dict",
"keys",
"dict_with_format_strings",
"dict_with_parsed_format_strings",
"subscription_name",
"list",
"script",
"unresolvable",
"leaf_name",
)