e92b1cd12a
A complete gutting of the internals of ytdl-sub to support functions in our variable syntax, in addition to being able to access a yt-dlp entry's .info.json fields using functions. Functionally, ytdl-sub should still look and behave the same from a user-perspective. With so many lines of code changed (+8927, -2708), no doubt there will be new issues. Please make a GH issue or reach out on Discord if your config/subscriptions break in any way/shape/form. Details on how to use function support will come soon in the form of proper documentation in our readthedocs.
296 lines
12 KiB
Python
296 lines
12 KiB
Python
from typing import Dict
|
|
from typing import List
|
|
from typing import Optional
|
|
from typing import Set
|
|
|
|
import mergedeep
|
|
|
|
from ytdl_sub.entries.script.custom_functions import CustomFunctions
|
|
from ytdl_sub.entries.script.variable_definitions import VARIABLES
|
|
from ytdl_sub.entries.script.variable_definitions import Metadata
|
|
from ytdl_sub.entries.script.variable_definitions import MetadataVariable
|
|
from ytdl_sub.entries.script.variable_definitions import Variable
|
|
from ytdl_sub.entries.script.variable_definitions import VariableDefinitions
|
|
|
|
###############################################################################################
|
|
# Helpers
|
|
|
|
v: VariableDefinitions = VARIABLES
|
|
|
|
|
|
def _pad_int(key: Variable, pad: int) -> str:
|
|
return f"{{%pad_zero({key.variable_name}, {pad})}}"
|
|
|
|
|
|
def _sanitized_plex(key: Variable) -> str:
|
|
return f"{{%sanitize_plex_episode({key.variable_name})}}"
|
|
|
|
|
|
def _date_metadata(date_key: Variable, metadata_key: str) -> str:
|
|
return f"{{%map_get(%to_date_metadata({date_key.variable_name}), '{metadata_key}')}}"
|
|
|
|
|
|
###############################################################################################
|
|
# Metadata Getters
|
|
|
|
|
|
def _get(
|
|
cast: str,
|
|
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
|
|
out = f"%map_get({metadata.variable_name}, '{key.metadata_key}')"
|
|
elif isinstance(default, Variable):
|
|
args = f"{metadata.variable_name}, '{key.metadata_key}', {default.variable_name}"
|
|
out = f"%map_get_non_empty({args})"
|
|
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})"
|
|
|
|
return f"{{ %legacy_bracket_safety(%{cast}({out})) }}"
|
|
|
|
|
|
###############################################################################################
|
|
# Entry Getters
|
|
|
|
|
|
def _entry_get_str(key: MetadataVariable, default: Optional[Variable | str] = None) -> str:
|
|
return _get("string", metadata=v.entry_metadata, key=key, default=default)
|
|
|
|
|
|
def _entry_get_int(key: MetadataVariable, default: Optional[Variable | int] = None) -> str:
|
|
return _get("int", metadata=v.entry_metadata, key=key, default=default)
|
|
|
|
|
|
def _entry_get_map(key: MetadataVariable, default: Optional[Variable | Dict] = None):
|
|
return _get("map", metadata=v.entry_metadata, key=key, default=default)
|
|
|
|
|
|
def _entry_get_array(key: MetadataVariable, default: Optional[Variable | List] = None):
|
|
return _get("array", metadata=v.entry_metadata, key=key, default=default)
|
|
|
|
|
|
###############################################################################################
|
|
# Playlist Getters
|
|
|
|
|
|
def _playlist_get_str(key: MetadataVariable, default: Optional[Variable | str] = None) -> str:
|
|
return _get("string", 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.playlist_metadata, key=key, default=default)
|
|
|
|
|
|
###############################################################################################
|
|
# Source Getters
|
|
|
|
|
|
def _source_get_str(key: MetadataVariable, default: Optional[Variable | str] = None) -> str:
|
|
return _get("string", metadata=v.source_metadata, key=key, default=default)
|
|
|
|
|
|
###############################################################################################
|
|
# Scripts
|
|
|
|
ENTRY_EMPTY_METADATA: Dict[Variable, str] = {v.entry_metadata: "{ {} }"}
|
|
|
|
ENTRY_HARDCODED_VARIABLES: Dict[Variable, str] = {
|
|
v.info_json_ext: "info.json",
|
|
v.thumbnail_ext: "jpg",
|
|
}
|
|
|
|
ENTRY_RELATIVE_VARIABLES: Dict[MetadataVariable, str] = {
|
|
v.playlist_metadata: _entry_get_map(v.playlist_metadata, {}),
|
|
v.source_metadata: _entry_get_map(v.source_metadata, {}),
|
|
v.sibling_metadata: _entry_get_array(v.sibling_metadata, []),
|
|
}
|
|
|
|
ENTRY_REQUIRED_VARIABLES: Dict[MetadataVariable, str] = {
|
|
v.uid: _entry_get_str(v.uid),
|
|
v.extractor_key: _entry_get_str(v.extractor_key),
|
|
v.epoch: _entry_get_int(v.epoch),
|
|
v.webpage_url: _entry_get_str(v.webpage_url),
|
|
v.ext: _entry_get_str(v.ext),
|
|
}
|
|
|
|
ENTRY_DEFAULT_VARIABLES: Dict[MetadataVariable, str] = {
|
|
v.title: _entry_get_str(v.title, v.uid),
|
|
v.extractor: _entry_get_str(v.extractor, v.extractor_key),
|
|
v.description: _entry_get_str(v.description, ""),
|
|
v.ie_key: _entry_get_str(v.ie_key, v.extractor_key),
|
|
v.uploader_id: _entry_get_str(v.uploader_id, v.uid),
|
|
v.uploader: _entry_get_str(v.uploader, v.uploader_id),
|
|
v.uploader_url: _entry_get_str(v.uploader_url, v.webpage_url),
|
|
v.upload_date: _entry_get_str(v.upload_date, v.epoch_date),
|
|
v.release_date: _entry_get_str(v.release_date, v.upload_date),
|
|
v.channel: _entry_get_str(v.channel, v.uploader),
|
|
v.creator: _entry_get_str(v.creator, v.channel),
|
|
v.channel_id: _entry_get_str(v.channel_id, v.uploader_id),
|
|
v.duration: _entry_get_int(v.duration, 0),
|
|
v.playlist_index: _entry_get_int(v.playlist_index, 1),
|
|
v.playlist_count: _entry_get_int(v.playlist_count, 1),
|
|
v.playlist_uid: _entry_get_str(v.playlist_uid, v.uid),
|
|
v.playlist_title: _entry_get_str(v.playlist_title, v.title),
|
|
v.playlist_uploader_id: _entry_get_str(v.playlist_uploader_id, v.uploader_id),
|
|
}
|
|
|
|
# MARK AS UNRESOLVABLE UNTIL THEY ARE ADDED IN THE DOWNLOADER
|
|
DOWNLOADER_INJECTED_VARIABLES: Dict[Variable, str] = {
|
|
v.download_index: "{%int(1)}",
|
|
v.upload_date_index: "{%int(1)}",
|
|
v.comments: "{ [] }",
|
|
v.requested_subtitles: "{ {} }",
|
|
v.chapters: "{ [] }",
|
|
v.sponsorblock_chapters: "{ [] }",
|
|
v.ytdl_sub_input_url: f"{{{v.source_webpage_url.variable_name}}}",
|
|
}
|
|
|
|
ENTRY_DERIVED_VARIABLES: Dict[Variable, str] = {
|
|
v.uid_sanitized_plex: _sanitized_plex(v.uid),
|
|
v.title_sanitized_plex: _sanitized_plex(v.title),
|
|
v.epoch_date: f"{{%datetime_strftime({v.epoch.variable_name}, '%Y%m%d')}}",
|
|
v.epoch_hour: f"{{%datetime_strftime({v.epoch.variable_name}, '%H')}}",
|
|
v.download_index_padded6: _pad_int(v.download_index, 6),
|
|
v.upload_date_index_padded: _pad_int(v.upload_date_index, 2),
|
|
v.upload_date_index_reversed: f"{{%sub(100, {v.upload_date_index.variable_name})}}",
|
|
v.upload_date_index_reversed_padded: _pad_int(v.upload_date_index_reversed, 2),
|
|
v.playlist_index_reversed: (
|
|
f"{{%sub({v.playlist_count.variable_name}, {v.playlist_index.variable_name}, -1)}}"
|
|
),
|
|
v.playlist_index_padded: _pad_int(v.playlist_index, 2),
|
|
v.playlist_index_reversed_padded: _pad_int(v.playlist_index_reversed, 2),
|
|
v.playlist_index_padded6: _pad_int(v.playlist_index, 6),
|
|
v.playlist_index_reversed_padded6: _pad_int(v.playlist_index_reversed, 6),
|
|
}
|
|
|
|
ENTRY_UPLOAD_DATE_VARIABLES: Dict[Variable, str] = {
|
|
v.upload_year: _date_metadata(v.upload_date, "year"),
|
|
v.upload_year_truncated: _date_metadata(v.upload_date, "year_truncated"),
|
|
v.upload_year_truncated_reversed: _date_metadata(v.upload_date, "year_truncated_reversed"),
|
|
v.upload_month_reversed: _date_metadata(v.upload_date, "month_reversed"),
|
|
v.upload_month_reversed_padded: _date_metadata(v.upload_date, "month_reversed_padded"),
|
|
v.upload_month_padded: _date_metadata(v.upload_date, "month_padded"),
|
|
v.upload_day_padded: _date_metadata(v.upload_date, "day_padded"),
|
|
v.upload_month: _date_metadata(v.upload_date, "month"),
|
|
v.upload_day: _date_metadata(v.upload_date, "day"),
|
|
v.upload_day_reversed: _date_metadata(v.upload_date, "day_reversed"),
|
|
v.upload_day_reversed_padded: _date_metadata(v.upload_date, "day_reversed_padded"),
|
|
v.upload_day_of_year: _date_metadata(v.upload_date, "day_of_year"),
|
|
v.upload_day_of_year_padded: _date_metadata(v.upload_date, "day_of_year_padded"),
|
|
v.upload_day_of_year_reversed: _date_metadata(v.upload_date, "day_of_year_reversed"),
|
|
v.upload_day_of_year_reversed_padded: _date_metadata(
|
|
v.upload_date, "day_of_year_reversed_padded"
|
|
),
|
|
v.upload_date_standardized: _date_metadata(v.upload_date, "date_standardized"),
|
|
}
|
|
|
|
ENTRY_RELEASE_DATE_VARIABLES: Dict[Variable, str] = {
|
|
v.release_year: _date_metadata(v.release_date, "year"),
|
|
v.release_year_truncated: _date_metadata(v.release_date, "year_truncated"),
|
|
v.release_year_truncated_reversed: _date_metadata(v.release_date, "year_truncated_reversed"),
|
|
v.release_month_reversed: _date_metadata(v.release_date, "month_reversed"),
|
|
v.release_month_reversed_padded: _date_metadata(v.release_date, "month_reversed_padded"),
|
|
v.release_month_padded: _date_metadata(v.release_date, "month_padded"),
|
|
v.release_day_padded: _date_metadata(v.release_date, "day_padded"),
|
|
v.release_month: _date_metadata(v.release_date, "month"),
|
|
v.release_day: _date_metadata(v.release_date, "day"),
|
|
v.release_day_reversed: _date_metadata(v.release_date, "day_reversed"),
|
|
v.release_day_reversed_padded: _date_metadata(v.release_date, "day_reversed_padded"),
|
|
v.release_day_of_year: _date_metadata(v.release_date, "day_of_year"),
|
|
v.release_day_of_year_padded: _date_metadata(v.release_date, "day_of_year_padded"),
|
|
v.release_day_of_year_reversed: _date_metadata(v.release_date, "day_of_year_reversed"),
|
|
v.release_day_of_year_reversed_padded: _date_metadata(
|
|
v.release_date, "day_of_year_reversed_padded"
|
|
),
|
|
v.release_date_standardized: _date_metadata(v.release_date, "date_standardized"),
|
|
}
|
|
|
|
PLAYLIST_VARIABLES: Dict[Variable, str] = {
|
|
v.playlist_webpage_url: _playlist_get_str(v.playlist_webpage_url, v.webpage_url),
|
|
v.playlist_description: _playlist_get_str(v.playlist_description, v.description),
|
|
v.playlist_uploader: _playlist_get_str(v.playlist_uploader, v.uploader),
|
|
v.playlist_uploader_url: _playlist_get_str(v.playlist_uploader_url, v.playlist_webpage_url),
|
|
v.source_index: _playlist_get_int(v.source_index, 1),
|
|
v.source_count: _playlist_get_int(v.source_count, 1),
|
|
}
|
|
|
|
|
|
SOURCE_VARIABLES: Dict[Variable, str] = {
|
|
v.source_uid: _source_get_str(v.source_uid, v.playlist_uid),
|
|
v.source_title: _source_get_str(v.source_title, v.playlist_title),
|
|
v.source_webpage_url: _source_get_str(v.source_webpage_url, v.playlist_webpage_url),
|
|
v.source_description: _source_get_str(v.source_description, v.playlist_description),
|
|
v.source_uploader_id: _source_get_str(v.source_uploader_id, v.playlist_uploader_id),
|
|
v.source_uploader: _source_get_str(v.source_uploader, v.playlist_uploader),
|
|
v.source_uploader_url: _source_get_str(v.source_uploader_url, v.source_webpage_url),
|
|
}
|
|
|
|
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_date: f"""{{
|
|
%array_reduce(
|
|
%if_passthrough(
|
|
%extract_field_from_siblings('{v.upload_date.variable_name}'),
|
|
[{v.upload_date.variable_name}]
|
|
),
|
|
%max
|
|
)
|
|
}}"""
|
|
}
|
|
|
|
SIBLING_DERIVED_VARIABLES: Dict[Variable, str] = {
|
|
v.playlist_max_upload_year: _date_metadata(v.playlist_max_upload_date, "year"),
|
|
v.playlist_max_upload_year_truncated: _date_metadata(
|
|
v.playlist_max_upload_date, "year_truncated"
|
|
),
|
|
}
|
|
|
|
_VARIABLE_SCRIPTS: Dict[Variable, str] = {}
|
|
mergedeep.merge(
|
|
_VARIABLE_SCRIPTS,
|
|
ENTRY_EMPTY_METADATA,
|
|
ENTRY_HARDCODED_VARIABLES,
|
|
ENTRY_RELATIVE_VARIABLES,
|
|
ENTRY_REQUIRED_VARIABLES,
|
|
ENTRY_DEFAULT_VARIABLES,
|
|
DOWNLOADER_INJECTED_VARIABLES,
|
|
ENTRY_DERIVED_VARIABLES,
|
|
ENTRY_UPLOAD_DATE_VARIABLES,
|
|
ENTRY_RELEASE_DATE_VARIABLES,
|
|
SIBLING_VARIABLES,
|
|
SIBLING_DERIVED_VARIABLES,
|
|
PLAYLIST_VARIABLES,
|
|
SOURCE_VARIABLES,
|
|
SOURCE_DERIVED_VARIABLES,
|
|
)
|
|
|
|
VARIABLE_SCRIPTS: Dict[str, str] = {
|
|
var.variable_name: script for var, script in _VARIABLE_SCRIPTS.items()
|
|
}
|
|
|
|
|
|
def _keys(*variables: Dict[Variable, str]) -> Set[str]:
|
|
keys: Set[str] = set()
|
|
for variable_set in variables:
|
|
keys.update(set(var.variable_name for var in variable_set.keys()))
|
|
return keys
|
|
|
|
|
|
UNRESOLVED_VARIABLES: Set[str] = _keys(
|
|
ENTRY_EMPTY_METADATA,
|
|
DOWNLOADER_INJECTED_VARIABLES,
|
|
)
|
|
|
|
CustomFunctions.register()
|