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

[FEAT] Add per extractor limits. Closes #421

Browse source
This commit is contained in:
arabcoders 2025-09-14 21:30:06 +03:00
parent a96be4c0fe
commit 7ad2bd6358
5 changed files with 101 additions and 12 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

9
FAQ.md
View file

@ -17,7 +17,8 @@ or the `environment:` section in `compose.yaml` file.
| YTP_INSTANCE_TITLE | The title of the instance | `empty string` |
| YTP_FILE_LOGGING | Whether to log to file | `false` |
| YTP_DOWNLOAD_PATH | Path to where the downloads will be saved | `/downloads` |
| YTP_MAX_WORKERS | How many works to use for downloads | `1` |
| YTP_MAX_WORKERS | The maximum number of workers to use for downloading | `20` |
| YTP_MAX_WORKERS_PER_EXTRACTOR | The maximum number of concurrent downloads per extractor | `2` |
| YTP_AUTH_USERNAME | Username for basic authentication | `empty string` |
| YTP_AUTH_PASSWORD | Password for basic authentication | `empty string` |
| YTP_CONSOLE_ENABLED | Whether to enable the console | `false` |
@ -50,6 +51,12 @@ or the `environment:` section in `compose.yaml` file.
| YTP_TEMP_DISABLED | Disable temp files handling. | `false` |
| YTP_DOWNLOAD_PATH_DEPTH | How many subdirectories to show in auto complete. | `1` |
> [!NOTE]
> To raise the maximum workers for specific extractor, you need to add a ENV variable that follows the pattern `YTP_MAX_WORKERS_FOR_<EXTRACTOR_NAME>`.
> The extractor name must be in uppercase, to know the extractor name, check the log for the specific extractor used for the download.
> The limit should not exceed the `YTP_MAX_WORKERS` value as it will be ignored.
# Browser extensions & bookmarklets
## Simple bookmarklet

75
app/library/DownloadQueue.py
View file

@ -2,6 +2,7 @@ import asyncio
import functools
import glob
import logging
import os
import time
import traceback
import uuid
@ -60,6 +61,8 @@ class DownloadQueue(metaclass=Singleton):
"Semaphore to limit the number of concurrent downloads."
self.processors = asyncio.Semaphore(self.config.playlist_items_concurrency)
"Semaphore to limit the number of concurrent processors."
self.limits: dict[str, asyncio.Semaphore] = {}
"Per-extractor semaphores to limit concurrent downloads per extractor."
self.paused = asyncio.Event()
"Event to pause the download queue."
self.event = asyncio.Event()
@ -82,6 +85,35 @@ class DownloadQueue(metaclass=Singleton):
"""
return DownloadQueue()
def _get_limit(self, extractor: str) -> asyncio.Semaphore:
"""
Get or create a semaphore for the given extractor.
Args:
extractor (str): The extractor name.
Returns:
asyncio.Semaphore: The semaphore for the extractor.
"""
if extractor not in self.limits:
env_limit: str | None = os.environ.get(f"YTP_MAX_WORKERS_FOR_{extractor.upper()}")
# Determine effective limit
if env_limit and env_limit.isdigit() and 1 <= int(env_limit):
limit: int = min(int(env_limit), self.config.max_workers)
else:
if env_limit:
LOG.warning(f"Invalid extractor limit '{env_limit}' for '{extractor}', using default limit.")
limit = self.config.max_workers_per_extractor
limit = min(limit, self.config.max_workers)
self.limits[extractor] = asyncio.Semaphore(limit)
LOG.info(f"Created limits container for extractor '{extractor}': {limit}")
return self.limits[extractor]
def attach(self, _: web.Application) -> None:
"""
Attach the download queue to the application.
@ -983,36 +1015,67 @@ class DownloadQueue(metaclass=Singleton):
"""
Create a pool of workers to download the files.
"""
adaptive_sleep = 0.2 # Start with base sleep
max_sleep = 5.0 # Maximum sleep to avoid excessive delays
while True:
while not self.queue.has_downloads():
LOG.info("Waiting for item to download.")
await self.event.wait()
self.event.clear()
adaptive_sleep = 0.2
if self.is_paused():
LOG.info("Download pool is paused.")
LOG.warning("Download pool is paused.")
await self.paused.wait()
LOG.info("Download pool resumed downloading.")
adaptive_sleep = 0.2
items_processed = 0
for _id, entry in list(self.queue.items()):
if entry.started() or entry.is_cancelled() or entry.info.auto_start is False:
continue
extractor: str = entry.info.get_extractor() or "unknown"
# Live downloads bypass all limits.
if entry.is_live:
task = asyncio.create_task(self._download_live(_id, entry), name=f"download_live_{_id}")
task: asyncio.Task[None] = asyncio.create_task(
self._download_live(_id, entry), name=f"download_live_{extractor}_{_id}"
)
task.add_done_callback(self._handle_task_exception)
items_processed += 1
else:
_limit: asyncio.Semaphore = self._get_limit(extractor)
# Skip this item in this iteration if no slots are available.
if self.workers.locked() or _limit.locked():
continue
await self.workers.acquire()
await _limit.acquire()
task = asyncio.create_task(self._download_file(_id, entry), name=f"download_file_{_id}")
task: asyncio.Task[None] = asyncio.create_task(
self._download_file(_id, entry), name=f"download_file_{extractor}_{_id}"
)
def _release_semaphore(t: asyncio.Task):
def _release(t: asyncio.Task, sem=_limit) -> None:
sem.release()
self.workers.release()
self._handle_task_exception(t)
task.add_done_callback(_release_semaphore)
task.add_done_callback(_release)
items_processed += 1
await asyncio.sleep(0.5)
# No items could be processed, back off a bit to avoid busy-waiting.
if 0 == items_processed:
adaptive_sleep: float = min(adaptive_sleep * 1.5, max_sleep)
LOG.info(f"No download slots available. Backing off for {adaptive_sleep:.2f}s before next attempt.")
else:
adaptive_sleep = 0.2
await asyncio.sleep(adaptive_sleep)
async def _download_live(self, _id: str, entry: Download) -> None:
LOG.info(f"Creating temporary worker for entry '{entry.info.name()}'.")

16
app/library/ItemDTO.py
View file

@ -167,11 +167,10 @@ class Item:
return Item(**data)
def get_archive_id(self) -> str | None:
if not self.url:
return None
return get_archive_id(self.url).get("archive_id") if self.url else None
idDict: dict = get_archive_id(self.url)
return idDict.get("archive_id")
def get_extractor(self) -> str | None:
return get_archive_id(self.url).get("ie_key") if self.url else None
def get_ytdlp_opts(self) -> YTDLPOpts:
params: YTDLPOpts = YTDLPOpts.get_instance()
@ -326,6 +325,15 @@ class ItemDTO:
return self.archive_id
def get_extractor(self) -> str | None:
if self.archive_id:
return self.archive_id.split(" ")[0]
idDict: dict[str, str | None] = get_archive_id(self.url)
self.archive_id = idDict.get("archive_id")
return idDict.get("ie_key") if self.url else None
def get_ytdlp_opts(self) -> YTDLPOpts:
"""
Get the yt-dlp options for the item.

6
app/library/Utils.py
View file

@ -87,6 +87,8 @@ DT_PATTERN: re.Pattern[str] = re.compile(r"^(\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}
T = TypeVar("T")
"Generic type variable."
ARCHIVE_IDS_CACHE: dict[str, dict] = {}
"Cache for archive IDs."
class StreamingError(Exception):
"""Raised when an error occurs during streaming."""
@ -1114,6 +1116,9 @@ def get_archive_id(url: str) -> dict[str, str | None]:
"archive_id": None,
}
if url in ARCHIVE_IDS_CACHE:
return ARCHIVE_IDS_CACHE[url]
if YTDLP_INFO_CLS is None:
YTDLP_INFO_CLS = YTDLP(
params={
@ -1146,6 +1151,7 @@ def get_archive_id(url: str) -> dict[str, str | None]:
LOG.exception(e)
LOG.error(f"Error getting archive ID: {e}")
ARCHIVE_IDS_CACHE.update({url: idDict})
return idDict

7
app/library/config.py
View file

@ -69,9 +69,12 @@ class Config(metaclass=Singleton):
base_path: str = "/"
"""The base path to use for the application."""
max_workers: int = 1
max_workers: int = 20
"""The maximum number of workers to use for downloading."""
max_workers_per_extractor: int = 2
"""The maximum number of concurrent downloads per extractor."""
streamer_vcodec: str = ""
"""The video codec to use for streaming. If empty, auto-detect."""
@ -205,6 +208,7 @@ class Config(metaclass=Singleton):
_int_vars: tuple = (
"port",
"max_workers",
"max_workers_per_extractor",
"extract_info_timeout",
"debugpy_port",
"playlist_items_concurrency",
@ -240,6 +244,7 @@ class Config(metaclass=Singleton):
"remove_files",
"ui_update_title",
"max_workers",
"max_workers_per_extractor",
"default_preset",
"instance_title",
"console_enabled",