Fixes
This commit is contained in:
ArabCoders
parent
a1aa7b02ba
commit
f9bb34fce7
5 changed files with 91 additions and 71 deletions
|
|
@ -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:
|
||||
|
|
|
|||
|
|
@ -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))
|
||||
|
|
|
|||
|
|
@ -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()
|
||||
|
||||
|
|
|
|||
|
|
@ -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())))
|
||||
|
|
|
|||
|
|
@ -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))
|
||||
|
|
|
|||
Loading…
Reference in a new issue