From c932c8deb6f0e36c2c0c0d8571eb7d8905017b09 Mon Sep 17 00:00:00 2001 From: ArabCoders Date: Tue, 9 Jan 2024 00:17:26 +0300 Subject: [PATCH] more changes to support multi downloads. --- .vscode/launch.json | 3 ++- app/Config.py | 6 +++++ app/DataStore.py | 33 ++++++++++++----------- app/Download.py | 62 ++++++++++++++++++++++++++------------------ app/DownloadQueue.py | 24 +++++++++-------- 5 files changed, 76 insertions(+), 52 deletions(-) diff --git a/.vscode/launch.json b/.vscode/launch.json index f51bf158..07ad7404 100644 --- a/.vscode/launch.json +++ b/.vscode/launch.json @@ -35,8 +35,9 @@ "YTP_YTDL_OPTIONS_FILE": "${workspaceFolder}/var/config/ytdlp.json", "YTP_URL_HOST": "http://localhost:8081", "YTP_YTDL_DEBUG": "false", - "YTP_TEMP_KEEP": "true", + "YTP_TEMP_KEEP": "false", "YTP_KEEP_ARCHIVE": "false", + "YTP_MAX_WORKERS": "0", } } ] diff --git a/app/Config.py b/app/Config.py index 1618c753..884d0963 100644 --- a/app/Config.py +++ b/app/Config.py @@ -44,6 +44,7 @@ class Config: 'keep_archive', 'ytdl_debug', 'temp_keep', 'allow_manifestless', ) + _int_vars: tuple = ('port', 'max_workers',) _immutable: tuple = ('version', '__instance', 'ytdl_options',) @staticmethod @@ -102,6 +103,9 @@ class Config: setattr(self, k, str(v).lower() in (True, 'true', 'on', '1')) + if k in self._int_vars: + setattr(self, k, int(v)) + if not self.url_prefix.endswith('/'): self.url_prefix += '/' @@ -142,6 +146,8 @@ class Config: self.ytdl_options['download_archive'] = os.path.join( self.config_path, 'archive.log') + log.info(f'Keep temp: {self.temp_keep}') + def _getAttributes(self) -> dict: attrs: dict = {} vclass: str = self.__class__ diff --git a/app/DataStore.py b/app/DataStore.py index fc9d3a2b..5e7422da 100644 --- a/app/DataStore.py +++ b/app/DataStore.py @@ -3,6 +3,7 @@ import copy from datetime import datetime, timezone from email.utils import formatdate import json +import logging import sqlite3 from Utils import calcDownloadPath from Config import Config @@ -45,10 +46,9 @@ class DataStore: if key and key in self.dict: return True - if url: - for key in self.dict: - if self.dict[key].info.url == url: - return True + for i in self.dict: + if (key and self.dict[i].info._id == key) or (url and self.dict[i].info.url == url): + return True return False @@ -56,13 +56,9 @@ class DataStore: if not key and not url: raise KeyError('key or url must be provided') - if key and key in self.dict: - return self.dict[key] - - if url: - for key in self.dict: - if self.dict[key].info.url == url: - return self.dict[key] + 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') @@ -91,10 +87,10 @@ class DataStore: return items def put(self, value: Download) -> Download: - for key in self.dict: - if self.dict[key].info.url == value.info.url: - value.info._id = key - return + # for key in self.dict: + # if self.dict[key].info.url == value.info.url: + # value.info._id = key + # return self.dict[value.info._id] = value self._updateStoreItem(self.type, value.info) @@ -124,6 +120,13 @@ class DataStore: return False + def getNextDownload(self) -> Download: + for key in self.dict: + if self.dict[key].started() is False and self.dict[key].is_canceled() is False: + return self.dict[key] + + return None + 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 1897ef68..d4e8246c 100644 --- a/app/Download.py +++ b/app/Download.py @@ -28,6 +28,8 @@ class Download: debug: bool = False tempPath: str = None notifier: Notifier = None + canceled: bool = False + _ytdlp_fields: tuple = ( 'tmpfilename', 'filename', @@ -94,12 +96,6 @@ class Download: 'filename': filename }) - # Create temp dir for each download. - self.tempPath = os.path.join( - self.temp_dir, self.info.id if self.info.id else self.info._id) - if not os.path.exists(self.tempPath): - os.makedirs(self.tempPath, exist_ok=True) - params: dict = { 'no_color': True, 'format': self.format, @@ -154,14 +150,6 @@ class Download: 'msg': str(exc), 'error': str(exc) }) - finally: - if self.tempKeep is False and self.tempPath and os.path.exists(self.tempPath): - if self.tempPath == self.temp_dir: - log.warning( - f'Attempted to delete video temp directory: {self.tempPath}, but it is the same as main temp directory.') - else: - log.debug(f'Deleting Temp directory: {self.tempPath}') - shutil.rmtree(self.tempPath, ignore_errors=True) async def start(self, notifier: Notifier): if self.manager is None: @@ -171,6 +159,11 @@ class Download: self.loop = asyncio.get_running_loop() self.notifier = notifier + # Create temp dir for each download. + self.tempPath = os.path.join(self.temp_dir, self.info._id) + if not os.path.exists(self.tempPath): + os.makedirs(self.tempPath, exist_ok=True) + self.proc = multiprocessing.Process(target=self._download) self.proc.start() self.info.status = 'preparing' @@ -178,25 +171,44 @@ class Download: asyncio.create_task(self.progress_update()) return await self.loop.run_in_executor(None, self.proc.join) + def started(self) -> bool: + return self.proc is not None + def cancel(self): - if self.running(): - self.proc.kill() + self.kill() self.canceled = True def close(self): if self.started(): self.proc.close() - if self.max_workers < 1: - self.status_queue.put(None) + + self.delete_temp() def running(self) -> bool: - try: - return self.proc is not None and self.proc.is_alive() - except ValueError: - return False + return self.started() and self.proc.is_alive() - def started(self) -> bool: - return self.proc is not None + def is_canceled(self) -> bool: + return self.canceled + + def kill(self): + if self.running(): + log.info(f'Killing download process: {self.proc.ident}') + self.proc.kill() + + def delete_temp(self): + if self.tempKeep is True or not self.tempPath: + return + + if not os.path.exists(self.tempPath): + return + + if self.tempPath == self.temp_dir: + log.warning( + f'Attempted to delete video temp directory: {self.tempPath}, but it is the same as main temp directory.') + return + + log.info(f'Deleting Temp directory: {self.tempPath}') + shutil.rmtree(self.tempPath, ignore_errors=True) async def progress_update(self): """ @@ -204,7 +216,7 @@ class Download: """ while True: status = await self.loop.run_in_executor(None, self.status_queue.get) - if not status or status.get('id') != self.id: + if status.get('id') != self.id or len(status) < 2: return if self.debug: diff --git a/app/DownloadQueue.py b/app/DownloadQueue.py index b78f4f90..b1362428 100644 --- a/app/DownloadQueue.py +++ b/app/DownloadQueue.py @@ -257,17 +257,18 @@ class DownloadQueue: async def cancel(self, ids): for id in ids: - if not self.queue.exists(id): + 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) - if item.started(): + 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=}') await self.notifier.canceled(id) @@ -326,11 +327,12 @@ class DownloadQueue: 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) + entry = self.queue.getNextDownload() + await asyncio.sleep(0.2) + + if entry.started() is False and entry.is_canceled() is False: + await executor.push(id=entry.info._id, entry=entry) + await asyncio.sleep(1) async def __downloadFile(self, id: str, entry: Download): log.info( @@ -349,12 +351,12 @@ class DownloadQueue: entry.close() - if self.queue.exists(id): - self.queue.delete(id) - if entry.canceled: + if self.queue.exists(key=id): + self.queue.delete(key=id) + if entry.is_canceled() is True: await self.notifier.canceled(id) else: - self.done.put(entry) + self.done.put(value=entry) await self.notifier.completed(entry.info) async def __download(self):