[FEAT] ADd new yt-dlp merger class.

This commit is contained in:
arabcoders 2025-10-27 19:57:05 +03:00
parent b888e67877
commit 2b9b36eacf
2 changed files with 702 additions and 38 deletions

View file

@ -1,13 +1,208 @@
import logging
import shlex
from pathlib import Path
from .config import Config
from .Presets import Preset, Presets
from .Utils import REMOVE_KEYS, arg_converter, calc_download_path, load_cookies, merge_dict
from .Utils import REMOVE_KEYS, arg_converter, calc_download_path, create_cookies_file, merge_dict
LOG: logging.Logger = logging.getLogger("YTDLPOpts")
class ARGSMerger:
def __init__(self):
self.args: list[str] = []
"Args merger"
def add(self, args: str) -> "ARGSMerger":
"""
Add command options for yt-dlp.
Args:
args (str): The command options for yt-dlp to add
Returns:
YTDLPCli: The instance of the class
"""
if not args or not isinstance(args, str) or len(args) < 2:
return self
_args = shlex.split(args)
if len(_args) > 0:
self.args.extend(_args)
return self
def as_string(self) -> str:
"""
Get all the options as a string.
Returns:
str: The options as a string
"""
return str(self)
def as_dict(self) -> dict:
"""
Get all the options as a dict.
Returns:
dict: The options as a dict
"""
return shlex.split(shlex.join(self.args))
def as_ytdlp(self) -> dict:
"""
Get all options as yt-dlp JSON options.
Returns:
dict: The options as a dict for yt-dlp
"""
return arg_converter(args=shlex.join(self.args), level=False)
def __str__(self) -> str:
"""
Get all the options as a string.
Returns:
str: The options as a string
"""
return shlex.join(self.args)
@staticmethod
def get_instance() -> "ARGSMerger":
"""
Get the instance of the class.
Returns:
Presets: The instance of the class
"""
return ARGSMerger()
def reset(self) -> "ARGSMerger":
"""
Reset the command options.
Returns:
YTDLPCli: The instance of the class
"""
self.args = []
return self
class YTDLPCli:
"""
Build yt-dlp CLI command.
Th logic is simple
1. User provided fields have highest priority
2. Preset provided fields have second priority
3. Defaults have lowest priority
"""
def __init__(self, item, config: Config | None = None):
"""
Initialize the CLI builder.
Args:
item: Item request
config (Config|None): The Config instance (optional)
"""
from .ItemDTO import Item
if not isinstance(item, Item):
msg = f"Expected Item instance, got {type(item).__name__}"
raise ValueError(msg)
self.item = item
self.preset = item.get_preset()
self._config = config or Config.get_instance()
def build(self) -> tuple[str, dict]:
"""
Build the CLI command following make_command logic.
Returns:
tuple[str, dict]: (command_string, info_dict)
"""
template: str | None = None
save_path: str | None = None
cookie_file: Path | None = None
cli_args = ARGSMerger.get_instance()
if self.item.cookies:
cookie_file = create_cookies_file(self.item.cookies)
if self.item.folder:
save_path = str(Path(self._config.download_path) / self.item.folder.lstrip("/"))
if self.item.template:
template = self.item.template
if self.preset:
if self.preset.cookies and cookie_file is None:
cookie_file = self.preset.get_cookie_file(config=self._config)
if self.preset.folder and save_path is None:
save_path = str(Path(self._config.download_path) / self.preset.folder.lstrip("/"))
if self.preset.template and template is None:
template = self.preset.template
if self.preset.cli:
cli_args.add(self.preset.cli)
if self.item.cli:
cli_args.add(self.item.cli)
if not save_path:
save_path = self._config.download_path
if not template:
template = self._config.output_template
if cookie_file:
cli_args.add(f'--cookies "{cookie_file!s}"')
if template:
cli_args.add(f'--output "{template}"')
if save_path:
cli_args.add(f'--paths "home:{save_path}"')
cli_args.add(f'--paths "temp:{self._config.temp_path}"')
if self.item.url:
cli_args.add(self.item.url)
command = str(cli_args)
for k, v in self._config.get_replacers().items():
command = command.replace(f"%({k})s", v if isinstance(v, str) else str(v))
info = {
"command": command,
"dict": cli_args.as_dict(),
"ytdlp": cli_args.as_ytdlp(),
"merged": {
"template": template,
"save_path": save_path,
"cookie_file": str(cookie_file) if cookie_file else None,
},
}
return command, info
class YTDLPOpts:
def __init__(self):
self._config: Config = Config.get_instance()
@ -102,22 +297,16 @@ class YTDLPOpts:
self._preset_opts = {}
self._preset_cli = preset.cli
except Exception as e:
msg = f"Invalid preset '{preset.name}' command options for yt-dlp. '{e!s}'."
msg: str = f"Invalid preset '{preset.name}' command options for yt-dlp. '{e!s}'."
raise ValueError(msg) from e
if preset.cookies:
file: Path = Path(self._config.config_path) / "cookies" / f"{preset.id}.txt"
if not file.parent.exists():
file.parent.mkdir(parents=True, exist_ok=True)
file.write_text(preset.cookies)
try:
load_cookies(file)
self._preset_opts["cookiefile"] = str(file)
file: Path | None = preset.get_cookies_file(config=self._config)
if file and file.exists():
self._preset_opts["cookiefile"] = str(file)
except ValueError as e:
LOG.error(f"Failed to load '{preset.name}' cookies from '{file}'. {e!s}")
LOG.error(f"Failed to load '{preset.name}' cookies. {e!s}")
if preset.template:
self._preset_opts["outtmpl"] = {"default": preset.template, "chapter": self._config.output_template_chapter}

View file

@ -172,7 +172,6 @@ class TestYTDLPOpts:
with (
patch("app.library.YTDLPOpts.Config") as mock_config,
patch("app.library.YTDLPOpts.Presets") as mock_presets,
patch("app.library.YTDLPOpts.load_cookies") as mock_load_cookies,
):
# Mock config
mock_config_instance = Mock()
@ -188,23 +187,23 @@ class TestYTDLPOpts:
mock_preset.template = None
mock_preset.folder = None
# Mock cookie_file path
mock_cookie_path = Mock()
mock_cookie_path.exists.return_value = True
mock_cookie_path.__str__ = Mock(return_value="/test/config/cookies/cookie_preset.txt")
mock_preset.get_cookies_file.return_value = mock_cookie_path
mock_presets_instance = Mock()
mock_presets_instance.get.return_value = mock_preset
mock_presets.get_instance.return_value = mock_presets_instance
# Use real Path but mock file operations
with (
patch("pathlib.Path.exists", return_value=False),
patch("pathlib.Path.mkdir"),
patch("pathlib.Path.write_text"),
):
opts = YTDLPOpts()
opts.preset("cookie_preset")
opts = YTDLPOpts()
opts.preset("cookie_preset")
# Check that the cookie file would be created at the right path
expected_path = "/test/config/cookies/cookie_preset.txt"
assert opts._preset_opts["cookiefile"] == expected_path
mock_load_cookies.assert_called_once()
# Check that the cookie file would be created at the right path
expected_path = "/test/config/cookies/cookie_preset.txt"
assert opts._preset_opts["cookiefile"] == expected_path
mock_preset.get_cookies_file.assert_called_once_with(config=mock_config_instance)
def test_preset_with_invalid_cli_raises_error(self):
"""Test that preset with invalid CLI raises ValueError."""
@ -443,7 +442,6 @@ class TestYTDLPOpts:
with (
patch("app.library.YTDLPOpts.Config") as mock_config,
patch("app.library.YTDLPOpts.Presets") as mock_presets,
patch("app.library.YTDLPOpts.load_cookies") as mock_load_cookies,
patch("app.library.YTDLPOpts.LOG") as mock_log,
):
# Mock config
@ -460,24 +458,24 @@ class TestYTDLPOpts:
mock_preset.template = None
mock_preset.folder = None
# Mock cookie loading failure
mock_preset.get_cookies_file.side_effect = ValueError("Invalid cookies")
mock_presets_instance = Mock()
mock_presets_instance.get.return_value = mock_preset
mock_presets.get_instance.return_value = mock_presets_instance
# Mock cookie loading failure
mock_load_cookies.side_effect = ValueError("Invalid cookies")
opts = YTDLPOpts()
opts.preset("cookie_preset")
with patch("pathlib.Path.exists", return_value=True), patch("pathlib.Path.write_text"):
opts = YTDLPOpts()
opts.preset("cookie_preset")
# Should log error but not raise
mock_log.error.assert_called_once()
error_args = mock_log.error.call_args[0][0]
assert "Failed to load" in error_args
assert "Cookie Preset" in error_args
# Should log error but not raise
mock_log.error.assert_called_once()
error_args = mock_log.error.call_args[0][0]
assert "Failed to load 'Cookie Preset' cookies" in error_args
# cookiefile should not be set
assert "cookiefile" not in opts._preset_opts
# cookiefile should not be set
assert "cookiefile" not in opts._preset_opts
def test_replacer_substitution_in_cli(self):
"""Test that CLI arguments get replacer substitution."""
@ -506,3 +504,480 @@ class TestYTDLPOpts:
# Should replace %(home)s and %(user)s
expected_cli = "--output /actual/home/testuser"
mock_converter.assert_called_once_with(args=expected_cli, level=True)
class TestARGSMerger:
"""Test the ARGSMerger class."""
def test_constructor_initializes_empty_args(self):
"""Test that ARGSMerger constructor initializes with empty args list."""
from app.library.YTDLPOpts import ARGSMerger
merger = ARGSMerger()
assert merger.args == []
def test_add_valid_args(self):
"""Test adding valid command-line arguments."""
from app.library.YTDLPOpts import ARGSMerger
merger = ARGSMerger()
result = merger.add("--format best")
assert result is merger # Returns self for chaining
assert merger.args == ["--format", "best"]
def test_add_multiple_args_with_chaining(self):
"""Test adding multiple arguments using method chaining."""
from app.library.YTDLPOpts import ARGSMerger
merger = ARGSMerger()
merger.add("--format best").add("--output test.mp4").add("--no-playlist")
assert merger.args == ["--format", "best", "--output", "test.mp4", "--no-playlist"]
def test_add_args_with_quotes(self):
"""Test adding arguments containing quoted strings."""
from app.library.YTDLPOpts import ARGSMerger
merger = ARGSMerger()
merger.add('--output "%(title)s.%(ext)s"')
assert merger.args == ["--output", "%(title)s.%(ext)s"]
def test_add_complex_args(self):
"""Test adding complex arguments with multiple values."""
from app.library.YTDLPOpts import ARGSMerger
merger = ARGSMerger()
merger.add("--format 'bestvideo[height<=1080]+bestaudio/best'")
assert "--format" in merger.args
assert "bestvideo[height<=1080]+bestaudio/best" in merger.args
def test_add_empty_string_returns_self(self):
"""Test that adding empty string returns self without modifying args."""
from app.library.YTDLPOpts import ARGSMerger
merger = ARGSMerger()
result = merger.add("")
assert result is merger
assert merger.args == []
def test_add_short_string_returns_self(self):
"""Test that adding short string (len < 2) returns self without modifying args."""
from app.library.YTDLPOpts import ARGSMerger
merger = ARGSMerger()
result = merger.add("a")
assert result is merger
assert merger.args == []
def test_add_non_string_returns_self(self):
"""Test that adding non-string returns self without modifying args."""
from app.library.YTDLPOpts import ARGSMerger
merger = ARGSMerger()
result = merger.add(123) # type: ignore
assert result is merger
assert merger.args == []
def test_as_string(self):
"""Test converting args to string."""
from app.library.YTDLPOpts import ARGSMerger
merger = ARGSMerger()
merger.add("--format best").add("--output test.mp4")
result = merger.as_string()
assert result == "--format best --output test.mp4"
def test_str_method(self):
"""Test __str__ method."""
from app.library.YTDLPOpts import ARGSMerger
merger = ARGSMerger()
merger.add("--format best").add("--output test.mp4")
result = str(merger)
assert result == "--format best --output test.mp4"
def test_as_dict(self):
"""Test converting args to dict."""
from app.library.YTDLPOpts import ARGSMerger
merger = ARGSMerger()
merger.add("--format best").add("--output test.mp4")
result = merger.as_dict()
assert isinstance(result, list)
assert result == ["--format", "best", "--output", "test.mp4"]
def test_as_ytdlp(self):
"""Test converting args to yt-dlp JSON options."""
from app.library.YTDLPOpts import ARGSMerger
merger = ARGSMerger()
merger.add("--format best")
with patch("app.library.YTDLPOpts.arg_converter") as mock_converter:
mock_converter.return_value = {"format": "best"}
result = merger.as_ytdlp()
assert result == {"format": "best"}
mock_converter.assert_called_once_with(args="--format best", level=False)
def test_get_instance(self):
"""Test get_instance static method."""
from app.library.YTDLPOpts import ARGSMerger
merger = ARGSMerger.get_instance()
assert isinstance(merger, ARGSMerger)
assert merger.args == []
def test_reset(self):
"""Test reset method clears args."""
from app.library.YTDLPOpts import ARGSMerger
merger = ARGSMerger()
merger.add("--format best").add("--output test.mp4")
assert len(merger.args) > 0
result = merger.reset()
assert result is merger
assert merger.args == []
def test_args_with_special_characters(self):
"""Test handling arguments with special characters."""
from app.library.YTDLPOpts import ARGSMerger
merger = ARGSMerger()
merger.add('--postprocessor-args "-movflags +faststart"')
assert "--postprocessor-args" in merger.args
assert "-movflags +faststart" in merger.args
class TestYTDLPCli:
"""Test the YTDLPCli class."""
@patch("app.library.Presets.Presets")
@patch("app.library.YTDLPOpts.Config")
def test_constructor_with_valid_item(self, mock_config, mock_presets):
"""Test YTDLPCli constructor with valid Item."""
from app.library.ItemDTO import Item
from app.library.YTDLPOpts import YTDLPCli
mock_config_instance = Mock()
mock_config_instance.download_path = "/downloads"
mock_config_instance.output_template = "%(title)s.%(ext)s"
mock_config_instance.get_replacers.return_value = {}
mock_config_instance.get_replacers.return_value = {}
mock_config.get_instance.return_value = mock_config_instance
mock_presets.get_instance.return_value.get.return_value = None
item = Item(url="https://example.com/video")
cli = YTDLPCli(item=item)
assert cli.item is item
assert cli._config is mock_config_instance
def test_constructor_with_invalid_type_raises_error(self):
"""Test YTDLPCli constructor raises error with non-Item type."""
from app.library.YTDLPOpts import YTDLPCli
with pytest.raises(ValueError, match="Expected Item instance"):
YTDLPCli(item="not an item") # type: ignore
@patch("app.library.Presets.Presets")
@patch("app.library.YTDLPOpts.Config")
def test_build_with_user_fields_only(self, mock_config, mock_presets):
"""Test build with only user-provided fields."""
from app.library.ItemDTO import Item
from app.library.YTDLPOpts import YTDLPCli
mock_config_instance = Mock()
mock_config_instance.download_path = "/downloads"
mock_config_instance.output_template = "%(title)s.%(ext)s"
mock_config_instance.get_replacers.return_value = {}
mock_config.get_instance.return_value = mock_config_instance
mock_presets.get_instance.return_value.get.return_value = None
item = Item(
url="https://example.com/video",
folder="myfolder",
template="%(id)s.%(ext)s",
cli="--format best",
)
cli = YTDLPCli(item=item)
command, info = cli.build()
assert isinstance(command, str)
assert isinstance(info, dict)
assert "command" in info
assert "dict" in info
assert "ytdlp" in info
assert "merged" in info
assert info["merged"]["template"] == "%(id)s.%(ext)s"
assert info["merged"]["save_path"] == "/downloads/myfolder"
assert "--format best" in command
@patch("app.library.Presets.Presets")
@patch("app.library.YTDLPOpts.Config")
def test_build_with_preset_fields_fallback(self, mock_config, mock_presets):
"""Test build falls back to preset fields when user doesn't provide them."""
from app.library.ItemDTO import Item
from app.library.Presets import Preset
from app.library.YTDLPOpts import YTDLPCli
mock_config_instance = Mock()
mock_config_instance.download_path = "/downloads"
mock_config_instance.output_template = "%(title)s.%(ext)s"
mock_config_instance.get_replacers.return_value = {}
mock_config.get_instance.return_value = mock_config_instance
# Create a preset with fields
preset = Preset(
name="test_preset",
folder="preset_folder",
template="%(channel)s/%(title)s.%(ext)s",
cli="--format 720p",
)
mock_presets.get_instance.return_value.get.return_value = preset
# Item without folder/template - should use preset's
item = Item(url="https://example.com/video", preset="test_preset")
cli = YTDLPCli(item=item)
command, info = cli.build()
# Should use preset values
assert info["merged"]["template"] == "%(channel)s/%(title)s.%(ext)s"
assert info["merged"]["save_path"] == "/downloads/preset_folder"
assert "--format 720p" in command
@patch("app.library.Presets.Presets")
@patch("app.library.YTDLPOpts.Config")
def test_build_user_fields_override_preset(self, mock_config, mock_presets):
"""Test that user fields override preset fields."""
from app.library.ItemDTO import Item
from app.library.Presets import Preset
from app.library.YTDLPOpts import YTDLPCli
mock_config_instance = Mock()
mock_config_instance.download_path = "/downloads"
mock_config_instance.output_template = "%(title)s.%(ext)s"
mock_config_instance.get_replacers.return_value = {}
mock_config.get_instance.return_value = mock_config_instance
preset = Preset(
name="test_preset",
folder="preset_folder",
template="%(channel)s/%(title)s.%(ext)s",
cli="--format 720p",
)
mock_presets.get_instance.return_value.get.return_value = preset
# Item with user fields - should override preset
item = Item(
url="https://example.com/video",
preset="test_preset",
folder="user_folder",
template="%(id)s.%(ext)s",
cli="--format best",
)
cli = YTDLPCli(item=item)
command, info = cli.build()
# Should use user values, not preset
assert info["merged"]["template"] == "%(id)s.%(ext)s"
assert info["merged"]["save_path"] == "/downloads/user_folder"
# User CLI should appear after preset CLI in command
assert "--format best" in command
@patch("app.library.Presets.Presets")
@patch("app.library.YTDLPOpts.Config")
def test_build_with_default_fallback(self, mock_config, mock_presets):
"""Test build falls back to defaults when neither user nor preset provide fields."""
from app.library.ItemDTO import Item
from app.library.YTDLPOpts import YTDLPCli
mock_config_instance = Mock()
mock_config_instance.download_path = "/default/downloads"
mock_config_instance.output_template = "%(title)s.%(ext)s"
mock_config_instance.get_replacers.return_value = {}
mock_config.get_instance.return_value = mock_config_instance
mock_presets.get_instance.return_value.get.return_value = None
# Item with minimal fields
item = Item(url="https://example.com/video")
cli = YTDLPCli(item=item)
_, info = cli.build()
# Should use default values
assert info["merged"]["template"] == "%(title)s.%(ext)s"
assert info["merged"]["save_path"] == "/default/downloads"
@patch("app.library.Presets.Presets")
@patch("app.library.YTDLPOpts.create_cookies_file")
@patch("app.library.YTDLPOpts.Config")
def test_build_with_cookies_from_user(self, mock_config, mock_create_cookies, mock_presets):
"""Test build with cookies from user."""
from app.library.ItemDTO import Item
from app.library.YTDLPOpts import YTDLPCli
mock_config_instance = Mock()
mock_config_instance.download_path = "/downloads"
mock_config_instance.output_template = "%(title)s.%(ext)s"
mock_config_instance.get_replacers.return_value = {}
mock_config.get_instance.return_value = mock_config_instance
mock_cookie_path = Mock()
mock_cookie_path.__str__ = Mock(return_value="/tmp/cookies.txt")
mock_create_cookies.return_value = mock_cookie_path
mock_presets.get_instance.return_value.get.return_value = None
item = Item(url="https://example.com/video", cookies="session_id=abc123")
cli = YTDLPCli(item=item)
command, info = cli.build()
mock_create_cookies.assert_called_once_with("session_id=abc123")
assert "--cookies" in command
assert "/tmp/cookies.txt" in command
assert info["merged"]["cookie_file"] == "/tmp/cookies.txt"
@patch("app.library.Presets.Presets")
@patch("app.library.YTDLPOpts.Config")
def test_build_with_cookies_from_preset(self, mock_config, mock_presets):
"""Test build with cookies from preset when user doesn't provide."""
from app.library.ItemDTO import Item
from app.library.Presets import Preset
from app.library.YTDLPOpts import YTDLPCli
mock_config_instance = Mock()
mock_config_instance.download_path = "/downloads"
mock_config_instance.output_template = "%(title)s.%(ext)s"
mock_config_instance.get_replacers.return_value = {}
mock_config.get_instance.return_value = mock_config_instance
mock_cookie_path = Mock()
mock_cookie_path.__str__ = Mock(return_value="/preset/cookies.txt")
preset = Mock(spec=Preset)
preset.name = "test_preset"
preset.cookies = "preset_cookies"
preset.get_cookie_file = Mock(return_value=mock_cookie_path)
preset.folder = None
preset.template = None
preset.cli = None
mock_presets.get_instance.return_value.get.return_value = preset
item = Item(url="https://example.com/video", preset="test_preset")
cli = YTDLPCli(item=item)
command, _ = cli.build()
preset.get_cookie_file.assert_called_once_with(config=mock_config_instance)
assert "--cookies" in command
assert "/preset/cookies.txt" in command
@patch("app.library.Presets.Presets")
@patch("app.library.YTDLPOpts.Config")
def test_build_with_absolute_folder_path(self, mock_config, mock_presets):
"""Test build with absolute folder path from user."""
from app.library.ItemDTO import Item
from app.library.YTDLPOpts import YTDLPCli
mock_config_instance = Mock()
mock_config_instance.download_path = "/downloads"
mock_config_instance.output_template = "%(title)s.%(ext)s"
mock_config_instance.get_replacers.return_value = {}
mock_config.get_instance.return_value = mock_config_instance
mock_presets.get_instance.return_value.get.return_value = None
# Absolute path gets leading slash stripped and joined with download_path
# This is the actual behavior: /absolute/path -> absolute/path -> /downloads/absolute/path
item = Item(url="https://example.com/video", folder="/absolute/path")
cli = YTDLPCli(item=item)
command, info = cli.build()
# The implementation strips leading slash and joins with download_path
assert info["merged"]["save_path"] == "/downloads/absolute/path"
assert "--paths" in command
@patch("app.library.Presets.Presets")
@patch("app.library.YTDLPOpts.Config")
def test_build_includes_url_in_command(self, mock_config, mock_presets):
"""Test that build includes the URL in the final command."""
from app.library.ItemDTO import Item
from app.library.YTDLPOpts import YTDLPCli
mock_config_instance = Mock()
mock_config_instance.download_path = "/downloads"
mock_config_instance.output_template = "%(title)s.%(ext)s"
mock_config_instance.get_replacers.return_value = {}
mock_config.get_instance.return_value = mock_config_instance
mock_presets.get_instance.return_value.get.return_value = None
item = Item(url="https://youtube.com/watch?v=test123")
cli = YTDLPCli(item=item)
command, _ = cli.build()
assert "https://youtube.com/watch?v=test123" in command
@patch("app.library.Presets.Presets")
@patch("app.library.YTDLPOpts.Config")
def test_build_cli_args_priority_order(self, mock_config, mock_presets):
"""Test that CLI args are added in correct priority order (preset first, user last)."""
from app.library.ItemDTO import Item
from app.library.Presets import Preset
from app.library.YTDLPOpts import YTDLPCli
mock_config_instance = Mock()
mock_config_instance.download_path = "/downloads"
mock_config_instance.output_template = "%(title)s.%(ext)s"
mock_config_instance.get_replacers.return_value = {}
mock_config.get_instance.return_value = mock_config_instance
preset = Preset(name="test_preset", cli="--format 720p --extract-audio")
mock_presets.get_instance.return_value.get.return_value = preset
item = Item(url="https://example.com/video", preset="test_preset", cli="--format best --no-playlist")
cli = YTDLPCli(item=item)
_, info = cli.build()
# User args should come after preset args in the command
# This ensures user args can override preset args
args_list = info["dict"]
preset_format_idx = args_list.index("720p") if "720p" in args_list else -1
user_format_idx = args_list.index("best") if "best" in args_list else -1
# User's 'best' should appear after preset's '720p'
assert user_format_idx > preset_format_idx