diff --git a/app/features/conditions/router.py b/app/features/conditions/router.py index 1a01588e..a4fcd049 100644 --- a/app/features/conditions/router.py +++ b/app/features/conditions/router.py @@ -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: diff --git a/app/features/conditions/schemas.py b/app/features/conditions/schemas.py index b2abdfff..9ec433f3 100644 --- a/app/features/conditions/schemas.py +++ b/app/features/conditions/schemas.py @@ -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 diff --git a/app/features/conditions/service.py b/app/features/conditions/service.py index 1ede3af8..d288e8bd 100644 --- a/app/features/conditions/service.py +++ b/app/features/conditions/service.py @@ -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") diff --git a/app/features/presets/migration.py b/app/features/presets/migration.py index f44c78e3..eb000334 100644 --- a/app/features/presets/migration.py +++ b/app/features/presets/migration.py @@ -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) diff --git a/app/features/presets/schemas.py b/app/features/presets/schemas.py index fd80f639..34ff8dc9 100644 --- a/app/features/presets/schemas.py +++ b/app/features/presets/schemas.py @@ -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}" diff --git a/app/features/tasks/definitions/handlers/generic.py b/app/features/tasks/definitions/handlers/generic.py index 9d94c41a..4efc71d4 100644 --- a/app/features/tasks/definitions/handlers/generic.py +++ b/app/features/tasks/definitions/handlers/generic.py @@ -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 diff --git a/app/features/tasks/definitions/handlers/rss.py b/app/features/tasks/definitions/handlers/rss.py index f5fc6431..084105fc 100644 --- a/app/features/tasks/definitions/handlers/rss.py +++ b/app/features/tasks/definitions/handlers/rss.py @@ -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 diff --git a/app/features/tasks/definitions/handlers/tver.py b/app/features/tasks/definitions/handlers/tver.py index 29aa3695..d2e5e19d 100644 --- a/app/features/tasks/definitions/handlers/tver.py +++ b/app/features/tasks/definitions/handlers/tver.py @@ -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 diff --git a/app/features/tasks/definitions/handlers/twitch.py b/app/features/tasks/definitions/handlers/twitch.py index 2da417a0..2cb9c43d 100644 --- a/app/features/tasks/definitions/handlers/twitch.py +++ b/app/features/tasks/definitions/handlers/twitch.py @@ -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 diff --git a/app/features/tasks/definitions/handlers/youtube.py b/app/features/tasks/definitions/handlers/youtube.py index a963e14a..1fb9d576 100644 --- a/app/features/tasks/definitions/handlers/youtube.py +++ b/app/features/tasks/definitions/handlers/youtube.py @@ -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 diff --git a/app/features/tasks/definitions/results.py b/app/features/tasks/definitions/results.py index 31d4fc3b..0377cd6d 100644 --- a/app/features/tasks/definitions/results.py +++ b/app/features/tasks/definitions/results.py @@ -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.") diff --git a/app/features/tasks/router.py b/app/features/tasks/router.py index d7ffc43b..e4043a1d 100644 --- a/app/features/tasks/router.py +++ b/app/features/tasks/router.py @@ -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 diff --git a/app/features/tasks/schemas.py b/app/features/tasks/schemas.py index a1b1fb0d..b3d52366 100644 --- a/app/features/tasks/schemas.py +++ b/app/features/tasks/schemas.py @@ -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}" diff --git a/app/features/ytdlp/__init__.py b/app/features/ytdlp/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/app/library/downloads/extractor.py b/app/features/ytdlp/extractor.py similarity index 90% rename from app/library/downloads/extractor.py rename to app/features/ytdlp/extractor.py index 88cdf093..74295500 100644 --- a/app/library/downloads/extractor.py +++ b/app/features/ytdlp/extractor.py @@ -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() diff --git a/app/library/mini_filter.py b/app/features/ytdlp/mini_filter.py similarity index 100% rename from app/library/mini_filter.py rename to app/features/ytdlp/mini_filter.py diff --git a/app/features/ytdlp/tests/test_ytdlp_extractor.py b/app/features/ytdlp/tests/test_ytdlp_extractor.py new file mode 100644 index 00000000..16fb77de --- /dev/null +++ b/app/features/ytdlp/tests/test_ytdlp_extractor.py @@ -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" diff --git a/app/tests/mini_filter_test.py b/app/features/ytdlp/tests/test_ytdlp_mini_filter.py similarity index 99% rename from app/tests/mini_filter_test.py rename to app/features/ytdlp/tests/test_ytdlp_mini_filter.py index 7641d349..00859c7d 100644 --- a/app/tests/mini_filter_test.py +++ b/app/features/ytdlp/tests/test_ytdlp_mini_filter.py @@ -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): diff --git a/app/tests/test_ytdlp_module.py b/app/features/ytdlp/tests/test_ytdlp_module.py similarity index 92% rename from app/tests/test_ytdlp_module.py rename to app/features/ytdlp/tests/test_ytdlp_module.py index 880a929c..07a1093b 100644 --- a/app/tests/test_ytdlp_module.py +++ b/app/features/ytdlp/tests/test_ytdlp_module.py @@ -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 diff --git a/app/tests/test_ytdlpopts.py b/app/features/ytdlp/tests/test_ytdlp_opts.py similarity index 87% rename from app/tests/test_ytdlpopts.py rename to app/features/ytdlp/tests/test_ytdlp_opts.py index 107a078f..bb84391b 100644 --- a/app/tests/test_ytdlpopts.py +++ b/app/features/ytdlp/tests/test_ytdlp_opts.py @@ -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" diff --git a/app/features/ytdlp/tests/test_ytdlp_utils.py b/app/features/ytdlp/tests/test_ytdlp_utils.py new file mode 100644 index 00000000..6e9d8bd1 --- /dev/null +++ b/app/features/ytdlp/tests/test_ytdlp_utils.py @@ -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) diff --git a/app/features/ytdlp/utils.py b/app/features/ytdlp/utils.py new file mode 100644 index 00000000..ff995934 --- /dev/null +++ b/app/features/ytdlp/utils.py @@ -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 diff --git a/app/library/ytdlp.py b/app/features/ytdlp/ytdlp.py similarity index 96% rename from app/library/ytdlp.py rename to app/features/ytdlp/ytdlp.py index 3b56f7b3..c0dba61b 100644 --- a/app/library/ytdlp.py +++ b/app/features/ytdlp/ytdlp.py @@ -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]]: diff --git a/app/library/YTDLPOpts.py b/app/features/ytdlp/ytdlp_opts.py similarity index 96% rename from app/library/YTDLPOpts.py rename to app/features/ytdlp/ytdlp_opts.py index d846e6a9..d30028a0 100644 --- a/app/library/YTDLPOpts.py +++ b/app/features/ytdlp/ytdlp_opts.py @@ -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: diff --git a/app/library/ItemDTO.py b/app/library/ItemDTO.py index d1439bbf..279cf7df 100644 --- a/app/library/ItemDTO.py +++ b/app/library/ItemDTO.py @@ -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: diff --git a/app/library/LogWrapper.py b/app/library/LogWrapper.py deleted file mode 100644 index 149242a9..00000000 --- a/app/library/LogWrapper.py +++ /dev/null @@ -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) diff --git a/app/library/Utils.py b/app/library/Utils.py index 66e178fc..6b75d416 100644 --- a/app/library/Utils.py +++ b/app/library/Utils.py @@ -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) diff --git a/app/library/downloads/core.py b/app/library/downloads/core.py index 4b512e3e..ef14b6a7 100644 --- a/app/library/downloads/core.py +++ b/app/library/downloads/core.py @@ -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 diff --git a/app/library/downloads/item_adder.py b/app/library/downloads/item_adder.py index 0db43f47..147a3c0a 100644 --- a/app/library/downloads/item_adder.py +++ b/app/library/downloads/item_adder.py @@ -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 diff --git a/app/library/downloads/playlist_processor.py b/app/library/downloads/playlist_processor.py index 0f3b638f..62166d10 100644 --- a/app/library/downloads/playlist_processor.py +++ b/app/library/downloads/playlist_processor.py @@ -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 diff --git a/app/library/downloads/queue_manager.py b/app/library/downloads/queue_manager.py index 1230d208..d68cc458 100644 --- a/app/library/downloads/queue_manager.py +++ b/app/library/downloads/queue_manager.py @@ -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]: """ diff --git a/app/library/downloads/video_processor.py b/app/library/downloads/video_processor.py index 93c16844..f0329065 100644 --- a/app/library/downloads/video_processor.py +++ b/app/library/downloads/video_processor.py @@ -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 diff --git a/app/routes/api/archiver.py b/app/routes/api/archiver.py index 4183deda..b8beb374 100644 --- a/app/routes/api/archiver.py +++ b/app/routes/api/archiver.py @@ -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__) diff --git a/app/routes/api/docs.py b/app/routes/api/docs.py index cbba2010..96e9ae69 100644 --- a/app/routes/api/docs.py +++ b/app/routes/api/docs.py @@ -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__) diff --git a/app/routes/api/images.py b/app/routes/api/images.py index 9d1b80bb..66dce6ae 100644 --- a/app/routes/api/images.py +++ b/app/routes/api/images.py @@ -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__) diff --git a/app/routes/api/yt_dlp.py b/app/routes/api/yt_dlp.py index f5fdd7d6..8acfe838 100644 --- a/app/routes/api/yt_dlp.py +++ b/app/routes/api/yt_dlp.py @@ -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( diff --git a/app/tests/test_itemdto.py b/app/tests/test_itemdto.py index 305402bf..72a27a9a 100644 --- a/app/tests/test_itemdto.py +++ b/app/tests/test_itemdto.py @@ -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"): diff --git a/app/tests/test_logwrapper.py b/app/tests/test_logwrapper.py deleted file mode 100644 index 9d8658f3..00000000 --- a/app/tests/test_logwrapper.py +++ /dev/null @@ -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) diff --git a/app/tests/test_utils.py b/app/tests/test_utils.py index f4ab0150..b8f7fc1f 100644 --- a/app/tests/test_utils.py +++ b/app/tests/test_utils.py @@ -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"