diff --git a/app/library/DownloadQueue.py b/app/library/DownloadQueue.py index fbd5320d..8407f729 100644 --- a/app/library/DownloadQueue.py +++ b/app/library/DownloadQueue.py @@ -1,4 +1,5 @@ import asyncio +import datetime import json import logging import os @@ -22,6 +23,7 @@ TYPE_QUEUE: str = "queue" class DownloadQueue: + paused: asyncio.Event event: asyncio.Event | None = None pool: AsyncPool | None = None @@ -32,6 +34,8 @@ class DownloadQueue: self.queue = DataStore(type=TYPE_QUEUE, connection=connection) self.done.load() self.queue.load() + self.paused = asyncio.Event() + self.paused.set() async def test(self) -> bool: await self.done.test() @@ -39,12 +43,23 @@ class DownloadQueue: async def initialize(self): self.event = asyncio.Event() - LOG.info(f"Using {self.config.max_workers} workers for downloading.") - - asyncio.create_task( - self.__download_pool() if self.config.max_workers > 1 else self.__download(), - name="download_pool" if self.config.max_workers > 1 else "download_worker", + LOG.info( + f"Using '{self.config.max_workers}' worker/s for downloading. Can be configured via `YTP_MAX_WORKERS` environment variable." ) + asyncio.create_task(self.__download_pool(), name="download_pool") + + def pause(self): + if self.paused.is_set(): + self.paused.clear() + LOG.warning(f"Download paused at. {datetime.datetime.now().isoformat()}") + + def resume(self): + if not self.paused.is_set(): + self.paused.set() + LOG.warning(f"Downloading resumed at. {datetime.datetime.now().isoformat()}") + + def isPaused(self) -> bool: + return False if self.paused.is_set() else True async def __add_entry( self, @@ -61,8 +76,8 @@ class DownloadQueue: options: dict = {} - error: str = None - live_in: str = None + error: str | None = None + live_in: str | None = None eventType = entry.get("_type") or "video" if eventType == "playlist": @@ -109,13 +124,13 @@ class DownloadQueue: f"Entry id '{entry.get('id', None)}' url '{entry.get('webpage_url', None)} - {entry.get('url', None)}'." ) - if self.done.exists(key=entry["id"], url=entry.get("webpage_url") or entry.get("url")): + if self.done.exists(key=entry["id"], url=str(entry.get("webpage_url") or entry.get("url"))): item = self.done.get(key=entry["id"], url=entry.get("webpage_url") or entry["url"]) LOG.warning(f"Item '{item.info.id}' - '{item.info.title}' already downloaded. Removing from history.") await self.clear([item.info._id], remove_file=False) try: - item = self.queue.get(key=entry.get("id"), url=entry.get("webpage_url") or entry.get("url")) + item = self.queue.get(key=str(entry.get("id")), url=str(entry.get("webpage_url") or entry.get("url"))) if item is not None: err_message = f"Item ID '{item.info.id}' - '{item.info.title}' already in download queue." LOG.info(err_message) @@ -127,7 +142,7 @@ class DownloadQueue: options.update({"is_manifestless": is_manifestless}) live_status: list = ["is_live", "is_upcoming"] - is_live = entry.get("is_live", None) or live_in or entry.get("live_status", None) in live_status + is_live = bool(entry.get("is_live", None) or live_in or entry.get("live_status", None) in live_status) try: download_dir = calcDownloadPath(basePath=self.config.download_path, folder=folder) @@ -142,9 +157,9 @@ class DownloadQueue: extras[field] = entry.get(field) dl = ItemDTO( - id=entry.get("id"), - title=entry.get("title"), - url=entry.get("webpage_url") or entry.get("url"), + id=str(entry.get("id")), + title=str(entry.get("title")), + url=str(entry.get("webpage_url") or entry.get("url")), preset=preset, thumbnail=entry.get("thumbnail", None), folder=folder, @@ -164,7 +179,7 @@ class DownloadQueue: for property, value in entry.items(): if property.startswith("playlist"): - dl.output_template = dl.output_template.replace(f"%({property})s", str(value)) + dl.output_template = str(dl.output_template).replace(f"%({property})s", str(value)) dlInfo: Download = Download(info=dl, info_dict=entry, debug=bool(self.config.ytdl_debug)) @@ -180,7 +195,8 @@ class DownloadQueue: else: NotifyEvent = "added" itemDownload = self.queue.put(dlInfo) - self.event.set() + if self.event: + self.event.set() asyncio.create_task( self.emitter.emit(NotifyEvent, itemDownload.info), name=f"notifier-{NotifyEvent}-{itemDownload.info.id}" @@ -189,7 +205,7 @@ class DownloadQueue: return {"status": "ok"} elif eventType.startswith("url"): return await self.add( - url=entry.get("url"), + url=str(entry.get("url")), preset=preset, folder=folder, ytdlp_config=ytdlp_config, @@ -233,7 +249,7 @@ class DownloadQueue: try: downloaded, id_dict = self.isDownloaded(url) - if downloaded is True: + if downloaded is True and id_dict: message = f"This url with ID '{id_dict.get('id')}' has been downloaded already and recorded in archive." LOG.info(message) return {"status": "error", "msg": message} @@ -304,9 +320,11 @@ class DownloadQueue: await item.close() LOG.debug(f"Deleting from queue {itemMessage}") self.queue.delete(id) - asyncio.create_task(self.emitter.canceled(id=id, dl=item), name=f"notifier-c-{id}") + asyncio.create_task(self.emitter.canceled(id=id, dl=item.info.serialize()), name=f"notifier-c-{id}") + item.info.status = "canceled" + item.info.error = "Canceled by user." self.done.put(item) - asyncio.create_task(self.emitter.completed(dl=item), name=f"notifier-d-{id}") + asyncio.create_task(self.emitter.completed(dl=item.info.serialize()), name=f"notifier-d-{id}") LOG.info(f"Deleted from queue {itemMessage}") status[id] = "ok" @@ -330,7 +348,7 @@ class DownloadQueue: filename: str = "" if remove_file and self.config.remove_files and "finished" == item.info.status: - filename = item.info.filename + filename = str(item.info.filename) if item.info.folder: filename = f"{item.info.folder}/{item.info.filename}" @@ -349,7 +367,7 @@ class DownloadQueue: LOG.error(f"Unable to remove '{itemRef}' local file '{filename}'. {str(e)}") self.done.delete(id) - asyncio.create_task(self.emitter.cleared(id, dl=item), name=f"notifier-c-{id}") + asyncio.create_task(self.emitter.cleared(id, dl=item.info.serialize()), name=f"notifier-c-{id}") msg = f"Deleted completed download '{itemRef}'." if fileDeleted and filename: msg += f" and removed local file '{filename}'." @@ -395,16 +413,22 @@ class DownloadQueue: while True: if self.pool.has_open_workers() is True: break - if time.time() - lastLog > 120: + if self.config.max_workers > 1 and time.time() - lastLog > 120: lastLog = time.time() LOG.info(f"Waiting for worker to be free. {self.pool.get_workers_status()}") await asyncio.sleep(1) while not self.queue.hasDownloads(): LOG.info(f"Waiting for item to download. '{self.pool.get_available_workers()}' free workers.") - await self.event.wait() - self.event.clear() - LOG.debug("Cleared wait event.") + if self.event: + await self.event.wait() + self.event.clear() + LOG.debug("Cleared wait event.") + + if self.paused and isinstance(self.paused, asyncio.Event): + LOG.info("Download pool is paused.") + await self.paused.wait() + LOG.info("Download pool resumed downloading.") entry = self.queue.getNextDownload() await asyncio.sleep(0.2) @@ -419,16 +443,6 @@ class DownloadQueue: LOG.debug(f"Pushed {entry=} to executor.") await asyncio.sleep(1) - async def __download(self): - while True: - while self.queue.empty(): - LOG.info("Waiting for item to download.") - await self.event.wait() - self.event.clear() - - id, entry = self.queue.next() - await self.__downloadFile(id, entry) - async def __downloadFile(self, id: str, entry: Download): LOG.info( f"Downloading 'id: {id}', 'Title: {entry.info.title}', 'URL: {entry.info.url}' to 'folder: {entry.info.folder}'." @@ -453,16 +467,17 @@ class DownloadQueue: self.queue.delete(key=id) if entry.is_canceled() is True: - asyncio.create_task(self.emitter.canceled(id, dl=entry.info), name=f"notifier-c-{id}") + asyncio.create_task(self.emitter.canceled(id, dl=entry.info.serialize()), name=f"notifier-c-{id}") entry.info.status = "canceled" entry.info.error = "Canceled by user." self.done.put(value=entry) - asyncio.create_task(self.emitter.completed(dl=entry.info), name=f"notifier-d-{id}") + asyncio.create_task(self.emitter.completed(dl=entry.info.serialize()), name=f"notifier-d-{id}") - self.event.set() + if self.event: + self.event.set() - def isDownloaded(self, url: str) -> tuple[bool, dict[str | None, str | None, str | None]]: + def isDownloaded(self, url: str) -> tuple[bool, dict | None]: if not url or not self.config.keep_archive: return False, None diff --git a/app/library/Emitter.py b/app/library/Emitter.py index d77a684c..e22b0893 100644 --- a/app/library/Emitter.py +++ b/app/library/Emitter.py @@ -9,9 +9,9 @@ class Emitter: This class is used to emit events to the registered emitters. """ - emitters: list[callable] = [] + emitters: list[callable] = [] # type: ignore - def add_emitter(self, emitter: callable): + def add_emitter(self, emitter: callable): # type: ignore """ Add an emitter to the list of emitters. @@ -29,10 +29,10 @@ class Emitter: async def completed(self, dl: dict, **kwargs): await self.emit("completed", dl, **kwargs) - async def canceled(self, id: str, dl: dict = None, **kwargs): + async def canceled(self, id: str, dl: dict | None = None, **kwargs): await self.emit("canceled", id, **kwargs) - async def cleared(self, id: str, dl: dict = None, **kwargs): + async def cleared(self, id: str, dl: dict | None = None, **kwargs): await self.emit("cleared", id, **kwargs) async def error(self, message: str, data: dict = {}, **kwargs): diff --git a/app/library/HttpSocket.py b/app/library/HttpSocket.py index 5a10e8f0..2b3ea60c 100644 --- a/app/library/HttpSocket.py +++ b/app/library/HttpSocket.py @@ -6,6 +6,7 @@ import os import pty import shlex from datetime import datetime +import time import socketio from aiohttp import web @@ -25,8 +26,8 @@ class HttpSocket(common): This class is used to handle WebSocket events. """ - config: Config = None - sio: socketio.AsyncServer = None + config: Config + sio: socketio.AsyncServer def __init__(self, queue: DownloadQueue, emitter: Emitter, encoder: Encoder): super().__init__(queue=queue, encoder=encoder) @@ -39,16 +40,16 @@ class HttpSocket(common): self.queue = queue self.emitter = emitter - def ws_event(func): + def ws_event(func): # type: ignore """ Decorator to mark a method as a socket event. """ - @functools.wraps(func) + @functools.wraps(func) # type: ignore async def wrapper(*args, **kwargs): - return await func(*args, **kwargs) + return await func(*args, **kwargs) # type: ignore - wrapper._ws_event = func.__name__ + wrapper._ws_event = func.__name__ # type: ignore return wrapper def attach(self, app: web.Application): @@ -56,11 +57,10 @@ class HttpSocket(common): for attr_name in dir(self): method = getattr(self, attr_name) - if hasattr(method, "_ws_event"): - event = method._ws_event - self.sio.on(event)(method) + if hasattr(method, "_ws_event") and self.sio: + self.sio.on(method._ws_event)(method) # type: ignore - @ws_event + @ws_event # type: ignore async def cli_post(self, sid: str, data): if not data: await self.emitter.emit("cli_close", {"exitcode": 0}, to=sid) @@ -153,12 +153,12 @@ class HttpSocket(common): ) await self.emitter.emit("cli_close", {"exitcode": -1}, to=sid) - @ws_event + @ws_event # type: ignore async def add_url(self, sid: str, data: dict): - url: str = data.get("url") + url: str | None = data.get("url") if not url: - self.emitter.warning("No URL provided.", to=sid) + await self.emitter.warning("No URL provided.", to=sid) return preset: str = data.get("preset", "default") @@ -180,7 +180,7 @@ class HttpSocket(common): await self.emitter.emit("status", status, to=sid) - @ws_event + @ws_event # type: ignore async def item_cancel(self, sid: str, id: str): if not id: await self.emitter.warning("Invalid request.", to=sid) @@ -192,13 +192,13 @@ class HttpSocket(common): await self.emitter.emit("item_cancel", status) - @ws_event + @ws_event # type: ignore async def item_delete(self, sid: str, data: dict): if not data: await self.emitter.warning("Invalid request.", to=sid) return - id: str = data.get("id") + id: str | None = data.get("id") if not id: await self.emitter.warning("Invalid request.", to=sid) return @@ -209,11 +209,14 @@ class HttpSocket(common): await self.emitter.emit("item_delete", status) - @ws_event + @ws_event # type: ignore async def archive_item(self, sid: str, data: dict): if not isinstance(data, dict) or "url" not in data or not self.config.keep_archive: return + if not isinstance(self.config.ytdl_options, dict): + self.config.ytdl_options = {} + file: str = self.config.ytdl_options.get("download_archive", None) if not file: @@ -242,13 +245,14 @@ class HttpSocket(common): LOG.info(f"Archiving url '{data['url']}' with id '{idDict['archive_id']}'.") - @ws_event + @ws_event # type: ignore async def connect(self, sid: str, _=None): data = { **self.queue.get(), "config": self.config.frontend(), "tasks": self.config.tasks, "presets": self.config.presets, + "paused": self.queue.isPaused(), } # get download folder listing @@ -256,3 +260,13 @@ class HttpSocket(common): data["folders"] = [name for name in os.listdir(downloadPath) if os.path.isdir(os.path.join(downloadPath, name))] await self.emitter.emit("initial_data", data, to=sid) + + @ws_event # type: ignore + async def pause(self, sid: str, _=None): + self.queue.pause() + await self.emitter.emit("paused", {"paused": True, "at": time.time()}) + + @ws_event # type: ignore + async def resume(self, sid: str, _=None): + self.queue.resume() + await self.emitter.emit("paused", {"paused": False, "at": time.time()}) diff --git a/app/library/ItemDTO.py b/app/library/ItemDTO.py index 48f4803b..f7f0da62 100644 --- a/app/library/ItemDTO.py +++ b/app/library/ItemDTO.py @@ -14,7 +14,7 @@ class ItemDTO: _id: str = field(default_factory=lambda: str(uuid.uuid4()), init=False) - error: str = None + error: str|None = None id: str title: str url: str diff --git a/app/library/common.py b/app/library/common.py index 956ece68..80396f3b 100644 --- a/app/library/common.py +++ b/app/library/common.py @@ -11,8 +11,8 @@ class common: This class is used to share common methods between the socket and the API gateways. """ - queue: DownloadQueue = None - encoder: Encoder = None + queue: DownloadQueue + encoder: Encoder def __init__(self, queue: DownloadQueue, encoder: Encoder): super().__init__() diff --git a/app/library/config.py b/app/library/config.py index 43b16537..ce4c1c36 100644 --- a/app/library/config.py +++ b/app/library/config.py @@ -60,7 +60,7 @@ class Config: # immutable config vars. version: str = APP_VERSION __instance = None - ytdl_options: dict | str = {} + ytdl_options: dict = {} tasks: list = [] new_version_available: bool = False ytdlp_version: str = YTDLP_VERSION @@ -159,6 +159,7 @@ class Config: "url_prefix", "remove_files", "ui_update_title", + "max_workers", ) @staticmethod diff --git a/ui/components/Queue.vue b/ui/components/Queue.vue index 3bc0e254..f24db2fb 100644 --- a/ui/components/Queue.vue +++ b/ui/components/Queue.vue @@ -52,7 +52,8 @@