diff --git a/app/DataStore.py b/app/DataStore.py index a92b3fdf..0de83233 100644 --- a/app/DataStore.py +++ b/app/DataStore.py @@ -28,13 +28,7 @@ class DataStore: def load(self) -> None: for id, item in self.saved_items(): - self.dict.update({id: Download( - info=item, - download_dir=calcDownloadPath(basePath=self.config.download_path, folder=item.folder), - temp_dir=self.config.temp_path, - output_template_chapter=self.config.output_template_chapter, - default_ytdl_opts=self.config.ytdl_options) - }) + self.dict.update({id: Download(info=item)}) def exists(self, key: str = None, url: str = None) -> bool: if not key and not url: diff --git a/app/Download.py b/app/Download.py index 6f51b7ec..b8610622 100644 --- a/app/Download.py +++ b/app/Download.py @@ -6,12 +6,12 @@ import os import re import shutil import yt_dlp +import hashlib + from Utils import Notifier, get_format, get_opts, jsonCookie, mergeConfig from ItemDTO import ItemDTO from Config import Config -import hashlib - LOG = logging.getLogger('download') @@ -60,25 +60,18 @@ class Download: tempKeep: bool = False "Keep temp directory after download." - def __init__( - self, - info: ItemDTO, - download_dir: str, - temp_dir: str, - output_template_chapter: str, - default_ytdl_opts: dict, - info_dict: dict = None, - debug: bool = False - ): - self.download_dir = download_dir - self.temp_dir = temp_dir - self.output_template_chapter = output_template_chapter + def __init__(self, info: ItemDTO, info_dict: dict = None, debug: bool = False): + config = Config.get_instance() + + self.download_dir = info.download_dir + self.temp_dir = info.temp_dir + self.output_template_chapter = info.output_template_chapter self.output_template = info.output_template self.format = get_format(info.format, info.quality) 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.default_ytdl_opts = config.ytdl_options self.debug = debug self.canceled = False @@ -87,8 +80,8 @@ class Download: self.proc = None self.loop = None self.notifier = None - self.max_workers = int(Config.get_instance().max_workers) - self.tempKeep = bool(Config.get_instance().temp_keep) + self.max_workers = int(config.max_workers) + self.tempKeep = bool(config.temp_keep) self.is_live = bool(info.is_live) or info.live_in is not None self.is_manifestless = 'is_manifestless' in self.info.options and self.info.options['is_manifestless'] is True self.info_dict = info_dict @@ -190,9 +183,7 @@ class Download: LOG.info(f'Finished {os.getpid()=} id="{self.info.id}" title="{self.info.title}".') async def start(self, notifier: Notifier): - self.manager = multiprocessing.Manager() if self.manager is None else self.manager - - self.status_queue = self.manager.Queue() + self.status_queue = multiprocessing.Manager().Queue() self.loop = asyncio.get_running_loop() self.notifier = notifier @@ -205,33 +196,59 @@ class Download: self.proc = multiprocessing.Process(target=self._download) self.proc.start() self.info.status = 'preparing' - await self.notifier.updated(self.info) + + asyncio.create_task(self.notifier.updated(self.info)) 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): - self.kill() - self.canceled = True + def cancel(self) -> bool: + if not self.started(): + return False - def close(self): - if self.started(): + if self.kill(): + self.canceled = True + + return True + + def close(self) -> bool: + if not self.started(): + return False + + try: + LOG.info(f"Closing download process: '{self.proc.ident}'.") self.proc.close() + self.delete_temp() + return True + except Exception as e: + LOG.error(f"Failed to close process: '{self.proc.ident}'. {e}") - self.delete_temp() + return False def running(self) -> bool: - return self.started() and self.proc.is_alive() + try: + return self.proc is not None and self.proc.is_alive() + except ValueError: + return False def is_canceled(self) -> bool: return self.canceled - def kill(self): - if self.running(): - LOG.info(f'Killing download process: {self.proc.ident}') + def kill(self) -> bool: + if not self.started(): + return False + + try: + LOG.info(f"Killing download process: '{self.proc.ident}'.") self.proc.kill() + return True + except Exception as e: + LOG.error(f"Failed to kill process: '{self.proc.ident}'. {e}") + + return False def delete_temp(self): if self.tempKeep is True or not self.tempPath: @@ -266,7 +283,7 @@ class Download: LOG.debug(f'Status Update: {self.info._id=} {status=}') if isinstance(status, str): - await self.notifier.updated(self.info) + asyncio.create_task(self.notifier.updated(self.info)) return self.tmpfilename = status.get('tmpfilename') @@ -286,7 +303,7 @@ class Download: if self.info.status == 'error' and 'error' in status: self.info.error = status.get('error') - await self.notifier.error(self.info, self.info.error) + asyncio.create_task(self.notifier.error(self.info, self.info.error)) if 'downloaded_bytes' in status: total = status.get('total_bytes') or status.get('total_bytes_estimate') @@ -305,4 +322,4 @@ class Download: self.info.file_size = None pass - await self.notifier.updated(self.info) + asyncio.create_task(self.notifier.updated(self.info)) diff --git a/app/DownloadQueue.py b/app/DownloadQueue.py index 087937a1..11ffbf57 100644 --- a/app/DownloadQueue.py +++ b/app/DownloadQueue.py @@ -126,6 +126,12 @@ class DownloadQueue: live_status: list = ['is_live', 'is_upcoming'] is_live = entry.get('is_live', None) or live_in or entry.get('live_status', None) in live_status + try: + download_dir = calcDownloadPath(basePath=self.config.download_path, folder=folder) + except Exception as e: + LOG.exception(e) + return {'status': 'error', 'msg': str(e)} + dl = ItemDTO( id=entry.get('id'), title=entry.get('title'), @@ -133,37 +139,24 @@ class DownloadQueue: quality=quality, format=format, folder=folder, + download_dir=download_dir, + temp_dir=self.config.temp_path, ytdlp_cookies=ytdlp_cookies, ytdlp_config=ytdlp_config, output_template=output_template if output_template else self.config.output_template, + output_template_chapter=self.config.output_template_chapter, datetime=formatdate(time.time()), error=error, is_live=is_live, live_in=live_in, - options=options + options=options, ) - try: - download_dir = calcDownloadPath(basePath=self.config.download_path, folder=folder) - except Exception as e: - LOG.exception(e) - return {'status': 'error', 'msg': str(e)} - - output_chapter = self.config.output_template_chapter - for property, value in entry.items(): if property.startswith("playlist"): dl.output_template = dl.output_template.replace(f"%({property})s", str(value)) - dlInfo: Download = Download( - info=dl, - download_dir=download_dir, - temp_dir=self.config.temp_path, - output_template_chapter=output_chapter, - default_ytdl_opts=self.config.ytdl_options, - info_dict=entry, - debug=bool(self.config.ytdl_debug) - ) + dlInfo: Download = Download(info=dl, info_dict=entry, debug=bool(self.config.ytdl_debug)) if dlInfo.info.live_in or 'is_upcoming' == entry.get('live_status', None): dlInfo.info.status = 'not_live' @@ -179,7 +172,7 @@ class DownloadQueue: itemDownload = self.queue.put(dlInfo) self.event.set() - await self.notifier.emit(NotifyEvent, itemDownload.info) + asyncio.create_task(self.notifier.emit(NotifyEvent, itemDownload.info)) return { 'status': 'ok' @@ -270,11 +263,15 @@ class DownloadQueue: already=already, ) - async def cancel(self, ids): + async def cancel(self, ids: list[str]) -> dict[str, str]: + + status: dict[str, str] = {"status": "ok"} for id in ids: try: item = self.queue.get(key=id) except KeyError as e: + status[id] = str(e) + status['status'] = 'error' LOG.warning(f'Requested cancel for non-existent download {id=}. {str(e)}') continue @@ -284,33 +281,39 @@ class DownloadQueue: LOG.debug(f'Canceling {itemMessage}') item.cancel() LOG.info(f'Cancelled {itemMessage}') - else: item.close() LOG.debug(f'Deleting from queue {itemMessage}') self.queue.delete(id) - await self.notifier.canceled(id) + asyncio.create_task(self.notifier.canceled(id)) self.done.put(item) - await self.notifier.completed(item) + asyncio.create_task(self.notifier.completed(item)) LOG.info(f'Deleted from queue {itemMessage}') - return {'status': 'ok'} + status[id] = 'ok' + + return status + + async def clear(self, ids: list[str]) -> dict[str, str]: + status: dict[str, str] = {"status": "ok"} - async def clear(self, ids): for id in ids: try: item = self.done.get(key=id) except KeyError as e: + status[id] = str(e) + status['status'] = 'error' LOG.warning(f'Requested delete for non-existent download {id=}. {str(e)}') continue itemMessage = f"{id=} {item.info.id=} {item.info.title=}" LOG.debug(f'Deleting completed download {itemMessage}') self.done.delete(id) - await self.notifier.cleared(id) + asyncio.create_task(self.notifier.cleared(id)) LOG.info(f'Deleted completed download {itemMessage}') + status[id] = 'ok' - return {'status': 'ok'} + return status def get(self) -> dict[str, list[dict[str, ItemDTO]]]: items = {'queue': {}, 'done': {}} @@ -391,6 +394,7 @@ class DownloadQueue: if entry.tmpfilename and os.path.isfile(entry.tmpfilename): try: os.remove(entry.tmpfilename) + entry.tmpfilename = None except: pass @@ -403,12 +407,12 @@ class DownloadQueue: self.queue.delete(key=id) if entry.is_canceled() is True: - await self.notifier.canceled(id) + asyncio.create_task(self.notifier.canceled(id)) entry.info.status = 'canceled' entry.info.error = 'Canceled by user.' self.done.put(value=entry) - await self.notifier.completed(entry.info) + asyncio.create_task(self.notifier.completed(entry.info)) self.event.set() diff --git a/app/ItemDTO.py b/app/ItemDTO.py index a34236fe..ae567dda 100644 --- a/app/ItemDTO.py +++ b/app/ItemDTO.py @@ -16,10 +16,13 @@ class ItemDTO: quality: str format: str folder: str + download_dir: str = None + temp_dir: str = None status: str = None ytdlp_cookies: str = None ytdlp_config: dict = field(default_factory=dict) output_template: str = None + output_template_chapter: str = None timestamp: float = time.time_ns() is_live: bool = None datetime: str = field(default_factory=lambda: str(formatdate(time.time()))) diff --git a/app/main.py b/app/main.py index 19a87f75..31459e95 100644 --- a/app/main.py +++ b/app/main.py @@ -291,6 +291,8 @@ class Main: if not ids or where not in ['queue', 'done']: raise web.HTTPBadRequest() + status: dict[str, str] = {} + status = await (self.queue.cancel(ids) if where == 'queue' else self.queue.clear(ids)) return web.Response(text=self.serializer.encode(status))