diff --git a/.vscode/launch.json b/.vscode/launch.json index 35b6d94d..f51bf158 100644 --- a/.vscode/launch.json +++ b/.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", } } ] diff --git a/app/AsyncPool.py b/app/AsyncPool.py index f8e2fafc..58d7bb54 100644 --- a/app/AsyncPool.py +++ b/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)) diff --git a/app/Config.py b/app/Config.py index 4b76444d..1618c753 100644 --- a/app/Config.py +++ b/app/Config.py @@ -36,6 +36,8 @@ class Config: allow_manifestless: bool = False + max_workers: int = 0 + version: str = APP_VERSION _boolean_vars: tuple = ( diff --git a/app/DataStore.py b/app/DataStore.py index 05ce99ce..fc9d3a2b 100644 --- a/app/DataStore.py +++ b/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") diff --git a/app/Download.py b/app/Download.py index 87602755..1897ef68 100644 --- a/app/Download.py +++ b/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( - {k: v for k, v in st.items() if k in self._ytdlp_fields}) + 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': @@ -83,12 +87,12 @@ class Download: ) else: filename = d['info_dict']['filepath'] - self.status_queue.put( - { - 'status': 'finished', - 'filename': filename - } - ) + + self.status_queue.put({ + 'id': self.id, + '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( - {'status': 'finished' if ret == 0 else 'error'} - ) + self.status_queue.put({ + '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: diff --git a/app/DownloadQueue.py b/app/DownloadQueue.py index a2d2bc0c..b78f4f90 100644 --- a/app/DownloadQueue.py +++ b/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( - 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) + await self.__downloadFile(id, entry) def isDownloaded(self, info: dict) -> bool: return self.config.keep_archive and isDownloaded(self.config.ytdl_options.get('download_archive', None), info) diff --git a/frontend/vue.config.js b/frontend/vue.config.js index 54761742..7ddbb1f7 100644 --- a/frontend/vue.config.js +++ b/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 + }) + } })