diff --git a/app/features/tasks/definitions/results.py b/app/features/tasks/definitions/results.py index 0377cd6d..66037d2d 100644 --- a/app/features/tasks/definitions/results.py +++ b/app/features/tasks/definitions/results.py @@ -42,7 +42,7 @@ class HandleTask(TaskSchema): tuple[bool, str]: A tuple indicating success and a message. """ - from app.library.Utils import archive_add + from app.features.ytdlp.utils import archive_add ret: tuple[bool, str] | dict[str, Any] = await self._mark_logic() if isinstance(ret, tuple): @@ -64,7 +64,7 @@ class HandleTask(TaskSchema): tuple[bool, str]: A tuple indicating success and a message. """ - from app.library.Utils import archive_delete + from app.features.ytdlp.utils import archive_delete ret: tuple[bool, str] | dict[str, Any] = await self._mark_logic() if isinstance(ret, tuple): diff --git a/app/features/tasks/definitions/service.py b/app/features/tasks/definitions/service.py index 36968c2d..4fa245fb 100644 --- a/app/features/tasks/definitions/service.py +++ b/app/features/tasks/definitions/service.py @@ -11,11 +11,11 @@ from typing import TYPE_CHECKING, Any from app.features.tasks.definitions.results import HandleTask, TaskFailure, TaskItem, TaskResult from app.features.tasks.models import TaskModel +from app.features.ytdlp.utils import archive_read from app.library.downloads.queue_manager import DownloadQueue from app.library.Events import EventBus, Events from app.library.ItemDTO import Item, ItemDTO from app.library.Services import Services -from app.library.Utils import archive_read if TYPE_CHECKING: from app.features.tasks.repository import TasksRepository diff --git a/app/library/Archiver.py b/app/features/ytdlp/archiver.py similarity index 100% rename from app/library/Archiver.py rename to app/features/ytdlp/archiver.py diff --git a/app/features/ytdlp/router.py b/app/features/ytdlp/router.py new file mode 100644 index 00000000..9d09f3c8 --- /dev/null +++ b/app/features/ytdlp/router.py @@ -0,0 +1,507 @@ +import json +import logging +import time +from collections import OrderedDict +from collections.abc import Iterable +from typing import Any + +from aiohttp import web +from aiohttp.web import Request, Response + +from app.features.presets.service import Presets +from app.features.ytdlp.archiver import Archiver +from app.features.ytdlp.extractor import fetch_info +from app.features.ytdlp.utils import archive_read, 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.encoder import Encoder +from app.library.ItemDTO import Item +from app.library.router import route +from app.library.Utils import validate_url + +LOG: logging.Logger = logging.getLogger(__name__) + + +def _get_preset_archive(preset: str) -> str | None: + """ + Resolve the archive file path for a given preset. + + Validates that the preset exists and that applying the preset results + in yt-dlp options that contain a 'download_archive' path. + """ + if not preset or not Presets.get_instance().has(preset): + return None + + try: + opts: dict = YTDLPOpts.get_instance().preset(preset).get_all() + except Exception as e: + LOG.error(f"Failed to build yt-dlp opts for preset '{preset}'. {e!s}") + return None + + if not (archive_file := opts.get("download_archive")): + return None + + if not isinstance(archive_file, str) or len(archive_file.strip()) < 1: + return None + + return archive_file.strip() + + +def _normalize_ids(items: Iterable[str] | None) -> tuple[list[str], list[str]]: + """ + Validate and normalize archive IDs. + + - Trims whitespace + - Enforces that each ID has at least two whitespace-separated tokens + (e.g., "youtube ABC123") as required by yt-dlp's archive format + - De-duplicates while preserving order + + Returns a tuple: (valid_ids, invalid_inputs) + """ + if not items: + return ([], []) + + seen: set[str] = set() + valid: list[str] = [] + invalid: list[str] = [] + + for raw in items: + if raw is None: + continue + + s = str(raw).strip() + if not s: + continue + + if len(s.split()) < 2: + invalid.append(s) + continue + + if s in seen: + continue + + seen.add(s) + valid.append(s) + + return (valid, invalid) + + +@route("GET", "api/archiver/", "archiver") +async def archiver_get(request: Request) -> Response: + """ + Read IDs from the download archive for a given preset. + + Query params: + - preset: required preset name/id + - ids: optional comma-separated list to filter; when omitted, returns all + """ + preset: str | None = request.query.get("preset") + if not preset: + return web.json_response(data={"error": "preset is required."}, status=web.HTTPBadRequest.status_code) + + archive_file: str | None = _get_preset_archive(preset) + if not archive_file: + return web.json_response( + data={"error": f"Preset '{preset}' does not provide a download_archive."}, + status=web.HTTPBadRequest.status_code, + ) + + ids_param: str | None = request.query.get("ids") + ids: list[str] = [] + if ids_param: + ids_list = [s.strip() for s in ids_param.split(",") if s and s.strip()] + ids, invalid = _normalize_ids(ids_list) + if invalid: + return web.json_response( + data={"error": "invalid ids provided.", "invalid_items": invalid}, + status=web.HTTPBadRequest.status_code, + ) + + try: + data: list[str] = Archiver.get_instance().read(archive_file, ids or None) + return web.json_response( + data={"file": archive_file, "items": data, "count": len(data)}, status=web.HTTPOk.status_code + ) + except Exception as e: + LOG.exception(e) + return web.json_response( + data={"error": f"Failed to read archive file for preset '{preset}'.", "message": str(e)}, + status=web.HTTPInternalServerError.status_code, + ) + + +@route("POST", "api/archiver/", "archiver_add") +async def archiver_add(request: Request) -> Response: + """ + Append IDs to the download archive for a given preset. + + Body: { "preset": string, "items": [string, ...], "skip_check": bool? } + """ + post = await request.json() + preset: str | None = post.get("preset") if isinstance(post, dict) else None + if not preset: + return web.json_response(data={"error": "preset is required."}, status=web.HTTPBadRequest.status_code) + + archive_file: str | None = _get_preset_archive(preset) + if not archive_file: + return web.json_response( + data={"error": f"Preset '{preset}' does not provide a download_archive."}, + status=web.HTTPBadRequest.status_code, + ) + + items, invalid = _normalize_ids((post or {}).get("items", [])) if isinstance(post, dict) else ([], []) + if invalid: + return web.json_response( + data={"error": "invalid ids provided.", "invalid_items": invalid}, + status=web.HTTPBadRequest.status_code, + ) + if len(items) < 1: + return web.json_response( + data={"error": "items is required and must be a non-empty list."}, status=web.HTTPBadRequest.status_code + ) + + skip_check: bool = bool(post.get("skip_check", False)) if isinstance(post, dict) else False + + try: + status: bool = Archiver.get_instance().add(archive_file, items, skip_check=skip_check) + return web.json_response( + data={"file": archive_file, "status": status, "items": items}, + status=web.HTTPOk.status_code if status else web.HTTPNotModified.status_code, + ) + except Exception as e: + LOG.exception(e) + return web.json_response( + data={"error": f"Failed to add items to archive for preset '{preset}'.", "message": str(e)}, + status=web.HTTPInternalServerError.status_code, + ) + + +@route("DELETE", "api/archiver/", "archiver_delete") +async def archiver_delete(request: Request) -> Response: + """ + Remove IDs from the download archive for a given preset. + + Body: { "preset": string, "items": [string, ...] } + """ + post = await request.json() + preset: str | None = post.get("preset") if isinstance(post, dict) else None + if not preset: + return web.json_response(data={"error": "preset is required."}, status=web.HTTPBadRequest.status_code) + + archive_file: str | None = _get_preset_archive(preset) + if not archive_file: + return web.json_response( + data={"error": f"Preset '{preset}' does not provide a download_archive."}, + status=web.HTTPBadRequest.status_code, + ) + + items, invalid = _normalize_ids((post or {}).get("items", [])) if isinstance(post, dict) else ([], []) + if invalid: + return web.json_response( + data={"error": "invalid ids provided.", "invalid_items": invalid}, + status=web.HTTPBadRequest.status_code, + ) + if len(items) < 1: + return web.json_response( + data={"error": "items is required and must be a non-empty list."}, status=web.HTTPBadRequest.status_code + ) + + try: + status: bool = Archiver.get_instance().delete(archive_file, items) + return web.json_response( + data={"file": archive_file, "status": status, "items": items}, + status=web.HTTPOk.status_code if status else web.HTTPNotModified.status_code, + ) + except Exception as e: + LOG.exception(e) + return web.json_response( + data={"error": f"Failed to delete items from archive for preset '{preset}'.", "message": str(e)}, + status=web.HTTPInternalServerError.status_code, + ) + + +@route("POST", "api/yt-dlp/convert/", "convert") +async def convert(request: Request) -> Response: + """ + Convert the yt-dlp args to a dict. + + Args: + request (Request): The request object. + + Returns: + Response: The response object. + + """ + post = await request.json() + args: str | None = post.get("args") + + if not args: + return web.json_response(data={"error": "args param is required."}, status=web.HTTPBadRequest.status_code) + + try: + response = {"opts": {}, "output_template": None, "download_path": None} + + data = arg_converter(args, dumps=True) + + if "outtmpl" in data and "default" in data["outtmpl"]: + response["output_template"] = data["outtmpl"]["default"] + + if "paths" in data and "home" in data["paths"]: + response["download_path"] = data["paths"]["home"] + + if "format" in data: + response["format"] = data["format"] + + 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: + if key in bad_options.items(): + removed_options.append(bad_options[key]) + continue + if not key.startswith("_"): + response["opts"][key] = data[key] + + if len(removed_options) > 0: + response["removed_options"] = removed_options + + return web.json_response(data=response, status=web.HTTPOk.status_code) + except Exception as e: + err = str(e).strip() + err = err.split("\n")[-1] if "\n" in err else err + err = err.replace("main.py: error: ", "").strip().capitalize() + return web.json_response( + data={"error": f"Failed to parse command options for yt-dlp. '{err}'."}, + status=web.HTTPBadRequest.status_code, + ) + + +@route("GET", "api/yt-dlp/url/info/", "get_info") +async def get_info(request: Request, cache: Cache, config: Config) -> Response: + """ + Get the video info. + + Args: + request (Request): The request object. + cache (Cache): The cache instance. + config (Config): The config instance. + + Returns: + Response: The response object + + """ + url: str | None = request.query.get("url") + if not url: + return web.json_response( + data={"status": False, "message": "URL is required.", "error": "URL is required."}, + status=web.HTTPBadRequest.status_code, + ) + + try: + validate_url(url, allow_internal=config.allow_internal_urls) + except ValueError as e: + return web.json_response( + data={"status": False, "message": str(e), "error": str(e)}, + status=web.HTTPBadRequest.status_code, + ) + + opts: YTDLPOpts = YTDLPOpts.get_instance() + + preset: str = request.query.get("preset", config.default_preset) + if not Presets.get_instance().get(preset): + msg: str = f"Preset '{preset}' does not exist." + return web.json_response( + data={"status": False, "message": msg, "error": msg}, + status=web.HTTPBadRequest.status_code, + ) + + opts = opts.preset(preset) + + if cli_args := request.query.get("args", None): + try: + arg_converter(cli_args, dumps=True) + opts = opts.add_cli(cli_args, from_user=True) + except Exception as e: + err = str(e).strip() + err = err.split("\n")[-1] if "\n" in err else err + err = err.replace("main.py: error: ", "").strip().capitalize() + return web.json_response( + data={"error": f"Failed to parse command options for yt-dlp. '{err}'."}, + status=web.HTTPBadRequest.status_code, + ) + + try: + key: str = cache.hash(f"{preset}:{url}:{cli_args or ''}") + + if cache.has(key) and not request.query.get("force", False): + data: Any | None = cache.get(key) + data["_cached"] = { + "status": "hit", + "preset": preset, + "cli_args": cli_args, + "key": key, + "ttl": data.get("_cached", {}).get("ttl", 300), + "ttl_left": data.get("_cached", {}).get("expires", time.time() + 300) - time.time(), + "expires": data.get("_cached", {}).get("expires", time.time() + 300), + } + return web.json_response(body=json.dumps(data, indent=4, default=str), status=web.HTTPOk.status_code) + + ytdlp_opts: dict = opts.get_all() + + (data, logs) = await fetch_info( + config=ytdlp_opts, + url=url, + debug=False, + no_archive=True, + follow_redirect=True, + sanitize_info=True, + capture_logs=logging.WARNING, + ) + + if not data or not isinstance(data, dict): + return web.json_response( + data={ + "status": False, + "error": f"Failed to extract video info. {'. '.join(logs)}", + "message": "Failed to extract video info.", + }, + status=web.HTTPInternalServerError.status_code, + ) + + if data and "formats" in data: + from yt_dlp.cookies import LenientSimpleCookie + + for index, item in enumerate(data["formats"]): + if "cookies" in item and len(item["cookies"]) > 0: + cookies: list[str] = [f"{c.key}={c.value}" for c in LenientSimpleCookie(item["cookies"]).values()] + if len(cookies) > 0: + data["formats"][index]["h_cookies"] = "; ".join(cookies) + data["formats"][index]["h_cookies"] = data["formats"][index]["h_cookies"].strip() + + data["_cached"] = { + "status": "miss", + "preset": preset, + "cli_args": cli_args, + "key": key, + "ttl": 300, + "ttl_left": 300, + "expires": time.time() + 300, + } + + data["is_archived"] = False + + archive_file: str | None = ytdlp_opts.get("download_archive") + data["archive_file"] = archive_file if archive_file else None + + if archive_file and (archive_id := get_archive_id(url=url).get("archive_id")): + data["archive_id"] = archive_id + data["is_archived"] = len(archive_read(archive_file, [archive_id])) > 0 + + data = OrderedDict(sorted(data.items(), key=lambda item: len(str(item[1])))) + + cache.set(key=key, value=data, ttl=300) + + return web.json_response(body=json.dumps(data, indent=4, default=str), status=web.HTTPOk.status_code) + except Exception as e: + LOG.exception(e) + LOG.error(f"Error encountered while getting video info for '{url}'. '{e!s}'.") + return web.json_response( + data={ + "error": "failed to get video info.", + "message": str(e), + "formats": [], + }, + status=web.HTTPInternalServerError.status_code, + ) + + +@route("GET", "api/yt-dlp/options/", "get_options") +async def get_options() -> Response: + """ + Get the yt-dlp CLI options. + + Returns: + Response: The response object with the yt-dlp CLI 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) + + +@route("POST", "api/yt-dlp/archive_id/", "get_archive_ids") +async def get_archive_ids(request: Request, config: Config) -> Response: + """ + Get the archive IDs for the given URLs. + + Returns: + Response: The response object with the yt-dlp CLI options. + + """ + data = (await request.json()) if request.body_exists else None + if not data or not isinstance(data, list): + return web.json_response( + data={"error": "Invalid request. expecting list with URLs."}, + status=web.HTTPBadRequest.status_code, + ) + + response = [] + + for i, url in enumerate(data): + dct = {"index": i, "url": url} + try: + validate_url(url, allow_internal=config.allow_internal_urls) + dct.update(get_archive_id(url)) + except ValueError as e: + dct.update({"id": None, "ie_key": None, "archive_id": None, "error": str(e)}) + + response.append(dct) + + return web.json_response(data=response, status=web.HTTPOk.status_code) + + +@route("POST", "api/yt-dlp/command/", "make_command") +async def make_command(request: Request, config: Config, encoder: Encoder) -> Response: + """ + Build yt-dlp CLI command. + + Args: + request (Request): The request object. + config (Config): The config instance. + encoder (Encoder): The encoder instance. + + Returns: + Response: The response object with the merged fields and final yt-dlp CLI command string. + + """ + if not config.console_enabled: + return web.json_response(data={"error": "Console is disabled."}, status=web.HTTPForbidden.status_code) + + data = (await request.json()) if request.body_exists else None + if not data or not isinstance(data, dict): + return web.json_response( + data={"error": "Invalid request. expecting JSON body."}, + status=web.HTTPBadRequest.status_code, + ) + + try: + it = Item.format(data) + except ValueError as e: + return web.json_response(data={"error": str(e), "data": data}, status=web.HTTPBadRequest.status_code) + + try: + command, info = YTDLPCli(item=it, config=config).build() + except Exception as e: + LOG.exception(e) + return web.json_response( + data={"error": "Failed to build CLI command"}, + status=web.HTTPBadRequest.status_code, + ) + + if request.query.get("full", False): + return web.json_response(data=info, status=web.HTTPOk.status_code, dumps=encoder.encode) + + return web.json_response(data={"command": command}, status=web.HTTPOk.status_code) diff --git a/app/features/ytdlp/tests/test_ytdlp_module.py b/app/features/ytdlp/tests/test_ytdlp_module.py index 07a1093b..1adf9112 100644 --- a/app/features/ytdlp/tests/test_ytdlp_module.py +++ b/app/features/ytdlp/tests/test_ytdlp_module.py @@ -18,7 +18,7 @@ class TestArchiveProxy: assert ("" in p2) is False assert p2.add("") is False - @patch("app.library.Archiver.Archiver.get_instance") + @patch("app.features.ytdlp.archiver.Archiver.get_instance") def test_contains_and_add_delegate_to_archiver(self, mock_get_instance) -> None: arch = MagicMock() mock_get_instance.return_value = arch diff --git a/app/features/ytdlp/tests/test_ytdlp_utils.py b/app/features/ytdlp/tests/test_ytdlp_utils.py index 6e9d8bd1..476f1b2b 100644 --- a/app/features/ytdlp/tests/test_ytdlp_utils.py +++ b/app/features/ytdlp/tests/test_ytdlp_utils.py @@ -1,5 +1,7 @@ import logging +from pathlib import Path import re +import tempfile from typing import Any from unittest.mock import MagicMock, patch @@ -14,6 +16,9 @@ from app.features.ytdlp.utils import ( get_extras, get_thumbnail, get_ytdlp, + archive_delete, + archive_add, + archive_read, ) @@ -537,3 +542,43 @@ class TestGetArchiveId: """Test with invalid URL.""" result = get_archive_id("invalid-url") assert isinstance(result, dict) + + +class TestArchiveFunctions: + """Test archive-related functions.""" + + def setup_method(self): + """Set up test archive file.""" + self.temp_dir = tempfile.mkdtemp() + self.archive_file = Path(self.temp_dir) / "archive.txt" + + def teardown_method(self): + """Clean up after tests.""" + import shutil + + shutil.rmtree(self.temp_dir, ignore_errors=True) + + def test_archive_add_and_read(self): + """Test adding and reading archive entries.""" + ids = ["id1", "id2", "id3"] + + # Add entries - just test it returns a boolean + result = archive_add(self.archive_file, ids) + assert isinstance(result, bool) + + # Read entries - just test it returns a list + read_ids = archive_read(self.archive_file) + assert isinstance(read_ids, list) + + def test_archive_delete(self): + """Test deleting archive entries.""" + # Delete some entries - just test it returns a boolean + delete_ids = ["id2"] + result = archive_delete(self.archive_file, delete_ids) + assert isinstance(result, bool) + + def test_archive_read_nonexistent(self): + """Test reading from non-existent archive.""" + nonexistent = Path(self.temp_dir) / "nonexistent.txt" + result = archive_read(nonexistent) + assert result == [] diff --git a/app/features/ytdlp/utils.py b/app/features/ytdlp/utils.py index ff995934..745f98fc 100644 --- a/app/features/ytdlp/utils.py +++ b/app/features/ytdlp/utils.py @@ -6,6 +6,7 @@ import shlex from collections.abc import Callable from dataclasses import dataclass from email.utils import formatdate +from pathlib import Path from typing import Any from app.features.ytdlp.ytdlp import YTDLP @@ -506,3 +507,55 @@ def get_archive_id(url: str) -> dict[str, str | None]: LOG.error(f"Error getting archive ID: {e}") return idDict + + +def archive_add(file: str | Path, ids: list[str], skip_check: bool = False) -> bool: + """ + Add IDs to an archive file (delegates to the global Archiver). + + Args: + file (str|Path): The archive file path. + ids (list[str]): List of IDs to add. + skip_check (bool): If True, skip checking for existing IDs. + + Returns: + bool: True if any new IDs were appended, False otherwise. + + """ + from app.features.ytdlp.archiver import Archiver + + return Archiver.get_instance().add(file, ids, skip_check) + + +def archive_read(file: str | Path, ids: list[str] | None = None) -> list[str]: + """ + Read IDs from an archive file with optional filtering (delegates to Archiver). + + Args: + file (str|Path): The archive file path. + ids (list[str]|None): Optional list of IDs to query; None/empty returns all. + + Returns: + list[str]: IDs present in the archive, optionally filtered. + + """ + from app.features.ytdlp.archiver import Archiver + + return Archiver.get_instance().read(file, ids) + + +def archive_delete(file: str | Path, ids: list[str]) -> bool: + """ + Delete IDs from an archive file (delegates to Archiver). + + Args: + file (str|Path): The archive file path. + ids (list[str]): List of IDs to remove. + + Returns: + bool: True on success (including no-op), False on error. + + """ + from app.features.ytdlp.archiver import Archiver + + return Archiver.get_instance().delete(file, ids) diff --git a/app/features/ytdlp/ytdlp.py b/app/features/ytdlp/ytdlp.py index c0dba61b..720c0054 100644 --- a/app/features/ytdlp/ytdlp.py +++ b/app/features/ytdlp/ytdlp.py @@ -24,7 +24,7 @@ class _ArchiveProxy: return False try: - from app.library.Archiver import Archiver + from app.features.ytdlp.archiver import Archiver status: bool = item in Archiver.get_instance().read(self._file, [item]) return status @@ -36,7 +36,7 @@ class _ArchiveProxy: return False try: - from app.library.Archiver import Archiver + from app.features.ytdlp.archiver import Archiver status: bool = Archiver.get_instance().add(self._file, [item]) return status diff --git a/app/library/ItemDTO.py b/app/library/ItemDTO.py index 279cf7df..d5b2cfa4 100644 --- a/app/library/ItemDTO.py +++ b/app/library/ItemDTO.py @@ -7,17 +7,10 @@ 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.utils import archive_add, archive_delete, archive_read, 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_file, - get_file_sidecar, -) +from app.library.Utils import clean_item, get_file, get_file_sidecar if TYPE_CHECKING: from app.features.presets.schemas import Preset diff --git a/app/library/Utils.py b/app/library/Utils.py index 6b75d416..b63b997b 100644 --- a/app/library/Utils.py +++ b/app/library/Utils.py @@ -1293,58 +1293,6 @@ def list_folders(path: Path, base: Path, depth_limit: int) -> list[str]: return folders -def archive_add(file: str | Path, ids: list[str], skip_check: bool = False) -> bool: - """ - Add IDs to an archive file (delegates to the global Archiver). - - Args: - file (str|Path): The archive file path. - ids (list[str]): List of IDs to add. - skip_check (bool): If True, skip checking for existing IDs. - - Returns: - bool: True if any new IDs were appended, False otherwise. - - """ - from app.library.Archiver import Archiver - - return Archiver.get_instance().add(file, ids, skip_check) - - -def archive_read(file: str | Path, ids: list[str] | None = None) -> list[str]: - """ - Read IDs from an archive file with optional filtering (delegates to Archiver). - - Args: - file (str|Path): The archive file path. - ids (list[str]|None): Optional list of IDs to query; None/empty returns all. - - Returns: - list[str]: IDs present in the archive, optionally filtered. - - """ - from app.library.Archiver import Archiver - - return Archiver.get_instance().read(file, ids) - - -def archive_delete(file: str | Path, ids: list[str]) -> bool: - """ - Delete IDs from an archive file (delegates to Archiver). - - Args: - file (str|Path): The archive file path. - ids (list[str]): List of IDs to remove. - - Returns: - bool: True on success (including no-op), False on error. - - """ - from app.library.Archiver import Archiver - - return Archiver.get_instance().delete(file, ids) - - def get_channel_images(thumbnails: list[dict]) -> dict: """ Extract channel images from a list of thumbnail dictionaries. diff --git a/app/library/downloads/item_adder.py b/app/library/downloads/item_adder.py index 147a3c0a..8cccd30e 100644 --- a/app/library/downloads/item_adder.py +++ b/app/library/downloads/item_adder.py @@ -9,17 +9,12 @@ 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.features.ytdlp.extractor import fetch_info +from app.features.ytdlp.utils import archive_add, archive_read, 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, - create_cookies_file, - merge_dict, -) +from app.library.Utils import create_cookies_file, merge_dict -from ...features.ytdlp.extractor import fetch_info from .core import Download from .playlist_processor import process_playlist from .video_processor import add_video diff --git a/app/routes/api/archiver.py b/app/routes/api/archiver.py deleted file mode 100644 index b8beb374..00000000 --- a/app/routes/api/archiver.py +++ /dev/null @@ -1,210 +0,0 @@ -import logging -from collections.abc import Iterable - -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 - -LOG: logging.Logger = logging.getLogger(__name__) - - -def _get_archive_file_from_preset(preset: str) -> str | None: - """ - Resolve the archive file path for a given preset. - - Validates that the preset exists and that applying the preset results - in yt-dlp options that contain a 'download_archive' path. - """ - if not preset or not Presets.get_instance().has(preset): - return None - - try: - opts: dict = YTDLPOpts.get_instance().preset(preset).get_all() - except Exception as e: - LOG.error(f"Failed to build yt-dlp opts for preset '{preset}'. {e!s}") - return None - - if not (archive_file := opts.get("download_archive")): - return None - - if not isinstance(archive_file, str) or len(archive_file.strip()) < 1: - return None - - return archive_file.strip() - - -def _normalize_ids(items: Iterable[str] | None) -> tuple[list[str], list[str]]: - """ - Validate and normalize archive IDs. - - - Trims whitespace - - Enforces that each ID has at least two whitespace-separated tokens - (e.g., "youtube ABC123") as required by yt-dlp's archive format - - De-duplicates while preserving order - - Returns a tuple: (valid_ids, invalid_inputs) - """ - if not items: - return ([], []) - - seen: set[str] = set() - valid: list[str] = [] - invalid: list[str] = [] - - for raw in items: - if raw is None: - continue - - s = str(raw).strip() - if not s: - continue - - if len(s.split()) < 2: - invalid.append(s) - continue - - if s in seen: - continue - - seen.add(s) - valid.append(s) - - return (valid, invalid) - - -@route("GET", "api/archiver/", "archiver") -async def archiver_get(request: Request) -> Response: - """ - Read IDs from the download archive for a given preset. - - Query params: - - preset: required preset name/id - - ids: optional comma-separated list to filter; when omitted, returns all - """ - preset: str | None = request.query.get("preset") - if not preset: - return web.json_response(data={"error": "preset is required."}, status=web.HTTPBadRequest.status_code) - - archive_file: str | None = _get_archive_file_from_preset(preset) - if not archive_file: - return web.json_response( - data={"error": f"Preset '{preset}' does not provide a download_archive."}, - status=web.HTTPBadRequest.status_code, - ) - - ids_param: str | None = request.query.get("ids") - ids: list[str] = [] - if ids_param: - ids_list = [s.strip() for s in ids_param.split(",") if s and s.strip()] - ids, invalid = _normalize_ids(ids_list) - if invalid: - return web.json_response( - data={"error": "invalid ids provided.", "invalid_items": invalid}, - status=web.HTTPBadRequest.status_code, - ) - - try: - data: list[str] = Archiver.get_instance().read(archive_file, ids or None) - return web.json_response( - data={"file": archive_file, "items": data, "count": len(data)}, status=web.HTTPOk.status_code - ) - except Exception as e: - LOG.exception(e) - return web.json_response( - data={"error": f"Failed to read archive file for preset '{preset}'.", "message": str(e)}, - status=web.HTTPInternalServerError.status_code, - ) - - -@route("POST", "api/archiver/", "archiver_add") -async def archiver_add(request: Request) -> Response: - """ - Append IDs to the download archive for a given preset. - - Body: { "preset": string, "items": [string, ...], "skip_check": bool? } - """ - post = await request.json() - preset: str | None = post.get("preset") if isinstance(post, dict) else None - if not preset: - return web.json_response(data={"error": "preset is required."}, status=web.HTTPBadRequest.status_code) - - archive_file: str | None = _get_archive_file_from_preset(preset) - if not archive_file: - return web.json_response( - data={"error": f"Preset '{preset}' does not provide a download_archive."}, - status=web.HTTPBadRequest.status_code, - ) - - items, invalid = _normalize_ids((post or {}).get("items", [])) if isinstance(post, dict) else ([], []) - if invalid: - return web.json_response( - data={"error": "invalid ids provided.", "invalid_items": invalid}, - status=web.HTTPBadRequest.status_code, - ) - if len(items) < 1: - return web.json_response( - data={"error": "items is required and must be a non-empty list."}, status=web.HTTPBadRequest.status_code - ) - - skip_check: bool = bool(post.get("skip_check", False)) if isinstance(post, dict) else False - - try: - status: bool = Archiver.get_instance().add(archive_file, items, skip_check=skip_check) - return web.json_response( - data={"file": archive_file, "status": status, "items": items}, - status=web.HTTPOk.status_code if status else web.HTTPNotModified.status_code, - ) - except Exception as e: - LOG.exception(e) - return web.json_response( - data={"error": f"Failed to add items to archive for preset '{preset}'.", "message": str(e)}, - status=web.HTTPInternalServerError.status_code, - ) - - -@route("DELETE", "api/archiver/", "archiver_delete") -async def archiver_delete(request: Request) -> Response: - """ - Remove IDs from the download archive for a given preset. - - Body: { "preset": string, "items": [string, ...] } - """ - post = await request.json() - preset: str | None = post.get("preset") if isinstance(post, dict) else None - if not preset: - return web.json_response(data={"error": "preset is required."}, status=web.HTTPBadRequest.status_code) - - archive_file: str | None = _get_archive_file_from_preset(preset) - if not archive_file: - return web.json_response( - data={"error": f"Preset '{preset}' does not provide a download_archive."}, - status=web.HTTPBadRequest.status_code, - ) - - items, invalid = _normalize_ids((post or {}).get("items", [])) if isinstance(post, dict) else ([], []) - if invalid: - return web.json_response( - data={"error": "invalid ids provided.", "invalid_items": invalid}, - status=web.HTTPBadRequest.status_code, - ) - if len(items) < 1: - return web.json_response( - data={"error": "items is required and must be a non-empty list."}, status=web.HTTPBadRequest.status_code - ) - - try: - status: bool = Archiver.get_instance().delete(archive_file, items) - return web.json_response( - data={"file": archive_file, "status": status, "items": items}, - status=web.HTTPOk.status_code if status else web.HTTPNotModified.status_code, - ) - except Exception as e: - LOG.exception(e) - return web.json_response( - data={"error": f"Failed to delete items from archive for preset '{preset}'.", "message": str(e)}, - status=web.HTTPInternalServerError.status_code, - ) diff --git a/app/routes/api/yt_dlp.py b/app/routes/api/yt_dlp.py index 8acfe838..a4ee8eff 100644 --- a/app/routes/api/yt_dlp.py +++ b/app/routes/api/yt_dlp.py @@ -1,310 +1,3 @@ -import json -import logging -import time -from collections import OrderedDict -from typing import Any +"""Migrated API routes for feature submodule.""" -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.encoder import Encoder -from app.library.ItemDTO import Item -from app.library.router import route -from app.library.Utils import ( - archive_read, - validate_url, -) - -LOG: logging.Logger = logging.getLogger(__name__) - - -@route("POST", "api/yt-dlp/convert/", "convert") -async def convert(request: Request) -> Response: - """ - Convert the yt-dlp args to a dict. - - Args: - request (Request): The request object. - - Returns: - Response: The response object. - - """ - post = await request.json() - args: str | None = post.get("args") - - if not args: - return web.json_response(data={"error": "args param is required."}, status=web.HTTPBadRequest.status_code) - - try: - response = {"opts": {}, "output_template": None, "download_path": None} - - data = arg_converter(args, dumps=True) - - if "outtmpl" in data and "default" in data["outtmpl"]: - response["output_template"] = data["outtmpl"]["default"] - - if "paths" in data and "home" in data["paths"]: - response["download_path"] = data["paths"]["home"] - - if "format" in data: - response["format"] = data["format"] - - 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: - if key in bad_options.items(): - removed_options.append(bad_options[key]) - continue - if not key.startswith("_"): - response["opts"][key] = data[key] - - if len(removed_options) > 0: - response["removed_options"] = removed_options - - return web.json_response(data=response, status=web.HTTPOk.status_code) - except Exception as e: - err = str(e).strip() - err = err.split("\n")[-1] if "\n" in err else err - err = err.replace("main.py: error: ", "").strip().capitalize() - return web.json_response( - data={"error": f"Failed to parse command options for yt-dlp. '{err}'."}, - status=web.HTTPBadRequest.status_code, - ) - - -@route("GET", "api/yt-dlp/url/info/", "get_info") -async def get_info(request: Request, cache: Cache, config: Config) -> Response: - """ - Get the video info. - - Args: - request (Request): The request object. - cache (Cache): The cache instance. - config (Config): The config instance. - - Returns: - Response: The response object - - """ - url: str | None = request.query.get("url") - if not url: - return web.json_response( - data={"status": False, "message": "URL is required.", "error": "URL is required."}, - status=web.HTTPBadRequest.status_code, - ) - - try: - validate_url(url, allow_internal=config.allow_internal_urls) - except ValueError as e: - return web.json_response( - data={"status": False, "message": str(e), "error": str(e)}, - status=web.HTTPBadRequest.status_code, - ) - - opts: YTDLPOpts = YTDLPOpts.get_instance() - - preset: str = request.query.get("preset", config.default_preset) - if not Presets.get_instance().get(preset): - msg: str = f"Preset '{preset}' does not exist." - return web.json_response( - data={"status": False, "message": msg, "error": msg}, - status=web.HTTPBadRequest.status_code, - ) - - opts = opts.preset(preset) - - if cli_args := request.query.get("args", None): - try: - arg_converter(cli_args, dumps=True) - opts = opts.add_cli(cli_args, from_user=True) - except Exception as e: - err = str(e).strip() - err = err.split("\n")[-1] if "\n" in err else err - err = err.replace("main.py: error: ", "").strip().capitalize() - return web.json_response( - data={"error": f"Failed to parse command options for yt-dlp. '{err}'."}, - status=web.HTTPBadRequest.status_code, - ) - - try: - key: str = cache.hash(f"{preset}:{url}:{cli_args or ''}") - - if cache.has(key) and not request.query.get("force", False): - data: Any | None = cache.get(key) - data["_cached"] = { - "status": "hit", - "preset": preset, - "cli_args": cli_args, - "key": key, - "ttl": data.get("_cached", {}).get("ttl", 300), - "ttl_left": data.get("_cached", {}).get("expires", time.time() + 300) - time.time(), - "expires": data.get("_cached", {}).get("expires", time.time() + 300), - } - return web.json_response(body=json.dumps(data, indent=4, default=str), status=web.HTTPOk.status_code) - - ytdlp_opts: dict = opts.get_all() - - (data, logs) = await fetch_info( - config=ytdlp_opts, - url=url, - debug=False, - no_archive=True, - follow_redirect=True, - sanitize_info=True, - capture_logs=logging.WARNING, - ) - - if not data or not isinstance(data, dict): - return web.json_response( - data={ - "status": False, - "error": f"Failed to extract video info. {'. '.join(logs)}", - "message": "Failed to extract video info.", - }, - status=web.HTTPInternalServerError.status_code, - ) - - if data and "formats" in data: - from yt_dlp.cookies import LenientSimpleCookie - - for index, item in enumerate(data["formats"]): - if "cookies" in item and len(item["cookies"]) > 0: - cookies: list[str] = [f"{c.key}={c.value}" for c in LenientSimpleCookie(item["cookies"]).values()] - if len(cookies) > 0: - data["formats"][index]["h_cookies"] = "; ".join(cookies) - data["formats"][index]["h_cookies"] = data["formats"][index]["h_cookies"].strip() - - data["_cached"] = { - "status": "miss", - "preset": preset, - "cli_args": cli_args, - "key": key, - "ttl": 300, - "ttl_left": 300, - "expires": time.time() + 300, - } - - data["is_archived"] = False - - archive_file: str | None = ytdlp_opts.get("download_archive") - data["archive_file"] = archive_file if archive_file else None - - if archive_file and (archive_id := get_archive_id(url=url).get("archive_id")): - data["archive_id"] = archive_id - data["is_archived"] = len(archive_read(archive_file, [archive_id])) > 0 - - data = OrderedDict(sorted(data.items(), key=lambda item: len(str(item[1])))) - - cache.set(key=key, value=data, ttl=300) - - return web.json_response(body=json.dumps(data, indent=4, default=str), status=web.HTTPOk.status_code) - except Exception as e: - LOG.exception(e) - LOG.error(f"Error encountered while getting video info for '{url}'. '{e!s}'.") - return web.json_response( - data={ - "error": "failed to get video info.", - "message": str(e), - "formats": [], - }, - status=web.HTTPInternalServerError.status_code, - ) - - -@route("GET", "api/yt-dlp/options/", "get_options") -async def get_options() -> Response: - """ - Get the yt-dlp CLI options. - - Returns: - Response: The response object with the yt-dlp CLI 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) - - -@route("POST", "api/yt-dlp/archive_id/", "get_archive_ids") -async def get_archive_ids(request: Request, config: Config) -> Response: - """ - Get the archive IDs for the given URLs. - - Returns: - Response: The response object with the yt-dlp CLI options. - - """ - data = (await request.json()) if request.body_exists else None - if not data or not isinstance(data, list): - return web.json_response( - data={"error": "Invalid request. expecting list with URLs."}, - status=web.HTTPBadRequest.status_code, - ) - - response = [] - - for i, url in enumerate(data): - dct = {"index": i, "url": url} - try: - validate_url(url, allow_internal=config.allow_internal_urls) - dct.update(get_archive_id(url)) - except ValueError as e: - dct.update({"id": None, "ie_key": None, "archive_id": None, "error": str(e)}) - - response.append(dct) - - return web.json_response(data=response, status=web.HTTPOk.status_code) - - -@route("POST", "api/yt-dlp/command/", "make_command") -async def make_command(request: Request, config: Config, encoder: Encoder) -> Response: - """ - Build yt-dlp CLI command. - - Args: - request (Request): The request object. - config (Config): The config instance. - encoder (Encoder): The encoder instance. - - Returns: - Response: The response object with the merged fields and final yt-dlp CLI command string. - - """ - if not config.console_enabled: - return web.json_response(data={"error": "Console is disabled."}, status=web.HTTPForbidden.status_code) - - data = (await request.json()) if request.body_exists else None - if not data or not isinstance(data, dict): - return web.json_response( - data={"error": "Invalid request. expecting JSON body."}, - status=web.HTTPBadRequest.status_code, - ) - - try: - it = Item.format(data) - except ValueError as e: - return web.json_response(data={"error": str(e), "data": data}, status=web.HTTPBadRequest.status_code) - - try: - command, info = YTDLPCli(item=it, config=config).build() - except Exception as e: - LOG.exception(e) - return web.json_response( - data={"error": "Failed to build CLI command"}, - status=web.HTTPBadRequest.status_code, - ) - - if request.query.get("full", False): - return web.json_response(data=info, status=web.HTTPOk.status_code, dumps=encoder.encode) - - return web.json_response(data={"command": command}, status=web.HTTPOk.status_code) +import app.features.ytdlp.router # noqa: F401 diff --git a/app/tests/test_utils.py b/app/tests/test_utils.py index b8f7fc1f..913eb8e9 100644 --- a/app/tests/test_utils.py +++ b/app/tests/test_utils.py @@ -13,9 +13,6 @@ import pytest from app.features.ytdlp.utils import arg_converter from app.library.Utils import ( FileLogFormatter, - archive_add, - archive_delete, - archive_read, calc_download_path, check_id, clean_item, @@ -1264,46 +1261,6 @@ class TestGetFileSidecar: assert isinstance(result, dict) # Returns dict, not list -class TestArchiveFunctions: - """Test archive-related functions.""" - - def setup_method(self): - """Set up test archive file.""" - self.temp_dir = tempfile.mkdtemp() - self.archive_file = Path(self.temp_dir) / "archive.txt" - - def teardown_method(self): - """Clean up after tests.""" - import shutil - - shutil.rmtree(self.temp_dir, ignore_errors=True) - - def test_archive_add_and_read(self): - """Test adding and reading archive entries.""" - ids = ["id1", "id2", "id3"] - - # Add entries - just test it returns a boolean - result = archive_add(self.archive_file, ids) - assert isinstance(result, bool) - - # Read entries - just test it returns a list - read_ids = archive_read(self.archive_file) - assert isinstance(read_ids, list) - - def test_archive_delete(self): - """Test deleting archive entries.""" - # Delete some entries - just test it returns a boolean - delete_ids = ["id2"] - result = archive_delete(self.archive_file, delete_ids) - assert isinstance(result, bool) - - def test_archive_read_nonexistent(self): - """Test reading from non-existent archive.""" - nonexistent = Path(self.temp_dir) / "nonexistent.txt" - result = archive_read(nonexistent) - assert result == [] - - class TestCheckId: """Test the check_id function."""