diff --git a/FAQ.md b/FAQ.md index c4f53db4..6115c1fc 100644 --- a/FAQ.md +++ b/FAQ.md @@ -167,6 +167,28 @@ YTP_YTDLP_VERSION=2025.07.21 or master or nightly Then restart the container to apply the changes. +# Custom output template placeholders + +YTPTube supports custom `ytp_*` placeholders in `yt-dlp` output template via the following syntax `%(ytp_*:)s`. + +## Currently available extra placeholders are: + +- `ytp_random`: random mixed letters and digits, + - `N` A number is required to specify the length of the random string, for example `%(ytp_random:8)s` will generate a random string of 8 characters. + - if the args followed by `:s` it will generate random letters only, if followed by `:d` it will generate random digits only. + +## Examples of the custom placeholders in action: + +- Template: `%(title)s [%(ytp_random:8)s].%(ext)s` + - Example result: `My Video [A7k2Pq9Z].mp4` +- Template: `%(uploader)s/%(ytp_random:6:d)s - %(title)s.%(ext)s` + - Example result: `MyChannel/483920 - My Video.mp4` +- Template: `%(playlist)s/%(ytp_random:10:s)s/%(title)s.%(ext)s` + - Example result: `Favorites/QwErTyUiOp/My Video.mp4` + +> [!NOTE] +> `%(ytp_` placeholders are a YTPTube extension and not avaliable via console or directly via yt-dlp. + # How can I monitor sites without RSS feeds? YTPTube includes a **generic task handler** that turns JSON definitions into site-specific scrapers. You can use it diff --git a/app/features/ytdlp/outtmpl.py b/app/features/ytdlp/outtmpl.py new file mode 100644 index 00000000..974b5477 --- /dev/null +++ b/app/features/ytdlp/outtmpl.py @@ -0,0 +1,113 @@ +from __future__ import annotations + +import re +import secrets +import string +from typing import TYPE_CHECKING, Any + +from yt_dlp.utils._utils import STR_FORMAT_RE_TMPL, STR_FORMAT_TYPES + +if TYPE_CHECKING: + from collections.abc import Callable + +OUTTMPL_RX: re.Pattern[str] = re.compile(STR_FORMAT_RE_TMPL.format("[^)]*", f"[{STR_FORMAT_TYPES}ljhqBUDS]")) +CALL_RX: re.Pattern[str] = re.compile(r"^(?Pytp_[A-Za-z0-9_]+)(?P(?::[A-Za-z0-9_]+)*)(?P.*)$") +OPERATORS: tuple[str, ...] = (",", "&", "|", ">", "+", "-", "*") + + +def random_text(args: tuple[str, ...], _info_dict: dict[str, Any], _state: dict[str, Any]) -> str: + if not args: + msg = "ytp_random requires a length argument. Use %(ytp_random:)s." + raise ValueError(msg) + + if len(args) > 2: + msg = "ytp_random accepts at most 2 arguments: length and optional mode." + raise ValueError(msg) + + try: + length = int(args[0]) + except ValueError as exc: + msg: str = f"ytp_random length must be an integer, got {args[0]!r}." + raise ValueError(msg) from exc + + if length < 1: + msg = "ytp_random length must be greater than 0." + raise ValueError(msg) + + mode = args[1].lower() if len(args) > 1 else "mixed" + if mode in ("mixed", "m"): + alphabet = string.ascii_letters + string.digits + elif mode in ("d", "digit", "digits", "int", "ints", "number", "numbers"): + alphabet = string.digits + elif mode in ("s", "str", "string", "strings", "alpha", "letter", "letters"): + alphabet = string.ascii_letters + else: + msg = f"ytp_random mode must be one of mixed, d, or s. Got {args[1]!r}." + raise ValueError(msg) + + return "".join(secrets.choice(alphabet) for _ in range(length)) + + +CALLS: dict[str, Callable[[tuple[str, ...], dict[str, Any], dict[str, Any]], Any]] = { + "ytp_random": random_text, +} + + +def split_call(key: str) -> tuple[str, tuple[str, ...], str] | None: + if not key.startswith("ytp_"): + return None + + if not (match := CALL_RX.match(key)): + msg = f"Invalid YTPTube output template callable {key!r}." + raise ValueError(msg) + + name: str | Any = match.group("name") + rest: str | Any = match.group("rest") or "" + if rest and rest[0] not in OPERATORS: + msg = f"Invalid YTPTube output template callable {key!r}." + raise ValueError(msg) + + if name not in CALLS: + msg = f"Unsupported YTPTube output template callable {name!r}." + raise ValueError(msg) + + raw_args: str | Any = match.group("args") + args: tuple[str | Any, ...] = tuple(part for part in raw_args.split(":") if part) + return name, args, rest + + +def rewrite_outtmpl( + outtmpl: str, + info_dict: dict[str, Any], + cache: dict[str, Any] | None = None, +) -> tuple[str, dict[str, Any]]: + if "%(ytp_" not in outtmpl: + return outtmpl, info_dict + + state: dict[str, Any] = {} + values: dict[str, Any] = {} if cache is None else cache + fields: dict[str, str] = {} + enriched = dict(info_dict) + + def replace(match: re.Match[str]) -> str: + if not (key := match.group("key")): + return match.group(0) + + parsed = split_call(key) + if not parsed: + return match.group(0) + + name, args, rest = parsed + call_key: str = ":".join((name, *args)) if args else name + + if call_key not in values: + values[call_key] = CALLS[name](args, enriched, state) + + if call_key not in fields: + fields[call_key] = f"__ytptube_outtmpl_{len(fields)}" + + synthetic_key: str = fields[call_key] + enriched[synthetic_key] = values[call_key] + return f"{match.group('prefix')}%({synthetic_key}{rest}){match.group('format')}" + + return OUTTMPL_RX.sub(replace, outtmpl), enriched diff --git a/app/features/ytdlp/tests/test_ytdlp_module.py b/app/features/ytdlp/tests/test_ytdlp_module.py index 66215795..46e8daf4 100644 --- a/app/features/ytdlp/tests/test_ytdlp_module.py +++ b/app/features/ytdlp/tests/test_ytdlp_module.py @@ -1,5 +1,8 @@ from unittest.mock import MagicMock, Mock, patch +import pytest + +from app.features.ytdlp.outtmpl import rewrite_outtmpl from app.features.ytdlp.patches import patch_windows_popen_wait from app.features.ytdlp.utils import _DATA from app.features.ytdlp.ytdlp import YTDLP, _ArchiveProxy, ytdlp_options @@ -327,3 +330,116 @@ class TestYTDLP: # Should not add anything ytdlp.archive.add.assert_not_called() + + def test_prepare_outtmpl_resolves_custom_callable(self) -> None: + ytdlp = YTDLP(params={"outtmpl": {"default": "%(title)s"}}) + + result = ytdlp.evaluate_outtmpl("%(ytp_random:8)s", {"title": "x"}) + + assert len(result) == 8 + assert result.isalnum() + + def test_prepare_outtmpl_reuses_same_callable_value_per_download(self) -> None: + ytdlp = YTDLP(params={"outtmpl": {"default": "%(title)s"}}) + + result = ytdlp.evaluate_outtmpl("%(ytp_random:8)s/%(ytp_random:8)s", {"title": "x"}) + first, second = result.split("/") + + assert first == second + + def test_prepare_outtmpl_does_not_reuse_callable_value_across_calls(self) -> None: + ytdlp = YTDLP(params={"outtmpl": {"default": "%(title)s"}}) + + first = ytdlp.evaluate_outtmpl("%(ytp_random:8)s", {"title": "x"}) + second = ytdlp.evaluate_outtmpl("%(ytp_random:8)s", {"title": "y"}) + + assert first != second + + def test_prepare_filename_reuses_custom_value_for_sidecars_of_same_entry(self) -> None: + ytdlp = YTDLP( + params={ + "outtmpl": { + "default": "%(ytp_random:6)s.%(ext)s", + "thumbnail": "%(ytp_random:6)s.%(ext)s", + "subtitle": "%(ytp_random:6)s.%(ext)s", + "infojson": "%(ytp_random:6)s.%(ext)s", + } + } + ) + + info = {"id": "abc123", "title": "Example", "ext": "mp4"} + + default_name = ytdlp.prepare_filename(info) + thumbnail_name = ytdlp.prepare_filename(info, "thumbnail") + subtitle_name = ytdlp.prepare_filename(info, "subtitle") + infojson_name = ytdlp.prepare_filename(info, "infojson") + + default_base = default_name.rsplit(".", 1)[0] + thumbnail_base = thumbnail_name.rsplit(".", 1)[0] + subtitle_base = subtitle_name.rsplit(".", 1)[0] + infojson_base = infojson_name.removesuffix(".info.json") + + assert default_base == thumbnail_base + assert default_base == subtitle_base + assert default_base == infojson_base + + def test_prepare_filename_resets_custom_value_for_different_info_dict_objects(self) -> None: + ytdlp = YTDLP(params={"outtmpl": {"default": "%(ytp_random:8)s.%(ext)s"}}) + + first = ytdlp.prepare_filename({"id": "one", "title": "One", "ext": "mp4"}) + second = ytdlp.prepare_filename({"id": "two", "title": "Two", "ext": "mp4"}) + + assert first != second + + @pytest.mark.parametrize( + ("template", "expected"), + [ + ("%(ytp_random:8|fallback)s", 8), + ("%(ytp_random:8&{} - |)s", 11), + ("%(ytp_random:8)S", 8), + ], + ) + def test_prepare_outtmpl_preserves_ytdlp_suffix_formatting(self, template: str, expected: int) -> None: + ytdlp = YTDLP( + params={ + "outtmpl": {"default": "%(title)s"}, + "restrictfilenames": True, + }, + ) + + result = ytdlp.evaluate_outtmpl(template, {"title": "x"}) + + assert len(result) == expected + + def test_prepare_outtmpl_supports_digit_and_string_modes(self) -> None: + ytdlp = YTDLP(params={"outtmpl": {"default": "%(title)s"}}) + + digits = ytdlp.evaluate_outtmpl("%(ytp_random:6:d)s", {"title": "x"}) + letters = ytdlp.evaluate_outtmpl("%(ytp_random:6:s)s", {"title": "x"}) + + assert digits.isdigit() + assert letters.isalpha() + + def test_prepare_outtmpl_rejects_unknown_callable(self) -> None: + ytdlp = YTDLP(params={"outtmpl": {"default": "%(title)s"}}) + + with pytest.raises(ValueError, match="Unsupported YTPTube output template callable"): + ytdlp.prepare_outtmpl("%(ytp_unknown:8)s", {"title": "x"}) + + def test_prepare_outtmpl_rejects_invalid_random_length(self) -> None: + ytdlp = YTDLP(params={"outtmpl": {"default": "%(title)s"}}) + + with pytest.raises(ValueError, match="ytp_random length must be an integer"): + ytdlp.prepare_outtmpl("%(ytp_random:nope)s", {"title": "x"}) + + +class TestOuttmpl: + def test_rewrite_outtmpl_uses_shared_cache_per_call_group(self) -> None: + cache: dict[str, object] = {} + template = "%(ytp_random:8)s/%(ytp_random:8)s.%(ext)s" + + rewritten, info = rewrite_outtmpl(template, {"ext": "mp4"}, cache=cache) + + assert rewritten == "%(__ytptube_outtmpl_0)s/%(__ytptube_outtmpl_0)s.%(ext)s" + assert len(cache) == 1 + assert info["__ytptube_outtmpl_0"] == next(iter(cache.values())) diff --git a/app/features/ytdlp/ytdlp.py b/app/features/ytdlp/ytdlp.py index cccba7e5..9f4394d9 100644 --- a/app/features/ytdlp/ytdlp.py +++ b/app/features/ytdlp/ytdlp.py @@ -5,6 +5,7 @@ from typing import Any import yt_dlp from yt_dlp.utils import make_archive_id +from app.features.ytdlp.outtmpl import rewrite_outtmpl from app.features.ytdlp.patches import apply_ytdlp_patches from app.library.cf_solver_handler import set_cf_handler @@ -76,6 +77,8 @@ class YTDLP(yt_dlp.YoutubeDL): except Exception: pass + self._ytptube_outtmpl_info: dict[str, Any] | None = None + self._ytptube_outtmpl_cache: dict[str, Any] = {} self.archive = _ArchiveProxy(orig_file) if not YTDLP._registered: try: @@ -94,6 +97,25 @@ class YTDLP(yt_dlp.YoutubeDL): return super()._delete_downloaded_files(*args, **kwargs) + def _reset_outtmpl_cache(self) -> None: + self._ytptube_outtmpl_info = None + self._ytptube_outtmpl_cache = {} + + def prepare_outtmpl(self, outtmpl, info_dict, sanitize=False): + if self._ytptube_outtmpl_info is not info_dict: + self._ytptube_outtmpl_info = info_dict + self._ytptube_outtmpl_cache = {} + + outtmpl, enriched = rewrite_outtmpl(outtmpl, info_dict, cache=self._ytptube_outtmpl_cache) + + return super().prepare_outtmpl(outtmpl, enriched, sanitize=sanitize) + + def process_info(self, info_dict): + try: + return super().process_info(info_dict) + finally: + self._reset_outtmpl_cache() + def record_download_archive(self, info_dict) -> None: if not self.params.get("download_archive"): return diff --git a/app/tests/test_download.py b/app/tests/test_download.py index f7731fc0..270f1296 100644 --- a/app/tests/test_download.py +++ b/app/tests/test_download.py @@ -310,8 +310,9 @@ class TestDownloadFlow: ) class FakeYTDLP: - def __init__(self, params): + def __init__(self, params, enable_custom_outtmpl=False): self.params = params + self.enable_custom_outtmpl = enable_custom_outtmpl self._download_retcode = 0 self._interrupted = False