diff --git a/app/library/config.py b/app/library/config.py index 3074e410..c139ed42 100644 --- a/app/library/config.py +++ b/app/library/config.py @@ -207,6 +207,9 @@ class Config(metaclass=Singleton): live_premiere_buffer: int = 5 """The buffer time in minutes to add to video duration to wait before starting premiere download.""" + playlist_items_concurrency: int = 4 + """The number of concurrent playlist items to be processed at same time.""" + auto_clear_history_days: int = 0 """Number of days after which completed download history is automatically cleared. 0 to disable.""" @@ -265,6 +268,7 @@ class Config(metaclass=Singleton): "max_workers_per_extractor", "extract_info_timeout", "debugpy_port", + "playlist_items_concurrency", "download_path_depth", "download_info_expires", "auto_clear_history_days", diff --git a/app/library/downloads/playlist_processor.py b/app/library/downloads/playlist_processor.py index 0f3b638f..2a7006d3 100644 --- a/app/library/downloads/playlist_processor.py +++ b/app/library/downloads/playlist_processor.py @@ -1,10 +1,13 @@ """Playlist processing.""" +import asyncio import logging from typing import TYPE_CHECKING, Any from app.library.Utils import merge_dict, ytdlp_reject +from .utils import handle_task_exception + if TYPE_CHECKING: from app.library.ItemDTO import Item @@ -56,54 +59,87 @@ async def process_playlist( "n_entries": len(entries), } - async def process_item(i: int, etr: dict) -> dict[str, str]: - """Process a single playlist item.""" - item_name: str = ( - f"'{entry.get('title')}: {i}/{playlist_keys['n_entries']}' - '{etr.get('id')}: {etr.get('title')}'" - ) - LOG.info(f"Processing '{item_name}'.") + async def playlist_processor(i: int, etr: dict): + acquired = False + try: + item_name: str = ( + f"'{entry.get('title')}: {i}/{playlist_keys['n_entries']}' - '{etr.get('id')}: {etr.get('title')}'" + ) + LOG.debug(f"Waiting to acquire lock for {item_name}") + await queue.processors.acquire() + acquired = True + LOG.debug(f"Acquired lock for {item_name}") - _status, _msg = ytdlp_reject(entry=etr, yt_params=yt_params) - if not _status: - return {"status": "error", "msg": _msg} + LOG.info(f"Processing '{item_name}'.") - extras: dict[str, Any] = { - **playlist_keys, - "playlist_index": i, - "playlist_index_number": i, - "playlist_autonumber": i, - } + _status, _msg = ytdlp_reject(entry=etr, yt_params=yt_params) + if not _status: + return {"status": "error", "msg": _msg} - for property in ("id", "title", "uploader", "uploader_id"): - if property in entry: - extras[f"playlist_{property}"] = entry.get(property) + extras: dict[str, Any] = { + **playlist_keys, + "playlist_index": i, + "playlist_index_number": i, + "playlist_autonumber": i, + } - extractor_key = entry.get("ie_key") or entry.get("extractor_key") or entry.get("extractor") or "" - if "thumbnail" not in etr and "youtube" in str(extractor_key).lower(): - extras["thumbnail"] = "https://img.youtube.com/vi/{id}/maxresdefault.jpg".format(**etr) + for property in ("id", "title", "uploader", "uploader_id"): + if property in entry: + extras[f"playlist_{property}"] = entry.get(property) - newItem: Item = item.new_with(url=etr.get("url") or etr.get("webpage_url"), extras=extras) + extractor_key = entry.get("ie_key") or entry.get("extractor_key") or entry.get("extractor") or "" + if "thumbnail" not in etr and "youtube" in str(extractor_key).lower(): + extras["thumbnail"] = "https://img.youtube.com/vi/{id}/maxresdefault.jpg".format(**etr) - if ("video" == etr.get("_type") and etr.get("url")) or ( - "formats" in etr and isinstance(etr["formats"], list) and len(etr["formats"]) > 0 - ): - dct = merge_dict(merge_dict({"_type": "video"}, etr), entry) - dct.pop("entries", None) - return await queue.add(item=newItem, entry=dct, already=already) + newItem: Item = item.new_with(url=etr.get("url") or etr.get("webpage_url"), extras=extras) - return await queue.add(item=newItem, already=already) + if ("video" == etr.get("_type") and etr.get("url")) or ( + "formats" in etr and isinstance(etr["formats"], list) and len(etr["formats"]) > 0 + ): + dct = merge_dict(merge_dict({"_type": "video"}, etr), entry) + dct.pop("entries", None) + return await queue.add(item=newItem, entry=dct, already=already) + + return await queue.add(item=newItem, already=already) + finally: + if acquired: + queue.processors.release() max_downloads: int = -1 ytdlp_opts: dict[str, Any] = item.get_ytdlp_opts().get_all() if ytdlp_opts.get("max_downloads") and isinstance(ytdlp_opts.get("max_downloads"), int): max_downloads: int = ytdlp_opts.get("max_downloads") - results: list[dict[str, str]] = [] + batch_size: int = max(50, int(queue.config.playlist_items_concurrency) * 10) + + async def run_batch(batch: list[tuple[int, dict]]) -> list[dict]: + tasks: list[asyncio.Task] = [] + for i, etr in batch: + task = asyncio.create_task( + playlist_processor(i, etr), + name=f"playlist_processor_{etr.get('id')}_{i}", + ) + task.add_done_callback(lambda t: handle_task_exception(t, LOG)) + tasks.append(task) + + batch_results: list[dict] = await asyncio.gather(*tasks) + await asyncio.sleep(0) + return batch_results + + results: list[dict] = [] + batch: list[tuple[int, dict]] = [] + for i, etr in enumerate(entries, start=1): if max_downloads > 0 and i > max_downloads: break - results.append(await process_item(i, etr)) + batch.append((i, etr)) + if len(batch) >= batch_size: + results.extend(await run_batch(batch)) + batch.clear() + + if batch: + results.extend(await run_batch(batch)) log_msg: str = f"Playlist '{playlist_name}' processing completed with '{len(results)}' entries." if max_downloads > 0 and len(entries) > max_downloads: diff --git a/app/library/downloads/queue_manager.py b/app/library/downloads/queue_manager.py index 1230d208..9f75c173 100644 --- a/app/library/downloads/queue_manager.py +++ b/app/library/downloads/queue_manager.py @@ -1,3 +1,4 @@ +import asyncio import functools import glob import logging @@ -41,6 +42,8 @@ class DownloadQueue(metaclass=Singleton): "DataStore for the completed downloads." self.queue = DataStore(type=StoreType.QUEUE, connection=SqliteStore.get_instance()) "DataStore for the download queue." + self.processors = asyncio.Semaphore(self.config.playlist_items_concurrency) + "Semaphore to limit the number of concurrent processors." self.pool = PoolManager(queue=self, config=self.config) "Pool manager for coordinating download execution."