From 25c44b4da91539c9abc28e1ee46cd450b8a21feb Mon Sep 17 00:00:00 2001 From: ArabCoders Date: Fri, 3 Nov 2023 14:02:37 +0300 Subject: [PATCH] Expose done/queue list as json endpoint --- .gitignore | 1 + app/main.py | 14 ++ app/ytdl.py | 703 ++++++++++++++++++++++++++-------------------------- 3 files changed, 368 insertions(+), 350 deletions(-) diff --git a/.gitignore b/.gitignore index f75d65d..ac87c29 100644 --- a/.gitignore +++ b/.gitignore @@ -46,3 +46,4 @@ Thumbs.db __pycache__ +.venv diff --git a/app/main.py b/app/main.py index 51caccb..9c9054b 100644 --- a/app/main.py +++ b/app/main.py @@ -132,6 +132,20 @@ async def delete(request): status = await (dqueue.cancel(ids) if where == 'queue' else dqueue.clear(ids)) return web.Response(text=serializer.encode(status)) +@routes.get(config.URL_PREFIX + 'history') +async def list_queue(_): + history = []; + for _ ,v in dqueue.queue.saved_items(): + obj = vars(v) + obj['list_type'] = 'queue' + history.append(obj) + for _ ,v in dqueue.done.saved_items(): + obj = vars(v) + obj['list_type'] = 'done' + history.append(obj) + + return web.Response(text=json.dumps(history)) + @sio.event async def connect(sid, environ): await sio.emit('all', serializer.encode(dqueue.get()), to=sid) diff --git a/app/ytdl.py b/app/ytdl.py index 1cf992e..8faf3d2 100644 --- a/app/ytdl.py +++ b/app/ytdl.py @@ -1,350 +1,353 @@ -import os -import yt_dlp -from collections import OrderedDict -import shelve -import time -import asyncio -import multiprocessing -import logging -import re -from dl_formats import get_format, get_opts, AUDIO_FORMATS - -log = logging.getLogger('ytdl') - -class DownloadQueueNotifier: - async def added(self, dl): - raise NotImplementedError - - async def updated(self, dl): - raise NotImplementedError - - async def completed(self, dl): - raise NotImplementedError - - async def canceled(self, id): - raise NotImplementedError - - async def cleared(self, id): - raise NotImplementedError - -class DownloadInfo: - def __init__(self, id, title, url, quality, format, folder, custom_name_prefix): - self.id = id if len(custom_name_prefix) == 0 else f'{custom_name_prefix}.{id}' - self.title = title if len(custom_name_prefix) == 0 else f'{custom_name_prefix}.{title}' - self.url = url - self.quality = quality - self.format = format - self.folder = folder - self.custom_name_prefix = custom_name_prefix - self.status = self.msg = self.percent = self.speed = self.eta = None - self.timestamp = time.time_ns() - -class Download: - manager = None - - def __init__(self, download_dir, temp_dir, output_template, output_template_chapter, quality, format, ytdl_opts, info): - self.download_dir = download_dir - self.temp_dir = temp_dir - self.output_template = output_template - self.output_template_chapter = output_template_chapter - self.format = get_format(format, quality) - self.ytdl_opts = get_opts(format, quality, ytdl_opts) - self.info = info - self.canceled = False - self.tmpfilename = None - self.status_queue = None - self.proc = None - self.loop = None - self.notifier = None - - def _download(self): - try: - def put_status(st): - self.status_queue.put({k: v for k, v in st.items() if k in ( - 'tmpfilename', - 'filename', - 'status', - 'msg', - 'total_bytes', - 'total_bytes_estimate', - 'downloaded_bytes', - 'speed', - 'eta', - )}) - 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({'status': 'finished', 'filename': filename}) - ret = yt_dlp.YoutubeDL(params={ - 'quiet': True, - 'no_color': True, - #'skip_download': True, - 'paths': {"home": self.download_dir, "temp": self.temp_dir}, - 'outtmpl': { "default": self.output_template, "chapter": self.output_template_chapter }, - 'format': self.format, - 'socket_timeout': 30, - 'progress_hooks': [put_status], - 'postprocessor_hooks': [put_status_postprocessor], - **self.ytdl_opts, - }).download([self.info.url]) - self.status_queue.put({'status': 'finished' if ret == 0 else 'error'}) - except yt_dlp.utils.YoutubeDLError as exc: - self.status_queue.put({'status': 'error', 'msg': str(exc)}) - - async def start(self, notifier): - if Download.manager is None: - Download.manager = multiprocessing.Manager() - self.status_queue = Download.manager.Queue() - self.proc = multiprocessing.Process(target=self._download) - self.proc.start() - self.loop = asyncio.get_running_loop() - self.notifier = notifier - self.info.status = 'preparing' - await self.notifier.updated(self.info) - asyncio.create_task(self.update_status()) - return await self.loop.run_in_executor(None, self.proc.join) - - def cancel(self): - if self.running(): - self.proc.kill() - self.canceled = True - - def close(self): - if self.started(): - self.proc.close() - self.status_queue.put(None) - - def running(self): - try: - return self.proc is not None and self.proc.is_alive() - except ValueError: - return False - - def started(self): - return self.proc is not None - - async def update_status(self): - while True: - status = await self.loop.run_in_executor(None, self.status_queue.get) - if status is None: - return - self.tmpfilename = status.get('tmpfilename') - if 'filename' in status: - self.info.filename = os.path.relpath(status.get('filename'), self.download_dir) - - # Set correct file extension for thumbnails - if(self.info.format == 'thumbnail'): - self.info.filename = re.sub(r'\.webm$', '.jpg', self.info.filename) - self.info.status = status['status'] - self.info.msg = status.get('msg') - if 'downloaded_bytes' in status: - total = status.get('total_bytes') or status.get('total_bytes_estimate') - if total: - self.info.percent = status['downloaded_bytes'] / total * 100 - self.info.speed = status.get('speed') - self.info.eta = status.get('eta') - await self.notifier.updated(self.info) - -class PersistentQueue: - def __init__(self, path): - pdir = os.path.dirname(path) - if not os.path.isdir(pdir): - os.mkdir(pdir) - with shelve.open(path, 'c'): - pass - self.path = path - self.dict = OrderedDict() - - def load(self): - for k, v in self.saved_items(): - self.dict[k] = Download(None, None, None, None, None, None, {}, v) - - def exists(self, key): - return key in self.dict - - def get(self, key): - return self.dict[key] - - def items(self): - return self.dict.items() - - def saved_items(self): - with shelve.open(self.path, 'r') as shelf: - return sorted(shelf.items(), key=lambda item: item[1].timestamp) - - def put(self, value): - key = value.info.url - self.dict[key] = value - with shelve.open(self.path, 'w') as shelf: - shelf[key] = value.info - - def delete(self, key): - del self.dict[key] - with shelve.open(self.path, 'w') as shelf: - shelf.pop(key) - - def next(self): - k, v = next(iter(self.dict.items())) - return k, v - - def empty(self): - return not bool(self.dict) - - -class DownloadQueue: - def __init__(self, config, notifier): - self.config = config - self.notifier = notifier - self.queue = PersistentQueue(self.config.STATE_DIR + '/queue') - self.done = PersistentQueue(self.config.STATE_DIR + '/completed') - self.done.load() - - async def __import_queue(self): - for k, v in self.queue.saved_items(): - await self.add(v.url, v.quality, v.format, v.folder, v.custom_name_prefix) - - async def initialize(self): - self.event = asyncio.Event() - asyncio.create_task(self.__download()) - asyncio.create_task(self.__import_queue()) - - def __extract_info(self, url): - return yt_dlp.YoutubeDL(params={ - 'quiet': True, - 'no_color': True, - 'extract_flat': True, - **self.config.YTDL_OPTIONS, - }).extract_info(url, download=False) - - def __calc_download_path(self, quality, format, folder): - """Calculates download path from quality, format and folder attributes. - - Returns: - Tuple dldirectory, error_message both of which might be None (but not at the same time) - """ - # Keep consistent with frontend - base_directory = self.config.DOWNLOAD_DIR if (quality != 'audio' and format not in AUDIO_FORMATS) else self.config.AUDIO_DOWNLOAD_DIR - if folder: - if not self.config.CUSTOM_DIRS: - return None, {'status': 'error', 'msg': f'A folder for the download was specified but CUSTOM_DIRS is not true in the configuration.'} - dldirectory = os.path.realpath(os.path.join(base_directory, folder)) - real_base_directory = os.path.realpath(base_directory) - if not dldirectory.startswith(real_base_directory): - return None, {'status': 'error', 'msg': f'Folder "{folder}" must resolve inside the base download directory "{real_base_directory}"'} - if not os.path.isdir(dldirectory): - if not self.config.CREATE_CUSTOM_DIRS: - return None, {'status': 'error', 'msg': f'Folder "{folder}" for download does not exist inside base directory "{real_base_directory}", and CREATE_CUSTOM_DIRS is not true in the configuration.'} - os.makedirs(dldirectory, exist_ok=True) - else: - dldirectory = base_directory - return dldirectory, None - - async def __add_entry(self, entry, quality, format, folder, custom_name_prefix, already): - etype = entry.get('_type') or 'video' - if etype == 'playlist': - entries = entry['entries'] - log.info(f'playlist detected with {len(entries)} entries') - playlist_index_digits = len(str(len(entries))) - results = [] - for index, etr in enumerate(entries, start=1): - etr["playlist"] = entry["id"] - etr["playlist_index"] = '{{0:0{0:d}d}}'.format(playlist_index_digits).format(index) - for property in ("id", "title", "uploader", "uploader_id"): - if property in entry: - etr[f"playlist_{property}"] = entry[property] - results.append(await self.__add_entry(etr, quality, format, folder, custom_name_prefix, already)) - if any(res['status'] == 'error' for res in results): - return {'status': 'error', 'msg': ', '.join(res['msg'] for res in results if res['status'] == 'error' and 'msg' in res)} - return {'status': 'ok'} - elif etype == 'video' or etype.startswith('url') and 'id' in entry and 'title' in entry: - if not self.queue.exists(entry['id']): - dl = DownloadInfo(entry['id'], entry['title'], entry.get('webpage_url') or entry['url'], quality, format, folder, custom_name_prefix) - dldirectory, error_message = self.__calc_download_path(quality, format, folder) - if error_message is not None: - return error_message - output = self.config.OUTPUT_TEMPLATE if len(custom_name_prefix) == 0 else f'{custom_name_prefix}.{self.config.OUTPUT_TEMPLATE}' - output_chapter = self.config.OUTPUT_TEMPLATE_CHAPTER - for property, value in entry.items(): - if property.startswith("playlist"): - output = output.replace(f"%({property})s", str(value)) - self.queue.put(Download(dldirectory, self.config.TEMP_DIR, output, output_chapter, quality, format, self.config.YTDL_OPTIONS, dl)) - self.event.set() - await self.notifier.added(dl) - return {'status': 'ok'} - elif etype.startswith('url'): - return await self.add(entry['url'], quality, format, folder, custom_name_prefix, already) - return {'status': 'error', 'msg': f'Unsupported resource "{etype}"'} - - async def add(self, url, quality, format, folder, custom_name_prefix, already=None): - log.info(f'adding {url}: {quality=} {format=} {already=} {folder=} {custom_name_prefix=}') - already = set() if already is None else already - if url in already: - log.info('recursion detected, skipping') - return {'status': 'ok'} - else: - already.add(url) - try: - entry = await asyncio.get_running_loop().run_in_executor(None, self.__extract_info, url) - except yt_dlp.utils.YoutubeDLError as exc: - return {'status': 'error', 'msg': str(exc)} - return await self.__add_entry(entry, quality, format, folder, custom_name_prefix, already) - - async def cancel(self, ids): - for id in ids: - if not self.queue.exists(id): - log.warn(f'requested cancel for non-existent download {id}') - continue - if self.queue.get(id).started(): - self.queue.get(id).cancel() - else: - 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(id): - log.warn(f'requested delete for non-existent download {id}') - continue - if self.config.DELETE_FILE_ON_TRASHCAN: - dl = self.done.get(id) - try: - dldirectory, _ = self.__calc_download_path(dl.info.quality, dl.info.format, dl.info.folder) - os.remove(os.path.join(dldirectory, dl.info.filename)) - except Exception as e: - log.warn(f'deleting file for download {id} failed with error message {e!r}') - self.done.delete(id) - await self.notifier.cleared(id) - return {'status': 'ok'} - - def get(self): - return(list((k, v.info) for k, v in self.queue.items()), - list((k, v.info) for k, v in self.done.items())) - - async def __download(self): - while True: - while self.queue.empty(): - log.info('waiting for item to download') - await self.event.wait() - self.event.clear() - id, entry = self.queue.next() - log.info(f'downloading {entry.info.title}') - 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) +import os +import yt_dlp +from collections import OrderedDict +import shelve +import time +import asyncio +import multiprocessing +import logging +import re +from dl_formats import get_format, get_opts, AUDIO_FORMATS + +log = logging.getLogger('ytdl') + +class DownloadQueueNotifier: + async def added(self, dl): + raise NotImplementedError + + async def updated(self, dl): + raise NotImplementedError + + async def completed(self, dl): + raise NotImplementedError + + async def canceled(self, id): + raise NotImplementedError + + async def cleared(self, id): + raise NotImplementedError + +class DownloadInfo: + def __init__(self, id, title, url, quality, format, folder, custom_name_prefix): + self.id = id if len(custom_name_prefix) == 0 else f'{custom_name_prefix}.{id}' + self.title = title if len(custom_name_prefix) == 0 else f'{custom_name_prefix}.{title}' + self.url = url + self.quality = quality + self.format = format + self.folder = folder + self.custom_name_prefix = custom_name_prefix + self.status = self.msg = self.percent = self.speed = self.eta = None + self.timestamp = time.time_ns() + +class Download: + manager = None + + def __init__(self, download_dir, temp_dir, output_template, output_template_chapter, quality, format, ytdl_opts, info): + self.download_dir = download_dir + self.temp_dir = temp_dir + self.output_template = output_template + self.output_template_chapter = output_template_chapter + self.format = get_format(format, quality) + self.ytdl_opts = get_opts(format, quality, ytdl_opts) + self.info = info + self.canceled = False + self.tmpfilename = None + self.status_queue = None + self.proc = None + self.loop = None + self.notifier = None + + def _download(self): + try: + def put_status(st): + self.status_queue.put({k: v for k, v in st.items() if k in ( + 'tmpfilename', + 'filename', + 'status', + 'msg', + 'total_bytes', + 'total_bytes_estimate', + 'downloaded_bytes', + 'speed', + 'eta', + )}) + 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({'status': 'finished', 'filename': filename}) + ret = yt_dlp.YoutubeDL(params={ + 'quiet': True, + 'no_color': True, + #'skip_download': True, + 'paths': {"home": self.download_dir, "temp": self.temp_dir}, + 'outtmpl': { "default": self.output_template, "chapter": self.output_template_chapter }, + 'format': self.format, + 'socket_timeout': 30, + 'progress_hooks': [put_status], + 'postprocessor_hooks': [put_status_postprocessor], + **self.ytdl_opts, + }).download([self.info.url]) + self.status_queue.put({'status': 'finished' if ret == 0 else 'error'}) + except yt_dlp.utils.YoutubeDLError as exc: + self.status_queue.put({'status': 'error', 'msg': str(exc)}) + + async def start(self, notifier): + if Download.manager is None: + Download.manager = multiprocessing.Manager() + self.status_queue = Download.manager.Queue() + self.proc = multiprocessing.Process(target=self._download) + self.proc.start() + self.loop = asyncio.get_running_loop() + self.notifier = notifier + self.info.status = 'preparing' + await self.notifier.updated(self.info) + asyncio.create_task(self.update_status()) + return await self.loop.run_in_executor(None, self.proc.join) + + def cancel(self): + if self.running(): + self.proc.kill() + self.canceled = True + + def close(self): + if self.started(): + self.proc.close() + self.status_queue.put(None) + + def running(self): + try: + return self.proc is not None and self.proc.is_alive() + except ValueError: + return False + + def started(self): + return self.proc is not None + + async def update_status(self): + while True: + status = await self.loop.run_in_executor(None, self.status_queue.get) + if status is None: + return + self.tmpfilename = status.get('tmpfilename') + if 'filename' in status: + self.info.filename = os.path.relpath(status.get('filename'), self.download_dir) + + # Set correct file extension for thumbnails + if(self.info.format == 'thumbnail'): + self.info.filename = re.sub(r'\.webm$', '.jpg', self.info.filename) + self.info.status = status['status'] + self.info.msg = status.get('msg') + if 'downloaded_bytes' in status: + total = status.get('total_bytes') or status.get('total_bytes_estimate') + if total: + self.info.percent = status['downloaded_bytes'] / total * 100 + self.info.speed = status.get('speed') + self.info.eta = status.get('eta') + await self.notifier.updated(self.info) + +class PersistentQueue: + def __init__(self, path): + pdir = os.path.dirname(path) + if not os.path.isdir(pdir): + os.mkdir(pdir) + with shelve.open(path, 'c'): + pass + self.path = path + self.dict = OrderedDict() + + def load(self): + for k, v in self.saved_items(): + self.dict[k] = Download(None, None, None, None, None, None, {}, v) + + def exists(self, key): + return key in self.dict + + def get(self, key): + return self.dict[key] + + def items(self): + return self.dict.items() + + def saved_items(self): + with shelve.open(self.path, 'r') as shelf: + return sorted(shelf.items(), key=lambda item: item[1].timestamp) + + def put(self, value): + key = value.info.url + self.dict[key] = value + with shelve.open(self.path, 'w') as shelf: + shelf[key] = value.info + + def delete(self, key): + del self.dict[key] + with shelve.open(self.path, 'w') as shelf: + shelf.pop(key) + + def next(self): + k, v = next(iter(self.dict.items())) + return k, v + + def empty(self): + return not bool(self.dict) + + +class DownloadQueue: + def __init__(self, config, notifier): + self.config = config + self.notifier = notifier + self.queue = PersistentQueue(self.config.STATE_DIR + '/queue') + self.done = PersistentQueue(self.config.STATE_DIR + '/completed') + self.done.load() + + async def __import_queue(self): + for k, v in self.queue.saved_items(): + await self.add(v.url, v.quality, v.format, v.folder, v.custom_name_prefix) + + async def initialize(self): + self.event = asyncio.Event() + asyncio.create_task(self.__download()) + asyncio.create_task(self.__import_queue()) + + def __extract_info(self, url): + return yt_dlp.YoutubeDL(params={ + 'quiet': True, + 'no_color': True, + 'extract_flat': True, + **self.config.YTDL_OPTIONS, + }).extract_info(url, download=False) + + def __calc_download_path(self, quality, format, folder): + """Calculates download path from quality, format and folder attributes. + + Returns: + Tuple dldirectory, error_message both of which might be None (but not at the same time) + """ + # Keep consistent with frontend + base_directory = self.config.DOWNLOAD_DIR if (quality != 'audio' and format not in AUDIO_FORMATS) else self.config.AUDIO_DOWNLOAD_DIR + if folder: + if not self.config.CUSTOM_DIRS: + return None, {'status': 'error', 'msg': f'A folder for the download was specified but CUSTOM_DIRS is not true in the configuration.'} + dldirectory = os.path.realpath(os.path.join(base_directory, folder)) + real_base_directory = os.path.realpath(base_directory) + if not dldirectory.startswith(real_base_directory): + return None, {'status': 'error', 'msg': f'Folder "{folder}" must resolve inside the base download directory "{real_base_directory}"'} + if not os.path.isdir(dldirectory): + if not self.config.CREATE_CUSTOM_DIRS: + return None, {'status': 'error', 'msg': f'Folder "{folder}" for download does not exist inside base directory "{real_base_directory}", and CREATE_CUSTOM_DIRS is not true in the configuration.'} + os.makedirs(dldirectory, exist_ok=True) + else: + dldirectory = base_directory + return dldirectory, None + + async def __add_entry(self, entry, quality, format, folder, custom_name_prefix, already): + if not entry: + return {'status': 'error', 'msg': "Invalid/empty data was given."} + + etype = entry.get('_type') or 'video' + if etype == 'playlist': + entries = entry['entries'] + log.info(f'playlist detected with {len(entries)} entries') + playlist_index_digits = len(str(len(entries))) + results = [] + for index, etr in enumerate(entries, start=1): + etr["playlist"] = entry["id"] + etr["playlist_index"] = '{{0:0{0:d}d}}'.format(playlist_index_digits).format(index) + for property in ("id", "title", "uploader", "uploader_id"): + if property in entry: + etr[f"playlist_{property}"] = entry[property] + results.append(await self.__add_entry(etr, quality, format, folder, custom_name_prefix, already)) + if any(res['status'] == 'error' for res in results): + return {'status': 'error', 'msg': ', '.join(res['msg'] for res in results if res['status'] == 'error' and 'msg' in res)} + return {'status': 'ok'} + elif etype == 'video' or etype.startswith('url') and 'id' in entry and 'title' in entry: + if not self.queue.exists(entry['id']): + dl = DownloadInfo(entry['id'], entry['title'], entry.get('webpage_url') or entry['url'], quality, format, folder, custom_name_prefix) + dldirectory, error_message = self.__calc_download_path(quality, format, folder) + if error_message is not None: + return error_message + output = self.config.OUTPUT_TEMPLATE if len(custom_name_prefix) == 0 else f'{custom_name_prefix}.{self.config.OUTPUT_TEMPLATE}' + output_chapter = self.config.OUTPUT_TEMPLATE_CHAPTER + for property, value in entry.items(): + if property.startswith("playlist"): + output = output.replace(f"%({property})s", str(value)) + self.queue.put(Download(dldirectory, self.config.TEMP_DIR, output, output_chapter, quality, format, self.config.YTDL_OPTIONS, dl)) + self.event.set() + await self.notifier.added(dl) + return {'status': 'ok'} + elif etype.startswith('url'): + return await self.add(entry['url'], quality, format, folder, custom_name_prefix, already) + return {'status': 'error', 'msg': f'Unsupported resource "{etype}"'} + + async def add(self, url, quality, format, folder, custom_name_prefix, already=None): + log.info(f'adding {url}: {quality=} {format=} {already=} {folder=} {custom_name_prefix=}') + already = set() if already is None else already + if url in already: + log.info('recursion detected, skipping') + return {'status': 'ok'} + else: + already.add(url) + try: + entry = await asyncio.get_running_loop().run_in_executor(None, self.__extract_info, url) + except yt_dlp.utils.YoutubeDLError as exc: + return {'status': 'error', 'msg': str(exc)} + return await self.__add_entry(entry, quality, format, folder, custom_name_prefix, already) + + async def cancel(self, ids): + for id in ids: + if not self.queue.exists(id): + log.warn(f'requested cancel for non-existent download {id}') + continue + if self.queue.get(id).started(): + self.queue.get(id).cancel() + else: + 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(id): + log.warn(f'requested delete for non-existent download {id}') + continue + if self.config.DELETE_FILE_ON_TRASHCAN: + dl = self.done.get(id) + try: + dldirectory, _ = self.__calc_download_path(dl.info.quality, dl.info.format, dl.info.folder) + os.remove(os.path.join(dldirectory, dl.info.filename)) + except Exception as e: + log.warn(f'deleting file for download {id} failed with error message {e!r}') + self.done.delete(id) + await self.notifier.cleared(id) + return {'status': 'ok'} + + def get(self): + return(list((k, v.info) for k, v in self.queue.items()), + list((k, v.info) for k, v in self.done.items())) + + async def __download(self): + while True: + while self.queue.empty(): + log.info('waiting for item to download') + await self.event.wait() + self.event.clear() + id, entry = self.queue.next() + log.info(f'downloading {entry.info.title}') + 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)