danvics/ytptube
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1

refactor: isolate yt_dlp related code into submodule

Browse source
This commit is contained in:
arabcoders 2026-01-27 18:15:12 +03:00
parent 3383d5d5dc
commit 2c5edc64f4
39 changed files with 1271 additions and 1388 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
app/features/conditions/router.py
View file

@ -87,8 +87,8 @@ async def conditions_test(request: Request, encoder: Encoder, cache: Cache, conf
preset: str = params.get("preset", config.default_preset)
key: str = cache.hash(url + str(preset))
if not cache.has(key):
from app.library.downloads.extractor import fetch_info
from app.library.YTDLPOpts import YTDLPOpts
from app.features.ytdlp.extractor import fetch_info
from app.features.ytdlp.ytdlp_opts import YTDLPOpts
(data, _) = await fetch_info(
config=YTDLPOpts.get_instance().preset(name=preset).get_all(),
@ -115,7 +115,7 @@ async def conditions_test(request: Request, encoder: Encoder, cache: Cache, conf
)
try:
from app.library.mini_filter import match_str
from app.features.ytdlp.mini_filter import match_str
status: bool = match_str(cond, data)
except Exception as e:

5
app/features/conditions/schemas.py
View file

@ -6,8 +6,7 @@ from pydantic import BaseModel, ConfigDict, Field, field_validator, model_valida
from app.features.core.schemas import Pagination
from app.features.core.utils import parse_int
from app.library.mini_filter import match_str
from app.library.Utils import arg_converter
from app.features.ytdlp.mini_filter import match_str
class Condition(BaseModel):
@ -38,6 +37,8 @@ class Condition(BaseModel):
if not value:
return ""
try:
from app.features.ytdlp.utils import arg_converter
arg_converter(args=value)
except ModuleNotFoundError:
return value

2
app/features/conditions/service.py
View file

@ -4,8 +4,8 @@ from aiohttp import web
from app.features.conditions.models import ConditionModel
from app.features.conditions.repository import ConditionsRepository
from app.features.ytdlp.mini_filter import match_str
from app.library.Events import EventBus, Events
from app.library.mini_filter import match_str
from app.library.Singleton import Singleton
LOG: logging.Logger = logging.getLogger("feature.conditions")

3
app/features/presets/migration.py
View file

@ -9,7 +9,6 @@ from app.features.core.migration import Migration as FeatureMigration
from app.features.presets.schemas import Preset
from app.features.presets.utils import preset_name
from app.library.config import Config
from app.library.Utils import arg_converter
if TYPE_CHECKING:
from app.features.presets.repository import PresetsRepository
@ -87,6 +86,8 @@ class Migration(FeatureMigration):
if cli:
try:
from app.features.ytdlp.utils import arg_converter
arg_converter(args=cli, level=True)
except Exception as exc:
LOG.warning("Skipping preset '%s' due to invalid CLI: %s", name, exc)

4
app/features/presets/schemas.py
View file

@ -9,7 +9,7 @@ from app.features.core.schemas import Pagination
from app.features.core.utils import parse_int
from app.features.presets.utils import preset_name
from app.library.config import Config
from app.library.Utils import arg_converter, create_cookies_file
from app.library.Utils import create_cookies_file
class Preset(BaseModel):
@ -58,6 +58,8 @@ class Preset(BaseModel):
return ""
try:
from app.features.ytdlp.utils import arg_converter
arg_converter(args=value, level=True)
except Exception as e:
msg = f"Invalid command options for yt-dlp: {e!s}"

4
app/features/tasks/definitions/handlers/generic.py
View file

@ -20,11 +20,11 @@ from app.features.tasks.definitions.schemas import (
ExtractionRule,
TaskDefinition,
)
from app.features.ytdlp.extractor import fetch_info
from app.features.ytdlp.utils import get_archive_id
from app.library.cache import Cache
from app.library.config import Config
from app.library.downloads.extractor import fetch_info
from app.library.httpx_client import Globals, build_request_headers, get_async_client, resolve_curl_transport
from app.library.Utils import get_archive_id
from ._base_handler import BaseHandler

4
app/features/tasks/definitions/handlers/rss.py
View file

@ -5,9 +5,9 @@ from typing import TYPE_CHECKING, Any
from xml.etree.ElementTree import Element
from app.features.tasks.definitions.results import HandleTask, TaskFailure, TaskItem, TaskResult
from app.features.ytdlp.extractor import fetch_info
from app.features.ytdlp.utils import get_archive_id
from app.library.cache import Cache
from app.library.downloads.extractor import fetch_info
from app.library.Utils import get_archive_id
from ._base_handler import BaseHandler

2
app/features/tasks/definitions/handlers/tver.py
View file

@ -2,7 +2,7 @@ import logging
import re
from app.features.tasks.definitions.results import HandleTask, TaskFailure, TaskItem, TaskResult
from app.library.Utils import get_archive_id
from app.features.ytdlp.utils import get_archive_id
from ._base_handler import BaseHandler

2
app/features/tasks/definitions/handlers/twitch.py
View file

@ -4,7 +4,7 @@ from typing import TYPE_CHECKING
from xml.etree.ElementTree import Element
from app.features.tasks.definitions.results import HandleTask, TaskFailure, TaskItem, TaskResult
from app.library.Utils import get_archive_id
from app.features.ytdlp.utils import get_archive_id
from ._base_handler import BaseHandler

2
app/features/tasks/definitions/handlers/youtube.py
View file

@ -4,7 +4,7 @@ from typing import TYPE_CHECKING, Any
from xml.etree.ElementTree import Element
from app.features.tasks.definitions.results import HandleTask, TaskFailure, TaskItem, TaskResult
from app.library.Utils import get_archive_id
from app.features.ytdlp.utils import get_archive_id
from ._base_handler import BaseHandler

8
app/features/tasks/definitions/results.py
View file

@ -5,7 +5,7 @@ from pathlib import Path
from typing import Any
from app.features.tasks.schemas import Task as TaskSchema
from app.library.YTDLPOpts import YTDLPOpts
from app.features.ytdlp.ytdlp_opts import YTDLPOpts
from .utils import split_inspect_metadata
@ -19,7 +19,7 @@ class HandleTask(TaskSchema):
YTDLPOpts: The yt-dlp options.
"""
from app.library.YTDLPOpts import YTDLPOpts
from app.features.ytdlp.ytdlp_opts import YTDLPOpts
params: YTDLPOpts = YTDLPOpts.get_instance()
@ -90,7 +90,7 @@ class HandleTask(TaskSchema):
indicating if the operation was successful, and a message.
"""
from app.library.ytdlp import fetch_info
from app.features.ytdlp.ytdlp import fetch_info
if not self.url:
return ({}, False, "No URL found in task parameters.")
@ -122,7 +122,7 @@ class HandleTask(TaskSchema):
tuple[bool, str] | dict[str, Any]: Either an error tuple or a dict with 'file' and 'items' keys.
"""
from app.library.ytdlp import fetch_info
from app.features.ytdlp.ytdlp import fetch_info
if not self.url:
return (False, "No URL found in task parameters.")

3
app/features/tasks/router.py
View file

@ -12,12 +12,13 @@ from app.features.tasks.definitions.results import TaskFailure, TaskResult
from app.features.tasks.definitions.service import TaskHandle
from app.features.tasks.repository import TasksRepository
from app.features.tasks.schemas import Task, TaskList, TaskPatch
from app.features.ytdlp.utils import parse_outtmpl
from app.library.ag_utils import ag
from app.library.config import Config
from app.library.encoder import Encoder
from app.library.Events import EventBus, Events
from app.library.router import route
from app.library.Utils import get_channel_images, get_file, parse_outtmpl, validate_url
from app.library.Utils import get_channel_images, get_file, validate_url
if TYPE_CHECKING:
from pathlib import Path

4
app/features/tasks/schemas.py
View file

@ -98,9 +98,9 @@ class Task(BaseModel):
if not value:
return ""
from app.library.Utils import arg_converter
try:
from app.features.ytdlp.utils import arg_converter
arg_converter(args=value)
except Exception as e:
msg = f"Invalid command options for yt-dlp: {e!s}"

0
app/features/ytdlp/__init__.py Normal file
View file

59
app/library/downloads/extractor.py → app/features/ytdlp/extractor.py
View file

@ -6,10 +6,12 @@ from concurrent.futures import ProcessPoolExecutor
from pathlib import Path
from typing import Any
from app.library.LogWrapper import LogWrapper
from app.library.mini_filter import match_str
from aiohttp import web
from app.features.ytdlp.utils import _DATA, LogWrapper, get_archive_id
from app.features.ytdlp.ytdlp import YTDLP
from app.library.Services import Services
from app.library.Singleton import Singleton
from app.library.ytdlp import YTDLP
LOG: logging.Logger = logging.getLogger("downloads.extractor")
@ -61,6 +63,11 @@ class ExtractorPool(metaclass=Singleton):
"""
return cls()
def attach(self, app: web.Application) -> None:
"""Attach the extractor pool to the application (no-op for now)."""
app.on_shutdown.append(self.shutdown)
Services.get_instance().add(__class__.__name__, self)
def _ensure_initialized(self, config: ExtractorConfig) -> None:
"""
Ensure pool and semaphore are initialized.
@ -190,22 +197,6 @@ def _sanitize_config(config: dict[str, Any]) -> dict[str, Any]:
return sanitized
def _get_archive_id_dict(url: str) -> dict[str, str | None]:
"""
Get archive ID for logging purposes.
Args:
url: URL to get archive ID for
Returns:
Dictionary with id, ie_key, and archive_id
"""
from app.library.Utils import get_archive_id
return get_archive_id(url=url)
def extract_info_sync(
config: dict[str, Any],
url: str,
@ -233,23 +224,14 @@ def extract_info_sync(
tuple[dict | None, list[dict]]: Extracted information and captured logs.
"""
params: dict[str, Any] = {
**config,
"simulate": True,
"color": "no_color",
"extract_flat": True,
"skip_download": True,
"ignoreerrors": True,
"ignore_no_formats_error": True,
}
params: dict[str, Any] = {**config, **_DATA.YTDLP_PARAMS, "simulate": True}
if debug:
params["verbose"] = True
else:
params["quiet"] = True
params.pop("quiet", None)
log_wrapper = LogWrapper()
id_dict: dict[str, str | None] = _get_archive_id_dict(url=url)
id_dict: dict[str, str | None] = get_archive_id(url=url)
archive_id: str | None = f".{id_dict['id']}" if id_dict.get("id") else None
log_wrapper.add_target(
@ -297,9 +279,14 @@ def extract_info_sync(
if not data:
return (data, captured_logs)
data["is_premiere"] = match_str("media_type=video & duration & is_live", data)
if not data["is_premiere"]:
data["is_premiere"] = "video" == data.get("media_type") and "is_upcoming" == data.get("live_status")
try:
from app.features.ytdlp.mini_filter import match_str
data["is_premiere"] = match_str("media_type=video & duration & is_live", data)
if not data["is_premiere"]:
data["is_premiere"] = "video" == data.get("media_type") and "is_upcoming" == data.get("live_status")
except ImportError:
pass
result = YTDLP.sanitize_info(data, remove_private_keys=True) if sanitize_info else data
@ -399,7 +386,3 @@ async def fetch_info(
)
finally:
semaphore.release()
async def shutdown_extractor() -> None:
await ExtractorPool.get_instance().shutdown()

0
app/library/mini_filter.py → app/features/ytdlp/mini_filter.py
View file

36
app/features/ytdlp/tests/test_ytdlp_extractor.py Normal file
View file

@ -0,0 +1,36 @@
from unittest.mock import MagicMock, patch
from app.features.ytdlp.extractor import extract_info_sync
class TestExtractInfo:
"""Test the extract_info function."""
@patch("app.features.ytdlp.extractor.YTDLP")
def test_extract_info_basic(self, mock_ytdlp_class):
"""Test basic extract_info functionality."""
mock_ytdlp = MagicMock()
mock_ytdlp.extract_info.return_value = {"title": "Test Video", "id": "test123"}
mock_ytdlp_class.return_value = mock_ytdlp
config = {"quiet": True}
url = "https://example.com/video"
(result, logs) = extract_info_sync(config, url)
assert isinstance(result, dict), "Result should be a dictionary"
assert isinstance(logs, list), "Logs should be a list"
mock_ytdlp.extract_info.assert_called_once()
@patch("app.features.ytdlp.extractor.YTDLP")
def test_extract_info_with_debug(self, mock_ytdlp_class):
"""Test extract_info with debug enabled."""
mock_ytdlp = MagicMock()
mock_ytdlp.extract_info.return_value = {"title": "Test Video"}
mock_ytdlp_class.return_value = mock_ytdlp
config = {}
url = "https://example.com/video"
(result, logs) = extract_info_sync(config, url, debug=True)
assert isinstance(result, dict), "Result should be a dictionary"
assert isinstance(logs, list), "Logs should be a list"

2
app/tests/mini_filter_test.py → app/features/ytdlp/tests/test_ytdlp_mini_filter.py
View file

@ -1,7 +1,7 @@
import unittest
from typing import Any
from app.library.mini_filter import MiniFilter
from app.features.ytdlp.mini_filter import MiniFilter
class TestMiniFilter(unittest.TestCase):

22
app/tests/test_ytdlp_module.py → app/features/ytdlp/tests/test_ytdlp_module.py
View file

@ -1,7 +1,7 @@
from unittest.mock import MagicMock, Mock, patch
from app.library.Utils import REMOVE_KEYS
from app.library.ytdlp import YTDLP, _ArchiveProxy, ytdlp_options
from app.features.ytdlp.utils import _DATA
from app.features.ytdlp.ytdlp import YTDLP, _ArchiveProxy, ytdlp_options
class TestArchiveProxy:
@ -61,7 +61,7 @@ class TestYtDlpOptions:
def test_ignored_flags_match_remove_keys(self) -> None:
# Collect the flags that should be ignored from REMOVE_KEYS
ignored_flags: set[str] = {
f.strip() for group in REMOVE_KEYS for v in group.values() for f in v.split(",") if f.strip()
f.strip() for group in _DATA.REMOVE_KEYS for v in group.values() for f in v.split(",") if f.strip()
}
opts = ytdlp_options()
@ -84,12 +84,12 @@ class TestYTDLP:
def _create_ytdlp(self, params=None):
"""Helper to create a YTDLP instance with mocked parent __init__."""
with patch("app.library.ytdlp.yt_dlp.YoutubeDL.__init__", return_value=None):
with patch("app.features.ytdlp.ytdlp.yt_dlp.YoutubeDL.__init__", return_value=None):
ytdlp = YTDLP(params=params)
ytdlp.params = params or {}
return ytdlp
@patch("app.library.ytdlp.yt_dlp.YoutubeDL.__init__")
@patch("app.features.ytdlp.ytdlp.yt_dlp.YoutubeDL.__init__")
def test_init_patches_download_archive_param(self, mock_super_init) -> None:
"""Test that __init__ removes download_archive before calling super, then restores it."""
mock_super_init.return_value = None
@ -116,7 +116,7 @@ class TestYTDLP:
assert isinstance(ytdlp.archive, _ArchiveProxy)
assert ytdlp.archive._file == "/tmp/archive.txt"
@patch("app.library.ytdlp.yt_dlp.YoutubeDL.__init__")
@patch("app.features.ytdlp.ytdlp.yt_dlp.YoutubeDL.__init__")
def test_init_handles_no_download_archive(self, mock_super_init) -> None:
"""Test __init__ works correctly when download_archive is not in params."""
mock_super_init.return_value = None
@ -134,7 +134,7 @@ class TestYTDLP:
assert isinstance(ytdlp.archive, _ArchiveProxy), "Verify archive proxy is falsey"
assert not ytdlp.archive
@patch("app.library.ytdlp.yt_dlp.YoutubeDL.__init__")
@patch("app.features.ytdlp.ytdlp.yt_dlp.YoutubeDL.__init__")
def test_init_handles_none_params(self, mock_super_init) -> None:
"""Test __init__ handles None params gracefully."""
mock_super_init.return_value = None
@ -145,10 +145,10 @@ class TestYTDLP:
assert isinstance(ytdlp.archive, _ArchiveProxy)
assert not ytdlp.archive
@patch("app.library.ytdlp.yt_dlp.YoutubeDL._delete_downloaded_files")
@patch("app.features.ytdlp.ytdlp.yt_dlp.YoutubeDL._delete_downloaded_files")
def test_delete_downloaded_files_skips_when_interrupted(self, mock_super_delete) -> None:
"""Test _delete_downloaded_files skips cleanup when _interrupted is True."""
with patch("app.library.ytdlp.yt_dlp.YoutubeDL.__init__", return_value=None):
with patch("app.features.ytdlp.ytdlp.yt_dlp.YoutubeDL.__init__", return_value=None):
ytdlp = YTDLP(params={})
ytdlp.to_screen = Mock()
@ -164,12 +164,12 @@ class TestYTDLP:
# Should return None
assert result is None
@patch("app.library.ytdlp.yt_dlp.YoutubeDL._delete_downloaded_files")
@patch("app.features.ytdlp.ytdlp.yt_dlp.YoutubeDL._delete_downloaded_files")
def test_delete_downloaded_files_calls_super_when_not_interrupted(self, mock_super_delete) -> None:
"""Test _delete_downloaded_files calls super when not interrupted."""
mock_super_delete.return_value = "cleanup_result"
with patch("app.library.ytdlp.yt_dlp.YoutubeDL.__init__", return_value=None):
with patch("app.features.ytdlp.ytdlp.yt_dlp.YoutubeDL.__init__", return_value=None):
ytdlp = YTDLP(params={})
ytdlp._interrupted = False

168
app/tests/test_ytdlpopts.py → app/features/ytdlp/tests/test_ytdlp_opts.py
View file

@ -5,7 +5,7 @@ from unittest.mock import Mock, patch
import pytest
from app.features.presets.schemas import Preset
from app.library.YTDLPOpts import YTDLPOpts
from app.features.ytdlp.ytdlp_opts import YTDLPOpts
class TestYTDLPOpts:
@ -13,7 +13,7 @@ class TestYTDLPOpts:
def test_constructor_initializes_correctly(self):
"""Test that YTDLPOpts constructor initializes all attributes."""
with patch("app.library.YTDLPOpts.Config") as mock_config:
with patch("app.features.ytdlp.ytdlp_opts.Config") as mock_config:
mock_config_instance = Mock()
mock_config.get_instance.return_value = mock_config_instance
@ -27,7 +27,7 @@ class TestYTDLPOpts:
def test_get_instance_returns_reset_instance(self):
"""Test that get_instance returns a reset YTDLPOpts instance."""
with patch("app.library.YTDLPOpts.Config"):
with patch("app.features.ytdlp.ytdlp_opts.Config"):
opts = YTDLPOpts.get_instance()
assert isinstance(opts, YTDLPOpts)
@ -38,7 +38,10 @@ class TestYTDLPOpts:
def test_add_cli_with_valid_args(self):
"""Test adding valid CLI arguments."""
with patch("app.library.YTDLPOpts.Config"), patch("app.library.YTDLPOpts.arg_converter") as mock_converter:
with (
patch("app.features.ytdlp.ytdlp_opts.Config"),
patch("app.features.ytdlp.ytdlp_opts.arg_converter") as mock_converter,
):
mock_converter.return_value = {"format": "best"}
opts = YTDLPOpts()
@ -50,7 +53,10 @@ class TestYTDLPOpts:
def test_add_cli_with_invalid_args_raises_error(self):
"""Test that invalid CLI arguments raise ValueError."""
with patch("app.library.YTDLPOpts.Config"), patch("app.library.YTDLPOpts.arg_converter") as mock_converter:
with (
patch("app.features.ytdlp.ytdlp_opts.Config"),
patch("app.features.ytdlp.ytdlp_opts.arg_converter") as mock_converter,
):
mock_converter.side_effect = Exception("Invalid argument")
opts = YTDLPOpts()
@ -59,7 +65,7 @@ class TestYTDLPOpts:
def test_add_cli_with_empty_args_returns_self(self):
"""Test that empty or invalid args return self without processing."""
with patch("app.library.YTDLPOpts.Config"):
with patch("app.features.ytdlp.ytdlp_opts.Config"):
opts = YTDLPOpts()
# Test empty string
@ -79,7 +85,7 @@ class TestYTDLPOpts:
def test_add_with_valid_config(self):
"""Test adding configuration options."""
with patch("app.library.YTDLPOpts.Config"):
with patch("app.features.ytdlp.ytdlp_opts.Config"):
opts = YTDLPOpts()
config = {"format": "best", "quality": "720p"}
@ -91,10 +97,7 @@ class TestYTDLPOpts:
def test_add_with_user_config_filters_bad_options(self):
"""Test that user config filters out dangerous options."""
with (
patch("app.library.YTDLPOpts.Config"),
patch("app.library.YTDLPOpts.REMOVE_KEYS", [{"paths": "-P, --paths", "outtmpl": "-o, --output"}]),
):
with patch("app.features.ytdlp.ytdlp_opts.Config"):
opts = YTDLPOpts()
config = {
@ -115,9 +118,9 @@ class TestYTDLPOpts:
def test_preset_with_valid_preset(self):
"""Test applying a valid preset."""
with (
patch("app.library.YTDLPOpts.Config") as mock_config,
patch("app.features.ytdlp.ytdlp_opts.Config") as mock_config,
patch("app.features.presets.service.Presets") as mock_presets,
patch("app.library.YTDLPOpts.arg_converter") as mock_converter,
patch("app.features.ytdlp.ytdlp_opts.arg_converter") as mock_converter,
):
# Mock config
mock_config_instance = Mock()
@ -142,7 +145,7 @@ class TestYTDLPOpts:
mock_converter.return_value = {"format": "best"}
with patch("app.library.YTDLPOpts.calc_download_path") as mock_calc_path:
with patch("app.features.ytdlp.ytdlp_opts.calc_download_path") as mock_calc_path:
mock_calc_path.return_value = "/test/downloads/custom_folder"
opts = YTDLPOpts()
@ -155,7 +158,10 @@ class TestYTDLPOpts:
def test_preset_with_nonexistent_preset(self):
"""Test applying a nonexistent preset returns self."""
with patch("app.library.YTDLPOpts.Config"), patch("app.features.presets.service.Presets") as mock_presets:
with (
patch("app.features.ytdlp.ytdlp_opts.Config"),
patch("app.features.presets.service.Presets") as mock_presets,
):
mock_presets_instance = Mock()
mock_presets_instance.get.return_value = None
mock_presets.get_instance.return_value = mock_presets_instance
@ -170,7 +176,7 @@ class TestYTDLPOpts:
def test_preset_with_cookies_creates_file(self):
"""Test that preset with cookies creates cookie file."""
with (
patch("app.library.YTDLPOpts.Config") as mock_config,
patch("app.features.ytdlp.ytdlp_opts.Config") as mock_config,
patch("app.features.presets.service.Presets") as mock_presets,
):
# Mock config
@ -208,9 +214,9 @@ class TestYTDLPOpts:
def test_preset_with_invalid_cli_raises_error(self):
"""Test that preset with invalid CLI raises ValueError."""
with (
patch("app.library.YTDLPOpts.Config"),
patch("app.features.ytdlp.ytdlp_opts.Config"),
patch("app.features.presets.service.Presets") as mock_presets,
patch("app.library.YTDLPOpts.arg_converter") as mock_converter,
patch("app.features.ytdlp.ytdlp_opts.arg_converter") as mock_converter,
):
mock_preset = Mock(spec=Preset)
mock_preset.id = "bad_preset"
@ -233,7 +239,7 @@ class TestYTDLPOpts:
def test_get_all_with_default_options(self):
"""Test get_all returns correct default options."""
with patch("app.library.YTDLPOpts.Config") as mock_config:
with patch("app.features.ytdlp.ytdlp_opts.Config") as mock_config:
mock_config_instance = Mock()
mock_config_instance.download_path = "/downloads"
mock_config_instance.temp_path = "/temp"
@ -242,7 +248,7 @@ class TestYTDLPOpts:
mock_config_instance.debug = False
mock_config.get_instance.return_value = mock_config_instance
with patch("app.library.YTDLPOpts.merge_dict") as mock_merge:
with patch("app.features.ytdlp.ytdlp_opts.merge_dict") as mock_merge:
mock_merge.return_value = {
"paths": {"home": "/downloads", "temp": "/temp"},
"outtmpl": {"default": "default_template", "chapter": "chapter_template"},
@ -257,9 +263,9 @@ class TestYTDLPOpts:
def test_get_all_processes_cli_arguments(self):
"""Test get_all processes CLI arguments correctly."""
with (
patch("app.library.YTDLPOpts.Config") as mock_config,
patch("app.library.YTDLPOpts.arg_converter") as mock_converter,
patch("app.library.YTDLPOpts.merge_dict") as mock_merge,
patch("app.features.ytdlp.ytdlp_opts.Config") as mock_config,
patch("app.features.ytdlp.ytdlp_opts.arg_converter") as mock_converter,
patch("app.features.ytdlp.ytdlp_opts.merge_dict") as mock_merge,
):
mock_config_instance = Mock()
mock_config_instance.download_path = "/downloads"
@ -286,8 +292,8 @@ class TestYTDLPOpts:
def test_get_all_handles_format_special_cases(self):
"""Test get_all handles special format values correctly."""
with (
patch("app.library.YTDLPOpts.Config") as mock_config,
patch("app.library.YTDLPOpts.merge_dict") as mock_merge,
patch("app.features.ytdlp.ytdlp_opts.Config") as mock_config,
patch("app.features.ytdlp.ytdlp_opts.merge_dict") as mock_merge,
):
mock_config_instance = Mock()
mock_config_instance.download_path = "/downloads"
@ -322,8 +328,8 @@ class TestYTDLPOpts:
def test_get_all_with_invalid_cli_raises_error(self):
"""Test get_all raises error for invalid CLI arguments."""
with (
patch("app.library.YTDLPOpts.Config") as mock_config,
patch("app.library.YTDLPOpts.arg_converter") as mock_converter,
patch("app.features.ytdlp.ytdlp_opts.Config") as mock_config,
patch("app.features.ytdlp.ytdlp_opts.arg_converter") as mock_converter,
):
mock_config_instance = Mock()
mock_config_instance.download_path = "/downloads"
@ -345,8 +351,8 @@ class TestYTDLPOpts:
def test_get_all_resets_unless_keep_true(self):
"""Test get_all resets instance unless keep=True."""
with (
patch("app.library.YTDLPOpts.Config") as mock_config,
patch("app.library.YTDLPOpts.merge_dict") as mock_merge,
patch("app.features.ytdlp.ytdlp_opts.Config") as mock_config,
patch("app.features.ytdlp.ytdlp_opts.merge_dict") as mock_merge,
):
mock_config_instance = Mock()
mock_config_instance.download_path = "/downloads"
@ -379,7 +385,7 @@ class TestYTDLPOpts:
def test_reset_clears_all_options(self):
"""Test reset clears all internal state."""
with patch("app.library.YTDLPOpts.Config"):
with patch("app.features.ytdlp.ytdlp_opts.Config"):
opts = YTDLPOpts()
# Set some state
@ -399,8 +405,8 @@ class TestYTDLPOpts:
def test_method_chaining(self):
"""Test that methods support chaining."""
with (
patch("app.library.YTDLPOpts.Config"),
patch("app.library.YTDLPOpts.arg_converter"),
patch("app.features.ytdlp.ytdlp_opts.Config"),
patch("app.features.ytdlp.ytdlp_opts.arg_converter"),
patch("app.features.presets.service.Presets") as mock_presets,
):
mock_presets_instance = Mock()
@ -417,9 +423,9 @@ class TestYTDLPOpts:
def test_debug_logging(self):
"""Test debug logging when enabled."""
with (
patch("app.library.YTDLPOpts.Config") as mock_config,
patch("app.library.YTDLPOpts.merge_dict") as mock_merge,
patch("app.library.YTDLPOpts.LOG") as mock_log,
patch("app.features.ytdlp.ytdlp_opts.Config") as mock_config,
patch("app.features.ytdlp.ytdlp_opts.merge_dict") as mock_merge,
patch("app.features.ytdlp.ytdlp_opts.LOG") as mock_log,
):
mock_config_instance = Mock()
mock_config_instance.download_path = "/downloads"
@ -440,9 +446,9 @@ class TestYTDLPOpts:
def test_cookie_loading_error_handling(self):
"""Test error handling when cookie loading fails."""
with (
patch("app.library.YTDLPOpts.Config") as mock_config,
patch("app.features.ytdlp.ytdlp_opts.Config") as mock_config,
patch("app.features.presets.service.Presets") as mock_presets,
patch("app.library.YTDLPOpts.LOG") as mock_log,
patch("app.features.ytdlp.ytdlp_opts.LOG") as mock_log,
):
# Mock config
mock_config_instance = Mock()
@ -479,9 +485,9 @@ class TestYTDLPOpts:
def test_replacer_substitution_in_cli(self):
"""Test that CLI arguments get replacer substitution."""
with (
patch("app.library.YTDLPOpts.Config") as mock_config,
patch("app.library.YTDLPOpts.arg_converter") as mock_converter,
patch("app.library.YTDLPOpts.merge_dict") as mock_merge,
patch("app.features.ytdlp.ytdlp_opts.Config") as mock_config,
patch("app.features.ytdlp.ytdlp_opts.arg_converter") as mock_converter,
patch("app.features.ytdlp.ytdlp_opts.merge_dict") as mock_merge,
):
mock_config_instance = Mock()
mock_config_instance.download_path = "/downloads"
@ -510,7 +516,7 @@ class TestARGSMerger:
def test_constructor_initializes_empty_args(self):
"""Test that ARGSMerger constructor initializes with empty args list."""
from app.library.YTDLPOpts import ARGSMerger
from app.features.ytdlp.ytdlp_opts import ARGSMerger
merger = ARGSMerger()
@ -518,7 +524,7 @@ class TestARGSMerger:
def test_add_valid_args(self):
"""Test adding valid command-line arguments."""
from app.library.YTDLPOpts import ARGSMerger
from app.features.ytdlp.ytdlp_opts import ARGSMerger
merger = ARGSMerger()
result = merger.add("--format best")
@ -528,7 +534,7 @@ class TestARGSMerger:
def test_add_filters_comment_lines(self):
"""Test that comment lines (starting with #) are filtered out."""
from app.library.YTDLPOpts import ARGSMerger
from app.features.ytdlp.ytdlp_opts import ARGSMerger
merger = ARGSMerger()
cli_with_comments = """--format best
@ -551,7 +557,7 @@ class TestARGSMerger:
def test_add_filters_indented_comment_lines(self):
"""Test that indented comment lines are filtered out."""
from app.library.YTDLPOpts import ARGSMerger
from app.features.ytdlp.ytdlp_opts import ARGSMerger
merger = ARGSMerger()
cli_with_indented_comments = """--format best
@ -574,7 +580,7 @@ class TestARGSMerger:
def test_add_filters_complex_commented_extractor_args(self):
"""Test filtering of complex real-world commented extractor-args."""
from app.library.YTDLPOpts import ARGSMerger
from app.features.ytdlp.ytdlp_opts import ARGSMerger
merger = ARGSMerger()
cli_with_complex_comments = """--extractor-args "youtube:player-client=default,tv,mweb,-web_safari;formats=incomplete"
@ -599,7 +605,7 @@ class TestARGSMerger:
def test_add_multiple_args_with_chaining(self):
"""Test adding multiple arguments using method chaining."""
from app.library.YTDLPOpts import ARGSMerger
from app.features.ytdlp.ytdlp_opts import ARGSMerger
merger = ARGSMerger()
merger.add("--format best").add("--output test.mp4").add("--no-playlist")
@ -608,7 +614,7 @@ class TestARGSMerger:
def test_add_args_with_quotes(self):
"""Test adding arguments containing quoted strings."""
from app.library.YTDLPOpts import ARGSMerger
from app.features.ytdlp.ytdlp_opts import ARGSMerger
merger = ARGSMerger()
merger.add('--output "%(title)s.%(ext)s"')
@ -617,7 +623,7 @@ class TestARGSMerger:
def test_add_complex_args(self):
"""Test adding complex arguments with multiple values."""
from app.library.YTDLPOpts import ARGSMerger
from app.features.ytdlp.ytdlp_opts import ARGSMerger
merger = ARGSMerger()
merger.add("--format 'bestvideo[height<=1080]+bestaudio/best'")
@ -627,7 +633,7 @@ class TestARGSMerger:
def test_add_empty_string_returns_self(self):
"""Test that adding empty string returns self without modifying args."""
from app.library.YTDLPOpts import ARGSMerger
from app.features.ytdlp.ytdlp_opts import ARGSMerger
merger = ARGSMerger()
result = merger.add("")
@ -637,7 +643,7 @@ class TestARGSMerger:
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
from app.features.ytdlp.ytdlp_opts import ARGSMerger
merger = ARGSMerger()
result = merger.add("a")
@ -647,7 +653,7 @@ class TestARGSMerger:
def test_add_non_string_returns_self(self):
"""Test that adding non-string returns self without modifying args."""
from app.library.YTDLPOpts import ARGSMerger
from app.features.ytdlp.ytdlp_opts import ARGSMerger
merger = ARGSMerger()
result = merger.add(123) # type: ignore
@ -657,7 +663,7 @@ class TestARGSMerger:
def test_as_string(self):
"""Test converting args to string."""
from app.library.YTDLPOpts import ARGSMerger
from app.features.ytdlp.ytdlp_opts import ARGSMerger
merger = ARGSMerger()
merger.add("--format best").add("--output test.mp4")
@ -668,7 +674,7 @@ class TestARGSMerger:
def test_str_method(self):
"""Test __str__ method."""
from app.library.YTDLPOpts import ARGSMerger
from app.features.ytdlp.ytdlp_opts import ARGSMerger
merger = ARGSMerger()
merger.add("--format best").add("--output test.mp4")
@ -679,7 +685,7 @@ class TestARGSMerger:
def test_as_dict(self):
"""Test converting args to dict."""
from app.library.YTDLPOpts import ARGSMerger
from app.features.ytdlp.ytdlp_opts import ARGSMerger
merger = ARGSMerger()
merger.add("--format best").add("--output test.mp4")
@ -691,12 +697,12 @@ class TestARGSMerger:
def test_as_ytdlp(self):
"""Test converting args to yt-dlp JSON options."""
from app.library.YTDLPOpts import ARGSMerger
from app.features.ytdlp.ytdlp_opts import ARGSMerger
merger = ARGSMerger()
merger.add("--format best")
with patch("app.library.YTDLPOpts.arg_converter") as mock_converter:
with patch("app.features.ytdlp.ytdlp_opts.arg_converter") as mock_converter:
mock_converter.return_value = {"format": "best"}
result = merger.as_ytdlp()
@ -706,7 +712,7 @@ class TestARGSMerger:
def test_get_instance(self):
"""Test get_instance static method."""
from app.library.YTDLPOpts import ARGSMerger
from app.features.ytdlp.ytdlp_opts import ARGSMerger
merger = ARGSMerger.get_instance()
@ -715,7 +721,7 @@ class TestARGSMerger:
def test_reset(self):
"""Test reset method clears args."""
from app.library.YTDLPOpts import ARGSMerger
from app.features.ytdlp.ytdlp_opts import ARGSMerger
merger = ARGSMerger()
merger.add("--format best").add("--output test.mp4")
@ -729,7 +735,7 @@ class TestARGSMerger:
def test_args_with_special_characters(self):
"""Test handling arguments with special characters."""
from app.library.YTDLPOpts import ARGSMerger
from app.features.ytdlp.ytdlp_opts import ARGSMerger
merger = ARGSMerger()
merger.add('--postprocessor-args "-movflags +faststart"')
@ -742,11 +748,11 @@ class TestYTDLPCli:
"""Test the YTDLPCli class."""
@patch("app.features.presets.service.Presets")
@patch("app.library.YTDLPOpts.Config")
@patch("app.features.ytdlp.ytdlp_opts.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
from app.features.ytdlp.ytdlp_opts import YTDLPCli
mock_config_instance = Mock()
mock_config_instance.download_path = "/downloads"
@ -765,17 +771,17 @@ class TestYTDLPCli:
def test_constructor_with_invalid_type_raises_error(self):
"""Test YTDLPCli constructor raises error with non-Item type."""
from app.library.YTDLPOpts import YTDLPCli
from app.features.ytdlp.ytdlp_opts import YTDLPCli
with pytest.raises(ValueError, match="Expected Item instance"):
YTDLPCli(item="not an item") # type: ignore
@patch("app.features.presets.service.Presets")
@patch("app.library.YTDLPOpts.Config")
@patch("app.features.ytdlp.ytdlp_opts.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
from app.features.ytdlp.ytdlp_opts import YTDLPCli
mock_config_instance = Mock()
mock_config_instance.download_path = "/downloads"
@ -806,12 +812,12 @@ class TestYTDLPCli:
assert "--format best" in command
@patch("app.features.presets.service.Presets")
@patch("app.library.YTDLPOpts.Config")
@patch("app.features.ytdlp.ytdlp_opts.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.features.presets.schemas import Preset
from app.library.YTDLPOpts import YTDLPCli
from app.features.ytdlp.ytdlp_opts import YTDLPCli
mock_config_instance = Mock()
mock_config_instance.download_path = "/downloads"
@ -840,12 +846,12 @@ class TestYTDLPCli:
assert "--format 720p" in command
@patch("app.features.presets.service.Presets")
@patch("app.library.YTDLPOpts.Config")
@patch("app.features.ytdlp.ytdlp_opts.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.features.presets.schemas import Preset
from app.library.YTDLPOpts import YTDLPCli
from app.features.ytdlp.ytdlp_opts import YTDLPCli
mock_config_instance = Mock()
mock_config_instance.download_path = "/downloads"
@ -879,11 +885,11 @@ class TestYTDLPCli:
assert "--format best" in command, "User CLI should appear after preset CLI in command"
@patch("app.features.presets.service.Presets")
@patch("app.library.YTDLPOpts.Config")
@patch("app.features.ytdlp.ytdlp_opts.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
from app.features.ytdlp.ytdlp_opts import YTDLPCli
mock_config_instance = Mock()
mock_config_instance.download_path = "/default/downloads"
@ -904,12 +910,12 @@ class TestYTDLPCli:
assert info["merged"]["save_path"] == "/default/downloads"
@patch("app.features.presets.service.Presets")
@patch("app.library.YTDLPOpts.create_cookies_file")
@patch("app.library.YTDLPOpts.Config")
@patch("app.features.ytdlp.ytdlp_opts.create_cookies_file")
@patch("app.features.ytdlp.ytdlp_opts.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
from app.features.ytdlp.ytdlp_opts import YTDLPCli
mock_config_instance = Mock()
mock_config_instance.download_path = "/downloads"
@ -934,12 +940,12 @@ class TestYTDLPCli:
assert info["merged"]["cookie_file"] == "/tmp/cookies.txt"
@patch("app.features.presets.service.Presets")
@patch("app.library.YTDLPOpts.Config")
@patch("app.features.ytdlp.ytdlp_opts.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.features.presets.schemas import Preset
from app.library.YTDLPOpts import YTDLPCli
from app.features.ytdlp.ytdlp_opts import YTDLPCli
mock_config_instance = Mock()
mock_config_instance.download_path = "/downloads"
@ -970,11 +976,11 @@ class TestYTDLPCli:
assert "/preset/cookies.txt" in command
@patch("app.features.presets.service.Presets")
@patch("app.library.YTDLPOpts.Config")
@patch("app.features.ytdlp.ytdlp_opts.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
from app.features.ytdlp.ytdlp_opts import YTDLPCli
mock_config_instance = Mock()
mock_config_instance.download_path = "/downloads"
@ -997,11 +1003,11 @@ class TestYTDLPCli:
assert "--paths" in command
@patch("app.features.presets.service.Presets")
@patch("app.library.YTDLPOpts.Config")
@patch("app.features.ytdlp.ytdlp_opts.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
from app.features.ytdlp.ytdlp_opts import YTDLPCli
mock_config_instance = Mock()
mock_config_instance.download_path = "/downloads"
@ -1019,12 +1025,12 @@ class TestYTDLPCli:
assert "https://youtube.com/watch?v=test123" in command
@patch("app.features.presets.service.Presets")
@patch("app.library.YTDLPOpts.Config")
@patch("app.features.ytdlp.ytdlp_opts.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.features.presets.schemas import Preset
from app.library.YTDLPOpts import YTDLPCli
from app.features.ytdlp.ytdlp_opts import YTDLPCli
mock_config_instance = Mock()
mock_config_instance.download_path = "/downloads"

539
app/features/ytdlp/tests/test_ytdlp_utils.py Normal file
View file

@ -0,0 +1,539 @@
import logging
import re
from typing import Any
from unittest.mock import MagicMock, patch
import pytest
from app.features.ytdlp.utils import (
LogWrapper,
extract_ytdlp_logs,
get_archive_id,
ytdlp_reject,
parse_outtmpl,
get_extras,
get_thumbnail,
get_ytdlp,
)
class CaptureHandler(logging.Handler):
def __init__(self) -> None:
super().__init__()
self.records: list[logging.LogRecord] = []
def emit(self, record: logging.LogRecord) -> None:
self.records.append(record)
def make_logger(name: str = "lw_test") -> tuple[logging.Logger, CaptureHandler]:
logger = logging.getLogger(name)
logger.setLevel(logging.DEBUG)
handler = CaptureHandler()
# Avoid duplicate handlers when tests run multiple times
for h in list(logger.handlers):
logger.removeHandler(h)
logger.addHandler(handler)
return logger, handler
class TestLogWrapper:
def test_add_target_type_validation(self) -> None:
lw = LogWrapper()
with pytest.raises(TypeError, match=r"Target must be a logging\.Logger instance or a callable"):
lw.add_target(123) # type: ignore[arg-type]
def test_add_target_name_inference_and_custom(self) -> None:
lw = LogWrapper()
logger, _ = make_logger("one")
# Name inferred from logger
lw.add_target(logger)
assert lw.targets[-1].name == "one"
assert lw.has_targets() is True
# Name inferred from callable
def sink(level: int, msg: str, *args: Any, **kwargs: Any) -> None: # noqa: ARG001
return None
lw.add_target(sink)
assert lw.targets[-1].name == "sink"
# Custom name overrides
lw.add_target(logger, name="custom")
assert lw.targets[-1].name == "custom"
def test_level_filtering_and_dispatch(self) -> None:
lw = LogWrapper()
logger, cap = make_logger("cap")
calls: list[tuple[int, str, tuple, dict]] = []
def sink(level: int, msg: str, *args: Any, **kwargs: Any) -> None:
calls.append((level, msg, args, kwargs))
# Logger target at INFO, callable target at WARNING
lw.add_target(logger, level=logging.INFO)
lw.add_target(sink, level=logging.WARNING)
# DEBUG should hit none
lw.debug("d1")
assert len(cap.records) == 0
assert len(calls) == 0
# INFO hits logger only
lw.info("hello %s", "X")
assert len(cap.records) == 1
assert cap.records[0].levelno == logging.INFO
assert cap.records[0].getMessage() == "hello X"
assert len(calls) == 0
# WARNING hits both
lw.warning("warn %s", "Y", extra={"k": 1})
assert len(cap.records) == 2
assert cap.records[1].levelno == logging.WARNING
assert cap.records[1].getMessage() == "warn Y"
assert len(calls) == 1
lvl, msg, args, kwargs = calls[0]
assert lvl == logging.WARNING
assert msg == "warn %s"
assert args == ("Y",)
assert "extra" in kwargs
assert kwargs["extra"] == {"k": 1}
# ERROR still hits both; CRITICAL too
lw.error("err")
lw.critical("boom")
assert any(r.levelno == logging.ERROR for r in cap.records)
assert any(r.levelno == logging.CRITICAL for r in cap.records)
assert any(c[0] == logging.ERROR for c in calls)
assert any(c[0] == logging.CRITICAL for c in calls)
class TestExtractYtdlpLogs:
"""Test YTDLP log extraction function."""
def test_extract_ytdlp_logs_basic(self):
"""Test basic log extraction."""
logs = ["This live event will begin soon", "ERROR: Failed", "WARNING: Deprecated"]
result = extract_ytdlp_logs(logs)
assert isinstance(result, list)
assert len(result) >= 1 # Should match "This live event will begin"
def test_extract_ytdlp_logs_with_filters(self):
"""Test log extraction with filters."""
logs = ["INFO: Downloading", "ERROR: Failed", "WARNING: Deprecated"]
filters = [re.compile(r"ERROR")]
result = extract_ytdlp_logs(logs, filters)
assert len(result) >= 0 # Should filter based on patterns
def test_extract_ytdlp_logs_empty(self):
"""Test with empty logs."""
result = extract_ytdlp_logs([])
assert result == []
class TestYtdlpReject:
"""Test the ytdlp_reject function."""
def test_ytdlp_reject_basic(self):
"""Test basic rejection logic."""
entry = {"title": "Test Video", "view_count": 1000}
yt_params = {}
passed, message = ytdlp_reject(entry, yt_params)
assert isinstance(passed, bool)
assert isinstance(message, str)
def test_ytdlp_reject_with_filters(self):
"""Test rejection with filters."""
entry = {"title": "Test Video", "upload_date": "20230101"}
yt_params = {"daterange": MagicMock()}
# Mock daterange to simulate rejection
yt_params["daterange"].__contains__ = MagicMock(return_value=False)
passed, message = ytdlp_reject(entry, yt_params)
assert isinstance(passed, bool)
assert isinstance(message, str)
class TestParseOuttmpl:
"""Test the parse_outtmpl function."""
def setup_method(self):
"""Reset YTDLP singleton state before each test."""
from app.features.ytdlp.utils import _DATA
_DATA.YTDLP_INFO_CLS = None
def test_parse_outtmpl_basic(self):
"""Test basic template parsing with simple placeholders."""
template = "%(title)s.%(ext)s"
info_dict = {
"title": "Test Video",
"ext": "mp4",
}
result = parse_outtmpl(template, info_dict)
assert result == "Test Video.mp4"
def test_parse_outtmpl_with_id(self):
"""Test template parsing with video ID."""
template = "[%(id)s] %(title)s.%(ext)s"
info_dict = {
"id": "dQw4w9WgXcQ",
"title": "Never Gonna Give You Up",
"ext": "webm",
}
result = parse_outtmpl(template, info_dict)
assert result == "[dQw4w9WgXcQ] Never Gonna Give You Up.webm"
def test_parse_outtmpl_with_uploader(self):
"""Test template parsing with uploader information."""
template = "%(uploader)s - %(title)s.%(ext)s"
info_dict = {
"uploader": "Rick Astley",
"title": "Never Gonna Give You Up",
"ext": "mp4",
}
result = parse_outtmpl(template, info_dict)
assert result == "Rick Astley - Never Gonna Give You Up.mp4"
def test_parse_outtmpl_with_nested_path(self):
"""Test template parsing with nested directory structure."""
template = "%(uploader)s/%(title)s.%(ext)s"
info_dict = {
"uploader": "Test Channel",
"title": "Test Video",
"ext": "mkv",
}
result = parse_outtmpl(template, info_dict)
assert result == "Test Channel/Test Video.mkv"
def test_parse_outtmpl_with_missing_field(self):
"""Test template parsing with missing field defaults to NA."""
template = "%(title)s - %(upload_date)s.%(ext)s"
info_dict = {
"title": "Test Video",
"ext": "mp4",
}
result = parse_outtmpl(template, info_dict)
assert result == "Test Video - NA.mp4", "Missing field upload_date should default to NA"
def test_parse_outtmpl_complex(self):
"""Test complex template with multiple fields."""
template = "%(uploader)s/%(playlist_title)s/%(playlist_index)03d - %(title)s [%(id)s].%(ext)s"
info_dict = {
"uploader": "Test Channel",
"playlist_title": "Best Videos",
"playlist_index": 5,
"title": "Amazing Content",
"id": "abc123xyz",
"ext": "mp4",
}
result = parse_outtmpl(template, info_dict)
assert result == "Test Channel/Best Videos/005 - Amazing Content [abc123xyz].mp4"
def test_parse_outtmpl_with_special_characters(self):
"""Test template parsing handles special characters in values."""
template = "%(title)s.%(ext)s"
info_dict = {
"title": "Test: Video / With \\ Special | Characters",
"ext": "mp4",
}
result = parse_outtmpl(template, info_dict)
assert ".mp4" in result, "yt-dlp should sanitize special characters but preserve extension"
assert "Test" in result, "yt-dlp should preserve safe parts of title"
def test_parse_outtmpl_with_playlist_info(self):
"""Test template parsing with playlist information."""
template = "%(playlist)s/%(title)s.%(ext)s"
info_dict = {
"playlist": "My Playlist",
"title": "Video Title",
"ext": "webm",
}
result = parse_outtmpl(template, info_dict)
assert result == "My Playlist/Video Title.webm"
def test_parse_outtmpl_with_restrict_filename(self):
"""Test template parsing with restrict_filename parameter."""
template = "%(uploader)s/%(title)s.%(ext)s"
info_dict = {
"uploader": "Foobar's Workshop",
"title": "Test Video",
"ext": "mp4",
}
result_unrestricted: str = parse_outtmpl(template, info_dict)
assert result_unrestricted == "Foobar's Workshop/Test Video.mp4"
result_restricted: str = parse_outtmpl(template, info_dict, params={"restrictfilenames": True})
assert result_restricted == "Foobar_s_Workshop/Test_Video.mp4"
class TestGetThumbnail:
def test_returns_none_for_empty_list(self):
"""Test that None is returned for an empty thumbnail list."""
assert get_thumbnail([]) is None
def test_returns_none_for_non_list(self):
"""Test that None is returned for non-list input."""
assert get_thumbnail(None) is None
assert get_thumbnail("not a list") is None
assert get_thumbnail({"not": "list"}) is None
def test_returns_highest_preference_thumbnail(self):
"""Test that the thumbnail with highest preference is returned."""
thumbnails = [
{"url": "low.jpg", "preference": 1, "width": 100, "height": 100},
{"url": "high.jpg", "preference": 10, "width": 200, "height": 200},
{"url": "medium.jpg", "preference": 5, "width": 150, "height": 150},
]
result = get_thumbnail(thumbnails)
assert result == {"url": "high.jpg", "preference": 10, "width": 200, "height": 200}
def test_returns_highest_width_when_preference_equal(self):
"""Test that the thumbnail with highest width is returned when preference is equal."""
thumbnails = [
{"url": "small.jpg", "preference": 1, "width": 100, "height": 100},
{"url": "large.jpg", "preference": 1, "width": 200, "height": 200},
{"url": "medium.jpg", "preference": 1, "width": 150, "height": 150},
]
result = get_thumbnail(thumbnails)
assert result == {"url": "large.jpg", "preference": 1, "width": 200, "height": 200}
def test_handles_missing_attributes(self):
"""Test that thumbnails with missing attributes are handled correctly."""
thumbnails = [
{"url": "no_pref.jpg", "width": 100},
{"url": "with_pref.jpg", "preference": 5, "width": 50},
]
result = get_thumbnail(thumbnails)
assert result["url"] == "with_pref.jpg"
def test_returns_first_when_all_equal(self):
"""Test that any thumbnail is returned when all attributes are equal."""
thumbnails = [
{"url": "first.jpg"},
{"url": "second.jpg"},
]
result = get_thumbnail(thumbnails)
assert result is not None
assert result["url"] in ["first.jpg", "second.jpg"]
class TestGetExtras:
def test_returns_empty_dict_for_none(self):
"""Test that empty dict is returned for None input."""
assert get_extras(None) == {}
def test_returns_empty_dict_for_non_dict(self):
"""Test that empty dict is returned for non-dict input."""
assert get_extras("not a dict") == {}
assert get_extras([]) == {}
def test_extracts_video_information(self):
"""Test extracting information from a video entry."""
entry = {
"id": "test123",
"title": "Test Video",
"uploader": "Test Uploader",
"channel": "Test Channel",
"thumbnails": [{"url": "thumb.jpg", "preference": 1}],
"duration": 120,
}
result = get_extras(entry, kind="video")
assert result["uploader"] == "Test Uploader"
assert result["channel"] == "Test Channel"
assert result["thumbnail"] == "thumb.jpg"
assert result["duration"] == 120
assert result["is_premiere"] is False
def test_extracts_playlist_information(self):
"""Test extracting information from a playlist entry."""
entry = {
"id": "playlist123",
"title": "Test Playlist",
"uploader": "Playlist Owner",
"uploader_id": "owner123",
}
result = get_extras(entry, kind="playlist")
assert result["playlist_id"] == "playlist123"
assert result["playlist_title"] == "Test Playlist"
assert result["playlist_uploader"] == "Playlist Owner"
assert result["playlist_uploader_id"] == "owner123"
def test_handles_release_timestamp(self):
"""Test handling of release_timestamp for upcoming content."""
entry = {
"release_timestamp": 1234567890,
}
result = get_extras(entry)
assert "release_in" in result
assert result["release_in"] == "Fri, 13 Feb 2009 23:31:30 GMT"
def test_handles_upcoming_live_stream(self):
"""Test handling of upcoming live stream."""
entry = {
"release_timestamp": 1234567890,
"live_status": "is_upcoming",
}
result = get_extras(entry)
assert result["is_live"] == 1234567890
assert "release_in" in result
def test_handles_premiere_flag(self):
"""Test handling of is_premiere flag."""
entry = {
"is_premiere": True,
}
result = get_extras(entry)
assert result["is_premiere"] is True
entry2 = {"is_premiere": False}
result2 = get_extras(entry2)
assert result2["is_premiere"] is False
def test_youtube_fallback_thumbnail(self):
"""Test fallback thumbnail generation for YouTube videos."""
entry = {
"id": "dQw4w9WgXcQ",
"ie_key": "Youtube",
}
result = get_extras(entry)
assert result["thumbnail"] == "https://img.youtube.com/vi/dQw4w9WgXcQ/maxresdefault.jpg"
def test_thumbnail_string_fallback(self):
"""Test fallback to thumbnail string when thumbnails list not available."""
entry = {
"thumbnail": "https://example.com/thumb.jpg",
}
result = get_extras(entry)
assert result["thumbnail"] == "https://example.com/thumb.jpg"
class TestGetStaticYtdlp:
"""Test the get_static_ytdlp function."""
def setup_method(self):
"""Reset YTDLP singleton state before each test."""
from app.features.ytdlp.utils import _DATA
_DATA.YTDLP_INFO_CLS = None
def test_get_static_ytdlp_returns_instance(self):
"""Test that get_static_ytdlp returns a YTDLP instance."""
from app.features.ytdlp.ytdlp import YTDLP
# Get the cached instance
instance = get_ytdlp()
assert instance is not None
assert isinstance(instance, YTDLP)
def test_get_static_ytdlp_returns_same_instance(self):
"""Test that get_static_ytdlp returns the same cached instance."""
instance1 = get_ytdlp()
instance2 = get_ytdlp()
assert instance1 is instance2
def test_get_static_ytdlp_with_params(self):
"""Test that get_static_ytdlp returns a new instance when params are provided."""
instance1 = get_ytdlp()
instance2 = get_ytdlp(params={"quiet": False})
assert instance1 is not instance2
assert instance2 is not None
def test_get_static_ytdlp_has_correct_params(self):
"""Test that get_static_ytdlp initializes with correct parameters."""
instance = get_ytdlp()
# Access the internal params
params = instance.params
assert params.get("color") == "no_color"
assert params.get("extract_flat") is True
assert params.get("skip_download") is True
assert params.get("ignoreerrors") is True
assert params.get("ignore_no_formats_error") is True
assert params.get("quiet") is True
class TestGetArchiveId:
"""Test the get_archive_id function."""
@patch("app.features.ytdlp.utils._DATA.YTDLP_INFO_CLS")
def test_get_archive_id_basic(self, mock_ytdlp):
"""Test basic archive ID extraction."""
mock_ytdlp._ies = {}
result = get_archive_id("https://youtube.com/watch?v=test123")
assert isinstance(result, dict)
assert "id" in result
assert "ie_key" in result
assert "archive_id" in result
def test_get_archive_id_invalid_url(self):
"""Test with invalid URL."""
result = get_archive_id("invalid-url")
assert isinstance(result, dict)

508
app/features/ytdlp/utils.py Normal file
View file

@ -0,0 +1,508 @@
import json
import logging
import os
import re
import shlex
from collections.abc import Callable
from dataclasses import dataclass
from email.utils import formatdate
from typing import Any
from app.features.ytdlp.ytdlp import YTDLP
from app.library.Utils import merge_dict, timed_lru_cache
LOG: logging.Logger = logging.getLogger("ytdlp.utils")
class _DATA:
YTDLP_INFO_CLS: Any = None
YTDLP_PARAMS: dict[str, Any] = {
"simulate": True,
"color": "no_color",
"extract_flat": True,
"skip_download": True,
"ignoreerrors": True,
"ignore_no_formats_error": True,
"quiet": True,
}
REMOVE_KEYS: list = [
{
"paths": "-P, --paths",
"outtmpl": "-o, --output",
"progress_hooks": "--progress_hooks",
"postprocessor_hooks": "--postprocessor_hooks",
"post_hooks": "--post_hooks",
},
{
"quiet": "-q, --quiet",
"no_warnings": "--no-warnings",
"skip_download": "--skip-download",
"forceprint": "-O, --print",
"simulate": "--simulate",
"noprogress": "--no-progress",
"wait_for_video": "--wait-for-video",
"progress_delta": " --progress-delta",
"progress_template": "--progress-template",
"consoletitle": "--console-title",
"progress_with_newline": "--newline",
"forcejson": "-j, --dump-single-json",
"opt_update_to": "--update-to",
"opt_ap_list_mso": "--ap-list-mso",
"opt_batch_file": "-a, --batch-file",
"opt_alias": "--alias",
"opt_list_extractors": "--list-extractors",
"opt_version": "--version",
"opt_help": "-h, --help",
"opt_update": "-U, --update",
"opt_list_subtitles": "--list-subs",
"opt_list_thumbnails": "--list-thumbnails",
"opt_list_format": "-F, --list-formats",
"opt_dump_agent": "--dump-user-agent",
"opt_extractor_descriptions": "--extractor-descriptions",
"opt_list_impersonate_targets": "--list-impersonate-targets",
},
]
"Keys to remove from yt-dlp options at various levels."
@dataclass(kw_only=True)
class LogTarget:
"""
A data class that represents a logging target with its level and type.
Attributes:
name (str): The name of the logging target.
target: The logging target, which can be a logging.Logger instance or a callable.
level (int): The logging level for the target.
logger (bool): True if the target is a logging.Logger instance, False otherwise.
type: The type of the target.
"""
name: str | None = None
target: logging.Logger | Callable
level: int
logger: bool
class LogWrapper:
def __init__(self):
self.targets: list[LogTarget] = []
def add_target(self, target: logging.Logger | Callable, level: int = logging.DEBUG, name: str | None = None):
"""
Adds a new logging target with the specified logging level.
Args:
target (logging.Logger|Callable): The logging target, which can be a logging.Logger instance or a Callable.
level (int): The logging level for the target. Defaults to logging.DEBUG.
name (str|None): The name of the logging target. Defaults to None.
"""
if not isinstance(target, logging.Logger | Callable):
msg = "Target must be a logging.Logger instance or a callable."
raise TypeError(msg)
if name is None:
name = target.name if isinstance(target, logging.Logger) else target.__name__
self.targets.append(
LogTarget(
name=name,
target=target,
level=level,
logger=isinstance(target, logging.Logger),
)
)
def has_targets(self):
return len(self.targets) > 0
def _log(self, level, msg, *args, **kwargs):
for target in self.targets:
if level < target.level:
continue
if target.logger:
target.target.log(level, msg, *args, **kwargs)
else:
target.target(level, msg, *args, **kwargs)
def debug(self, msg, *args, **kwargs):
self._log(logging.DEBUG, msg, *args, **kwargs)
def info(self, msg, *args, **kwargs):
self._log(logging.INFO, msg, *args, **kwargs)
def warning(self, msg, *args, **kwargs):
self._log(logging.WARNING, msg, *args, **kwargs)
def error(self, msg, *args, **kwargs):
self._log(logging.ERROR, msg, *args, **kwargs)
def critical(self, msg, *args, **kwargs):
self._log(logging.CRITICAL, msg, *args, **kwargs)
def patch_metadataparser() -> None:
"""
Patches yt_dlp MetadataParserPP action to handle subprocess pickling issues.
"""
try:
from yt_dlp.postprocessor.metadataparser import MetadataParserPP
from yt_dlp.utils import Namespace
except Exception as exc:
LOG.warning(f"Unable to import yt_dlp metadata parser for patching: {exc!s}")
return
if getattr(MetadataParserPP.Actions, "_ytptube_patched", False):
return
class _ActionNS(Namespace):
_ACTIONS_STR: list[str] = []
@staticmethod
def _get_name(func) -> str | None:
if not callable(func):
return None
target = getattr(func, "__func__", func)
module_name = getattr(target, "__module__", None)
qual_name = getattr(target, "__qualname__", getattr(target, "__name__", None))
return f"{module_name}.{qual_name}" if module_name and qual_name else None
def __contains__(self, candidate: object) -> bool:
if candidate in self.__dict__.values():
return True
if func_name := _ActionNS._get_name(candidate):
if len(_ActionNS._ACTIONS_STR) < 1:
_ActionNS._ACTIONS_STR.extend([_ActionNS._get_name(value) for value in self.__dict__.values()])
return func_name in _ActionNS._ACTIONS_STR
return False
actions_dict: dict[str, Any] = dict(MetadataParserPP.Actions.items_)
MetadataParserPP.Actions = _ActionNS(**actions_dict)
MetadataParserPP.Actions._ytptube_patched = True
LOG.debug("MetadataParserPP action namespace patch applied successfully.")
def arg_converter(
args: str,
level: int | bool | None = None,
dumps: bool = False,
removed_options: list | None = None,
keep_defaults: bool = False,
) -> dict:
"""
Convert yt-dlp options to a dictionary.
Args:
args (str): yt-dlp options string.
level (int|bool|None): Level of options to remove, True for all.
dumps (bool): Dump options as JSON.
removed_options (list|None): List of removed options.
keep_defaults (bool): Keep default options.
Returns:
dict: yt-dlp options dictionary.
"""
import yt_dlp.options
create_parser = yt_dlp.options.create_parser
def _default_opts(args: str):
patched_parser = create_parser()
try:
yt_dlp.options.create_parser = lambda: patched_parser
return yt_dlp.parse_options(args)
finally:
yt_dlp.options.create_parser = create_parser
try:
patch_metadataparser()
except Exception as exc:
LOG.debug("Metadata parser patch failed to apply: %s", exc)
default_opts = _default_opts([]).ydl_opts
if args:
# important to ignore external config files.
args = "--ignore-config " + args
opts = yt_dlp.parse_options(shlex.split(args, posix=os.name != "nt")).ydl_opts
diff = {k: v for k, v in opts.items() if default_opts[k] != v} if not keep_defaults else opts.items()
if "postprocessors" in diff:
diff["postprocessors"] = [pp for pp in diff["postprocessors"] if pp not in default_opts["postprocessors"]]
if "_warnings" in diff:
diff.pop("_warnings", None)
if level is True or isinstance(level, int):
bad_options = {}
if isinstance(level, bool) or not isinstance(level, int):
level = len(_DATA.REMOVE_KEYS)
for i, item in enumerate(_DATA.REMOVE_KEYS):
if i > level:
break
bad_options.update(item.items())
for key in diff.copy():
if key not in bad_options:
continue
if isinstance(removed_options, list):
removed_options.append(bad_options[key])
diff.pop(key, None)
if dumps is True:
from app.library.encoder import Encoder
if "match_filter" in diff:
import inspect
matchFilter = inspect.getclosurevars(diff["match_filter"].func).nonlocals["filters"]
if isinstance(matchFilter, set):
diff["match_filter"] = {"filters": list(matchFilter)}
return json.loads(json.dumps(diff, cls=Encoder, default=str))
return diff
def extract_ytdlp_logs(logs: list[str], filters: list[str | re.Pattern] = None) -> list[str]:
"""
Extract yt-dlp log lines matching built-in filters plus any extras.
Args:
logs (list): log strings.
filters (list[str|Re.Pattern]): Optional extra filters of strings and/or regex.
Returns:
(list): List of matching log lines.
"""
all_patterns: list[str | re.Pattern] = [
"This live event will begin",
"Video unavailable. This video is private",
"This video is available to this channel",
"Private video. Sign in if you've been granted access to this video",
"[youtube] Premieres in",
"Falling back on generic information extractor",
"URL could be a direct video link, returning it as such",
] + (filters or [])
compiled: list[re.Pattern] = [
p if isinstance(p, re.Pattern) else re.compile(re.escape(p), re.IGNORECASE) for p in all_patterns
]
matched: list[str] = []
matched.extend(line for line in logs if line and any(p.search(line) for p in compiled))
return list(dict.fromkeys(matched))
def ytdlp_reject(entry: dict, yt_params: dict) -> tuple[bool, str]:
"""
Implement yt-dlp reject filter logic.
Args:
entry (dict): The entry to check.
yt_params (dict): The yt-dlp parameters containing filters.
Returns:
tuple[bool, str]: A tuple where the first element is True if the entry passes the filters, or False if it does not,
and the second element is a message explaining the reason for rejection or an empty string if it passes.
"""
if title := entry.get("title"):
if (matchtitle := yt_params.get("matchtitle")) and not re.search(matchtitle, title, re.IGNORECASE):
return (False, f'"{title}" title did not match pattern "{matchtitle}". Skipping download.')
if (rejecttitle := yt_params.get("rejecttitle")) and re.search(rejecttitle, title, re.IGNORECASE):
return (False, f'"{title}" title matched reject pattern "{rejecttitle}". Skipping download.')
date = entry.get("upload_date")
date_range = yt_params.get("daterange")
if (date and date_range) and date not in date_range:
return (False, f"Upload date '{date}' is not in range '{date_range}'.")
view_count = entry.get("view_count")
if view_count is not None:
min_views = yt_params.get("min_views")
if min_views is not None and view_count < min_views:
return (
False,
f"Skipping {entry.get('title', 'video')}, because it has not reached minimum view count ({view_count}/{min_views}).",
)
max_views = yt_params.get("max_views")
if max_views is not None and view_count > max_views:
return (
False,
f"Skipping {entry.get('title', 'video')}, because it has exceeded maximum view count ({view_count}/{max_views}).",
)
try:
from yt_dlp.utils import age_restricted
if entry.get("age_limit") and age_restricted(entry.get("age_limit"), yt_params.get("age_limit")):
return (False, f'Video "{entry.get("title", "unknown")}" is age restricted.')
except ImportError:
pass
return (True, "")
def get_ytdlp(params: dict | None = None) -> YTDLP:
if params:
return YTDLP(params=merge_dict(params, _DATA.YTDLP_PARAMS))
if _DATA.YTDLP_INFO_CLS is None:
_DATA.YTDLP_INFO_CLS = YTDLP(params=_DATA.YTDLP_PARAMS)
return _DATA.YTDLP_INFO_CLS
def get_thumbnail(thumbnails: list) -> str | None:
"""
Extract thumbnail URL from a yt-dlp entry.
Args:
thumbnails (list): The list of thumbnail dictionaries from yt-dlp entry.
Returns:
str | None: The thumbnail URL if available, otherwise None.
"""
if not thumbnails or not isinstance(thumbnails, list):
return None
def _thumb_sort_key(thumb: dict) -> tuple:
return (
thumb.get("preference") if thumb.get("preference") is not None else -1,
thumb.get("width") if thumb.get("width") is not None else -1,
thumb.get("height") if thumb.get("height") is not None else -1,
thumb.get("id") if thumb.get("id") is not None else "",
thumb.get("url"),
)
return max(thumbnails, key=_thumb_sort_key, default=None)
def get_extras(entry: dict, kind: str = "video") -> dict:
"""
Extract useful information from a yt-dlp entry.
Args:
entry (dict): The entry data from yt-dlp.
kind (str): The type of the item (e.g., "video", "playlist").
Returns:
dict: The extracted data.
"""
extras = {}
if not entry or not isinstance(entry, dict):
return extras
if "playlist" == kind:
for property in ("id", "title", "uploader", "uploader_id"):
if val := entry.get(property):
extras[f"playlist_{property}"] = val
if thumbnail := get_thumbnail(entry.get("thumbnails", [])):
extras["thumbnail"] = thumbnail.get("url")
elif thumbnail := entry.get("thumbnail"):
extras["thumbnail"] = thumbnail
for property in ("uploader", "channel"):
if val := entry.get(property):
extras[property] = val
if release_in := entry.get("release_timestamp"):
extras["release_in"] = formatdate(release_in, usegmt=True)
if release_in and "is_upcoming" == entry.get("live_status"):
extras["is_live"] = release_in
if duration := entry.get("duration"):
extras["duration"] = duration
extras["is_premiere"] = bool(entry.get("is_premiere", False))
extractor_key = entry.get("ie_key") or entry.get("extractor_key") or entry.get("extractor") or ""
if "thumbnail" not in extras and "youtube" in str(extractor_key).lower():
extras["thumbnail"] = "https://img.youtube.com/vi/{id}/maxresdefault.jpg".format(**entry)
return extras
def parse_outtmpl(output_template: str, info_dict: dict, params: dict | None = None) -> str:
"""
Parse yt-dlp output template with given info_dict.
Args:
output_template (str): The output template string.
info_dict (dict): The info dictionary from yt-dlp.
params (dict|None): Additional parameters for yt-dlp.
Returns:
str: The parsed output string.
"""
return get_ytdlp(params=params).prepare_filename(info_dict=info_dict, outtmpl=output_template)
@timed_lru_cache(ttl_seconds=300, max_size=256)
def get_archive_id(url: str) -> dict[str, str | None]:
"""
Get the archive ID for a given URL.
Args:
url (str): URL to check.
Returns:
dict: {
"id": str | None,
"ie_key": str | None,
"archive_id": str | None
}
"""
idDict: dict[str, None] = {
"id": None,
"ie_key": None,
"archive_id": None,
}
from yt_dlp.utils import make_archive_id
for key, _ie in get_ytdlp()._ies.items():
try:
if not _ie.suitable(url):
continue
if not _ie.working():
continue
temp_id = _ie.get_temp_id(url)
if not temp_id:
continue
idDict["id"] = temp_id
idDict["ie_key"] = key
idDict["archive_id"] = make_archive_id(_ie, temp_id)
break
except Exception as e:
LOG.exception(e)
LOG.error(f"Error getting archive ID: {e}")
return idDict

4
app/library/ytdlp.py → app/features/ytdlp/ytdlp.py
View file

@ -112,12 +112,12 @@ def ytdlp_options() -> list[dict[str, Any]]:
"""
from yt_dlp.options import create_parser
from app.library.Utils import REMOVE_KEYS
from app.features.ytdlp.utils import _DATA
parser = create_parser()
ignored_flags: set[str] = {
f.strip() for group in REMOVE_KEYS for v in group.values() for f in v.split(",") if f.strip()
f.strip() for group in _DATA.REMOVE_KEYS for v in group.values() for f in v.split(",") if f.strip()
}
def collect(opts, group: str) -> list[dict[str, Any]]:

14
app/library/YTDLPOpts.py → app/features/ytdlp/ytdlp_opts.py
View file

@ -4,11 +4,11 @@ from pathlib import Path
from typing import Any
from app.features.presets.schemas import Preset
from app.features.ytdlp.utils import arg_converter
from app.library.config import Config
from app.library.Utils import calc_download_path, create_cookies_file, merge_dict
from .config import Config
from .Utils import REMOVE_KEYS, arg_converter, calc_download_path, create_cookies_file, merge_dict
LOG: logging.Logger = logging.getLogger("YTDLPOpts")
LOG: logging.Logger = logging.getLogger("ytdlp.ytdlp_opts")
class ARGSMerger:
@ -123,7 +123,7 @@ class YTDLPCli:
config (Config|None): The Config instance (optional)
"""
from .ItemDTO import Item
from app.library.ItemDTO import Item
if not isinstance(item, Item):
msg = f"Expected Item instance, got {type(item).__name__}"
@ -293,7 +293,9 @@ class YTDLPOpts:
"""
bad_options: dict = {}
if from_user:
bad_options: dict[str, str] = {k: v for d in REMOVE_KEYS for k, v in d.items()}
from app.features.ytdlp.utils import _DATA
bad_options: dict[str, str] = {k: v for d in _DATA.REMOVE_KEYS for k, v in d.items()}
for key, value in config.items():
if from_user and key in bad_options:

8
app/library/ItemDTO.py
View file

@ -7,17 +7,17 @@ from email.utils import formatdate
from pathlib import Path
from typing import TYPE_CHECKING, Any
from app.features.ytdlp.utils import get_archive_id
from app.features.ytdlp.ytdlp_opts import YTDLPOpts
from app.library.encoder import Encoder
from app.library.Utils import (
archive_add,
archive_delete,
archive_read,
clean_item,
get_archive_id,
get_file,
get_file_sidecar,
)
from app.library.YTDLPOpts import YTDLPOpts
if TYPE_CHECKING:
from app.features.presets.schemas import Preset
@ -213,9 +213,9 @@ class Item:
cli: str | None = item.get("cli")
if cli and len(cli) > 2:
from .Utils import arg_converter
try:
from app.features.ytdlp.utils import arg_converter
arg_converter(args=cli, level=True)
data["cli"] = cli
except Exception as e:

180
app/library/LogWrapper.py
View file

@ -1,180 +0,0 @@
import logging
from collections.abc import Callable
from dataclasses import dataclass
LOG = logging.getLogger(__name__)
@dataclass(kw_only=True)
class LogTarget:
"""
A data class that represents a logging target with its level and type.
Attributes:
name (str): The name of the logging target.
target: The logging target, which can be a logging.Logger instance or a callable.
level (int): The logging level for the target.
logger (bool): True if the target is a logging.Logger instance, False otherwise.
type: The type of the target.
"""
name: str | None = None
target: logging.Logger | Callable
level: int
logger: bool
class LogWrapper:
"""
A wrapper class for logging that allows adding multiple logging targets with different logging levels.
Attributes:
targets (list[dict]): A list of dictionaries where each dictionary represents a logging target with its level and type.
Methods:
add_target(target, level=logging.DEBUG):
Adds a new logging target with the specified logging level.
has_targets():
Checks if there are any logging targets added.
_log(level, msg, *args, **kwargs):
Logs a message to all targets that have a logging level less than or equal to the specified level.
debug(msg, *args, **kwargs):
Logs a debug message to all targets.
info(msg, *args, **kwargs):
Logs an info message to all targets.
warning(msg, *args, **kwargs):
Logs a warning message to all targets.
error(msg, *args, **kwargs):
Logs an error message to all targets.
critical(msg, *args, **kwargs):
Logs a critical message to all targets.
"""
def __init__(self):
self.targets: list[LogTarget] = []
"""A list of dictionaries where each dictionary represents a logging target with its level and type."""
def add_target(self, target: logging.Logger | Callable, level: int = logging.DEBUG, name: str | None = None):
"""
Adds a new logging target with the specified logging level.
Args:
target (logging.Logger|Callable): The logging target, which can be a logging.Logger instance or a Callable.
level (int): The logging level for the target. Defaults to logging.DEBUG.
name (str|None): The name of the logging target. Defaults to None.
"""
if not isinstance(target, logging.Logger | Callable):
msg = "Target must be a logging.Logger instance or a callable."
raise TypeError(msg)
if name is None:
name = target.name if isinstance(target, logging.Logger) else target.__name__
self.targets.append(
LogTarget(
name=name,
target=target,
level=level,
logger=isinstance(target, logging.Logger),
)
)
def has_targets(self):
"""
Checks if there are any logging targets added.
Returns:
bool: True if there are targets, False otherwise.
"""
return len(self.targets) > 0
def _log(self, level, msg, *args, **kwargs):
"""
Logs a message to all targets that have a logging level less than or equal to the specified level.
Args:
level (int): The logging level of the message.
msg (str): The message to log.
*args: Additional positional arguments to pass to the logging method.
**kwargs: Additional keyword arguments to pass to the logging method.
"""
for target in self.targets:
if level < target.level:
continue
if target.logger:
target.target.log(level, msg, *args, **kwargs)
else:
target.target(level, msg, *args, **kwargs)
def debug(self, msg, *args, **kwargs):
"""
Logs a debug message to all targets.
Args:
msg (str): The message to log.
*args: Additional positional arguments to pass to the logging method.
**kwargs: Additional keyword arguments to pass to the logging method.
"""
self._log(logging.DEBUG, msg, *args, **kwargs)
def info(self, msg, *args, **kwargs):
"""
Logs an info message to all targets.
Args:
msg (str): The message to log.
*args: Additional positional arguments to pass to the logging method.
**kwargs: Additional keyword arguments to pass to the logging method.
"""
self._log(logging.INFO, msg, *args, **kwargs)
def warning(self, msg, *args, **kwargs):
"""
Logs a warning message to all targets.
Args:
msg (str): The message to log.
*args: Additional positional arguments to pass to the logging method.
**kwargs: Additional keyword arguments to pass to the logging method.
"""
self._log(logging.WARNING, msg, *args, **kwargs)
def error(self, msg, *args, **kwargs):
"""
Logs an error message to all targets.
Args:
msg (str): The message to log.
*args: Additional positional arguments to pass to the logging method.
**kwargs: Additional keyword arguments to pass to the logging method.
"""
self._log(logging.ERROR, msg, *args, **kwargs)
def critical(self, msg, *args, **kwargs):
"""
Logs a critical message to all targets.
Args:
msg (str): The message to log.
*args: Additional positional arguments to pass to the logging method.
**kwargs: Additional keyword arguments to pass to the logging method.
"""
self._log(logging.CRITICAL, msg, *args, **kwargs)

427
app/library/Utils.py
View file

@ -2,70 +2,22 @@ import base64
import copy
import glob
import ipaddress
import json
import logging
import os
import re
import shlex
import socket
import time
import uuid
from datetime import UTC, datetime, timedelta
from email.utils import formatdate
from functools import lru_cache, wraps
from http.cookiejar import MozillaCookieJar
from pathlib import Path
from typing import Any
from Crypto.Cipher import AES
from yt_dlp.utils import age_restricted
from .ytdlp import YTDLP, make_archive_id
LOG: logging.Logger = logging.getLogger("Utils")
REMOVE_KEYS: list = [
{
"paths": "-P, --paths",
"outtmpl": "-o, --output",
"progress_hooks": "--progress_hooks",
"postprocessor_hooks": "--postprocessor_hooks",
"post_hooks": "--post_hooks",
},
{
"quiet": "-q, --quiet",
"no_warnings": "--no-warnings",
"skip_download": "--skip-download",
"forceprint": "-O, --print",
"simulate": "--simulate",
"noprogress": "--no-progress",
"wait_for_video": "--wait-for-video",
"progress_delta": " --progress-delta",
"progress_template": "--progress-template",
"consoletitle": "--console-title",
"progress_with_newline": "--newline",
"forcejson": "-j, --dump-single-json",
"opt_update_to": "--update-to",
"opt_ap_list_mso": "--ap-list-mso",
"opt_batch_file": "-a, --batch-file",
"opt_alias": "--alias",
"opt_list_extractors": "--list-extractors",
"opt_version": "--version",
"opt_help": "-h, --help",
"opt_update": "-U, --update",
"opt_list_subtitles": "--list-subs",
"opt_list_thumbnails": "--list-thumbnails",
"opt_list_format": "-F, --list-formats",
"opt_dump_agent": "--dump-user-agent",
"opt_extractor_descriptions": "--extractor-descriptions",
"opt_list_impersonate_targets": "--list-impersonate-targets",
},
]
"Keys to remove from yt-dlp options at various levels."
YTDLP_INFO_CLS: YTDLP | None = None
"Cached YTDLP info class."
ALLOWED_SUBS_EXTENSIONS: set[str] = {".srt", ".vtt", ".ass"}
"Allowed subtitle file extensions."
@ -90,83 +42,6 @@ class FileLogFormatter(logging.Formatter):
return datetime.fromtimestamp(record.created).astimezone().isoformat(timespec="milliseconds")
def get_ytdlp(params: dict | None = None) -> YTDLP:
"""
Get a static YTDLP instance for info extraction.
Args:
params (dict|None): YTDLP parameters.
Returns:
YTDLP: A static YTDLP instance.
"""
global YTDLP_INFO_CLS # noqa: PLW0603
default_params: dict[str, Any] = {
"color": "no_color",
"extract_flat": True,
"skip_download": True,
"ignoreerrors": True,
"ignore_no_formats_error": True,
"quiet": True,
}
if params:
return YTDLP(params=merge_dict(params, default_params))
if YTDLP_INFO_CLS is None:
YTDLP_INFO_CLS = YTDLP(params=default_params)
return YTDLP_INFO_CLS
def patch_metadataparser() -> None:
"""
Patches yt_dlp MetadataParserPP action to handle subprocess pickling issues.
"""
try:
from yt_dlp.postprocessor.metadataparser import MetadataParserPP
from yt_dlp.utils import Namespace
except Exception as exc:
LOG.warning(f"Unable to import yt_dlp metadata parser for patching: {exc!s}")
return
if getattr(MetadataParserPP.Actions, "_ytptube_patched", False):
return
class _ActionNS(Namespace):
_ACTIONS_STR: list[str] = []
@staticmethod
def _get_name(func) -> str | None:
if not callable(func):
return None
target = getattr(func, "__func__", func)
module_name = getattr(target, "__module__", None)
qual_name = getattr(target, "__qualname__", getattr(target, "__name__", None))
return f"{module_name}.{qual_name}" if module_name and qual_name else None
def __contains__(self, candidate: object) -> bool:
if candidate in self.__dict__.values():
return True
if func_name := _ActionNS._get_name(candidate):
if len(_ActionNS._ACTIONS_STR) < 1:
_ActionNS._ACTIONS_STR.extend([_ActionNS._get_name(value) for value in self.__dict__.values()])
return func_name in _ActionNS._ACTIONS_STR
return False
actions_dict: dict[str, Any] = dict(MetadataParserPP.Actions.items_)
MetadataParserPP.Actions = _ActionNS(**actions_dict)
MetadataParserPP.Actions._ytptube_patched = True
LOG.debug("MetadataParserPP action namespace patch applied successfully.")
def timed_lru_cache(ttl_seconds: int, max_size: int = 128):
"""
Decorator that applies an LRU cache with a time-to-live (TTL) to a function.
@ -525,93 +400,6 @@ def validate_url(url: str, allow_internal: bool = False) -> bool:
return True
def arg_converter(
args: str,
level: int | bool | None = None,
dumps: bool = False,
removed_options: list | None = None,
keep_defaults: bool = False,
) -> dict:
"""
Convert yt-dlp options to a dictionary.
Args:
args (str): yt-dlp options string.
level (int|bool|None): Level of options to remove, True for all.
dumps (bool): Dump options as JSON.
removed_options (list|None): List of removed options.
keep_defaults (bool): Keep default options.
Returns:
dict: yt-dlp options dictionary.
"""
import yt_dlp.options
create_parser = yt_dlp.options.create_parser
def _default_opts(args: str):
patched_parser = create_parser()
try:
yt_dlp.options.create_parser = lambda: patched_parser
return yt_dlp.parse_options(args)
finally:
yt_dlp.options.create_parser = create_parser
try:
patch_metadataparser()
except Exception as exc:
LOG.debug("Metadata parser patch failed to apply: %s", exc)
default_opts = _default_opts([]).ydl_opts
if args:
# important to ignore external config files.
args = "--ignore-config " + args
opts = yt_dlp.parse_options(shlex.split(args, posix=os.name != "nt")).ydl_opts
diff = {k: v for k, v in opts.items() if default_opts[k] != v} if not keep_defaults else opts.items()
if "postprocessors" in diff:
diff["postprocessors"] = [pp for pp in diff["postprocessors"] if pp not in default_opts["postprocessors"]]
if "_warnings" in diff:
diff.pop("_warnings", None)
if level is True or isinstance(level, int):
bad_options = {}
if isinstance(level, bool) or not isinstance(level, int):
level = len(REMOVE_KEYS)
for i, item in enumerate(REMOVE_KEYS):
if i > level:
break
bad_options.update(item.items())
for key in diff.copy():
if key not in bad_options:
continue
if isinstance(removed_options, list):
removed_options.append(bad_options[key])
diff.pop(key, None)
if dumps is True:
from .encoder import Encoder
if "match_filter" in diff:
import inspect
matchFilter = inspect.getclosurevars(diff["match_filter"].func).nonlocals["filters"]
if isinstance(matchFilter, set):
diff["match_filter"] = {"filters": list(matchFilter)}
return json.loads(json.dumps(diff, cls=Encoder, default=str))
return diff
def validate_uuid(uuid_str: str, version: int = 4) -> bool:
"""
Validate if the UUID is valid.
@ -1314,51 +1102,6 @@ def load_cookies(file: str | Path) -> tuple[bool, MozillaCookieJar]:
raise ValueError(msg) from e
@timed_lru_cache(ttl_seconds=300, max_size=256)
def get_archive_id(url: str) -> dict[str, str | None]:
"""
Get the archive ID for a given URL.
Args:
url (str): URL to check.
Returns:
dict: {
"id": str | None,
"ie_key": str | None,
"archive_id": str | None
}
"""
idDict: dict[str, None] = {
"id": None,
"ie_key": None,
"archive_id": None,
}
for key, _ie in get_ytdlp()._ies.items():
try:
if not _ie.suitable(url):
continue
if not _ie.working():
continue
temp_id = _ie.get_temp_id(url)
if not temp_id:
continue
idDict["id"] = temp_id
idDict["ie_key"] = key
idDict["archive_id"] = make_archive_id(_ie, temp_id)
break
except Exception as e:
LOG.exception(e)
LOG.error(f"Error getting archive ID: {e}")
return idDict
def dt_delta(delta: timedelta) -> str:
"""
Convert a timedelta object to a human-readable string.
@ -1392,38 +1135,6 @@ def dt_delta(delta: timedelta) -> str:
return " ".join(parts)
def extract_ytdlp_logs(logs: list[str], filters: list[str | re.Pattern] = None) -> list[str]:
"""
Extract yt-dlp log lines matching built-in filters plus any extras.
Args:
logs (list): log strings.
filters (list[str|Re.Pattern]): Optional extra filters of strings and/or regex.
Returns:
(list): List of matching log lines.
"""
all_patterns: list[str | re.Pattern] = [
"This live event will begin",
"Video unavailable. This video is private",
"This video is available to this channel",
"Private video. Sign in if you've been granted access to this video",
"[youtube] Premieres in",
"Falling back on generic information extractor",
"URL could be a direct video link, returning it as such",
] + (filters or [])
compiled: list[re.Pattern] = [
p if isinstance(p, re.Pattern) else re.compile(re.escape(p), re.IGNORECASE) for p in all_patterns
]
matched: list[str] = []
matched.extend(line for line in logs if line and any(p.search(line) for p in compiled))
return list(dict.fromkeys(matched))
def delete_dir(dir: Path) -> bool:
"""
Delete a directory and all its contents.
@ -1553,53 +1264,6 @@ def str_to_dt(time_str: str, now=None) -> datetime:
return dt
def ytdlp_reject(entry: dict, yt_params: dict) -> tuple[bool, str]:
"""
Implement yt-dlp reject filter logic.
Args:
entry (dict): The entry to check.
yt_params (dict): The yt-dlp parameters containing filters.
Returns:
tuple[bool, str]: A tuple where the first element is True if the entry passes the filters, or False if it does not,
and the second element is a message explaining the reason for rejection or an empty string if it passes.
"""
if title := entry.get("title"):
if (matchtitle := yt_params.get("matchtitle")) and not re.search(matchtitle, title, re.IGNORECASE):
return (False, f'"{title}" title did not match pattern "{matchtitle}". Skipping download.')
if (rejecttitle := yt_params.get("rejecttitle")) and re.search(rejecttitle, title, re.IGNORECASE):
return (False, f'"{title}" title matched reject pattern "{rejecttitle}". Skipping download.')
date = entry.get("upload_date")
date_range = yt_params.get("daterange")
if (date and date_range) and date not in date_range:
return (False, f"Upload date '{date}' is not in range '{date_range}'.")
view_count = entry.get("view_count")
if view_count is not None:
min_views = yt_params.get("min_views")
if min_views is not None and view_count < min_views:
return (
False,
f"Skipping {entry.get('title', 'video')}, because it has not reached minimum view count ({view_count}/{min_views}).",
)
max_views = yt_params.get("max_views")
if max_views is not None and view_count > max_views:
return (
False,
f"Skipping {entry.get('title', 'video')}, because it has exceeded maximum view count ({view_count}/{max_views}).",
)
if entry.get("age_limit") and age_restricted(entry.get("age_limit"), yt_params.get("age_limit")):
return (False, f'Video "{entry.get("title", "unknown")}" is age restricted.')
return (True, "")
def list_folders(path: Path, base: Path, depth_limit: int) -> list[str]:
"""
List all folders relative to a base path, up to a specified depth limit.
@ -1761,94 +1425,3 @@ def create_cookies_file(cookies: str, file: Path | None = None) -> Path:
load_cookies(file)
return file
def get_thumbnail(thumbnails: list) -> str | None:
"""
Extract thumbnail URL from a yt-dlp entry.
Args:
thumbnails (list): The list of thumbnail dictionaries from yt-dlp entry.
Returns:
str | None: The thumbnail URL if available, otherwise None.
"""
if not thumbnails or not isinstance(thumbnails, list):
return None
def _thumb_sort_key(thumb: dict) -> tuple:
return (
thumb.get("preference") if thumb.get("preference") is not None else -1,
thumb.get("width") if thumb.get("width") is not None else -1,
thumb.get("height") if thumb.get("height") is not None else -1,
thumb.get("id") if thumb.get("id") is not None else "",
thumb.get("url"),
)
return max(thumbnails, key=_thumb_sort_key, default=None)
def get_extras(entry: dict, kind: str = "video") -> dict:
"""
Extract useful information from a yt-dlp entry.
Args:
entry (dict): The entry data from yt-dlp.
kind (str): The type of the item (e.g., "video", "playlist").
Returns:
dict: The extracted data.
"""
extras = {}
if not entry or not isinstance(entry, dict):
return extras
if "playlist" == kind:
for property in ("id", "title", "uploader", "uploader_id"):
if val := entry.get(property):
extras[f"playlist_{property}"] = val
if thumbnail := get_thumbnail(entry.get("thumbnails", [])):
extras["thumbnail"] = thumbnail.get("url")
elif thumbnail := entry.get("thumbnail"):
extras["thumbnail"] = thumbnail
for property in ("uploader", "channel"):
if val := entry.get(property):
extras[property] = val
if release_in := entry.get("release_timestamp"):
extras["release_in"] = formatdate(release_in, usegmt=True)
if release_in and "is_upcoming" == entry.get("live_status"):
extras["is_live"] = release_in
if duration := entry.get("duration"):
extras["duration"] = duration
extras["is_premiere"] = bool(entry.get("is_premiere", False))
extractor_key = entry.get("ie_key") or entry.get("extractor_key") or entry.get("extractor") or ""
if "thumbnail" not in extras and "youtube" in str(extractor_key).lower():
extras["thumbnail"] = "https://img.youtube.com/vi/{id}/maxresdefault.jpg".format(**entry)
return extras
def parse_outtmpl(output_template: str, info_dict: dict, params: dict | None = None) -> str:
"""
Parse yt-dlp output template with given info_dict.
Args:
output_template (str): The output template string.
info_dict (dict): The info dictionary from yt-dlp.
params (dict|None): Additional parameters for yt-dlp.
Returns:
str: The parsed output string.
"""
return get_ytdlp(params=params).prepare_filename(info_dict=info_dict, outtmpl=output_template)

7
app/library/downloads/core.py
View file

@ -13,12 +13,13 @@ from typing import TYPE_CHECKING, Any
import yt_dlp.utils
from app.features.ytdlp.utils import extract_ytdlp_logs
from app.features.ytdlp.ytdlp import YTDLP
from app.library.config import Config
from app.library.Events import EventBus, Events
from app.library.Utils import create_cookies_file, extract_ytdlp_logs
from app.library.ytdlp import YTDLP
from app.library.Utils import create_cookies_file
from .extractor import extract_info_sync
from ...features.ytdlp.extractor import extract_info_sync
from .hooks import HookHandlers, NestedLogger
from .process_manager import ProcessManager
from .status_tracker import StatusTracker

6
app/library/downloads/item_adder.py
View file

@ -9,20 +9,18 @@ import yt_dlp.utils
from app.features.conditions.service import Conditions
from app.features.presets.service import Presets
from app.features.ytdlp.utils import arg_converter, get_extras, ytdlp_reject
from app.library.Events import Events
from app.library.ItemDTO import ItemDTO
from app.library.Utils import (
archive_add,
archive_read,
arg_converter,
create_cookies_file,
get_extras,
merge_dict,
ytdlp_reject,
)
from ...features.ytdlp.extractor import fetch_info
from .core import Download
from .extractor import fetch_info
from .playlist_processor import process_playlist
from .video_processor import add_video

3
app/library/downloads/playlist_processor.py
View file

@ -3,7 +3,8 @@
import logging
from typing import TYPE_CHECKING, Any
from app.library.Utils import merge_dict, ytdlp_reject
from app.features.ytdlp.utils import ytdlp_reject
from app.library.Utils import merge_dict
if TYPE_CHECKING:
from app.library.ItemDTO import Item

8
app/library/downloads/queue_manager.py
View file

@ -8,7 +8,6 @@ from typing import TYPE_CHECKING
from aiohttp import web
from app.library.config import Config
from app.library.downloads.extractor import shutdown_extractor
from app.library.Events import EventBus, Events
from app.library.ItemDTO import Item, ItemDTO
from app.library.Scheduler import Scheduler
@ -48,7 +47,7 @@ class DownloadQueue(metaclass=Singleton):
def get_instance(config: Config | None = None) -> "DownloadQueue":
return DownloadQueue(config=config)
def attach(self, app: web.Application) -> None:
def attach(self, _: web.Application) -> None:
Services.get_instance().add("queue", self)
async def event_handler(_, __):
@ -75,7 +74,7 @@ class DownloadQueue(metaclass=Singleton):
id=delete_old_history.__name__,
)
app.on_shutdown.append(self.on_shutdown)
# app.on_shutdown.append(self.on_shutdown)
async def test(self) -> bool:
await self.done.test()
@ -217,8 +216,7 @@ class DownloadQueue(metaclass=Singleton):
return self.pool.is_paused()
async def on_shutdown(self, _: web.Application):
# await self.pool.shutdown()
await shutdown_extractor()
await self.pool.shutdown()
async def add(self, item: Item, already: set | None = None, entry: dict | None = None) -> dict[str, str]:
"""

6
app/library/downloads/video_processor.py
View file

@ -6,11 +6,11 @@ from datetime import UTC, datetime, timedelta
from email.utils import formatdate
from typing import TYPE_CHECKING
from app.features.ytdlp.utils import extract_ytdlp_logs, get_extras
from app.library.downloads import Download
from app.library.Events import Events
from app.library.ItemDTO import ItemDTO
from app.library.Utils import calc_download_path, extract_ytdlp_logs, get_extras, merge_dict, str_to_dt
from .core import Download
from app.library.Utils import calc_download_path, merge_dict, str_to_dt
if TYPE_CHECKING:
from app.library.ItemDTO import Item

2
app/routes/api/archiver.py
View file

@ -5,9 +5,9 @@ from aiohttp import web
from aiohttp.web import Request, Response
from app.features.presets.service import Presets
from app.features.ytdlp.ytdlp_opts import YTDLPOpts
from app.library.Archiver import Archiver
from app.library.router import route
from app.library.YTDLPOpts import YTDLPOpts
LOG: logging.Logger = logging.getLogger(__name__)

2
app/routes/api/docs.py
View file

@ -5,11 +5,11 @@ from datetime import UTC, datetime
from aiohttp import web
from aiohttp.web import Request, Response
from app.features.ytdlp.ytdlp_opts import YTDLPOpts
from app.library.cache import Cache
from app.library.config import Config
from app.library.httpx_client import Globals, build_request_headers, get_async_client, resolve_curl_transport
from app.library.router import add_route, route
from app.library.YTDLPOpts import YTDLPOpts
LOG: logging.Logger = logging.getLogger(__name__)

2
app/routes/api/images.py
View file

@ -8,13 +8,13 @@ from urllib.parse import urlparse
from aiohttp import web
from aiohttp.web import Request, Response
from app.features.ytdlp.ytdlp_opts import YTDLPOpts
from app.library.ag_utils import ag
from app.library.cache import Cache
from app.library.config import Config
from app.library.httpx_client import Globals, build_request_headers, get_async_client, resolve_curl_transport
from app.library.router import route
from app.library.Utils import validate_url
from app.library.YTDLPOpts import YTDLPOpts
LOG: logging.Logger = logging.getLogger(__name__)

16
app/routes/api/yt_dlp.py
View file

@ -8,20 +8,18 @@ from aiohttp import web
from aiohttp.web import Request, Response
from app.features.presets.service import Presets
from app.features.ytdlp.extractor import fetch_info
from app.features.ytdlp.utils import arg_converter, get_archive_id
from app.features.ytdlp.ytdlp_opts import YTDLPCli, YTDLPOpts
from app.library.cache import Cache
from app.library.config import Config
from app.library.downloads.extractor import fetch_info
from app.library.encoder import Encoder
from app.library.ItemDTO import Item
from app.library.router import route
from app.library.Utils import (
REMOVE_KEYS,
archive_read,
arg_converter,
get_archive_id,
validate_url,
)
from app.library.YTDLPOpts import YTDLPCli, YTDLPOpts
LOG: logging.Logger = logging.getLogger(__name__)
@ -58,7 +56,9 @@ async def convert(request: Request) -> Response:
if "format" in data:
response["format"] = data["format"]
bad_options = {k: v for d in REMOVE_KEYS for k, v in d.items()}
from app.features.ytdlp.utils import _DATA
bad_options = {k: v for d in _DATA.REMOVE_KEYS for k, v in d.items()}
removed_options = []
for key in data:
@ -230,7 +230,7 @@ async def get_options() -> Response:
Response: The response object with the yt-dlp CLI options.
"""
from app.library.ytdlp import ytdlp_options
from app.features.ytdlp.ytdlp import ytdlp_options
return web.json_response(body=json.dumps(ytdlp_options(), indent=4, default=str), status=web.HTTPOk.status_code)
@ -244,8 +244,6 @@ async def get_archive_ids(request: Request, config: Config) -> Response:
Response: The response object with the yt-dlp CLI options.
"""
from app.library.Utils import get_archive_id
data = (await request.json()) if request.body_exists else None
if not data or not isinstance(data, list):
return web.json_response(

4
app/tests/test_itemdto.py
View file

@ -26,7 +26,7 @@ class TestItemFormatAndBasics:
}
with (
patch("app.library.ItemDTO.Item._default_preset", return_value="default"),
patch("app.library.Utils.arg_converter") as mock_arg_conv,
patch("app.features.ytdlp.utils.arg_converter") as mock_arg_conv,
):
mock_arg_conv.return_value = None
item = Item.format(data)
@ -56,7 +56,7 @@ class TestItemFormatAndBasics:
):
Item.format({"url": "https://example.com", "preset": "bad"})
@patch("app.library.Utils.arg_converter")
@patch("app.features.ytdlp.utils.arg_converter")
def test_format_cli_parse_error(self, mock_arg_conv):
mock_arg_conv.side_effect = RuntimeError("bad cli")
with pytest.raises(ValueError, match="Failed to parse command options"):

98
app/tests/test_logwrapper.py
View file

@ -1,98 +0,0 @@
import logging
from typing import Any
import pytest
from app.library.LogWrapper import LogWrapper
class CaptureHandler(logging.Handler):
def __init__(self) -> None:
super().__init__()
self.records: list[logging.LogRecord] = []
def emit(self, record: logging.LogRecord) -> None:
self.records.append(record)
def make_logger(name: str = "lw_test") -> tuple[logging.Logger, CaptureHandler]:
logger = logging.getLogger(name)
logger.setLevel(logging.DEBUG)
handler = CaptureHandler()
# Avoid duplicate handlers when tests run multiple times
for h in list(logger.handlers):
logger.removeHandler(h)
logger.addHandler(handler)
return logger, handler
class TestLogWrapper:
def test_add_target_type_validation(self) -> None:
lw = LogWrapper()
with pytest.raises(TypeError, match=r"Target must be a logging\.Logger instance or a callable"):
lw.add_target(123) # type: ignore[arg-type]
def test_add_target_name_inference_and_custom(self) -> None:
lw = LogWrapper()
logger, _ = make_logger("one")
# Name inferred from logger
lw.add_target(logger)
assert lw.targets[-1].name == "one"
assert lw.has_targets() is True
# Name inferred from callable
def sink(level: int, msg: str, *args: Any, **kwargs: Any) -> None: # noqa: ARG001
return None
lw.add_target(sink)
assert lw.targets[-1].name == "sink"
# Custom name overrides
lw.add_target(logger, name="custom")
assert lw.targets[-1].name == "custom"
def test_level_filtering_and_dispatch(self) -> None:
lw = LogWrapper()
logger, cap = make_logger("cap")
calls: list[tuple[int, str, tuple, dict]] = []
def sink(level: int, msg: str, *args: Any, **kwargs: Any) -> None:
calls.append((level, msg, args, kwargs))
# Logger target at INFO, callable target at WARNING
lw.add_target(logger, level=logging.INFO)
lw.add_target(sink, level=logging.WARNING)
# DEBUG should hit none
lw.debug("d1")
assert len(cap.records) == 0
assert len(calls) == 0
# INFO hits logger only
lw.info("hello %s", "X")
assert len(cap.records) == 1
assert cap.records[0].levelno == logging.INFO
assert cap.records[0].getMessage() == "hello X"
assert len(calls) == 0
# WARNING hits both
lw.warning("warn %s", "Y", extra={"k": 1})
assert len(cap.records) == 2
assert cap.records[1].levelno == logging.WARNING
assert cap.records[1].getMessage() == "warn Y"
assert len(calls) == 1
lvl, msg, args, kwargs = calls[0]
assert lvl == logging.WARNING
assert msg == "warn %s"
assert args == ("Y",)
assert "extra" in kwargs
assert kwargs["extra"] == {"k": 1}
# ERROR still hits both; CRITICAL too
lw.error("err")
lw.critical("boom")
assert any(r.levelno == logging.ERROR for r in cap.records)
assert any(r.levelno == logging.CRITICAL for r in cap.records)
assert any(c[0] == logging.ERROR for c in calls)
assert any(c[0] == logging.CRITICAL for c in calls)

489
app/tests/test_utils.py
View file

@ -10,13 +10,12 @@ from pathlib import Path
from unittest.mock import MagicMock, patch
import pytest
from app.features.ytdlp.utils import arg_converter
from app.library.Utils import (
FileLogFormatter,
archive_add,
archive_delete,
archive_read,
arg_converter,
calc_download_path,
check_id,
clean_item,
@ -24,15 +23,12 @@ from app.library.Utils import (
delete_dir,
dt_delta,
encrypt_data,
extract_ytdlp_logs,
get,
get_archive_id,
get_file,
get_file_sidecar,
get_files,
get_mime_type,
get_possible_images,
get_ytdlp,
init_class,
is_private_address,
list_folders,
@ -40,7 +36,6 @@ from app.library.Utils import (
load_modules,
merge_dict,
move_file,
parse_outtmpl,
parse_tags,
read_logfile,
rename_file,
@ -50,9 +45,7 @@ from app.library.Utils import (
timed_lru_cache,
validate_url,
validate_uuid,
ytdlp_reject,
)
from app.library.downloads.extractor import extract_info_sync
class TestTimedLruCache:
@ -1242,29 +1235,6 @@ class TestValidateUrl:
assert True
class TestExtractYtdlpLogs:
"""Test YTDLP log extraction function."""
def test_extract_ytdlp_logs_basic(self):
"""Test basic log extraction."""
logs = ["This live event will begin soon", "ERROR: Failed", "WARNING: Deprecated"]
result = extract_ytdlp_logs(logs)
assert isinstance(result, list)
assert len(result) >= 1 # Should match "This live event will begin"
def test_extract_ytdlp_logs_with_filters(self):
"""Test log extraction with filters."""
logs = ["INFO: Downloading", "ERROR: Failed", "WARNING: Deprecated"]
filters = [re.compile(r"ERROR")]
result = extract_ytdlp_logs(logs, filters)
assert len(result) >= 0 # Should filter based on patterns
def test_extract_ytdlp_logs_empty(self):
"""Test with empty logs."""
result = extract_ytdlp_logs([])
assert result == []
class TestGetFileSidecar:
"""Test file sidecar function."""
@ -1334,39 +1304,6 @@ class TestArchiveFunctions:
assert result == []
class TestExtractInfo:
"""Test the extract_info function."""
@patch("app.library.downloads.extractor.YTDLP")
def test_extract_info_basic(self, mock_ytdlp_class):
"""Test basic extract_info functionality."""
mock_ytdlp = MagicMock()
mock_ytdlp.extract_info.return_value = {"title": "Test Video", "id": "test123"}
mock_ytdlp_class.return_value = mock_ytdlp
config = {"quiet": True}
url = "https://example.com/video"
(result, logs) = extract_info_sync(config, url)
assert isinstance(result, dict), "Result should be a dictionary"
assert isinstance(logs, list), "Logs should be a list"
mock_ytdlp.extract_info.assert_called_once()
@patch("app.library.downloads.extractor.YTDLP")
def test_extract_info_with_debug(self, mock_ytdlp_class):
"""Test extract_info with debug enabled."""
mock_ytdlp = MagicMock()
mock_ytdlp.extract_info.return_value = {"title": "Test Video"}
mock_ytdlp_class.return_value = mock_ytdlp
config = {}
url = "https://example.com/video"
(result, logs) = extract_info_sync(config, url, debug=True)
assert isinstance(result, dict), "Result should be a dictionary"
assert isinstance(logs, list), "Logs should be a list"
class TestCheckId:
"""Test the check_id function."""
@ -1699,26 +1636,6 @@ class TestLoadCookies:
assert True
class TestGetArchiveId:
"""Test the get_archive_id function."""
@patch("app.library.Utils.YTDLP_INFO_CLS")
def test_get_archive_id_basic(self, mock_ytdlp):
"""Test basic archive ID extraction."""
mock_ytdlp._ies = {}
result = get_archive_id("https://youtube.com/watch?v=test123")
assert isinstance(result, dict)
assert "id" in result
assert "ie_key" in result
assert "archive_id" in result
def test_get_archive_id_invalid_url(self):
"""Test with invalid URL."""
result = get_archive_id("invalid-url")
assert isinstance(result, dict)
class TestStrToDt:
"""Test the str_to_dt function."""
@ -1740,31 +1657,6 @@ class TestStrToDt:
assert True
class TestYtdlpReject:
"""Test the ytdlp_reject function."""
def test_ytdlp_reject_basic(self):
"""Test basic rejection logic."""
entry = {"title": "Test Video", "view_count": 1000}
yt_params = {}
passed, message = ytdlp_reject(entry, yt_params)
assert isinstance(passed, bool)
assert isinstance(message, str)
def test_ytdlp_reject_with_filters(self):
"""Test rejection with filters."""
entry = {"title": "Test Video", "upload_date": "20230101"}
yt_params = {"daterange": MagicMock()}
# Mock daterange to simulate rejection
yt_params["daterange"].__contains__ = MagicMock(return_value=False)
passed, message = ytdlp_reject(entry, yt_params)
assert isinstance(passed, bool)
assert isinstance(message, str)
class TestInitClass:
"""Test the init_class function."""
@ -2455,382 +2347,3 @@ class TestMoveFile:
# Original file should still exist
assert test_file.exists()
class TestGetThumbnail:
def test_returns_none_for_empty_list(self):
"""Test that None is returned for an empty thumbnail list."""
from app.library.Utils import get_thumbnail
assert get_thumbnail([]) is None
def test_returns_none_for_non_list(self):
"""Test that None is returned for non-list input."""
from app.library.Utils import get_thumbnail
assert get_thumbnail(None) is None
assert get_thumbnail("not a list") is None
assert get_thumbnail({"not": "list"}) is None
def test_returns_highest_preference_thumbnail(self):
"""Test that the thumbnail with highest preference is returned."""
from app.library.Utils import get_thumbnail
thumbnails = [
{"url": "low.jpg", "preference": 1, "width": 100, "height": 100},
{"url": "high.jpg", "preference": 10, "width": 200, "height": 200},
{"url": "medium.jpg", "preference": 5, "width": 150, "height": 150},
]
result = get_thumbnail(thumbnails)
assert result == {"url": "high.jpg", "preference": 10, "width": 200, "height": 200}
def test_returns_highest_width_when_preference_equal(self):
"""Test that the thumbnail with highest width is returned when preference is equal."""
from app.library.Utils import get_thumbnail
thumbnails = [
{"url": "small.jpg", "preference": 1, "width": 100, "height": 100},
{"url": "large.jpg", "preference": 1, "width": 200, "height": 200},
{"url": "medium.jpg", "preference": 1, "width": 150, "height": 150},
]
result = get_thumbnail(thumbnails)
assert result == {"url": "large.jpg", "preference": 1, "width": 200, "height": 200}
def test_handles_missing_attributes(self):
"""Test that thumbnails with missing attributes are handled correctly."""
from app.library.Utils import get_thumbnail
thumbnails = [
{"url": "no_pref.jpg", "width": 100},
{"url": "with_pref.jpg", "preference": 5, "width": 50},
]
result = get_thumbnail(thumbnails)
assert result["url"] == "with_pref.jpg"
def test_returns_first_when_all_equal(self):
"""Test that any thumbnail is returned when all attributes are equal."""
from app.library.Utils import get_thumbnail
thumbnails = [
{"url": "first.jpg"},
{"url": "second.jpg"},
]
result = get_thumbnail(thumbnails)
assert result is not None
assert result["url"] in ["first.jpg", "second.jpg"]
class TestGetExtras:
def test_returns_empty_dict_for_none(self):
"""Test that empty dict is returned for None input."""
from app.library.Utils import get_extras
assert get_extras(None) == {}
def test_returns_empty_dict_for_non_dict(self):
"""Test that empty dict is returned for non-dict input."""
from app.library.Utils import get_extras
assert get_extras("not a dict") == {}
assert get_extras([]) == {}
def test_extracts_video_information(self):
"""Test extracting information from a video entry."""
from app.library.Utils import get_extras
entry = {
"id": "test123",
"title": "Test Video",
"uploader": "Test Uploader",
"channel": "Test Channel",
"thumbnails": [{"url": "thumb.jpg", "preference": 1}],
"duration": 120,
}
result = get_extras(entry, kind="video")
assert result["uploader"] == "Test Uploader"
assert result["channel"] == "Test Channel"
assert result["thumbnail"] == "thumb.jpg"
assert result["duration"] == 120
assert result["is_premiere"] is False
def test_extracts_playlist_information(self):
"""Test extracting information from a playlist entry."""
from app.library.Utils import get_extras
entry = {
"id": "playlist123",
"title": "Test Playlist",
"uploader": "Playlist Owner",
"uploader_id": "owner123",
}
result = get_extras(entry, kind="playlist")
assert result["playlist_id"] == "playlist123"
assert result["playlist_title"] == "Test Playlist"
assert result["playlist_uploader"] == "Playlist Owner"
assert result["playlist_uploader_id"] == "owner123"
def test_handles_release_timestamp(self):
"""Test handling of release_timestamp for upcoming content."""
from app.library.Utils import get_extras
entry = {
"release_timestamp": 1234567890,
}
result = get_extras(entry)
assert "release_in" in result
assert result["release_in"] == "Fri, 13 Feb 2009 23:31:30 GMT"
def test_handles_upcoming_live_stream(self):
"""Test handling of upcoming live stream."""
from app.library.Utils import get_extras
entry = {
"release_timestamp": 1234567890,
"live_status": "is_upcoming",
}
result = get_extras(entry)
assert result["is_live"] == 1234567890
assert "release_in" in result
def test_handles_premiere_flag(self):
"""Test handling of is_premiere flag."""
from app.library.Utils import get_extras
entry = {
"is_premiere": True,
}
result = get_extras(entry)
assert result["is_premiere"] is True
entry2 = {"is_premiere": False}
result2 = get_extras(entry2)
assert result2["is_premiere"] is False
def test_youtube_fallback_thumbnail(self):
"""Test fallback thumbnail generation for YouTube videos."""
from app.library.Utils import get_extras
entry = {
"id": "dQw4w9WgXcQ",
"ie_key": "Youtube",
}
result = get_extras(entry)
assert result["thumbnail"] == "https://img.youtube.com/vi/dQw4w9WgXcQ/maxresdefault.jpg"
def test_thumbnail_string_fallback(self):
"""Test fallback to thumbnail string when thumbnails list not available."""
from app.library.Utils import get_extras
entry = {
"thumbnail": "https://example.com/thumb.jpg",
}
result = get_extras(entry)
assert result["thumbnail"] == "https://example.com/thumb.jpg"
class TestGetStaticYtdlp:
"""Test the get_static_ytdlp function."""
def setup_method(self):
"""Reset YTDLP singleton state before each test."""
import app.library.Utils as Utils
Utils.YTDLP_INFO_CLS = None
def test_get_static_ytdlp_returns_instance(self):
"""Test that get_static_ytdlp returns a YTDLP instance."""
from app.library.ytdlp import YTDLP
# Get the cached instance
instance = get_ytdlp()
assert instance is not None
assert isinstance(instance, YTDLP)
def test_get_static_ytdlp_returns_same_instance(self):
"""Test that get_static_ytdlp returns the same cached instance."""
instance1 = get_ytdlp()
instance2 = get_ytdlp()
assert instance1 is instance2
def test_get_static_ytdlp_with_params(self):
"""Test that get_static_ytdlp returns a new instance when params are provided."""
instance1 = get_ytdlp()
instance2 = get_ytdlp(params={"quiet": False})
assert instance1 is not instance2
assert instance2 is not None
def test_get_static_ytdlp_has_correct_params(self):
"""Test that get_static_ytdlp initializes with correct parameters."""
instance = get_ytdlp()
# Access the internal params
params = instance.params
assert params.get("color") == "no_color"
assert params.get("extract_flat") is True
assert params.get("skip_download") is True
assert params.get("ignoreerrors") is True
assert params.get("ignore_no_formats_error") is True
assert params.get("quiet") is True
class TestParseOuttmpl:
"""Test the parse_outtmpl function."""
def setup_method(self):
"""Reset YTDLP singleton state before each test."""
import app.library.Utils as Utils
Utils.YTDLP_INFO_CLS = None
def test_parse_outtmpl_basic(self):
"""Test basic template parsing with simple placeholders."""
template = "%(title)s.%(ext)s"
info_dict = {
"title": "Test Video",
"ext": "mp4",
}
result = parse_outtmpl(template, info_dict)
assert result == "Test Video.mp4"
def test_parse_outtmpl_with_id(self):
"""Test template parsing with video ID."""
template = "[%(id)s] %(title)s.%(ext)s"
info_dict = {
"id": "dQw4w9WgXcQ",
"title": "Never Gonna Give You Up",
"ext": "webm",
}
result = parse_outtmpl(template, info_dict)
assert result == "[dQw4w9WgXcQ] Never Gonna Give You Up.webm"
def test_parse_outtmpl_with_uploader(self):
"""Test template parsing with uploader information."""
template = "%(uploader)s - %(title)s.%(ext)s"
info_dict = {
"uploader": "Rick Astley",
"title": "Never Gonna Give You Up",
"ext": "mp4",
}
result = parse_outtmpl(template, info_dict)
assert result == "Rick Astley - Never Gonna Give You Up.mp4"
def test_parse_outtmpl_with_nested_path(self):
"""Test template parsing with nested directory structure."""
template = "%(uploader)s/%(title)s.%(ext)s"
info_dict = {
"uploader": "Test Channel",
"title": "Test Video",
"ext": "mkv",
}
result = parse_outtmpl(template, info_dict)
assert result == "Test Channel/Test Video.mkv"
def test_parse_outtmpl_with_missing_field(self):
"""Test template parsing with missing field defaults to NA."""
template = "%(title)s - %(upload_date)s.%(ext)s"
info_dict = {
"title": "Test Video",
"ext": "mp4",
}
result = parse_outtmpl(template, info_dict)
assert result == "Test Video - NA.mp4", "Missing field upload_date should default to NA"
def test_parse_outtmpl_complex(self):
"""Test complex template with multiple fields."""
template = "%(uploader)s/%(playlist_title)s/%(playlist_index)03d - %(title)s [%(id)s].%(ext)s"
info_dict = {
"uploader": "Test Channel",
"playlist_title": "Best Videos",
"playlist_index": 5,
"title": "Amazing Content",
"id": "abc123xyz",
"ext": "mp4",
}
result = parse_outtmpl(template, info_dict)
assert result == "Test Channel/Best Videos/005 - Amazing Content [abc123xyz].mp4"
def test_parse_outtmpl_with_special_characters(self):
"""Test template parsing handles special characters in values."""
template = "%(title)s.%(ext)s"
info_dict = {
"title": "Test: Video / With \\ Special | Characters",
"ext": "mp4",
}
result = parse_outtmpl(template, info_dict)
assert ".mp4" in result, "yt-dlp should sanitize special characters but preserve extension"
assert "Test" in result, "yt-dlp should preserve safe parts of title"
def test_parse_outtmpl_with_playlist_info(self):
"""Test template parsing with playlist information."""
template = "%(playlist)s/%(title)s.%(ext)s"
info_dict = {
"playlist": "My Playlist",
"title": "Video Title",
"ext": "webm",
}
result = parse_outtmpl(template, info_dict)
assert result == "My Playlist/Video Title.webm"
def test_parse_outtmpl_with_restrict_filename(self):
"""Test template parsing with restrict_filename parameter."""
template = "%(uploader)s/%(title)s.%(ext)s"
info_dict = {
"uploader": "Foobar's Workshop",
"title": "Test Video",
"ext": "mp4",
}
result_unrestricted: str = parse_outtmpl(template, info_dict)
assert result_unrestricted == "Foobar's Workshop/Test Video.mp4"
result_restricted: str = parse_outtmpl(template, info_dict, params={"restrictfilenames": True})
assert result_restricted == "Foobar_s_Workshop/Test_Video.mp4"