More code changes to support multi downloads.

app/AsyncPool.py

@@ -143,7 +143,7 @@ class AsyncPool:
         if not self._workers:
             return
 
-        self._logger.debug('Joining {}'.format(self._name))
+        self._logger.info('Joining {}'.format(self._name))
         # The Terminators will kick each worker from being blocked against the _queue.get() and allow
         # each one to exit
         for _ in range(self._num_workers):
@@ -156,7 +156,7 @@ class AsyncPool:
             self._logger.exception('Exception joining {}'.format(self._name))
             raise
         finally:
-            self._logger.debug('Completed {}'.format(self._name))
+            self._logger.info('Completed {}'.format(self._name))
 
         if self._exceptions and self._raise_on_join:
             raise Exception("Exception occurred in pool {}".format(self._name))

app/DataStore.py

@@ -28,7 +28,7 @@ class DataStore:
 
     def load(self) -> None:
         for id, item in self.saved_items():
-            self.dict[id] = Download(
+            self.dict.update({id: Download(
                 info=item,
                 download_dir=calcDownloadPath(
                     basePath=self.config.download_path,
@@ -36,8 +36,8 @@ class DataStore:
                 ),
                 temp_dir=self.config.temp_path,
                 output_template_chapter=self.config.output_template_chapter,
-                default_ytdl_opts=self.config.ytdl_options,
+                default_ytdl_opts=self.config.ytdl_options)
-            )
+            })
 
     def exists(self, key: str = None, url: str = None) -> bool:
         if not key and not url:
@@ -54,13 +54,13 @@ class DataStore:
 
     def get(self, key: str, url: str = None) -> Download:
         if not key and not url:
-            raise KeyError('key or url must be provided')
+            raise KeyError('key or url must be provided.')
 
         for i in self.dict:
             if (key and self.dict[i].info._id == key) or (url and self.dict[i].info.url == url):
                 return self.dict[i]
 
-        raise KeyError('key or url not found')
+        raise KeyError(f'{key=} or {url=} not found.')
 
     def items(self) -> list[tuple[str, Download]]:
         return self.dict.items()
@@ -92,16 +92,13 @@ class DataStore:
         #         value.info._id = key
         #         return
 
-        self.dict[value.info._id] = value
+        self.dict.update({value.info._id: value})
         self._updateStoreItem(self.type, value.info)
 
         return self.dict[value.info._id]
 
     def delete(self, key: str) -> None:
-        if not key in self.dict:
+        self.dict.pop(key, None)
-            return
-
-        del self.dict[key]
         self._deleteStoreItem(key)
 
     def next(self) -> tuple[str, Download]:

app/Download.py

@@ -3,7 +3,6 @@ import json
 import logging
 import multiprocessing
 import os
-import queue
 import re
 import shutil
 import yt_dlp
@@ -75,30 +74,32 @@ class Download:
         self.max_workers = int(config.max_workers)
         self.tempKeep = bool(config.temp_keep)
 
+    def _progress_hook(self, data: dict):
+        dicto = {k: v for k, v in data.items() if k in self._ytdlp_fields}
+        self.status_queue.put({'id': self.id, **dicto})
+
+    def _postprocessor_hook(self, data: dict):
+        if data.get('postprocessor') != 'MoveFiles' or data.get('status') != 'finished':
+            return
+
+        if '__finaldir' in data['info_dict']:
+            filename = os.path.join(
+                data['info_dict']['__finaldir'],
+                os.path.basename(data['info_dict']['filepath'])
+            )
+        else:
+            filename = data['info_dict']['filepath']
+
+        self.status_queue.put({
+            'id': self.id,
+            'status': 'finished',
+            'filename': filename
+        })
+
     def _download(self):
         try:
-            def put_status(st):
-                dicto = {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':
-                    if '__finaldir' in d['info_dict']:
-                        filename = os.path.join(
-                            d['info_dict']['__finaldir'],
-                            os.path.basename(d['info_dict']['filepath'])
-                        )
-                    else:
-                        filename = d['info_dict']['filepath']
-
-                    self.status_queue.put({
-                        'id': self.id,
-                        'status': 'finished',
-                        'filename': filename
-                    })
-
             params: dict = {
-                'no_color': True,
+                'color': 'no_color',
                 'format': self.format,
                 'paths': {
                     'home': self.download_dir,
@@ -108,11 +109,11 @@ class Download:
                     'default': self.output_template,
                     'chapter': self.output_template_chapter
                 },
-                'socket_timeout': 60,
+                'noprogress': True if self.max_workers < 1 else False,
-                'break_per_url': True,
+                'break_on_existing': True,
                 'ignoreerrors': False,
-                'progress_hooks': [put_status],
+                'progress_hooks': [self._progress_hook],
-                'postprocessor_hooks': [put_status_postprocessor],
+                'postprocessor_hooks': [self._postprocessor_hook],
                 **mergeConfig(self.default_ytdl_opts, self.ytdl_opts),
             }
 
@@ -136,7 +137,7 @@ class Download:
                         f'Invalid cookies: was provided for {self.info.title} - {str(e)}')
 
             log.info(
-                f'Downloading id="{self.info.id}" title="{self.info.title}".')
+                f'Downloading {os.getpid()=} id="{self.info.id}" title="{self.info.title}".')
 
             ret = yt_dlp.YoutubeDL(params=params).download([self.info.url])
 
@@ -152,6 +153,9 @@ class Download:
                 'error': str(exc)
             })
 
+        log.info(
+            f'Finished {os.getpid()=} id="{self.info.id}" title="{self.info.title}".')
+
     async def start(self, notifier: Notifier):
         if self.manager is None:
             self.manager = multiprocessing.Manager()

app/DownloadQueue.py

@@ -257,30 +257,39 @@ class DownloadQueue:
 
     async def cancel(self, ids):
         for id in ids:
-            if self.queue.exists(key=id) is False:
+            # if self.queue.exists(key=id) is False:
-                log.warn(f'Requested cancel for non-existent download {id=}')
+            #     log.warn(f'Requested cancel for non-existent download {id=}')
-                continue
+            #     continue
 
-            item = self.queue.get(key=id)
+            try:
+                item = self.queue.get(key=id)
+            except KeyError as e:
+                log.warn(
+                    f'Requested cancel for non-existent download {id=}. {str(e)}')
+                continue
 
             if item.started() is True:
                 log.info(f'Canceling {id=} {item.info.title=}')
                 item.cancel()
             else:
-                self.queue.delete(id)
                 item.close()
-                log.info(f'Deleting {id=} {item.info.title=}')
+                log.info(f'Deleting from queue {id=} {item.info.title=}')
+                self.queue.delete(id)
                 await self.notifier.canceled(id)
 
         return {'status': 'ok'}
 
     async def clear(self, ids):
         for id in ids:
-            if not self.done.exists(key=id):
+
-                log.warn(f'Requested delete for non-existent download {id}')
+            try:
+                item = self.done.get(key=id)
+            except KeyError as e:
+                log.warn(
+                    f'Requested delete for non-existent download {id=}. {str(e)}')
                 continue
-            item = self.done.get(key=id)
+
-            log.info(f'Deleting {id=} {item.info.title=}')
+            log.info(f'Deleting completed download {id=} {item.info.title=}')
             self.done.delete(id)
             await self.notifier.cleared(id)
 
@@ -311,7 +320,7 @@ class DownloadQueue:
             num_workers=self.config.max_workers,
             worker_co=self.__downloadFile,
             name='WorkerPool',
-            logger=logging.getLogger('WorkerPool')
+            logger=log,
         ) as executor:
             while True:
                 while not self.queue.hasDownloads():
@@ -321,8 +330,6 @@ class DownloadQueue:
 
                 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

app/Utils.py

@@ -117,7 +117,7 @@ def get_opts(format: str, quality: str, ytdl_opts: dict) -> dict:
 def getVideoInfo(url: str, ytdlp_opts: dict = None) -> (Any | dict[str, Any] | None):
     params: dict = {
         'quiet': True,
-        'no_color': True,
+        'color': 'no_color',
         'extract_flat': True,
     }
 
@@ -143,7 +143,7 @@ def getAttributes(vclass: str | type) -> dict:
 
 
 def calcDownloadPath(basePath: str, folder: str = None, createPath: bool = True) -> str:
-    """Calculates download path And prevents directory traversal attacks.
+    """Calculates download path and prevents directory traversal.
 
     Returns:
         Dir with base dir factored in.
@@ -166,7 +166,7 @@ def calcDownloadPath(basePath: str, folder: str = None, createPath: bool = True)
 
 def ExtractInfo(config: dict, url: str, debug: bool = False) -> dict:
     params: dict = {
-        'no_color': True,
+        'color': 'no_color',
         'extract_flat': True,
         'skip_download': True,
         'ignoreerrors': True,

app/main.py

@@ -107,6 +107,7 @@ class Main:
             port=self.config.port,
             reuse_port=True,
             loop=self.loop,
+            access_log=None,
             print=lambda _: print(
                 f'YTPTube v{self.config.version} - listening on http://{self.config.host}:{self.config.port}'),
         )

frontend/src/App.vue

@@ -92,7 +92,7 @@ onMounted(() => {
       return
     }
 
-    toast.info('Download canceled: ' + completed[id]?.title);
+    toast.info('Download canceled: ' + downloading[id]?.title);
     delete downloading[id];
   });