Merge pull request #155 from arabcoders/dev

Added the ability to pause download pool

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.")
+        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() if self.config.max_workers > 1 else self.__download(),
-            name="download_pool" if self.config.max_workers > 1 else "download_worker",
         )
+        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
+        error: str | None = None
-        live_in: str = 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"),
+                id=str(entry.get("id")),
-                title=entry.get("title"),
+                title=str(entry.get("title")),
-                url=entry.get("webpage_url") or entry.get("url"),
+                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()
+                if self.event:
-                self.event.clear()
+                    await self.event.wait()
-                LOG.debug("Cleared wait event.")
+                    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
 

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):

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
+    config: Config
-    sio: socketio.AsyncServer = None
+    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"):
+            if hasattr(method, "_ws_event") and self.sio:
-                event = method._ws_event
+                self.sio.on(method._ws_event)(method)  # type: ignore
-                self.sio.on(event)(method)
 
-    @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()})

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

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
+    queue: DownloadQueue
-    encoder: Encoder = None
+    encoder: Encoder
 
     def __init__(self, queue: DownloadQueue, encoder: Encoder):
         super().__init__()

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

ui/components/Queue.vue

@@ -52,7 +52,8 @@
           <div v-if="false === hideThumbnail" class="card-image">
             <figure class="image is-3by1" v-if="item.extras?.thumbnail">
               <NuxtLink v-tooltip="item.title" :href="item.url" target="_blank">
-                <img :src="config.app.url_host + config.app.url_prefix + 'thumbnail?url=' + encodePath(item.extras.thumbnail)"
+                <img
+                  :src="config.app.url_host + config.app.url_prefix + 'thumbnail?url=' + encodePath(item.extras.thumbnail)"
                   :alt="item.title" />
               </NuxtLink>
             </figure>
@@ -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';
   }
 

ui/layouts/default.vue

@@ -11,29 +11,33 @@
         </NuxtLink>
       </div>
       <div class="navbar-end is-flex">
+
+        <div class="navbar-item" v-if="socket.isConnected">
+          <button class="button is-dark" @click="socket.emit('pause', {})" v-if="false === config.paused"
+            v-tooltip="'Pause non-active downloads.'">
+            <span class="icon has-text-warning"><i class="fas fa-pause"></i></span>
+          </button>
+          <button class="button is-dark" @click="socket.emit('resume', {})" v-else v-tooltip="'Resume downloading.'">
+            <span class="icon has-text-success"><i class="fas fa-play"></i></span>
+          </button>
+        </div>
+
         <div class="navbar-item">
-          <NuxtLink class="button is-dark has-tooltip-bottom" to="/console">
+          <NuxtLink class="button is-dark has-tooltip-bottom" to="/console" v-tooltip.bottom="'Terminal'">
-            <span class="icon-text">
+            <span class="icon"><i class="fa-solid fa-terminal" /></span>
-              <span class="icon"><i class="fa-solid fa-terminal" /></span>
-              <span>Terminal</span>
-            </span>
           </NuxtLink>
         </div>
 
         <div class="navbar-item">
           <button v-tooltip.bottom="'Toggle Add Form'" class="button is-dark has-tooltip-bottom"
             @click="config.showForm = !config.showForm">
-            <span class="icon-text">
+            <span class="icon"><i class="fa-solid fa-plus" /></span>
-              <span class="icon"><i class="fa-solid fa-plus" /></span>
-              <span>Add</span>
-            </span>
           </button>
         </div>
 
-        <div class="navbar-item" v-if="config.tasks.length > 0">
+        <div class="navbar-item" v-if="config.tasks.length > 0" v-tooltip.bottom="'Tasks'">
           <NuxtLink class="button is-dark has-tooltip-bottom" to="/tasks">
             <span class="icon"><i class="fa-solid fa-tasks" /></span>
-            <span class="is-hidden-mobile">Tasks</span>
           </NuxtLink>
         </div>
 

ui/pages/index.vue

@@ -7,19 +7,28 @@
 
 <script setup>
 const config = useConfigStore()
-useHead({ title: 'YTPTube' })
+const stateStore = useStateStore()
 
-watch(() => config.app.ui_update_title, value => {
+onMounted(() => {
-  if (true !== value) {
+  if (!config.app.ui_update_title) {
+    useHead({ title: 'YTPTube' })
     return
   }
+  useHead({ title: `YTPTube: ( ${Object.keys(stateStore.queue).length || 0}/${config.app.max_workers} | ${Object.keys(stateStore.history).length || 0} )` })
+})
 
-  const s = useStateStore()
+watch(() => stateStore.history, () => {
-  useHead({ title: `YTPTube: ( ${Object.keys(s.queue).length || 0} | ${Object.keys(s.history).length || 0} )` })
+  if (!config.app.ui_update_title) {
-  watch([s.queue, s.history], () => {
+    return
-    const title = `YTPTube: ( ${Object.keys(s.queue).length || 0} | ${Object.keys(s.history).length || 0} )`
+  }
-    useHead({ title })
+  useHead({ title: `YTPTube: ( ${Object.keys(stateStore.queue).length || 0}/${config.app.max_workers}  | ${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}/${config.app.max_workers}  | ${Object.keys(stateStore.history).length || 0} )` })
+}, { deep: true })
 
-}, { immediate: true })
 </script>

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: '',
@@ -21,6 +22,7 @@ const CONFIG_KEYS = {
   ],
   folders: [],
   tasks: [],
+  paused: false,
 };
 
 export const useConfigStore = defineStore('config', () => {

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);