diff --git a/app/library/Tasks.py b/app/library/Tasks.py index 056eae21..92a22c44 100644 --- a/app/library/Tasks.py +++ b/app/library/Tasks.py @@ -77,6 +77,9 @@ class Task: if self.cli: params = params.add_cli(self.cli, from_user=True) + if self.template: + params = params.add({"outtmpl": {"default": self.template}}, from_user=False) + return params def mark(self) -> tuple[bool, str]: @@ -140,7 +143,9 @@ class Task: params = params.get_all() - ie_info: dict | None = extract_info(params, self.url, no_archive=True, follow_redirect=False, cache=True) + ie_info: dict | None = extract_info( + params, self.url, no_archive=True, follow_redirect=False, sanitize_info=True + ) if not ie_info or not isinstance(ie_info, dict): return ({}, False, "Failed to extract information from URL.") diff --git a/app/library/Utils.py b/app/library/Utils.py index f0e69cbb..ead7069b 100644 --- a/app/library/Utils.py +++ b/app/library/Utils.py @@ -65,7 +65,7 @@ REMOVE_KEYS: list = [ ] "Keys to remove from yt-dlp options at various levels." -YTDLP_INFO_CLS: YTDLP = None +YTDLP_INFO_CLS: YTDLP | None = None "Cached YTDLP info class." ALLOWED_SUBS_EXTENSIONS: set[str] = {".srt", ".vtt", ".ass"} @@ -96,6 +96,32 @@ class FileLogFormatter(logging.Formatter): return datetime.fromtimestamp(record.created).astimezone().isoformat(timespec="milliseconds") +def get_static_ytdlp(reload: bool = False) -> YTDLP: + """ + Get a static YTDLP instance for info extraction. + + Args: + reload (bool): If True, forces re-creation of the instance. + + Returns: + YTDLP: A static YTDLP instance. + + """ + global YTDLP_INFO_CLS # noqa: PLW0603 + if YTDLP_INFO_CLS is None or reload: + YTDLP_INFO_CLS = YTDLP( + params={ + "color": "no_color", + "extract_flat": True, + "skip_download": True, + "ignoreerrors": True, + "ignore_no_formats_error": True, + "quiet": True, + } + ) + return YTDLP_INFO_CLS + + def patch_metadataparser() -> None: """ Patches yt_dlp MetadataParserPP action to handle subprocess pickling issues. @@ -382,7 +408,7 @@ def extract_info( if not data["is_premiere"]: data["is_premiere"] = "video" == data.get("media_type") and "is_upcoming" == data.get("live_status") - return YTDLP.sanitize_info(data) if sanitize_info else data + return YTDLP.sanitize_info(data, remove_private_keys=True) if sanitize_info else data def _is_safe_key(key: any) -> bool: @@ -1406,27 +1432,13 @@ def get_archive_id(url: str) -> dict[str, str | None]: } """ - global YTDLP_INFO_CLS # noqa: PLW0603 - idDict: dict[str, None] = { "id": None, "ie_key": None, "archive_id": None, } - if YTDLP_INFO_CLS is None: - YTDLP_INFO_CLS = YTDLP( - params={ - "color": "no_color", - "extract_flat": True, - "skip_download": True, - "ignoreerrors": True, - "ignore_no_formats_error": True, - "quiet": True, - } - ) - - for key, _ie in YTDLP_INFO_CLS._ies.items(): + for key, _ie in get_static_ytdlp()._ies.items(): try: if not _ie.suitable(url): continue @@ -1923,3 +1935,18 @@ def get_extras(entry: dict, kind: str = "video") -> dict: extras["thumbnail"] = "https://img.youtube.com/vi/{id}/maxresdefault.jpg".format(**entry) return extras + + +def parse_outtmpl(output_template: str, info_dict: dict) -> 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. + + Returns: + str: The parsed output string. + + """ + return get_static_ytdlp().prepare_filename(info_dict=info_dict, outtmpl=output_template) diff --git a/app/routes/api/tasks.py b/app/routes/api/tasks.py index b0848484..983d4d0c 100644 --- a/app/routes/api/tasks.py +++ b/app/routes/api/tasks.py @@ -1,3 +1,5 @@ +import asyncio +import functools import logging import uuid from typing import TYPE_CHECKING, Any @@ -10,7 +12,7 @@ from app.library.config import Config from app.library.encoder import Encoder from app.library.router import route from app.library.Tasks import Task, TaskFailure, TaskResult, Tasks -from app.library.Utils import get_channel_images, get_file, init_class, validate_url, validate_uuid +from app.library.Utils import get_channel_images, get_file, init_class, parse_outtmpl, validate_url, validate_uuid if TYPE_CHECKING: from pathlib import Path @@ -252,10 +254,38 @@ async def task_metadata(request: Request, config: Config, encoder: Encoder) -> R if not save_path.exists(): save_path.mkdir(parents=True, exist_ok=True) - metadata, status, message = task.fetch_metadata(full=False) + metadata, status, message = await asyncio.wait_for( + fut=asyncio.get_running_loop().run_in_executor( + None, + functools.partial(task.fetch_metadata, full=False), + ), + timeout=120, + ) if not status: return web.json_response(data={"error": message}, status=web.HTTPBadRequest.status_code) + if not task.folder: + try: + outtmpl = parse_outtmpl( + output_template=task.get_ytdlp_opts().get_all().get("outtmpl", {}).get("default", "{title} [{id}]"), + info_dict=metadata, + ) + if outtmpl: + _path = save_path / outtmpl + if not _path.is_dir(): + _path = _path.parent + + (save_path, _) = get_file(config.download_path, _path.relative_to(config.download_path)) + if not str(save_path or "").startswith(str(config.download_path)): + return web.json_response( + data={"error": "Invalid final path folder."}, status=web.HTTPBadRequest.status_code + ) + + if not save_path.exists(): + save_path.mkdir(parents=True, exist_ok=True) + except Exception as e: + LOG.warning(f"Failed to resolve final path from outtmpl. '{e!s}'") + info = { "id": ag(metadata, ["id", "channel_id"]), "id_type": metadata.get("extractor", "").split(":")[0].lower() if metadata.get("extractor") else None, @@ -277,12 +307,12 @@ async def task_metadata(request: Request, config: Config, encoder: Encoder) -> R from app.yt_dlp_plugins.postprocessor.nfo_maker import NFOMakerPP title: str = sanitize_filename(info.get("title")) - filename: Path = save_path / f"{title} [{info.get('id')}].info.json" - filename.write_text(encoder.encode(metadata), encoding="utf-8") - info["json_file"] = str(filename.relative_to(config.download_path)) + info_file: Path = save_path / f"{title} [{info.get('id')}].info.json" + info_file.write_text(encoder.encode(metadata), encoding="utf-8") + info["json_file"] = str(info_file.relative_to(config.download_path)) - filename = save_path / "tvshow.nfo" - info["nfo_file"] = f"{save_path}/{filename}" + xml_file: Path = save_path / "tvshow.nfo" + info["nfo_file"] = str(xml_file.relative_to(config.download_path)) xml_content = "\n" xml_content += f" {NFOMakerPP._escape_text(info.get('title'))}\n" @@ -303,8 +333,7 @@ async def task_metadata(request: Request, config: Config, encoder: Encoder) -> R xml_content += f" {info.get('year')}\n" xml_content += " Continuing\n" xml_content += "\n" - - filename.write_text(xml_content, encoding="utf-8") + xml_file.write_text(xml_content, encoding="utf-8") try: from yt_dlp.utils.networking import random_user_agent @@ -348,9 +377,9 @@ async def task_metadata(request: Request, config: Config, encoder: Encoder) -> R resp = await client.request(method="GET", url=url, follow_redirects=True) - filename = save_path / f"{key}.jpg" - filename.write_bytes(resp.content) - info["thumbnails"][key] = str(filename.relative_to(config.download_path)) + img_file = save_path / f"{key}.jpg" + img_file.write_bytes(resp.content) + info["thumbnails"][key] = str(img_file.relative_to(config.download_path)) except Exception as e: LOG.warning(f"Failed to fetch thumbnail '{key}' from '{url}'. '{e!s}'") continue diff --git a/app/tests/test_utils.py b/app/tests/test_utils.py index 7c1221f5..666ea48e 100644 --- a/app/tests/test_utils.py +++ b/app/tests/test_utils.py @@ -34,6 +34,7 @@ from app.library.Utils import ( get_files, get_mime_type, get_possible_images, + get_static_ytdlp, init_class, is_private_address, list_folders, @@ -41,6 +42,7 @@ from app.library.Utils import ( load_modules, merge_dict, move_file, + parse_outtmpl, parse_tags, read_logfile, rename_file, @@ -2675,3 +2677,180 @@ class TestGetExtras: result = get_extras(entry) assert result["thumbnail"] == "https://example.com/thumb.jpg" + + +class TestGetStaticYtdlp: + """Test the get_static_ytdlp function.""" + + def test_get_static_ytdlp_returns_instance(self): + """Test that get_static_ytdlp returns a YTDLP instance.""" + from app.library.Utils import get_static_ytdlp + from app.library.ytdlp import YTDLP + + # Force reload to ensure we get a real instance, not a mock + instance = get_static_ytdlp(reload=True) + + 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.""" + from app.library.Utils import get_static_ytdlp + + instance1 = get_static_ytdlp() + instance2 = get_static_ytdlp() + + assert instance1 is instance2 + + def test_get_static_ytdlp_reload(self): + """Test that get_static_ytdlp can reload and return a new instance.""" + from app.library.Utils import get_static_ytdlp + + instance1 = get_static_ytdlp() + instance2 = get_static_ytdlp(reload=True) + + 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.""" + from app.library.Utils import get_static_ytdlp + + instance = get_static_ytdlp(reload=True) + + # 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 test_parse_outtmpl_basic(self): + """Test basic template parsing with simple placeholders.""" + from app.library.Utils import parse_outtmpl + + 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.""" + from app.library.Utils import parse_outtmpl + + 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.""" + from app.library.Utils import parse_outtmpl + + 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.""" + from app.library.Utils import parse_outtmpl + + 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.""" + from app.library.Utils import parse_outtmpl + + template = "%(title)s - %(upload_date)s.%(ext)s" + info_dict = { + "title": "Test Video", + "ext": "mp4", + # upload_date is missing + } + + result = parse_outtmpl(template, info_dict) + + assert result == "Test Video - NA.mp4" + + def test_parse_outtmpl_complex(self): + """Test complex template with multiple fields.""" + from app.library.Utils import parse_outtmpl + + 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.""" + from app.library.Utils import parse_outtmpl + + template = "%(title)s.%(ext)s" + info_dict = { + "title": "Test: Video / With \\ Special | Characters", + "ext": "mp4", + } + + result = parse_outtmpl(template, info_dict) + + # yt-dlp sanitizes special characters in filenames + assert ".mp4" in result + assert "Test" in result + + def test_parse_outtmpl_with_playlist_info(self): + """Test template parsing with playlist information.""" + from app.library.Utils import parse_outtmpl + + 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" diff --git a/ui/app/pages/tasks.vue b/ui/app/pages/tasks.vue index 5fd3bef5..1e0009ae 100644 --- a/ui/app/pages/tasks.vue +++ b/ui/app/pages/tasks.vue @@ -72,7 +72,8 @@ -
+