201 lines
7.8 KiB
Python
201 lines
7.8 KiB
Python
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
|
|
from ytdl_sub.config.plugin.plugin_operation import PluginOperation
|
|
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.script.variable_definitions import VARIABLE_SCRIPTS
|
|
from ytdl_sub.entries.variables.override_variables import SubscriptionVariables
|
|
from ytdl_sub.script.script import Script
|
|
from ytdl_sub.validators.string_formatter_validators import validate_formatters
|
|
|
|
|
|
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):
|
|
dummy_overrides[override_name] = ""
|
|
# pylint: disable=protected-access
|
|
for variable_dependency in overrides.script._variables[override_name].variables:
|
|
dummy_overrides[override_name] += f"{{ {variable_dependency.name } }}"
|
|
# pylint: enable=protected-access
|
|
return dummy_overrides
|
|
|
|
|
|
def _get_added_and_modified_variables(
|
|
plugins: PresetPlugins, downloader_options: MultiUrlValidator
|
|
) -> 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)
|
|
|
|
for plugin_options in options:
|
|
added_variables: Set[str] = set()
|
|
modified_variables: Set[str] = set()
|
|
|
|
for plugin_added_variables in plugin_options.added_variables(
|
|
resolved_variables=set(),
|
|
unresolved_variables=set(),
|
|
plugin_op=PluginOperation.ANY,
|
|
).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())) | {
|
|
SubscriptionVariables.subscription_name()
|
|
}
|
|
|
|
|
|
def _entry_variables() -> Set[str]:
|
|
return set(list(VARIABLE_SCRIPTS.keys()))
|
|
|
|
|
|
class VariableValidation:
|
|
def __init__(
|
|
self,
|
|
downloader_options: MultiUrlValidator,
|
|
output_options: OutputOptions,
|
|
plugins: PresetPlugins,
|
|
):
|
|
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_overrides(
|
|
self, subscription_name: str, overrides: Overrides
|
|
) -> "VariableValidation":
|
|
"""
|
|
Do some gymnastics to initialize the Overrides script.
|
|
"""
|
|
entry_variables = _entry_variables()
|
|
override_variables = _override_variables(overrides)
|
|
|
|
# Set resolved variables as all entry + override variables
|
|
# at this point to generate every possible added/modified variable
|
|
self.resolved_variables = entry_variables | override_variables
|
|
|
|
for (
|
|
plugin_options,
|
|
added_variables,
|
|
modified_variables,
|
|
) in _get_added_and_modified_variables(
|
|
plugins=self.plugins,
|
|
downloader_options=self.downloader_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
|
|
|
|
# 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(
|
|
subscription_name=subscription_name, unresolved_variables=self.unresolved_variables
|
|
)
|
|
|
|
# copy the script and mock entry variables
|
|
self.script = copy.deepcopy(overrides.script).add(_add_dummy_variables(entry_variables))
|
|
self.script.add(
|
|
variables=_add_dummy_overrides(overrides=overrides),
|
|
unresolvable=self.unresolved_variables,
|
|
)
|
|
|
|
return self
|
|
|
|
def _update_script(self) -> None:
|
|
_ = self.script.resolve(unresolvable=self.unresolved_variables, update=True)
|
|
|
|
def _add_variables(self, plugin_op: PluginOperation, options: OptionsValidator) -> Set[str]:
|
|
added_variables = options.added_variables(
|
|
resolved_variables=self.resolved_variables,
|
|
unresolved_variables=self.unresolved_variables,
|
|
plugin_op=plugin_op,
|
|
).get(plugin_op, set())
|
|
modified_variables = options.modified_variables().get(plugin_op, set())
|
|
|
|
resolved_variables = added_variables | modified_variables
|
|
|
|
self.script.add(_add_dummy_variables(resolved_variables))
|
|
self.resolved_variables |= resolved_variables
|
|
self.unresolved_variables -= resolved_variables
|
|
|
|
return added_variables
|
|
|
|
def ensure_proper_usage(self) -> None:
|
|
"""
|
|
Validate variables resolve as plugins are executed, and return
|
|
a mock script which contains actualized added variables from the plugins
|
|
"""
|
|
self._add_variables(PluginOperation.DOWNLOADER, options=self.downloader_options)
|
|
|
|
# Metadata variables to be added
|
|
for plugin_options in PluginMapping.order_options_by(
|
|
self.plugins.zipped(), PluginOperation.MODIFY_ENTRY_METADATA
|
|
):
|
|
self._add_variables(PluginOperation.MODIFY_ENTRY_METADATA, options=plugin_options)
|
|
|
|
self._update_script()
|
|
for plugin_options in PluginMapping.order_options_by(
|
|
self.plugins.zipped(), PluginOperation.MODIFY_ENTRY
|
|
):
|
|
added = self._add_variables(PluginOperation.MODIFY_ENTRY, options=plugin_options)
|
|
if added:
|
|
self._update_script()
|
|
|
|
# Validate that any formatter in the plugin options can resolve
|
|
validate_formatters(
|
|
script=self.script,
|
|
unresolved_variables=self.unresolved_variables,
|
|
validator=plugin_options,
|
|
)
|
|
|
|
validate_formatters(
|
|
script=self.script,
|
|
unresolved_variables=self.unresolved_variables,
|
|
validator=self.output_options,
|
|
)
|
|
|
|
assert not self.unresolved_variables
|