initial code to support multi downloads.

.vscode/launch.json

@@ -34,9 +34,9 @@
         "YTP_TEMP_PATH": "${workspaceFolder}/var/tmp",
         "YTP_YTDL_OPTIONS_FILE": "${workspaceFolder}/var/config/ytdlp.json",
         "YTP_URL_HOST": "http://localhost:8081",
-        "YTP_YTDL_DEBUG": "true",
+        "YTP_YTDL_DEBUG": "false",
         "YTP_TEMP_KEEP": "true",
-        "YTP_KEEP_ARCHIVE": "true",
+        "YTP_KEEP_ARCHIVE": "false",
       }
     }
   ]

app/AsyncPool.py

@@ -32,7 +32,7 @@ class AsyncPool:
         @param expected_total: (optional) expected total number of jobs (used for `log_event_n` logging)
         @return: instance of AsyncWorkerPool
         """
-        loop = loop or asyncio.get_event_loop()
+        loop = loop if loop else asyncio.get_event_loop()
         self._loop = loop
         self._num_workers = num_workers
         self._logger = logger
@@ -65,7 +65,7 @@ class AsyncPool:
                 future, args, kwargs = item
                 # the wait_for will cancel the task (task sees CancelledError) and raises a TimeoutError from here
                 # so be wary of catching TimeoutErrors in this loop
-                result = await asyncio.wait_for(self._worker_co(*args, **kwargs), self._max_task_time, loop=self._loop)
+                result = await asyncio.wait_for(self._worker_co(*args, **kwargs), self._max_task_time)
 
                 if future:
                     future.set_result(result)
@@ -94,6 +94,12 @@ class AsyncPool:
     def total_queued(self):
         return self._total_queued
 
+    def has_open_workers(self) -> bool:
+        """
+        :return: True if there are open workers.
+        """
+        return self._queue.qsize() < self._num_workers
+
     async def __aenter__(self):
         self.start()
         return self
@@ -144,7 +150,7 @@ class AsyncPool:
             await self._queue.put(Terminator())
 
         try:
-            await asyncio.gather(*self._workers, loop=self._loop)
+            await asyncio.gather(*self._workers)
             self._workers = None
         except:
             self._logger.exception('Exception joining {}'.format(self._name))

app/Config.py

@@ -36,6 +36,8 @@ class Config:
 
     allow_manifestless: bool = False
 
+    max_workers: int = 0
+
     version: str = APP_VERSION
 
     _boolean_vars: tuple = (

app/DataStore.py

@@ -114,6 +114,16 @@ class DataStore:
     def empty(self):
         return not bool(self.dict)
 
+    def hasDownloads(self):
+        if 0 == len(self.dict):
+            return False
+
+        for key in self.dict:
+            if self.dict[key].started() is False:
+                return True
+
+        return False
+
     def _updateStoreItem(self, type: str, item: ItemDTO) -> None:
         sqlStatement = """
         INSERT INTO "history" ("id", "type", "url", "data")

app/Download.py

@@ -1,9 +1,9 @@
-
 import asyncio
 import json
 import logging
 import multiprocessing
 import os
+import queue
 import re
 import shutil
 import yt_dlp
@@ -15,6 +15,7 @@ log = logging.getLogger('download')
 
 
 class Download:
+    id: str = None
     manager = None
     download_dir: str = None
     temp_dir: str = None
@@ -57,6 +58,7 @@ class Download:
         self.ytdl_opts = get_opts(
             info.format, info.quality, info.ytdlp_config if info.ytdlp_config else {})
         self.info = info
+        self.id = info._id
         self.default_ytdl_opts = default_ytdl_opts
         self.debug = debug
 
@@ -66,13 +68,15 @@ class Download:
         self.proc = None
         self.loop = None
         self.notifier = None
-        self.tempKeep = bool(Config.get_instance().temp_keep)
+        config = Config.get_instance()
+        self.max_workers = int(config.max_workers)
+        self.tempKeep = bool(config.temp_keep)
 
     def _download(self):
         try:
             def put_status(st):
-                self.status_queue.put(
+                dicto = {k: v for k, v in st.items() if k in self._ytdlp_fields}
-                    {k: v for k, v in st.items() if k in self._ytdlp_fields})
+                self.status_queue.put({'id': self.id, **dicto})
 
             def put_status_postprocessor(d):
                 if d['postprocessor'] == 'MoveFiles' and d['status'] == 'finished':
@@ -83,12 +87,12 @@ class Download:
                         )
                     else:
                         filename = d['info_dict']['filepath']
-                    self.status_queue.put(
+
-                        {
+                    self.status_queue.put({
-                            'status': 'finished',
+                        'id': self.id,
-                            'filename': filename
+                        'status': 'finished',
-                        }
+                        'filename': filename
-                    )
+                    })
 
             # Create temp dir for each download.
             self.tempPath = os.path.join(
@@ -136,13 +140,16 @@ class Download:
 
             log.info(
                 f'Downloading id="{self.info.id}" title="{self.info.title}".')
+
             ret = yt_dlp.YoutubeDL(params=params).download([self.info.url])
 
-            self.status_queue.put(
+            self.status_queue.put({
-                {'status': 'finished' if ret == 0 else 'error'}
+                'id': self.id,
-            )
+                'status': 'finished' if ret == 0 else 'error'
+            })
         except Exception as exc:
             self.status_queue.put({
+                'id': self.id,
                 'status': 'error',
                 'msg': str(exc),
                 'error': str(exc)
@@ -179,15 +186,16 @@ class Download:
     def close(self):
         if self.started():
             self.proc.close()
-            self.status_queue.put(None)
+            if self.max_workers < 1:
+                self.status_queue.put(None)
 
-    def running(self):
+    def running(self) -> bool:
         try:
             return self.proc is not None and self.proc.is_alive()
         except ValueError:
             return False
 
-    def started(self):
+    def started(self) -> bool:
         return self.proc is not None
 
     async def progress_update(self):
@@ -196,7 +204,7 @@ class Download:
         """
         while True:
             status = await self.loop.run_in_executor(None, self.status_queue.get)
-            if not status:
+            if not status or status.get('id') != self.id:
                 return
 
             if self.debug:

app/DownloadQueue.py

@@ -10,8 +10,11 @@ from Download import Download
 from ItemDTO import ItemDTO
 from DataStore import DataStore
 from Utils import Notifier, ObjectSerializer, calcDownloadPath, ExtractInfo, isDownloaded, mergeConfig
+from AsyncPool import AsyncPool
+from concurrent.futures import ProcessPoolExecutor
 
 log = logging.getLogger('DownloadQueue')
+dl_queue = asyncio.Queue()
 
 
 class DownloadQueue:
@@ -35,7 +38,15 @@ class DownloadQueue:
 
     async def initialize(self):
         self.event = asyncio.Event()
-        asyncio.create_task(self.__download())
+        if self.config.max_workers > 0:
+            log.info(
+                f'Using {self.config.max_workers} workers for downloading.')
+        else:
+            log.info('Using single threaded download.')
+
+        asyncio.create_task(
+            self.__download_pool() if self.config.max_workers > 0 else self.__download()
+        )
 
     async def __add_entry(
         self,
@@ -252,9 +263,9 @@ class DownloadQueue:
 
             item = self.queue.get(key=id)
 
-            if self.queue.get(id).started():
+            if item.started():
                 log.info(f'Canceling {id=} {item.info.title=}')
-                self.queue.get(id).cancel()
+                item.cancel()
             else:
                 self.queue.delete(id)
                 log.info(f'Deleting {id=} {item.info.title=}')
@@ -293,6 +304,59 @@ class DownloadQueue:
 
         return items
 
+    async def __download_pool(self):
+        async with AsyncPool(
+            loop=asyncio.get_running_loop(),
+            num_workers=self.config.max_workers,
+            worker_co=self.__downloadFile,
+            name='WorkerPool',
+            logger=logging.getLogger('WorkerPool')
+        ) as executor:
+            while True:
+                while not self.queue.hasDownloads():
+                    log.info('waiting for item to download.')
+                    await self.event.wait()
+                    self.event.clear()
+
+                while True:
+                    if executor.has_open_workers() is False:
+                        log.info(
+                            f'Waiting for workers to finish. {executor.total_queued} items in queue.')
+                        await asyncio.sleep(1)
+                    else:
+                        break
+
+                for id in self.queue.dict:
+                    entry = self.queue.get(key=id)
+                    if entry.started() is False and entry.canceled is False:
+                        await executor.push(id=id, entry=entry)
+                        await asyncio.sleep(1)
+
+    async def __downloadFile(self, id: str, entry: Download):
+        log.info(
+            f'Queuing {id=} - {entry.info.title=} - {entry.info.url=} - {entry.info.folder=}.')
+
+        await entry.start(self.notifier)
+
+        if entry.info.status != 'finished':
+            if entry.tmpfilename and os.path.isfile(entry.tmpfilename):
+                try:
+                    os.remove(entry.tmpfilename)
+                except:
+                    pass
+
+            entry.info.status = 'error'
+
+        entry.close()
+
+        if self.queue.exists(id):
+            self.queue.delete(id)
+            if entry.canceled:
+                await self.notifier.canceled(id)
+            else:
+                self.done.put(entry)
+                await self.notifier.completed(entry.info)
+
     async def __download(self):
         while True:
             while self.queue.empty():
@@ -301,29 +365,7 @@ class DownloadQueue:
                 self.event.clear()
 
             id, entry = self.queue.next()
-            log.info(
+            await self.__downloadFile(id, entry)
-                f'Queuing {id=} - {entry.info.title=} - {entry.info.url=} - {entry.info.folder=}.')
-
-            await entry.start(self.notifier)
-
-            if entry.info.status != 'finished':
-                if entry.tmpfilename and os.path.isfile(entry.tmpfilename):
-                    try:
-                        os.remove(entry.tmpfilename)
-                    except:
-                        pass
-
-                entry.info.status = 'error'
-
-            entry.close()
-
-            if self.queue.exists(id):
-                self.queue.delete(id)
-                if entry.canceled:
-                    await self.notifier.canceled(id)
-                else:
-                    self.done.put(entry)
-                    await self.notifier.completed(entry.info)
 
     def isDownloaded(self, info: dict) -> bool:
         return self.config.keep_archive and isDownloaded(self.config.ytdl_options.get('download_archive', None), info)

frontend/vue.config.js

@@ -1,5 +1,15 @@
 const { defineConfig } = require('@vue/cli-service')
 module.exports = defineConfig({
   publicPath: './',
-  transpileDependencies: true
+  transpileDependencies: true,
+  chainWebpack: (config) => {
+    config.plugin('define').tap((definitions) => {
+      Object.assign(definitions[0], {
+        __VUE_OPTIONS_API__: 'true',
+        __VUE_PROD_DEVTOOLS__: 'false',
+        __VUE_PROD_HYDRATION_MISMATCH_DETAILS__: 'false'
+      })
+      return definitions
+    })
+  }
 })