diff --git a/app/features/tasks/definitions/handlers/generic.py b/app/features/tasks/definitions/handlers/generic.py index 6f4cfac3..9d94c41a 100644 --- a/app/features/tasks/definitions/handlers/generic.py +++ b/app/features/tasks/definitions/handlers/generic.py @@ -34,7 +34,7 @@ if TYPE_CHECKING: import httpx from parsel.selector import SelectorList -LOG: logging.Logger = logging.getLogger(__name__) +LOG: logging.Logger = logging.getLogger("handlers.generic") CACHE: Cache = Cache() diff --git a/app/features/tasks/definitions/handlers/rss.py b/app/features/tasks/definitions/handlers/rss.py index b7849b8d..f5fc6431 100644 --- a/app/features/tasks/definitions/handlers/rss.py +++ b/app/features/tasks/definitions/handlers/rss.py @@ -14,7 +14,7 @@ from ._base_handler import BaseHandler if TYPE_CHECKING: from xml.etree.ElementTree import Element -LOG: logging.Logger = logging.getLogger(__name__) +LOG: logging.Logger = logging.getLogger("handlers.rss") CACHE: Cache = Cache() diff --git a/app/features/tasks/definitions/handlers/tver.py b/app/features/tasks/definitions/handlers/tver.py index 81c2a4e8..29aa3695 100644 --- a/app/features/tasks/definitions/handlers/tver.py +++ b/app/features/tasks/definitions/handlers/tver.py @@ -6,7 +6,7 @@ from app.library.Utils import get_archive_id from ._base_handler import BaseHandler -LOG: logging.Logger = logging.getLogger(__name__) +LOG: logging.Logger = logging.getLogger("handlers.tver") class TverHandler(BaseHandler): diff --git a/app/features/tasks/definitions/handlers/twitch.py b/app/features/tasks/definitions/handlers/twitch.py index 25714bc2..2da417a0 100644 --- a/app/features/tasks/definitions/handlers/twitch.py +++ b/app/features/tasks/definitions/handlers/twitch.py @@ -11,7 +11,7 @@ from ._base_handler import BaseHandler if TYPE_CHECKING: from xml.etree.ElementTree import Element -LOG: logging.Logger = logging.getLogger(__name__) +LOG: logging.Logger = logging.getLogger("handlers.twitch") class TwitchHandler(BaseHandler): diff --git a/app/features/tasks/definitions/handlers/youtube.py b/app/features/tasks/definitions/handlers/youtube.py index 887ba78a..a963e14a 100644 --- a/app/features/tasks/definitions/handlers/youtube.py +++ b/app/features/tasks/definitions/handlers/youtube.py @@ -11,7 +11,7 @@ from ._base_handler import BaseHandler if TYPE_CHECKING: from xml.etree.ElementTree import Element -LOG: logging.Logger = logging.getLogger(__name__) +LOG: logging.Logger = logging.getLogger("handlers.youtube") class YoutubeHandler(BaseHandler): diff --git a/app/features/tasks/definitions/service.py b/app/features/tasks/definitions/service.py index f7702feb..36968c2d 100644 --- a/app/features/tasks/definitions/service.py +++ b/app/features/tasks/definitions/service.py @@ -22,7 +22,7 @@ if TYPE_CHECKING: from app.library.config import Config from app.library.Scheduler import Scheduler -LOG: logging.Logger = logging.getLogger("tasks.definitions.service") +LOG: logging.Logger = logging.getLogger("definitions.service") class TaskHandle: diff --git a/app/features/tasks/service.py b/app/features/tasks/service.py index 2e4900d7..82d5cd50 100644 --- a/app/features/tasks/service.py +++ b/app/features/tasks/service.py @@ -14,7 +14,7 @@ from app.library.Singleton import Singleton if TYPE_CHECKING: from aiohttp import web -LOG: logging.Logger = logging.getLogger(__name__) +LOG: logging.Logger = logging.getLogger("tasks.service") class Tasks(metaclass=Singleton): diff --git a/app/library/HttpSocket.py b/app/library/HttpSocket.py index 33163277..ceb8264c 100644 --- a/app/library/HttpSocket.py +++ b/app/library/HttpSocket.py @@ -1,3 +1,4 @@ +import asyncio import functools import json import logging @@ -186,10 +187,20 @@ class HttpSocket: await handle_connect(sid) + async def handle_message_safe(sid: str, message: str) -> None: + """Wrapper to handle message with error logging""" + try: + await handle_message(sid, message) + except Exception as e: + LOG.exception(e) + LOG.error(f"Error handling WebSocket message from client '{sid}': {e}") + try: + i: int = 0 async for msg in ws: if msg.type == web.WSMsgType.TEXT: - await handle_message(sid, msg.data) + i = i + 1 + asyncio.create_task(handle_message_safe(sid, msg.data), name=f"ws_msg_{sid}_{i}") elif msg.type == web.WSMsgType.ERROR: LOG.error(f"WebSocket connection closed with exception {ws.exception()}") finally: diff --git a/app/library/config.py b/app/library/config.py index c139ed42..3074e410 100644 --- a/app/library/config.py +++ b/app/library/config.py @@ -207,9 +207,6 @@ 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.""" @@ -268,7 +265,6 @@ 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 2a7006d3..0f3b638f 100644 --- a/app/library/downloads/playlist_processor.py +++ b/app/library/downloads/playlist_processor.py @@ -1,13 +1,10 @@ """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 @@ -59,87 +56,54 @@ async def process_playlist( "n_entries": len(entries), } - 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}") + 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}'.") - LOG.info(f"Processing '{item_name}'.") + _status, _msg = ytdlp_reject(entry=etr, yt_params=yt_params) + if not _status: + return {"status": "error", "msg": _msg} - _status, _msg = ytdlp_reject(entry=etr, yt_params=yt_params) - if not _status: - return {"status": "error", "msg": _msg} + extras: dict[str, Any] = { + **playlist_keys, + "playlist_index": i, + "playlist_index_number": i, + "playlist_autonumber": i, + } - extras: dict[str, Any] = { - **playlist_keys, - "playlist_index": i, - "playlist_index_number": i, - "playlist_autonumber": i, - } + for property in ("id", "title", "uploader", "uploader_id"): + if property in entry: + extras[f"playlist_{property}"] = entry.get(property) - for property in ("id", "title", "uploader", "uploader_id"): - if property in entry: - extras[f"playlist_{property}"] = entry.get(property) + 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) - 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) + newItem: Item = item.new_with(url=etr.get("url") or etr.get("webpage_url"), extras=extras) - newItem: Item = item.new_with(url=etr.get("url") or etr.get("webpage_url"), extras=extras) + 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) - 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() + return await queue.add(item=newItem, already=already) 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") - 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]] = [] - + results: list[dict[str, str]] = [] for i, etr in enumerate(entries, start=1): if max_downloads > 0 and i > max_downloads: break - 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)) + results.append(await process_item(i, etr)) 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 9f75c173..1230d208 100644 --- a/app/library/downloads/queue_manager.py +++ b/app/library/downloads/queue_manager.py @@ -1,4 +1,3 @@ -import asyncio import functools import glob import logging @@ -42,8 +41,6 @@ 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."