feat: extra placeholders for output template

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_*:<args>)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

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"^(?P<name>ytp_[A-Za-z0-9_]+)(?P<args>(?::[A-Za-z0-9_]+)*)(?P<rest>.*)$")
+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:<length>)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

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()))

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

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