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

Refactor: replace direct extract_info calls to async fetch_info

Browse source
This commit is contained in:
arabcoders 2026-01-16 20:45:10 +03:00
parent ea572ec4c5
commit 795943196e
15 changed files with 245 additions and 109 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
app/library/Events.py
View file

@ -9,7 +9,7 @@ from typing import Any
from .BackgroundWorker import BackgroundWorker
from .Singleton import Singleton
LOG: logging.Logger = logging.getLogger("library.events")
LOG: logging.Logger = logging.getLogger("events")
class Events:

2
app/library/HttpAPI.py
View file

@ -28,7 +28,7 @@ class HttpAPI:
self.config: Config = Config.get_instance()
self._notify: EventBus = EventBus.get_instance()
self.rootPath: Path = root_path
self.cache = Cache()
self.cache: Cache = Cache.get_instance()
self.app: web.Application | None = None
services: Services = Services.get_instance()

25
app/library/Tasks.py
View file

@ -21,7 +21,7 @@ from .ItemDTO import Item, ItemDTO
from .Scheduler import Scheduler
from .Services import Services
from .Singleton import Singleton
from .Utils import archive_add, archive_delete, archive_read, extract_info, init_class, validate_url
from .Utils import archive_add, archive_delete, archive_read, fetch_info, init_class, validate_url
from .YTDLPOpts import YTDLPOpts
LOG: logging.Logger = logging.getLogger("tasks")
@ -82,7 +82,7 @@ class Task:
return params
def mark(self) -> tuple[bool, str]:
async def mark(self) -> tuple[bool, str]:
"""
Mark the task's items as downloaded in the archive file.
@ -90,7 +90,7 @@ class Task:
tuple[bool, str]: A tuple indicating success and a message.
"""
ret = self._mark_logic()
ret: tuple[bool, str] | set[tuple[Path, set[str]]] = await self._mark_logic()
if isinstance(ret, tuple):
return ret
@ -102,7 +102,7 @@ class Task:
return (True, f"Task '{self.name}' items marked as downloaded.")
def unmark(self) -> tuple[bool, str]:
async def unmark(self) -> tuple[bool, str]:
"""
Unmark the task's items from the archive file.
@ -110,7 +110,7 @@ class Task:
tuple[bool, str]: A tuple indicating success and a message.
"""
ret: tuple[bool, str] | set[tuple[Path, set[str]]] = self._mark_logic()
ret: tuple[bool, str] | set[tuple[Path, set[str]]] = await self._mark_logic()
if isinstance(ret, tuple):
return ret
@ -122,7 +122,7 @@ class Task:
return (True, f"Removed '{self.name}' items from archive file.")
def fetch_metadata(self, full: bool = False) -> tuple[dict[str, Any] | None, bool, str]:
async def fetch_metadata(self, full: bool = False) -> tuple[dict[str, Any] | None, bool, str]:
"""
Fetch metadata for the task's URL.
@ -143,15 +143,20 @@ class Task:
params = params.get_all()
ie_info: dict | None = extract_info(
params, self.url, no_archive=True, follow_redirect=False, sanitize_info=True
ie_info: dict | None = await fetch_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.")
return (ie_info, True, "")
def _mark_logic(self) -> tuple[bool, str] | set[tuple[Path, set[str]]]:
async def _mark_logic(self) -> tuple[bool, str] | set[tuple[Path, set[str]]]:
if not self.url:
return (False, "No URL found in task parameters.")
@ -161,7 +166,7 @@ class Task:
archive_file: Path = Path(archive_file)
ie_info: dict | None = extract_info(params, self.url, no_archive=True, follow_redirect=True)
ie_info: dict | None = await fetch_info(params, self.url, no_archive=True, follow_redirect=True)
if not ie_info or not isinstance(ie_info, dict):
return (False, "Failed to extract information from URL.")

2
app/library/UpdateChecker.py
View file

@ -45,7 +45,7 @@ class UpdateChecker(metaclass=Singleton):
self._notify: EventBus = notify or EventBus.get_instance()
"Instance of EventBus for notifications."
self._cache: Cache = Cache()
self._cache: Cache = Cache.get_instance()
"Instance of Cache for caching check results."
self._job_id: str | None = None

57
app/library/Utils.py
View file

@ -1,5 +1,7 @@
import asyncio
import base64
import copy
import functools
import glob
import ipaddress
import json
@ -86,6 +88,9 @@ TAG_REGEX: re.Pattern[str] = re.compile(r"%{([^:}]+)(?::([^}]*))?}c")
DT_PATTERN: re.Pattern[str] = re.compile(r"^(\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d+)?(?:[+-]\d{2}:\d{2}))\s?")
"Regex to match ISO 8601 datetime strings."
EXTRACTORS_SEMAPHORE: asyncio.Semaphore | None = None
"Global semaphore for limiting concurrent extractor operations."
class StreamingError(Exception):
"""Raised when an error occurs during streaming."""
@ -416,6 +421,58 @@ def extract_info(
return YTDLP.sanitize_info(data, remove_private_keys=True) if sanitize_info else data
async def fetch_info(
config: dict,
url: str,
debug: bool = False,
no_archive: bool = False,
follow_redirect: bool = False,
sanitize_info: bool = False,
**kwargs,
) -> dict:
"""
Extracts video information from the given URL.
Args:
config (dict): Configuration options.
url (str): URL to extract information from.
debug (bool): Enable debug logging.
no_archive (bool): Disable download archive.
follow_redirect (bool): Follow URL redirects.
sanitize_info (bool): Sanitize the extracted information
**kwargs: Additional arguments.
Returns:
dict: Video information.
"""
from .config import Config
global EXTRACTORS_SEMAPHORE # noqa: PLW0603
conf = Config.get_instance()
if EXTRACTORS_SEMAPHORE is None:
EXTRACTORS_SEMAPHORE = asyncio.Semaphore(conf.extract_info_concurrency)
async with EXTRACTORS_SEMAPHORE:
return await asyncio.wait_for(
fut=asyncio.get_running_loop().run_in_executor(
None,
functools.partial(
extract_info,
config=config,
url=url,
debug=debug,
no_archive=no_archive,
follow_redirect=follow_redirect,
sanitize_info=sanitize_info,
**kwargs,
),
),
timeout=conf.extract_info_timeout,
)
def _is_safe_key(key: any) -> bool:
"""
Check if a dictionary key is safe for merging.

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

@ -11,7 +11,6 @@ This module handles the complete flow of adding items to the download queue:
"""
import asyncio
import functools
import logging
import time
import uuid
@ -29,7 +28,7 @@ from app.library.Utils import (
archive_read,
arg_converter,
create_cookies_file,
extract_info,
fetch_info,
get_extras,
merge_dict,
ytdlp_reject,
@ -205,22 +204,14 @@ async def add(
LOG.info(f"[P] Extracting '{item.url}'{' with cookies' if yt_conf.get('cookiefile') else ''}.")
if not entry:
async with queue.extractors:
LOG.info(f"Extracting '{item.url}'{' with cookies' if yt_conf.get('cookiefile') else ''}.")
entry: dict | None = await asyncio.wait_for(
fut=asyncio.get_running_loop().run_in_executor(
None,
functools.partial(
extract_info,
config=yt_conf,
url=item.url,
debug=bool(queue.config.ytdlp_debug),
no_archive=False,
follow_redirect=True,
),
),
timeout=queue.config.extract_info_timeout,
)
LOG.info(f"Extracting '{item.url}'{' with cookies' if yt_conf.get('cookiefile') else ''}.")
entry: dict | None = await fetch_info(
config=yt_conf,
url=item.url,
debug=bool(queue.config.ytdlp_debug),
no_archive=False,
follow_redirect=True,
)
if not entry:
LOG.error(f"Unable to extract info for '{item.url}'. Logs: {logs}")

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

@ -43,8 +43,6 @@ class DownloadQueue(metaclass=Singleton):
"DataStore for the download queue."
self.processors = asyncio.Semaphore(self.config.playlist_items_concurrency)
"Semaphore to limit the number of concurrent processors."
self.extractors = asyncio.Semaphore(self.config.extract_info_concurrency)
"Semaphore to limit the number of concurrent extract_info calls."
self.pool = PoolManager(queue=self, config=self.config)
"Pool manager for coordinating download execution."

4
app/library/task_handlers/generic.py
View file

@ -23,7 +23,7 @@ from app.library.cache import Cache
from app.library.config import Config
from app.library.httpx_client import async_client
from app.library.Tasks import Task, TaskFailure, TaskItem, TaskResult
from app.library.Utils import extract_info, get_archive_id
from app.library.Utils import fetch_info, get_archive_id
from ._base_handler import BaseHandler
@ -714,7 +714,7 @@ class GenericTaskHandler(BaseHandler):
f"[{definition.name}]: '{task.name}': Unable to generate static archive id for '{url}' in feed. Doing real request to fetch yt-dlp archive id."
)
info = extract_info(
info = await fetch_info(
config=task.get_ytdlp_opts().get_all(),
url=url,
no_archive=True,

4
app/library/task_handlers/rss.py
View file

@ -6,7 +6,7 @@ from xml.etree.ElementTree import Element
from app.library.cache import Cache
from app.library.Tasks import Task, TaskFailure, TaskItem, TaskResult
from app.library.Utils import extract_info, get_archive_id
from app.library.Utils import fetch_info, get_archive_id
from ._base_handler import BaseHandler
@ -179,7 +179,7 @@ class RssGenericHandler(BaseHandler):
"Doing real request to fetch yt-dlp archive ID."
)
info = extract_info(
info = await fetch_info(
config=params,
url=url,
no_archive=True,

25
app/routes/api/conditions.py
View file

@ -1,5 +1,3 @@
import asyncio
import functools
import logging
import uuid
from collections import OrderedDict
@ -14,7 +12,7 @@ from app.library.encoder import Encoder
from app.library.Events import EventBus, Events
from app.library.mini_filter import match_str
from app.library.router import route
from app.library.Utils import extract_info, init_class, validate_uuid
from app.library.Utils import fetch_info, init_class, validate_uuid
from app.library.YTDLPOpts import YTDLPOpts
LOG: logging.Logger = logging.getLogger(__name__)
@ -164,20 +162,13 @@ 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):
data: dict | None = await asyncio.wait_for(
fut=asyncio.get_running_loop().run_in_executor(
None,
functools.partial(
extract_info,
config=YTDLPOpts.get_instance().preset(name=preset).get_all(),
url=url,
debug=False,
no_archive=True,
follow_redirect=True,
sanitize_info=True,
),
),
timeout=120,
data: dict | None = await fetch_info(
config=YTDLPOpts.get_instance().preset(name=preset).get_all(),
url=url,
debug=False,
no_archive=True,
follow_redirect=True,
sanitize_info=True,
)
if not data:

14
app/routes/api/tasks.py
View file

@ -1,5 +1,3 @@
import asyncio
import functools
import logging
import uuid
from pathlib import Path
@ -177,7 +175,7 @@ async def task_mark(request: Request, encoder: Encoder) -> Response:
data={"error": f"Task '{task_id}' does not exist."}, status=web.HTTPNotFound.status_code
)
_status, _message = task.mark()
_status, _message = await task.mark()
if not _status:
return web.json_response(data={"error": _message}, status=web.HTTPBadRequest.status_code)
@ -212,7 +210,7 @@ async def task_unmark(request: Request, encoder: Encoder) -> Response:
data={"error": f"Task '{task_id}' does not exist."}, status=web.HTTPNotFound.status_code
)
_status, _message = task.unmark()
_status, _message = await task.unmark()
if not _status:
return web.json_response(data={"error": _message}, status=web.HTTPBadRequest.status_code)
@ -255,13 +253,7 @@ 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 = await asyncio.wait_for(
fut=asyncio.get_running_loop().run_in_executor(
None,
functools.partial(task.fetch_metadata, full=False),
),
timeout=120,
)
metadata, status, message = await task.fetch_metadata()
if not status:
return web.json_response(data={"error": message}, status=web.HTTPBadRequest.status_code)

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

@ -1,5 +1,3 @@
import asyncio
import functools
import json
import logging
import time
@ -19,7 +17,7 @@ from app.library.Utils import (
REMOVE_KEYS,
archive_read,
arg_converter,
extract_info,
fetch_info,
get_archive_id,
validate_url,
)
@ -165,20 +163,13 @@ async def get_info(request: Request, cache: Cache, config: Config) -> Response:
},
}
data: dict | None = await asyncio.wait_for(
fut=asyncio.get_running_loop().run_in_executor(
None,
functools.partial(
extract_info,
config=ytdlp_opts,
url=url,
debug=False,
no_archive=True,
follow_redirect=True,
sanitize_info=True,
),
),
timeout=120,
data: dict | None = await fetch_info(
config=ytdlp_opts,
url=url,
debug=False,
no_archive=True,
follow_redirect=True,
sanitize_info=True,
)
if not data or not isinstance(data, dict):

6
app/tests/test_generic_task_handler.py
View file

@ -309,11 +309,11 @@ async def test_generic_task_handler_inspect(monkeypatch):
monkeypatch.setattr(GenericTaskHandler, "_fetch_content", staticmethod(fake_fetch_content))
# Mock extract_info to return valid info with required fields for archive ID generation
def fake_extract_info(config, url, **kwargs): # noqa: ARG001
# Mock fetch_info to return valid info with required fields for archive ID generation
async def fake_fetch_info(config, url, **kwargs): # noqa: ARG001
return {"id": "test_video_1", "extractor_key": "Example"}
with patch("app.library.task_handlers.generic.extract_info", side_effect=fake_extract_info):
with patch("app.library.task_handlers.generic.fetch_info", side_effect=fake_fetch_info):
task = Task(id="inspect", name="Inspect", url="https://example.com/api")
result: TaskResult | TaskFailure = await GenericTaskHandler.extract(task)

157
app/tests/test_tasks.py
View file

@ -140,12 +140,13 @@ class TestTask:
mock_opts_instance.preset.assert_called_once_with(name="test_preset")
mock_preset_opts.add_cli.assert_called_once_with("--extract-flat", from_user=True)
@pytest.mark.asyncio
@patch("app.library.Tasks.archive_add")
@patch("app.library.Tasks.extract_info")
def test_task_mark_success(self, mock_extract_info, mock_archive_add):
@patch("app.library.Tasks.fetch_info")
async def test_task_mark_success(self, mock_fetch_info, mock_archive_add):
"""Test successful task marking."""
# Mock extract_info to return playlist data
mock_extract_info.return_value = {
# Mock fetch_info to return playlist data
mock_fetch_info.return_value = {
"_type": "playlist",
"entries": [
{
@ -167,7 +168,7 @@ class TestTask:
task = Task(id="test_id", name="test_task", url="https://example.com/playlist")
success, message = task.mark()
success, message = await task.mark()
assert success is True
assert "marked as downloaded" in message
@ -180,33 +181,36 @@ class TestTask:
assert "youtube video1" in items
assert "youtube video2" in items
def test_task_mark_no_url(self):
@pytest.mark.asyncio
async def test_task_mark_no_url(self):
"""Test task marking with no URL."""
task = Task(id="test_id", name="test_task", url="")
success, message = task.mark()
success, message = await task.mark()
assert success is False
assert "No URL found" in message
def test_task_mark_no_archive_file(self):
@pytest.mark.asyncio
async def test_task_mark_no_archive_file(self):
"""Test task marking with no archive file configured."""
with patch.object(Task, "get_ytdlp_opts") as mock_get_opts:
mock_get_opts.return_value.get_all.return_value = {}
task = Task(id="test_id", name="test_task", url="https://example.com/video")
success, message = task.mark()
success, message = await task.mark()
assert success is False
assert "No archive file found" in message
@pytest.mark.asyncio
@patch("app.library.Tasks.archive_delete")
@patch("app.library.Tasks.extract_info")
def test_task_unmark_success(self, mock_extract_info, mock_archive_delete):
@patch("app.library.Tasks.fetch_info")
async def test_task_unmark_success(self, mock_fetch_info, mock_archive_delete):
"""Test successful task unmarking."""
# Mock extract_info to return playlist data
mock_extract_info.return_value = {
# Mock fetch_info to return playlist data
mock_fetch_info.return_value = {
"_type": "playlist",
"entries": [
{
@ -223,34 +227,36 @@ class TestTask:
task = Task(id="test_id", name="test_task", url="https://example.com/playlist")
success, message = task.unmark()
success, message = await task.unmark()
assert success is True
assert "Removed" in message
assert "items from archive file" in message
mock_archive_delete.assert_called_once()
@patch("app.library.Tasks.extract_info")
def test_task_mark_logic_invalid_extract_info(self, mock_extract_info):
"""Test _mark_logic with invalid extract_info response."""
mock_extract_info.return_value = None
@pytest.mark.asyncio
@patch("app.library.Tasks.fetch_info")
async def test_task_mark_logic_invalid_extract_info(self, mock_fetch_info):
"""Test _mark_logic with invalid fetch_info response."""
mock_fetch_info.return_value = None
with patch.object(Task, "get_ytdlp_opts") as mock_get_opts:
mock_get_opts.return_value.get_all.return_value = {"download_archive": "/tmp/archive.txt"}
task = Task(id="test_id", name="test_task", url="https://example.com/video")
result = task._mark_logic()
result = await task._mark_logic()
assert isinstance(result, tuple)
success, message = result
assert success is False
assert "Failed to extract information" in message
@patch("app.library.Tasks.extract_info")
def test_task_mark_logic_not_playlist(self, mock_extract_info):
@pytest.mark.asyncio
@patch("app.library.Tasks.fetch_info")
async def test_task_mark_logic_not_playlist(self, mock_fetch_info):
"""Test _mark_logic with non-playlist type."""
mock_extract_info.return_value = {
mock_fetch_info.return_value = {
"_type": "video",
"id": "video1",
}
@ -260,13 +266,118 @@ class TestTask:
task = Task(id="test_id", name="test_task", url="https://example.com/video")
result = task._mark_logic()
result = await task._mark_logic()
assert isinstance(result, tuple)
success, message = result
assert success is False
assert "Expected a playlist type" in message
@pytest.mark.asyncio
async def test_task_fetch_metadata_no_url(self):
"""Test fetch_metadata with no URL."""
task = Task(id="test_id", name="test_task", url="")
metadata, success, message = await task.fetch_metadata()
assert success is False, "Should return False when URL is missing"
assert metadata == {}, "Should return empty dict for metadata"
assert "No URL found" in message, "Should indicate missing URL"
@pytest.mark.asyncio
@patch("app.library.Tasks.fetch_info")
async def test_task_fetch_metadata_success_not_full(self, mock_fetch_info):
"""Test fetch_metadata without full flag."""
mock_fetch_info.return_value = {
"_type": "playlist",
"id": "playlist123",
"title": "Test Playlist",
"entries": [{"id": "video1"}],
}
with patch.object(Task, "get_ytdlp_opts") as mock_get_opts:
mock_opts = Mock()
mock_opts.add_cli.return_value = mock_opts
mock_opts.get_all.return_value = {"download_archive": "/tmp/archive.txt"}
mock_get_opts.return_value = mock_opts
task = Task(id="test_id", name="test_task", url="https://example.com/playlist")
metadata, success, message = await task.fetch_metadata(full=False)
assert success is True, "Should return True on successful fetch"
assert metadata["_type"] == "playlist", "Should return playlist metadata"
assert metadata["id"] == "playlist123", "Should return correct ID"
assert message == "", "Should have empty message on success"
mock_opts.add_cli.assert_called_once_with("-I0", from_user=False)
mock_fetch_info.assert_called_once()
call_kwargs = mock_fetch_info.call_args[1]
assert call_kwargs["no_archive"] is True, "Should disable archive"
assert call_kwargs["follow_redirect"] is False, "Should not follow redirects"
assert call_kwargs["sanitize_info"] is True, "Should sanitize info"
@pytest.mark.asyncio
@patch("app.library.Tasks.fetch_info")
async def test_task_fetch_metadata_success_full(self, mock_fetch_info):
"""Test fetch_metadata with full flag."""
mock_fetch_info.return_value = {
"_type": "playlist",
"id": "playlist123",
"title": "Test Playlist",
"entries": [
{"id": "video1", "title": "Video 1"},
{"id": "video2", "title": "Video 2"},
],
}
with patch.object(Task, "get_ytdlp_opts") as mock_get_opts:
mock_opts = Mock()
mock_opts.get_all.return_value = {}
mock_get_opts.return_value = mock_opts
task = Task(id="test_id", name="test_task", url="https://example.com/playlist")
metadata, success, message = await task.fetch_metadata(full=True)
assert success is True, "Should return True on successful fetch"
assert len(metadata["entries"]) == 2, "Should return full entries"
assert message == "", "Should have empty message on success"
mock_opts.add_cli.assert_not_called()
@pytest.mark.asyncio
@patch("app.library.Tasks.fetch_info")
async def test_task_fetch_metadata_fetch_info_returns_none(self, mock_fetch_info):
"""Test fetch_metadata when fetch_info returns None."""
mock_fetch_info.return_value = None
with patch.object(Task, "get_ytdlp_opts") as mock_get_opts:
mock_get_opts.return_value.get_all.return_value = {}
task = Task(id="test_id", name="test_task", url="https://example.com/video")
metadata, success, message = await task.fetch_metadata()
assert success is False, "Should return False when fetch_info fails"
assert metadata == {}, "Should return empty dict for metadata"
assert "Failed to extract information" in message, "Should indicate extraction failure"
@pytest.mark.asyncio
@patch("app.library.Tasks.fetch_info")
async def test_task_fetch_metadata_fetch_info_returns_invalid_type(self, mock_fetch_info):
"""Test fetch_metadata when fetch_info returns non-dict."""
mock_fetch_info.return_value = "invalid"
with patch.object(Task, "get_ytdlp_opts") as mock_get_opts:
mock_get_opts.return_value.get_all.return_value = {}
task = Task(id="test_id", name="test_task", url="https://example.com/video")
metadata, success, message = await task.fetch_metadata()
assert success is False, "Should return False when fetch_info returns invalid type"
assert metadata == {}, "Should return empty dict for metadata"
assert "Failed to extract information" in message, "Should indicate extraction failure"
class TestTasks:
"""Test the Tasks singleton manager."""

2
ui/app/components/Dropdown.vue
View file

@ -14,7 +14,7 @@
<Teleport to="body">
<div v-if="isOpen" class="dropdown is-active dropdown-portal" ref="menu" :style="menuStyle">
<div class="dropdown-menu" role="menu">
<div class="dropdown-content" @click="handle_slot_click">
<div class="dropdown-content is-unselectable" @click="handle_slot_click">
<template v-if="hideLabel">
<div class="dropdown-label">
<span class="icon-text">