ugh
This commit is contained in:
parent
97949b3e13
commit
1d465a7d69
6 changed files with 145 additions and 182 deletions
|
|
@ -31,6 +31,7 @@ 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
|
||||
|
|
@ -182,9 +183,9 @@ class Preset(_PresetShell):
|
|||
def _script_builder(self) -> ScriptBuilder:
|
||||
# Set the formatter variables to be the overrides
|
||||
script = ScriptBuilder(
|
||||
Scriptable.add_sanitized_variables(self.overrides.dict_with_format_strings)
|
||||
ScriptUtils.add_sanitized_variables(self.overrides.dict_with_format_strings)
|
||||
)
|
||||
script.add_resolved(Scriptable.add_dummy_variables(self._source_variables))
|
||||
script.add_resolved(ScriptUtils.add_dummy_variables(self._source_variables))
|
||||
return script
|
||||
|
||||
@functools.cached_property
|
||||
|
|
@ -193,7 +194,7 @@ class Preset(_PresetShell):
|
|||
Contains actualized script which should hold all Override variables
|
||||
"""
|
||||
return self._script_builder.add_resolved(
|
||||
Scriptable.add_dummy_variables(self._added_variables)
|
||||
ScriptUtils.add_dummy_variables(self._added_variables)
|
||||
).partial_build()
|
||||
|
||||
def __validate_and_get_plugins(self) -> PresetPlugins:
|
||||
|
|
@ -214,7 +215,7 @@ class Preset(_PresetShell):
|
|||
script_builder = copy.deepcopy(self._script_builder)
|
||||
self.downloader_options.validate_with_variables(script=script_builder.partial_build())
|
||||
script_builder.add_resolved(
|
||||
Scriptable.add_dummy_variables(self.downloader_options.added_source_variables())
|
||||
ScriptUtils.add_dummy_variables(self.downloader_options.added_source_variables())
|
||||
)
|
||||
|
||||
for _, plugin_options in sorted(
|
||||
|
|
@ -223,7 +224,7 @@ class Preset(_PresetShell):
|
|||
# Validate current plugin using source + added plugin variables
|
||||
plugin_options.validate_with_variables(script=script_builder.partial_build())
|
||||
script_builder.add_resolved(
|
||||
Scriptable.add_dummy_variables(self.downloader_options.added_source_variables())
|
||||
ScriptUtils.add_dummy_variables(self.downloader_options.added_source_variables())
|
||||
)
|
||||
|
||||
@functools.cache
|
||||
|
|
|
|||
|
|
@ -1,52 +1,23 @@
|
|||
import math
|
||||
from typing import Any
|
||||
from typing import Dict
|
||||
from typing import List
|
||||
from typing import Optional
|
||||
|
||||
import mergedeep
|
||||
|
||||
from ytdl_sub.entries.base_entry import BaseEntry
|
||||
from ytdl_sub.entries.base_entry import TBaseEntry
|
||||
from ytdl_sub.entries.entry import Entry
|
||||
from ytdl_sub.entries.variables.kwargs import DESCRIPTION
|
||||
from ytdl_sub.entries.variables.kwargs import PLAYLIST_COUNT
|
||||
from ytdl_sub.entries.variables.kwargs import PLAYLIST_DESCRIPTION
|
||||
from ytdl_sub.entries.variables.kwargs import PLAYLIST_ENTRY
|
||||
from ytdl_sub.entries.variables.kwargs import PLAYLIST_INDEX
|
||||
from ytdl_sub.entries.variables.kwargs import PLAYLIST_MAX_UPLOAD_YEAR
|
||||
from ytdl_sub.entries.variables.kwargs import PLAYLIST_MAX_UPLOAD_YEAR_TRUNCATED
|
||||
from ytdl_sub.entries.variables.kwargs import PLAYLIST_TITLE
|
||||
from ytdl_sub.entries.variables.kwargs import PLAYLIST_UID
|
||||
from ytdl_sub.entries.variables.kwargs import PLAYLIST_UPLOADER
|
||||
from ytdl_sub.entries.variables.kwargs import PLAYLIST_UPLOADER_ID
|
||||
from ytdl_sub.entries.variables.kwargs import PLAYLIST_UPLOADER_URL
|
||||
from ytdl_sub.entries.variables.kwargs import PLAYLIST_WEBPAGE_URL
|
||||
from ytdl_sub.entries.variables.kwargs import SOURCE_COUNT
|
||||
from ytdl_sub.entries.variables.kwargs import SOURCE_DESCRIPTION
|
||||
from ytdl_sub.entries.variables.kwargs import SOURCE_ENTRY
|
||||
from ytdl_sub.entries.variables.kwargs import SOURCE_INDEX
|
||||
from ytdl_sub.entries.variables.kwargs import SOURCE_TITLE
|
||||
from ytdl_sub.entries.variables.kwargs import SOURCE_UID
|
||||
from ytdl_sub.entries.variables.kwargs import SOURCE_UPLOADER
|
||||
from ytdl_sub.entries.variables.kwargs import SOURCE_UPLOADER_ID
|
||||
from ytdl_sub.entries.variables.kwargs import SOURCE_UPLOADER_URL
|
||||
from ytdl_sub.entries.variables.kwargs import SOURCE_WEBPAGE_URL
|
||||
from ytdl_sub.entries.variables.kwargs import TITLE
|
||||
from ytdl_sub.entries.variables.kwargs import UID
|
||||
from ytdl_sub.entries.variables.kwargs import UPLOADER
|
||||
from ytdl_sub.entries.variables.kwargs import UPLOADER_ID
|
||||
from ytdl_sub.entries.variables.kwargs import UPLOADER_URL
|
||||
from ytdl_sub.entries.variables.kwargs import WEBPAGE_URL
|
||||
|
||||
|
||||
class ParentType:
|
||||
PLAYLIST = "playlist"
|
||||
SOURCE = "source"
|
||||
from ytdl_sub.entries.script.variable_definitions import VARIABLES as v
|
||||
from ytdl_sub.entries.script.variable_definitions import MetadataVariable
|
||||
from ytdl_sub.entries.script.variable_scripts import ENTRY_DEFAULT_VARIABLES
|
||||
from ytdl_sub.entries.script.variable_scripts import ENTRY_REQUIRED_VARIABLES
|
||||
|
||||
|
||||
def _sort_entries(entries: List[TBaseEntry]) -> List[TBaseEntry]:
|
||||
"""Try sorting by playlist_id first, then fall back to uid"""
|
||||
return sorted(entries, key=lambda ent: (ent.kwargs_get(PLAYLIST_INDEX, math.inf), ent.uid))
|
||||
return sorted(
|
||||
entries, key=lambda ent: (ent.kwargs_get(v.playlist_index.metadata_key, math.inf), ent.uid)
|
||||
)
|
||||
|
||||
|
||||
class EntryParent(BaseEntry):
|
||||
|
|
@ -73,86 +44,38 @@ class EntryParent(BaseEntry):
|
|||
self.entry_children()
|
||||
)
|
||||
|
||||
def _playlist_variables(self, idx: int, children: List[TBaseEntry], parent_type: str) -> Dict:
|
||||
_count = self.kwargs_get(PLAYLIST_COUNT, len(children))
|
||||
_index = children[idx].kwargs_get(PLAYLIST_INDEX, idx + 1)
|
||||
|
||||
if parent_type == ParentType.SOURCE:
|
||||
return {SOURCE_INDEX: _index, SOURCE_COUNT: _count}
|
||||
return {
|
||||
SOURCE_INDEX: self.kwargs_get(SOURCE_INDEX, 1),
|
||||
SOURCE_COUNT: self.kwargs_get(SOURCE_INDEX, 1),
|
||||
PLAYLIST_INDEX: _index,
|
||||
PLAYLIST_COUNT: _count,
|
||||
}
|
||||
|
||||
def _parent_variables(self, parent_type: str) -> Dict:
|
||||
def _(source_key: str, playlist_key: str) -> str:
|
||||
return playlist_key if parent_type == ParentType.PLAYLIST else source_key
|
||||
|
||||
def __(key: str) -> Optional[str]:
|
||||
return self.kwargs_get(key=key)
|
||||
|
||||
return {
|
||||
_(SOURCE_ENTRY, PLAYLIST_ENTRY): self._kwargs,
|
||||
_(SOURCE_TITLE, PLAYLIST_TITLE): __(TITLE),
|
||||
_(SOURCE_WEBPAGE_URL, PLAYLIST_WEBPAGE_URL): __(WEBPAGE_URL),
|
||||
_(SOURCE_UID, PLAYLIST_UID): __(UID),
|
||||
_(SOURCE_DESCRIPTION, PLAYLIST_DESCRIPTION): __(DESCRIPTION),
|
||||
_(SOURCE_UPLOADER, PLAYLIST_UPLOADER): __(UPLOADER),
|
||||
_(SOURCE_UPLOADER_ID, PLAYLIST_UPLOADER_ID): __(UPLOADER_ID),
|
||||
_(SOURCE_UPLOADER_URL, PLAYLIST_UPLOADER_URL): __(UPLOADER_URL),
|
||||
}
|
||||
|
||||
def _get_entry_children_variable_list(self, variable_name: str) -> List[str | int]:
|
||||
return [getattr(entry_child, variable_name) for entry_child in self.entry_children()]
|
||||
|
||||
def _entry_aggregate_variables(self) -> Dict:
|
||||
if not self.entry_children():
|
||||
return {}
|
||||
|
||||
return {
|
||||
PLAYLIST_MAX_UPLOAD_YEAR: max(self._get_entry_children_variable_list("upload_year")),
|
||||
PLAYLIST_MAX_UPLOAD_YEAR_TRUNCATED: max(
|
||||
self._get_entry_children_variable_list("upload_year_truncated")
|
||||
),
|
||||
}
|
||||
|
||||
# pylint: disable=protected-access
|
||||
def _sibling_entry_metadata(self) -> List[Dict[str, Any]]:
|
||||
sibling_entry_metadata: List[Dict[str, Any]] = []
|
||||
variable_filter: List[MetadataVariable] = list(ENTRY_REQUIRED_VARIABLES.keys()) + list(
|
||||
ENTRY_DEFAULT_VARIABLES.keys()
|
||||
)
|
||||
for entry in self.entry_children():
|
||||
sibling_entry_metadata.append(
|
||||
{var.metadata_key: entry.kwargs_get(var.metadata_key) for var in variable_filter}
|
||||
)
|
||||
return sibling_entry_metadata
|
||||
|
||||
def _set_child_variables(self, parents: Optional[List["EntryParent"]] = None) -> "EntryParent":
|
||||
if parents is None:
|
||||
parents = [self]
|
||||
self.add_kwargs(
|
||||
self._playlist_variables(idx=0, children=parents, parent_type=ParentType.SOURCE)
|
||||
)
|
||||
|
||||
kwargs_to_add: Dict = {}
|
||||
kwargs_to_add: Dict[str, Any] = {
|
||||
v.sibling_entry_metadata.metadata_key: self._sibling_entry_metadata()
|
||||
}
|
||||
if len(parents) >= 1:
|
||||
mergedeep.merge(kwargs_to_add, parents[-1]._parent_variables(ParentType.PLAYLIST))
|
||||
kwargs_to_add[v.playlist_metadata.metadata_key] = parents[-1]._kwargs
|
||||
if len(parents) >= 2:
|
||||
mergedeep.merge(kwargs_to_add, parents[-2]._parent_variables(ParentType.SOURCE))
|
||||
kwargs_to_add[v.source_metadata.metadata_key] = parents[-2]._kwargs
|
||||
if len(parents) >= 3:
|
||||
raise ValueError(
|
||||
"ytdl-sub currently does support more than 3 layers of playlists/entries. "
|
||||
"If you encounter this error, please file a ticket with the URLs used."
|
||||
)
|
||||
|
||||
mergedeep.merge(kwargs_to_add, self._entry_aggregate_variables())
|
||||
for idx, entry_child in enumerate(self.entry_children()):
|
||||
entry_child.add_kwargs(
|
||||
self._playlist_variables(
|
||||
idx=idx, children=self.entry_children(), parent_type=ParentType.PLAYLIST
|
||||
)
|
||||
)
|
||||
for entry_child in self.entry_children():
|
||||
entry_child.add_kwargs(kwargs_to_add)
|
||||
|
||||
for idx, parent_child in enumerate(self.parent_children()):
|
||||
parent_child.add_kwargs(
|
||||
self._playlist_variables(
|
||||
idx=idx, children=self.parent_children(), parent_type=ParentType.SOURCE
|
||||
)
|
||||
)
|
||||
for parent_child in self.parent_children():
|
||||
parent_child._set_child_variables(parents=parents + [parent_child])
|
||||
|
||||
return self
|
||||
|
|
|
|||
|
|
@ -27,7 +27,12 @@ class MetadataVariable(Variable):
|
|||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class DerivedMetadata(MetadataVariable, Metadata):
|
||||
class RelativeMetadata(MetadataVariable, Metadata):
|
||||
pass
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class SiblingMetadata(MetadataVariable):
|
||||
pass
|
||||
|
||||
|
||||
|
|
@ -42,6 +47,18 @@ class _Variables:
|
|||
"""
|
||||
return Metadata("entry_metadata")
|
||||
|
||||
@property
|
||||
def playlist_metadata(self) -> RelativeMetadata:
|
||||
return RelativeMetadata("playlist_metadata", metadata_key="playlist_metadata")
|
||||
|
||||
@property
|
||||
def source_metadata(self) -> RelativeMetadata:
|
||||
return RelativeMetadata("source_metadata", metadata_key="source_metadata")
|
||||
|
||||
@property
|
||||
def sibling_entry_metadata(self) -> SiblingMetadata:
|
||||
return SiblingMetadata("sibling_entry_metadata", metadata_key="sibling_entry_metadata")
|
||||
|
||||
@property
|
||||
def uid(self) -> MetadataVariable:
|
||||
"""
|
||||
|
|
@ -223,7 +240,7 @@ class _Variables:
|
|||
Name of the source (i.e. channel with multiple playlists) if it exists, otherwise
|
||||
returns its playlist_title.
|
||||
"""
|
||||
return MetadataVariable("source_title", metadata_key="source_title")
|
||||
return MetadataVariable("source_title", metadata_key=self.title.metadata_key)
|
||||
|
||||
@property
|
||||
def source_title_sanitized(self) -> Variable:
|
||||
|
|
@ -243,7 +260,7 @@ class _Variables:
|
|||
str
|
||||
The source unique id if it exists, otherwise returns the playlist unique ID.
|
||||
"""
|
||||
return MetadataVariable("source_uid", metadata_key="source_uid")
|
||||
return MetadataVariable("source_uid", metadata_key=self.uid.metadata_key)
|
||||
|
||||
@property
|
||||
def source_index(self) -> MetadataVariable:
|
||||
|
|
@ -256,7 +273,7 @@ class _Variables:
|
|||
It is recommended to not use this unless you know the source will never add new content
|
||||
(it is easy for this value to change).
|
||||
"""
|
||||
return MetadataVariable("source_index", metadata_key="source_index")
|
||||
return MetadataVariable("source_index", metadata_key=self.playlist_index.metadata_key)
|
||||
|
||||
@property
|
||||
def source_index_padded(self) -> Variable:
|
||||
|
|
@ -276,7 +293,7 @@ class _Variables:
|
|||
int
|
||||
The source count if it exists, otherwise returns the playlist count.
|
||||
"""
|
||||
return MetadataVariable("source_count", metadata_key="source_count")
|
||||
return MetadataVariable("source_count", metadata_key=self.playlist_count.metadata_key)
|
||||
|
||||
@property
|
||||
def source_webpage_url(self) -> MetadataVariable:
|
||||
|
|
@ -286,7 +303,7 @@ class _Variables:
|
|||
str
|
||||
The source webpage url if it exists, otherwise returns the playlist webpage url.
|
||||
"""
|
||||
return MetadataVariable("source_webpage_url", metadata_key="source_webpage_url")
|
||||
return MetadataVariable("source_webpage_url", metadata_key=self.webpage_url.metadata_key)
|
||||
|
||||
@property
|
||||
def source_description(self) -> MetadataVariable:
|
||||
|
|
@ -296,7 +313,7 @@ class _Variables:
|
|||
str
|
||||
The source description if it exists, otherwise returns the playlist description.
|
||||
"""
|
||||
return MetadataVariable("source_description", metadata_key="source_description")
|
||||
return MetadataVariable("source_description", metadata_key=self.description.metadata_key)
|
||||
|
||||
@property
|
||||
def playlist_uid(self) -> MetadataVariable:
|
||||
|
|
@ -306,7 +323,7 @@ class _Variables:
|
|||
str
|
||||
The playlist unique ID if it exists, otherwise return Variable("")
|
||||
"""
|
||||
return MetadataVariable(variable_name="playlist_uid", metadata_key="playlist_uid")
|
||||
return MetadataVariable(variable_name="playlist_uid", metadata_key=self.uid.metadata_key)
|
||||
|
||||
@property
|
||||
def playlist_title(self) -> MetadataVariable:
|
||||
|
|
@ -316,7 +333,9 @@ class _Variables:
|
|||
str
|
||||
Name of its parent playlist/channel if it exists, otherwise returns its title.
|
||||
"""
|
||||
return MetadataVariable(variable_name="playlist_title", metadata_key="playlist_title")
|
||||
return MetadataVariable(
|
||||
variable_name="playlist_title", metadata_key=self.title.metadata_key
|
||||
)
|
||||
|
||||
@property
|
||||
def playlist_title_sanitized(self) -> Variable:
|
||||
|
|
@ -413,7 +432,7 @@ class _Variables:
|
|||
The playlist description if it exists, otherwise returns the entry's description.
|
||||
"""
|
||||
return MetadataVariable(
|
||||
variable_name="playlist_description", metadata_key="playlist_description"
|
||||
variable_name="playlist_description", metadata_key=self.description.metadata_key
|
||||
)
|
||||
|
||||
@property
|
||||
|
|
@ -425,7 +444,7 @@ class _Variables:
|
|||
The playlist webpage url if it exists. Otherwise, returns the entry webpage url.
|
||||
"""
|
||||
return MetadataVariable(
|
||||
variable_name="playlist_webpage_url", metadata_key="playlist_webpage_url"
|
||||
variable_name="playlist_webpage_url", metadata_key=self.webpage_url.metadata_key
|
||||
)
|
||||
|
||||
@property
|
||||
|
|
@ -461,7 +480,7 @@ class _Variables:
|
|||
str
|
||||
The playlist uploader id if it exists, otherwise returns the entry uploader ID.
|
||||
"""
|
||||
return MetadataVariable("playlist_uploader_id", metadata_key="playlist_uploader_id")
|
||||
return MetadataVariable("playlist_uploader_id", metadata_key=self.uploader_id.metadata_key)
|
||||
|
||||
@property
|
||||
def playlist_uploader(self) -> MetadataVariable:
|
||||
|
|
@ -471,7 +490,7 @@ class _Variables:
|
|||
str
|
||||
The playlist uploader if it exists, otherwise return Variable("")
|
||||
"""
|
||||
return MetadataVariable("playlist_uploader", metadata_key="playlist_uploader")
|
||||
return MetadataVariable("playlist_uploader", metadata_key=self.uploader.metadata_key)
|
||||
|
||||
@property
|
||||
def playlist_uploader_sanitized(self) -> Variable:
|
||||
|
|
@ -491,7 +510,9 @@ class _Variables:
|
|||
str
|
||||
The playlist uploader url if it exists, otherwise returns the playlist webpage_url.
|
||||
"""
|
||||
return MetadataVariable("playlist_uploader_url", metadata_key="playlist_uploader_url")
|
||||
return MetadataVariable(
|
||||
"playlist_uploader_url", metadata_key=self.uploader_url.metadata_key
|
||||
)
|
||||
|
||||
@property
|
||||
def source_uploader_id(self) -> MetadataVariable:
|
||||
|
|
@ -501,7 +522,7 @@ class _Variables:
|
|||
str
|
||||
The source uploader id if it exists, otherwise returns the playlist_uploader_id
|
||||
"""
|
||||
return MetadataVariable("source_uploader_id", metadata_key="source_uploader_id")
|
||||
return MetadataVariable("source_uploader_id", metadata_key=self.uploader_id.metadata_key)
|
||||
|
||||
@property
|
||||
def source_uploader(self) -> MetadataVariable:
|
||||
|
|
@ -511,7 +532,7 @@ class _Variables:
|
|||
str
|
||||
The source uploader if it exists, otherwise return Variable("")
|
||||
"""
|
||||
return MetadataVariable("source_uploader", metadata_key="source_uploader")
|
||||
return MetadataVariable("source_uploader", metadata_key=self.uploader.metadata_key)
|
||||
|
||||
@property
|
||||
def source_uploader_url(self) -> MetadataVariable:
|
||||
|
|
@ -521,7 +542,7 @@ class _Variables:
|
|||
str
|
||||
The source uploader url if it exists, otherwise returns the source webpage_url.
|
||||
"""
|
||||
return MetadataVariable("source_uploader_url", metadata_key="source_uploader_url")
|
||||
return MetadataVariable("source_uploader_url", metadata_key=self.uploader_url.metadata_key)
|
||||
|
||||
@property
|
||||
def creator(self) -> MetadataVariable:
|
||||
|
|
|
|||
|
|
@ -1,4 +1,5 @@
|
|||
from typing import Dict
|
||||
from typing import List
|
||||
from typing import Optional
|
||||
from typing import Set
|
||||
|
||||
|
|
@ -35,7 +36,7 @@ def date_metadata(date_key: Variable, metadata_key: str) -> str:
|
|||
|
||||
|
||||
def _get_internal(
|
||||
metadata: Metadata, key: MetadataVariable, default: Optional[Variable | str | int]
|
||||
metadata: Metadata, key: MetadataVariable, default: Optional[Variable | str | int | Dict | List]
|
||||
) -> str:
|
||||
if default is None:
|
||||
# TODO: assert with good error message if key DNE
|
||||
|
|
@ -44,6 +45,10 @@ def _get_internal(
|
|||
out = f"%map_get_non_empty({metadata.variable_name}, '{key.metadata_key}', {default.variable_name})"
|
||||
elif isinstance(default, str):
|
||||
out = f"%map_get_non_empty({metadata.variable_name}, '{key.metadata_key}', '{default}')"
|
||||
elif isinstance(default, dict):
|
||||
out = f"%map_get_non_empty({metadata.variable_name}, '{key.metadata_key}', {{}})"
|
||||
elif isinstance(default, list):
|
||||
out = f"%map_get_non_empty({metadata.variable_name}, '{key.metadata_key}', [])"
|
||||
else:
|
||||
out = f"%map_get_non_empty({metadata.variable_name}, '{key.metadata_key}', {default})"
|
||||
|
||||
|
|
@ -56,7 +61,9 @@ def _get_int(
|
|||
return f"{{%int({_get_internal(metadata=metadata, key=key, default=default)})}}"
|
||||
|
||||
|
||||
def _get(metadata: Metadata, key: MetadataVariable, default: Optional[Variable | str | int]) -> str:
|
||||
def _get(
|
||||
metadata: Metadata, key: MetadataVariable, default: Optional[Variable | str | int | Dict | List]
|
||||
) -> str:
|
||||
return f"{{{_get_internal(metadata=metadata, key=key, default=default)}}}"
|
||||
|
||||
|
||||
|
|
@ -64,7 +71,9 @@ def _get(metadata: Metadata, key: MetadataVariable, default: Optional[Variable |
|
|||
# Entry Getters
|
||||
|
||||
|
||||
def entry_get(key: MetadataVariable, default: Optional[Variable | str | int] = None) -> str:
|
||||
def entry_get(
|
||||
key: MetadataVariable, default: Optional[Variable | str | int | Dict | List] = None
|
||||
) -> str:
|
||||
return _get(metadata=v.entry_metadata, key=key, default=default)
|
||||
|
||||
|
||||
|
|
@ -77,11 +86,11 @@ def entry_get_int(key: MetadataVariable, default: Optional[Variable | int] = Non
|
|||
|
||||
|
||||
def playlist_get(key: MetadataVariable, default: Optional[Variable | str | int] = None) -> str:
|
||||
return _get(metadata=v.entry_metadata, key=key, default=default)
|
||||
return _get(metadata=v.playlist_metadata, key=key, default=default)
|
||||
|
||||
|
||||
def playlist_get_int(key: MetadataVariable, default: Optional[Variable | int] = None) -> str:
|
||||
return _get_int(metadata=v.entry_metadata, key=key, default=default)
|
||||
return _get_int(metadata=v.playlist_metadata, key=key, default=default)
|
||||
|
||||
|
||||
###############################################################################################
|
||||
|
|
@ -89,11 +98,11 @@ def playlist_get_int(key: MetadataVariable, default: Optional[Variable | int] =
|
|||
|
||||
|
||||
def source_get(key: MetadataVariable, default: Optional[Variable | str | int] = None) -> str:
|
||||
return _get(metadata=v.entry_metadata, key=key, default=default)
|
||||
return _get(metadata=v.source_metadata, key=key, default=default)
|
||||
|
||||
|
||||
def source_get_int(key: MetadataVariable, default: Optional[Variable | int] = None) -> str:
|
||||
return _get_int(metadata=v.entry_metadata, key=key, default=default)
|
||||
return _get_int(metadata=v.source_metadata, key=key, default=default)
|
||||
|
||||
|
||||
###############################################################################################
|
||||
|
|
@ -106,6 +115,12 @@ ENTRY_HARDCODED_VARIABLES: Dict[Variable, str] = {
|
|||
v.thumbnail_ext: "jpg",
|
||||
}
|
||||
|
||||
ENTRY_RELATIVE_VARIABLES: Dict[MetadataVariable, str] = {
|
||||
v.playlist_metadata: entry_get(v.playlist_metadata, {}),
|
||||
v.source_metadata: entry_get(v.source_metadata, {}),
|
||||
v.sibling_entry_metadata: entry_get(v.sibling_entry_metadata, "{ [] }"),
|
||||
}
|
||||
|
||||
ENTRY_REQUIRED_VARIABLES: Dict[MetadataVariable, str] = {
|
||||
v.uid: entry_get(v.uid),
|
||||
v.extractor_key: entry_get(v.extractor_key),
|
||||
|
|
@ -133,7 +148,6 @@ ENTRY_DEFAULT_VARIABLES: Dict[MetadataVariable, str] = {
|
|||
ENTRY_INJECTED_VARIABLES: Dict[Variable, str] = {
|
||||
v.download_index: "{%int(1)}",
|
||||
v.upload_date_index: "{%int(1)}",
|
||||
v.playlist_max_upload_year: f"{{{v.upload_year.variable_name}}}",
|
||||
v.comments: "",
|
||||
v.requested_subtitles: "",
|
||||
v.sponsorblock_chapters: "",
|
||||
|
|
@ -216,7 +230,6 @@ PLAYLIST_DERIVED_VARIABLES: Dict[Variable, str] = {
|
|||
v.playlist_index_padded6: pad_int(v.playlist_index, 6),
|
||||
v.playlist_index_reversed_padded6: pad_int(v.playlist_index_reversed, 6),
|
||||
v.playlist_uploader_sanitized: sanitized(v.playlist_uploader),
|
||||
v.playlist_max_upload_year_truncated: f"{{%int(%slice(%string({v.playlist_max_upload_year.variable_name}), 2))}}",
|
||||
}
|
||||
|
||||
|
||||
|
|
@ -237,11 +250,20 @@ SOURCE_DERIVED_VARIABLES: Dict[Variable, str] = {
|
|||
v.source_index_padded: pad_int(v.source_index, 2),
|
||||
}
|
||||
|
||||
SIBLING_VARIABLES: Dict[Variable, str] = {
|
||||
v.playlist_max_upload_year: "TODO!!!",
|
||||
}
|
||||
|
||||
SIBLING_DERIVED_VARIABLES: Dict[Variable, str] = {
|
||||
v.playlist_max_upload_year_truncated: f"{{%int(%slice(%string({v.playlist_max_upload_year.variable_name}), 2))}}",
|
||||
}
|
||||
|
||||
_VARIABLE_SCRIPTS: Dict[Variable, str] = {}
|
||||
mergedeep.merge(
|
||||
_VARIABLE_SCRIPTS,
|
||||
ENTRY_EMPTY_METADATA,
|
||||
ENTRY_HARDCODED_VARIABLES,
|
||||
ENTRY_RELATIVE_VARIABLES,
|
||||
ENTRY_REQUIRED_VARIABLES,
|
||||
ENTRY_DEFAULT_VARIABLES,
|
||||
ENTRY_INJECTED_VARIABLES,
|
||||
|
|
|
|||
42
src/ytdl_sub/utils/script.py
Normal file
42
src/ytdl_sub/utils/script.py
Normal file
|
|
@ -0,0 +1,42 @@
|
|||
import json
|
||||
from typing import Any
|
||||
from typing import Dict
|
||||
from typing import Iterable
|
||||
|
||||
from ytdl_sub.script.types.resolvable import Resolvable
|
||||
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] = {}
|
||||
for var in variables:
|
||||
dummy_variables[var] = String("dummy_string")
|
||||
dummy_variables[f"{var}_sanitized"] = String("dummy_string")
|
||||
|
||||
return dummy_variables
|
||||
|
||||
@classmethod
|
||||
def add_sanitized_variables(cls, variables: Dict[str, str]) -> Dict[str, str]:
|
||||
sanitized_variables = {
|
||||
f"{name}_sanitized": f"{{%sanitize({name})}}" for name in variables.keys()
|
||||
}
|
||||
return dict(variables, **sanitized_variables)
|
||||
|
||||
@classmethod
|
||||
def to_script(cls, value: Any) -> str:
|
||||
if value is None:
|
||||
out = ""
|
||||
elif isinstance(value, str):
|
||||
out = value
|
||||
elif isinstance(value, int):
|
||||
out = f"{{%int({value})}}"
|
||||
elif isinstance(value, float):
|
||||
out = f"{{%float({value})}}"
|
||||
elif isinstance(value, bool):
|
||||
out = f"{{%bool({value})}}"
|
||||
else:
|
||||
out = json.dumps(value)
|
||||
|
||||
return out
|
||||
|
|
@ -1,62 +1,16 @@
|
|||
import copy
|
||||
import json
|
||||
from abc import ABC
|
||||
from typing import Any
|
||||
from typing import Dict
|
||||
from typing import Iterable
|
||||
from typing import Set
|
||||
|
||||
from ytdl_sub.entries.script.variable_scripts import UNRESOLVED_VARIABLES
|
||||
from ytdl_sub.entries.script.variable_scripts import VARIABLE_SCRIPTS
|
||||
from ytdl_sub.script.parser import parse
|
||||
from ytdl_sub.script.script import Script
|
||||
from ytdl_sub.script.types.resolvable import Resolvable
|
||||
from ytdl_sub.script.types.resolvable import String
|
||||
from ytdl_sub.utils.script import ScriptUtils
|
||||
|
||||
|
||||
class Scriptable(ABC):
|
||||
@classmethod
|
||||
def add_dummy_variables(cls, variables: Iterable[str]) -> Dict[str, Resolvable]:
|
||||
dummy_variables: Dict[str, Resolvable] = {}
|
||||
for var in variables:
|
||||
dummy_variables[var] = String("dummy_string")
|
||||
dummy_variables[f"{var}_sanitized"] = String("dummy_string")
|
||||
|
||||
return dummy_variables
|
||||
|
||||
@classmethod
|
||||
def add_sanitized_variables(cls, variables: Dict[str, str]) -> Dict[str, str]:
|
||||
sanitized_variables = {
|
||||
f"{name}_sanitized": f"{{%sanitize({name})}}" for name in variables.keys()
|
||||
}
|
||||
return dict(variables, **sanitized_variables)
|
||||
|
||||
@classmethod
|
||||
def wrappable_format_string(cls, format_string: str) -> str:
|
||||
parsed = parse(format_string)
|
||||
|
||||
if resolvable := parsed.maybe_resolvable:
|
||||
return f"'{str(resolvable)}'"
|
||||
|
||||
stripped_format_string = format_string.strip()
|
||||
if stripped_format_string.startswith("{") and stripped_format_string.endswith("}"):
|
||||
return stripped_format_string[1:-1]
|
||||
return format_string
|
||||
|
||||
@classmethod
|
||||
def to_script(cls, value: Any) -> str:
|
||||
if value is None:
|
||||
return ""
|
||||
if isinstance(value, str):
|
||||
return value
|
||||
if isinstance(value, int):
|
||||
return f"{{%int({value})}}"
|
||||
if isinstance(value, float):
|
||||
return f"{{%float({value})}}"
|
||||
if isinstance(value, bool):
|
||||
return f"{{%bool({value})}}"
|
||||
return f"{{{json.dumps(value)}}}"
|
||||
|
||||
def __init__(self):
|
||||
self.script = Script(copy.deepcopy(VARIABLE_SCRIPTS))
|
||||
self.unresolvable: Set[str] = copy.deepcopy(UNRESOLVED_VARIABLES)
|
||||
|
|
@ -66,8 +20,8 @@ class Scriptable(ABC):
|
|||
|
||||
def add(self, values: Dict[str, Any]) -> None:
|
||||
self.script.add(
|
||||
self.add_sanitized_variables(
|
||||
{name: self.to_script(value) for name, value in values.items()}
|
||||
ScriptUtils.add_sanitized_variables(
|
||||
{name: ScriptUtils.to_script(value) for name, value in values.items()}
|
||||
)
|
||||
)
|
||||
|
||||
|
|
|
|||
Loading…
Reference in a new issue