mirror of
https://github.com/unshackle-dl/unshackle.git
synced 2026-11-14 21:15:06 +00:00
459 lines
18 KiB
Python
459 lines
18 KiB
Python
from __future__ import annotations
|
|
|
|
import re
|
|
import warnings
|
|
from pathlib import Path
|
|
from typing import Any, Optional, Union
|
|
|
|
import yaml
|
|
from appdirs import AppDirs
|
|
|
|
from unshackle.core.service_repo import is_repo_spec
|
|
from unshackle.core.utils.collections import ci_get
|
|
|
|
# YAML reads a bare `no` (Norwegian) or `yes` (Nyankpa) as a boolean, so `lang: no` would
|
|
# silently drop the language. These two are the only valid language tags that YAML coerces,
|
|
# and no option under a LANGUAGE_KEYS name holds a boolean.
|
|
BOOL_LANGUAGE_TAGS = {False: "no", True: "yes"}
|
|
LANGUAGE_KEYS = frozenset(
|
|
{
|
|
"lang",
|
|
"v_lang",
|
|
"a_lang",
|
|
"s_lang",
|
|
"forced_s_lang",
|
|
"require_audio",
|
|
"require_video",
|
|
"require_subs",
|
|
"language_priority",
|
|
"default_language",
|
|
"video",
|
|
"audio",
|
|
"subtitle",
|
|
"subs_contain",
|
|
"subs_contain_all",
|
|
}
|
|
)
|
|
|
|
OPAQUE_KEYS = frozenset({"tag_rules"})
|
|
|
|
|
|
def restore_bool_languages(data: Any) -> Any:
|
|
"""Put back the language tags ``no`` and ``yes``, which YAML reads as booleans."""
|
|
if isinstance(data, list):
|
|
return [restore_bool_languages(item) for item in data]
|
|
if not isinstance(data, dict):
|
|
return data
|
|
|
|
fixed = {}
|
|
for key, value in data.items():
|
|
if key in OPAQUE_KEYS:
|
|
fixed[key] = value
|
|
continue
|
|
if key in LANGUAGE_KEYS:
|
|
if isinstance(value, bool):
|
|
value = BOOL_LANGUAGE_TAGS[value]
|
|
elif isinstance(value, list):
|
|
value = [BOOL_LANGUAGE_TAGS[i] if isinstance(i, bool) else i for i in value]
|
|
fixed[key] = restore_bool_languages(value)
|
|
|
|
return fixed
|
|
|
|
|
|
def resolve_decryption(decryption_map: dict, default: str, service: str) -> str:
|
|
"""Pick the decryption backend for a service (case-insensitive), falling back to default."""
|
|
return ci_get(decryption_map, service, default)
|
|
|
|
|
|
def resolve_cdm_name(cdm: dict, service: str, override: Any = None) -> Any:
|
|
"""Find a service's top-level CDM entry (case-insensitive), with default fallback."""
|
|
return override or ci_get(cdm, service) or ci_get(cdm, "default")
|
|
|
|
|
|
def cdm_entry_names_device(entry: Any, profile: Optional[str] = None) -> bool:
|
|
"""Whether a per-service ``cdm`` entry names a device when no quality is known.
|
|
|
|
An entry that selects only by quality, or by a different profile, names none.
|
|
"""
|
|
if not isinstance(entry, dict):
|
|
return bool(entry)
|
|
systems = {"widevine", "playready"} & {str(key).lower() for key in entry}
|
|
return bool(systems or entry.get(profile) or entry.get("default"))
|
|
|
|
|
|
class Config:
|
|
class _Directories:
|
|
# default directories, do not modify here, set via config
|
|
app_dirs = AppDirs("unshackle", False)
|
|
core_dir = Path(__file__).resolve().parent
|
|
namespace_dir = core_dir.parent
|
|
commands = namespace_dir / "commands"
|
|
services = [namespace_dir / "services"]
|
|
vaults = namespace_dir / "vaults"
|
|
fonts = namespace_dir / "fonts"
|
|
user_configs = core_dir.parent
|
|
data = core_dir.parent
|
|
downloads = core_dir.parent.parent / "downloads"
|
|
temp = core_dir.parent.parent / "temp"
|
|
cache = data / "cache"
|
|
cookies = data / "cookies"
|
|
logs = data / "logs"
|
|
exports = data / "exports"
|
|
wvds = data / "WVDs"
|
|
prds = data / "PRDs"
|
|
dcsl = data / "DCSL"
|
|
|
|
class _Filenames:
|
|
# default filenames, do not modify here, set via config
|
|
log = "unshackle_{name}_{time}.log" # Directories.logs
|
|
debug_log = "unshackle_debug_{service}_{time}.jsonl" # Directories.logs
|
|
config = "config.yaml" # Directories.services / tag
|
|
root_config = "unshackle.yaml" # Directories.user_configs
|
|
chapters = "Chapters_{title}_{random}.txt" # Directories.temp
|
|
subtitle = "Subtitle_{id}_{language}.srt" # Directories.temp
|
|
|
|
def __init__(self, **kwargs: Any):
|
|
self.dl: dict = kwargs.get("dl") or {}
|
|
self.cdm: dict = kwargs.get("cdm") or {}
|
|
self.chapter_fallback_name: str = kwargs.get("chapter_fallback_name") or ""
|
|
self.network: dict = kwargs.get("network") or kwargs.get("curl_impersonate") or {}
|
|
self.curl_impersonate: dict = self.network
|
|
if "curl_impersonate" in kwargs and "network" not in kwargs:
|
|
warnings.warn(
|
|
"the 'curl_impersonate' config section is deprecated, rename it to 'network'.",
|
|
DeprecationWarning,
|
|
stacklevel=2,
|
|
)
|
|
self.remote_cdm: list[dict] = kwargs.get("remote_cdm") or []
|
|
self.credentials: dict = kwargs.get("credentials") or {}
|
|
self.firefox_cookies: dict = kwargs.get("firefox_cookies") or {}
|
|
self.subtitle: dict = kwargs.get("subtitle") or {}
|
|
|
|
self.directories = self._Directories()
|
|
for name, path in (kwargs.get("directories") or {}).items():
|
|
if name.lower() in ("app_dirs", "core_dir", "namespace_dir", "user_configs", "data"):
|
|
# these must not be modified by the user
|
|
continue
|
|
if name == "services" and isinstance(path, list):
|
|
# repo specs (git URLs / owner-repo) stay raw strings; resolved lazily in services.py
|
|
setattr(self.directories, name, [p if is_repo_spec(p) else Path(p).expanduser() for p in path])
|
|
else:
|
|
setattr(self.directories, name, Path(path).expanduser())
|
|
|
|
downloader_cfg = kwargs.get("downloader")
|
|
if downloader_cfg and downloader_cfg != "requests":
|
|
warnings.warn(
|
|
f"downloader '{downloader_cfg}' is deprecated. The unified requests downloader is now used.",
|
|
DeprecationWarning,
|
|
stacklevel=2,
|
|
)
|
|
|
|
self.filenames = self._Filenames()
|
|
for name, filename in (kwargs.get("filenames") or {}).items():
|
|
setattr(self.filenames, name, filename)
|
|
|
|
self.audio: dict = kwargs.get("audio") or {}
|
|
self.headers: dict = kwargs.get("headers") or {}
|
|
self.key_vaults: list[dict[str, Any]] = kwargs.get("key_vaults", [])
|
|
self.vault_timeout: float = kwargs.get("vault_timeout", 10.0)
|
|
self.muxing: dict = kwargs.get("muxing") or {}
|
|
self.proxy_providers: dict = kwargs.get("proxy_providers") or {}
|
|
self.remote_services: dict = kwargs.get("remote_services") or {}
|
|
self.serve: dict = kwargs.get("serve") or {}
|
|
self.services: dict = kwargs.get("services") or {}
|
|
decryption_cfg = kwargs.get("decryption") or {}
|
|
if isinstance(decryption_cfg, dict):
|
|
self.decryption_map = {k.upper(): v for k, v in decryption_cfg.items()}
|
|
self.decryption = self.decryption_map.get("DEFAULT", "shaka")
|
|
else:
|
|
self.decryption_map = {}
|
|
self.decryption = decryption_cfg or "shaka"
|
|
self.decrypt_segments: bool = bool(kwargs.get("decrypt_segments", False))
|
|
self.merge_segments: bool = bool(kwargs.get("merge_segments", False))
|
|
|
|
self.theme: str = kwargs.get("theme") or "default"
|
|
self.set_terminal_bg: bool = kwargs.get("set_terminal_bg", False)
|
|
self.tag: str = kwargs.get("tag") or ""
|
|
self.tag_group_name: bool = kwargs.get("tag_group_name", True)
|
|
self.tag_imdb_tmdb: bool = kwargs.get("tag_imdb_tmdb", True)
|
|
self.omdb_api_key: str = kwargs.get("omdb_api_key") or ""
|
|
self.tmdb_api_key: str = kwargs.get("tmdb_api_key") or ""
|
|
self.tvdb_api_key: str = kwargs.get("tvdb_api_key") or ""
|
|
self.tvdb_pin: str = kwargs.get("tvdb_pin") or ""
|
|
self.tvdb_order: str = (kwargs.get("tvdb_order") or "").lower()
|
|
self.metadata_providers: Union[list, dict] = kwargs.get("metadata_providers") or []
|
|
self.anilist_title_language: str = (kwargs.get("anilist_title_language") or "english").lower()
|
|
self.disable_metadata: bool = kwargs.get("disable_metadata", False)
|
|
self.simkl_client_id: str = kwargs.get("simkl_client_id") or ""
|
|
self.decrypt_labs_api_key: str = kwargs.get("decrypt_labs_api_key") or ""
|
|
self.ipinfo_api_key: str = kwargs.get("ipinfo_api_key") or ""
|
|
self.update_checks: bool = kwargs.get("update_checks", True)
|
|
self.services_repo_force: bool = bool(kwargs.get("services_repo_force", False))
|
|
self.update_check_interval: int = kwargs.get("update_check_interval", 24)
|
|
# mask local base dirs (install root/venv/home) in logged paths; False shows full paths
|
|
self.redact_paths: bool = kwargs.get("redact_paths", True)
|
|
self.continue_downloads: bool = kwargs.get("continue_downloads", False)
|
|
self.sample: Optional[tuple[int, int]] = None
|
|
self.post_scripts: list = kwargs.get("post_scripts") or []
|
|
|
|
self.language_tags: dict = kwargs.get("language_tags") or {}
|
|
self.tag_rules: list = kwargs.get("tag_rules") or []
|
|
self.dual_multi_mode: str = (kwargs.get("dual_multi_mode") or "strict").lower()
|
|
self.generate_dv_mode: str = str(kwargs.get("generate_dv_mode") or "v29").lower()
|
|
self.hdr10plus_probe: bool = kwargs.get("hdr10plus_probe", True)
|
|
if not isinstance(self.hdr10plus_probe, bool):
|
|
raise SystemExit(f"ERROR: hdr10plus_probe must be true or false, not {self.hdr10plus_probe!r}")
|
|
self.output_template: dict = kwargs.get("output_template") or {}
|
|
folder_cfg = self.output_template.pop("folder", "")
|
|
self.folder_template: str = ""
|
|
self.folder_templates: dict = {}
|
|
if isinstance(folder_cfg, dict):
|
|
self.folder_templates = {k: v for k, v in folder_cfg.items() if isinstance(v, str) and v}
|
|
elif isinstance(folder_cfg, str):
|
|
self.folder_template = folder_cfg or ""
|
|
|
|
if kwargs.get("scene_naming") is not None:
|
|
raise SystemExit(
|
|
"ERROR: The 'scene_naming' option has been removed.\n"
|
|
"Please configure 'output_template' in your unshackle.yaml instead.\n"
|
|
"See unshackle-example.yaml for examples."
|
|
)
|
|
|
|
if self.output_template:
|
|
self.validate_output_templates()
|
|
|
|
self.unicode_filenames: bool = kwargs.get("unicode_filenames", False)
|
|
|
|
self.title_cache_time: int = kwargs.get("title_cache_time", 1800) # 30 minutes default
|
|
self.title_cache_max_retention: int = kwargs.get("title_cache_max_retention", 86400) # 24 hours default
|
|
self.title_cache_enabled: bool = kwargs.get("title_cache_enabled", True)
|
|
|
|
self.debug: bool = kwargs.get("debug", False)
|
|
self.debug_keys: bool = kwargs.get("debug_keys", False)
|
|
self.debug_requests: bool = kwargs.get("debug_requests", False)
|
|
|
|
def validate_output_templates(self) -> None:
|
|
"""Validate output template configurations and warn about potential issues."""
|
|
if not self.output_template:
|
|
return
|
|
|
|
valid_variables = {
|
|
"title",
|
|
"year",
|
|
"season",
|
|
"episode",
|
|
"season_episode",
|
|
"episode_name",
|
|
"part",
|
|
"absolute",
|
|
"date",
|
|
"quality",
|
|
"resolution",
|
|
"source",
|
|
"tag",
|
|
"track_number",
|
|
"artist",
|
|
"album_artist",
|
|
"album",
|
|
"disc",
|
|
"track_total",
|
|
"disc_total",
|
|
"release_type",
|
|
"genre",
|
|
"explicit",
|
|
"isrc",
|
|
"upc",
|
|
"label",
|
|
"audio",
|
|
"audio_channels",
|
|
"audio_full",
|
|
"atmos",
|
|
"dual",
|
|
"multi",
|
|
"dubbed",
|
|
"video",
|
|
"hdr",
|
|
"hfr",
|
|
"edition",
|
|
"repack",
|
|
"lang_tag",
|
|
"title_type",
|
|
}
|
|
|
|
type_only = {
|
|
"series": {"season", "episode", "season_episode", "episode_name", "part", "absolute", "date"},
|
|
"songs": {
|
|
"track_number",
|
|
"artist",
|
|
"album_artist",
|
|
"album",
|
|
"disc",
|
|
"track_total",
|
|
"disc_total",
|
|
"release_type",
|
|
"genre",
|
|
"explicit",
|
|
"isrc",
|
|
"upc",
|
|
"label",
|
|
},
|
|
}
|
|
type_only["albums"] = type_only["songs"]
|
|
shared = valid_variables - set().union(*type_only.values())
|
|
|
|
unsafe_chars = r'[<>:"/\\|?*]'
|
|
|
|
all_templates = dict(self.output_template)
|
|
if self.folder_template:
|
|
all_templates["folder"] = self.folder_template
|
|
for kind, tmpl in self.folder_templates.items():
|
|
if kind not in {"movies", "series", "songs", "albums"}:
|
|
warnings.warn(f"Unknown folder template kind '{kind}' (expected movies/series/songs/albums)")
|
|
continue
|
|
all_templates[f"folder.{kind}"] = tmpl
|
|
|
|
for template_type, template_str in all_templates.items():
|
|
if not isinstance(template_str, str):
|
|
warnings.warn(f"Template '{template_type}' must be a string, got {type(template_str).__name__}")
|
|
continue
|
|
|
|
variables = re.findall(r"\{([^}]+)\}", template_str)
|
|
|
|
kind = template_type.split(".")[-1]
|
|
allowed = shared | type_only.get(kind, set())
|
|
for var in variables:
|
|
var_clean = var.rstrip("?")
|
|
if var_clean not in valid_variables:
|
|
warnings.warn(f"Unknown template variable '{var}' in {template_type} template")
|
|
elif var_clean not in allowed:
|
|
warnings.warn(f"Template variable '{var}' is not available in the {template_type} template")
|
|
|
|
test_template = re.sub(r"\{[^}]+\}", "TEST", template_str)
|
|
if template_type.startswith("folder"):
|
|
unsafe_segment = any(re.search(unsafe_chars, seg) for seg in re.split(r"[\\/]", test_template))
|
|
else:
|
|
unsafe_segment = bool(re.search(unsafe_chars, test_template))
|
|
if unsafe_segment:
|
|
warnings.warn(f"Template '{template_type}' may contain filesystem-unsafe characters")
|
|
|
|
if not template_str.strip():
|
|
warnings.warn(f"Template '{template_type}' is empty")
|
|
|
|
def get_folder_template(self, kind: str) -> str:
|
|
"""Find the folder template for the given title kind.
|
|
|
|
kind: one of "movies", "series", "songs", "albums".
|
|
Falls back to the legacy single-string folder template, then "".
|
|
"""
|
|
if self.folder_templates:
|
|
tmpl = self.folder_templates.get(kind)
|
|
if tmpl:
|
|
return tmpl
|
|
return self.folder_template or ""
|
|
|
|
def get_template_separator(self, template_type: str = "movies") -> str:
|
|
"""Get the filename separator for the given template type.
|
|
|
|
Analyzes the active template to determine whether it uses dots or spaces
|
|
between variables. Falls back to dot separator (scene-style) by default.
|
|
|
|
Args:
|
|
template_type: One of "movies", "series", or "songs".
|
|
"""
|
|
template = self.output_template.get(template_type) or ""
|
|
between_vars = re.findall(r"\}([^{]*)\{", template)
|
|
separator_text = "".join(between_vars)
|
|
dot_count = separator_text.count(".")
|
|
space_count = separator_text.count(" ")
|
|
|
|
return " " if space_count > dot_count else "."
|
|
|
|
@classmethod
|
|
def from_yaml(cls, path: Path) -> Config:
|
|
if not path.exists():
|
|
raise FileNotFoundError(f"Config file path ({path}) was not found")
|
|
if not path.is_file():
|
|
raise FileNotFoundError(f"Config file path ({path}) is not to a file.")
|
|
return cls(**restore_bool_languages(load_yaml(path) or {}))
|
|
|
|
|
|
def duplicate_key_lines(root: Any) -> list[tuple[int, int]]:
|
|
"""
|
|
Return (first line, repeat line), counted from 1, for each repeated scalar key in a PyYAML or ruamel node tree.
|
|
|
|
Keys compare as written, so two spellings that load as the same key (yes and true, 1 and 0x1) are not found.
|
|
"""
|
|
pairs, stack, visited = [], [root], set()
|
|
while stack:
|
|
node = stack.pop()
|
|
if node in visited:
|
|
continue
|
|
visited.add(node)
|
|
if node.id == "mapping":
|
|
first: dict[tuple[str, str], int] = {}
|
|
for key_node, value_node in node.value:
|
|
if key_node.id == "scalar" and key_node.tag != "tag:yaml.org,2002:merge":
|
|
key = (key_node.tag, key_node.value)
|
|
line = key_node.start_mark.line + 1
|
|
if key in first:
|
|
pairs.append((first[key], line))
|
|
else:
|
|
first[key] = line
|
|
stack += [key_node, value_node]
|
|
elif node.id == "sequence":
|
|
stack += node.value
|
|
return sorted(pairs)
|
|
|
|
|
|
def duplicate_key_message(name: str, pairs: list[tuple[int, int]], line: int = 0) -> str:
|
|
"""Say where a file repeats a key: each (first, repeat) pair, or `line` when no pair is known."""
|
|
where = ", ".join(f"lines {first} and {repeat}" for first, repeat in pairs) or f"line {line}"
|
|
return f"{name}: duplicate key on {where}. Unshackle uses the later one."
|
|
|
|
|
|
def load_yaml(path: Path) -> Any:
|
|
"""yaml.safe_load a file, and warn with the lines of each duplicate key, which PyYAML takes silently."""
|
|
loader = yaml.SafeLoader(path.read_text(encoding="utf8"))
|
|
try:
|
|
node = loader.get_single_node()
|
|
if node is None:
|
|
return None
|
|
if pairs := duplicate_key_lines(node):
|
|
warnings.warn(duplicate_key_message(str(path), pairs), stacklevel=2)
|
|
return loader.construct_document(node)
|
|
finally:
|
|
loader.dispose()
|
|
|
|
|
|
# noinspection PyProtectedMember
|
|
PACKAGE_CONFIG_PATH = Config._Directories.namespace_dir / Config._Filenames.root_config
|
|
USER_CONFIG_PATH = Path(Config._Directories.app_dirs.user_config_dir) / Config._Filenames.root_config
|
|
# noinspection PyProtectedMember
|
|
POSSIBLE_CONFIG_PATHS = (
|
|
PACKAGE_CONFIG_PATH,
|
|
# The Parent Folder to the unshackle Namespace Folder (e.g., %appdata%/Python/Python311/site-packages)
|
|
Config._Directories.namespace_dir.parent / Config._Filenames.root_config,
|
|
USER_CONFIG_PATH,
|
|
)
|
|
|
|
|
|
def get_config_path() -> Optional[Path]:
|
|
"""
|
|
Get Path to Config from any one of the possible locations.
|
|
|
|
Returns None if no config file could be found.
|
|
"""
|
|
for path in POSSIBLE_CONFIG_PATHS:
|
|
if path.exists():
|
|
return path
|
|
return None
|
|
|
|
|
|
config_path = get_config_path()
|
|
if config_path:
|
|
config = Config.from_yaml(config_path)
|
|
else:
|
|
config = Config()
|
|
|
|
__all__ = ("config",)
|