From 2528d3df25bcb2d039408cc632036ca9b7246f25 Mon Sep 17 00:00:00 2001 From: ArabCoders Date: Mon, 30 Dec 2024 10:33:06 +0300 Subject: [PATCH 1/6] attempt to fix title update --- ui/pages/index.vue | 23 +++++++++++++---------- 1 file changed, 13 insertions(+), 10 deletions(-) diff --git a/ui/pages/index.vue b/ui/pages/index.vue index c1c62e22..1bdda7e1 100644 --- a/ui/pages/index.vue +++ b/ui/pages/index.vue @@ -7,19 +7,22 @@ From 1bcad666be9ac1a23a27d897459fa02c4115b4d1 Mon Sep 17 00:00:00 2001 From: ArabCoders Date: Mon, 30 Dec 2024 10:42:02 +0300 Subject: [PATCH 2/6] another attempt at fixing title update --- ui/pages/index.vue | 11 +++++++++-- 1 file changed, 9 insertions(+), 2 deletions(-) diff --git a/ui/pages/index.vue b/ui/pages/index.vue index 1bdda7e1..690dece6 100644 --- a/ui/pages/index.vue +++ b/ui/pages/index.vue @@ -17,12 +17,19 @@ onMounted(() => { useHead({ title: `YTPTube: ( ${Object.keys(stateStore.queue).length || 0} | ${Object.keys(stateStore.history).length || 0} )` }) }) -watch([stateStore.queue, stateStore.history], () => { +watch(() => stateStore.history, () => { if (!config.app.ui_update_title) { return } - console.log('logging watch event') useHead({ title: `YTPTube: ( ${Object.keys(stateStore.queue).length || 0} | ${Object.keys(stateStore.history).length || 0} )` }) }, { deep: true }) +watch(() => stateStore.queue, () => { + if (!config.app.ui_update_title) { + return + } + useHead({ title: `YTPTube: ( ${Object.keys(stateStore.queue).length || 0} | ${Object.keys(stateStore.history).length || 0} )` }) +}, { deep: true }) + + From 448a145ed0bbd71b83e69b9034ff9797d734d964 Mon Sep 17 00:00:00 2001 From: ArabCoders Date: Mon, 30 Dec 2024 10:51:14 +0300 Subject: [PATCH 3/6] Added maxWorkers to title --- app/library/config.py | 1 + ui/pages/index.vue | 8 ++++---- ui/stores/ConfigStore.js | 1 + 3 files changed, 6 insertions(+), 4 deletions(-) diff --git a/app/library/config.py b/app/library/config.py index 43b16537..c7891b90 100644 --- a/app/library/config.py +++ b/app/library/config.py @@ -159,6 +159,7 @@ class Config: "url_prefix", "remove_files", "ui_update_title", + "max_workers", ) @staticmethod diff --git a/ui/pages/index.vue b/ui/pages/index.vue index 690dece6..0d1ac62d 100644 --- a/ui/pages/index.vue +++ b/ui/pages/index.vue @@ -8,28 +8,28 @@ diff --git a/ui/stores/ConfigStore.js b/ui/stores/ConfigStore.js index 93e25ccc..f5ff28f9 100644 --- a/ui/stores/ConfigStore.js +++ b/ui/stores/ConfigStore.js @@ -7,6 +7,7 @@ const CONFIG_KEYS = { ui_update_title: true, output_template: '', ytdlp_version: '', + max_workers: 1, version: '', url_host: '', url_prefix: '', From 0359eafbec93802afbd963ec8830f2e4bc9f191b Mon Sep 17 00:00:00 2001 From: ArabCoders Date: Mon, 30 Dec 2024 11:03:01 +0300 Subject: [PATCH 4/6] fix updating max_workers in UI --- ui/pages/index.vue | 7 +++---- 1 file changed, 3 insertions(+), 4 deletions(-) diff --git a/ui/pages/index.vue b/ui/pages/index.vue index 0d1ac62d..aead7ed2 100644 --- a/ui/pages/index.vue +++ b/ui/pages/index.vue @@ -8,28 +8,27 @@ From ed02c3ec01bdcac0f156890e6456fa06858c645e Mon Sep 17 00:00:00 2001 From: ArabCoders Date: Mon, 30 Dec 2024 21:39:24 +0300 Subject: [PATCH 5/6] Added the ability to pause download pool --- app/library/DownloadQueue.py | 96 ++++++++++++++++++++++-------------- app/library/Emitter.py | 8 +-- app/library/HttpSocket.py | 50 ++++++++++++------- app/library/ItemDTO.py | 2 +- app/library/common.py | 4 +- ui/components/Queue.vue | 17 +++++-- ui/layouts/default.vue | 26 +++++----- ui/stores/ConfigStore.js | 1 + ui/stores/SocketStore.js | 17 +++++++ 9 files changed, 143 insertions(+), 78 deletions(-) diff --git a/app/library/DownloadQueue.py b/app/library/DownloadQueue.py index fbd5320d..62d8bb3b 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} @@ -241,6 +257,9 @@ class DownloadQueue: started = time.perf_counter() LOG.debug(f"extract_info: checking {url=}") + if not isinstance(self.config.ytdl_options, dict): + self.config.ytdl_options = {} + entry = await asyncio.wait_for( fut=asyncio.get_running_loop().run_in_executor( None, @@ -304,9 +323,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 +351,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 +370,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 +416,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 +446,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,17 +470,20 @@ 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() - 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 + if not isinstance(self.config.ytdl_options, dict): + self.config.ytdl_options = {} + return isDownloaded(self.config.ytdl_options.get("download_archive", None), url) 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/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 @@
-
@@ -171,14 +172,18 @@ const hasSelected = computed(() => selectedElms.value.length > 0) const hasQueuedItems = computed(() => stateStore.count('queue') > 0) const setIcon = item => { - if (item.status === 'downloading' && item.is_live) { + if ('downloading' === item.status && item.is_live) { return 'fa-solid fa-globe'; } - if (item.status === 'downloading') { + if ('downloading' === item.status) { return 'fa-solid fa-circle-check'; } + if (null === item.status && true === config.paused) { + return 'fa-solid fa-pause-circle'; + } + return 'fa-solid fa-spinner fa-spin'; } @@ -219,7 +224,11 @@ const percentPipe = value => { const updateProgress = (item) => { let string = ''; - if (item.status == 'preparing') { + if (null === item.status && true === config.paused) { + return 'Paused'; + } + + if ('preparing' === item.status) { return 'Preparing'; } diff --git a/ui/layouts/default.vue b/ui/layouts/default.vue index 9b9d03c6..b68e7e9a 100644 --- a/ui/layouts/default.vue +++ b/ui/layouts/default.vue @@ -11,29 +11,33 @@
+ +
+ + +
+
- - - - Terminal - + +
-
+
- Tasks
diff --git a/ui/stores/ConfigStore.js b/ui/stores/ConfigStore.js index f5ff28f9..07d98d8d 100644 --- a/ui/stores/ConfigStore.js +++ b/ui/stores/ConfigStore.js @@ -22,6 +22,7 @@ const CONFIG_KEYS = { ], folders: [], tasks: [], + paused: false, }; export const useConfigStore = defineStore('config', () => { diff --git a/ui/stores/SocketStore.js b/ui/stores/SocketStore.js index 3f286cb0..dcba21b0 100644 --- a/ui/stores/SocketStore.js +++ b/ui/stores/SocketStore.js @@ -24,6 +24,7 @@ export const useSocketStore = defineStore('socket', () => { tasks: initialData['tasks'], folders: initialData['folders'], presets: initialData['presets'], + paused: Boolean(initialData['paused']) }) stateStore.addAll('queue', initialData['queue'] ?? {}) @@ -89,6 +90,22 @@ export const useSocketStore = defineStore('socket', () => { return; } }); + + socket.value.on('paused', data => { + const json = JSON.parse(data); + const pausedState = Boolean(json.paused); + config.update('paused', pausedState); + + if (false === pausedState) { + toast.success('Download queue resumed.'); + return; + } + + toast.warning('Download queue paused.', { + timeout: 10000, + }); + }); + } const on = (event, callback) => socket.value.on(event, callback); From d40237134c57106cba6a6cb9de0eaa578ddc3ba3 Mon Sep 17 00:00:00 2001 From: ArabCoders Date: Mon, 30 Dec 2024 21:58:01 +0300 Subject: [PATCH 6/6] minor fixes --- app/library/DownloadQueue.py | 9 ++------- app/library/config.py | 2 +- 2 files changed, 3 insertions(+), 8 deletions(-) diff --git a/app/library/DownloadQueue.py b/app/library/DownloadQueue.py index 62d8bb3b..8407f729 100644 --- a/app/library/DownloadQueue.py +++ b/app/library/DownloadQueue.py @@ -257,9 +257,6 @@ class DownloadQueue: started = time.perf_counter() LOG.debug(f"extract_info: checking {url=}") - if not isinstance(self.config.ytdl_options, dict): - self.config.ytdl_options = {} - entry = await asyncio.wait_for( fut=asyncio.get_running_loop().run_in_executor( None, @@ -477,13 +474,11 @@ class DownloadQueue: self.done.put(value=entry) 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 | None]: if not url or not self.config.keep_archive: return False, None - if not isinstance(self.config.ytdl_options, dict): - self.config.ytdl_options = {} - return isDownloaded(self.config.ytdl_options.get("download_archive", None), url) diff --git a/app/library/config.py b/app/library/config.py index c7891b90..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