From 6e20930af7e1a1181e54efd0ec6cdca4519e8a17 Mon Sep 17 00:00:00 2001 From: ArabCoders Date: Tue, 9 Jan 2024 23:38:55 +0300 Subject: [PATCH] More code changes to support multi downloads. --- app/AsyncPool.py | 4 +-- app/DataStore.py | 17 ++++++------- app/Download.py | 58 +++++++++++++++++++++++--------------------- app/DownloadQueue.py | 33 +++++++++++++++---------- app/Utils.py | 6 ++--- app/main.py | 1 + frontend/src/App.vue | 2 +- 7 files changed, 65 insertions(+), 56 deletions(-) diff --git a/app/AsyncPool.py b/app/AsyncPool.py index 58d7bb54..13c2fded 100644 --- a/app/AsyncPool.py +++ b/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)) diff --git a/app/DataStore.py b/app/DataStore.py index 5e7422da..2e02a9f0 100644 --- a/app/DataStore.py +++ b/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: - return - - del self.dict[key] + self.dict.pop(key, None) self._deleteStoreItem(key) def next(self) -> tuple[str, Download]: diff --git a/app/Download.py b/app/Download.py index 323fc2c9..191f6876 100644 --- a/app/Download.py +++ b/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, - 'break_per_url': True, + 'noprogress': True if self.max_workers < 1 else False, + 'break_on_existing': True, 'ignoreerrors': False, - 'progress_hooks': [put_status], - 'postprocessor_hooks': [put_status_postprocessor], + 'progress_hooks': [self._progress_hook], + '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() diff --git a/app/DownloadQueue.py b/app/DownloadQueue.py index b1362428..568e2582 100644 --- a/app/DownloadQueue.py +++ b/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: - log.warn(f'Requested cancel for non-existent download {id=}') - continue + # if self.queue.exists(key=id) is False: + # log.warn(f'Requested cancel for non-existent download {id=}') + # 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 diff --git a/app/Utils.py b/app/Utils.py index f2436c2f..f5737e2b 100644 --- a/app/Utils.py +++ b/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, diff --git a/app/main.py b/app/main.py index 6e7c4418..2f8fdfe0 100644 --- a/app/main.py +++ b/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}'), ) diff --git a/frontend/src/App.vue b/frontend/src/App.vue index 81d309d7..041cf5f5 100644 --- a/frontend/src/App.vue +++ b/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]; });