From f9bb34fce7f2f5e4615f1c5cfcb550b32aacb8dd Mon Sep 17 00:00:00 2001 From: ArabCoders Date: Sun, 24 Mar 2024 20:32:41 +0300 Subject: [PATCH 1/2] Fixes --- app/DataStore.py | 8 +--- app/Download.py | 87 ++++++++++++++++++++++++++------------------ app/DownloadQueue.py | 62 ++++++++++++++++--------------- app/ItemDTO.py | 3 ++ app/main.py | 2 + 5 files changed, 91 insertions(+), 71 deletions(-) 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)) From 70cff70247a02bca43da9825379f577ecf2eb052 Mon Sep 17 00:00:00 2001 From: ArabCoders Date: Mon, 25 Mar 2024 23:15:37 +0300 Subject: [PATCH 2/2] Converted the video player to be async --- app/main.py | 2 +- app/player/M3u8.py | 7 +- app/player/Segments.py | 116 +++++++++++++---------- app/player/ffprobe.py | 87 +++++++++-------- frontend/src/components/Video-Player.vue | 1 + 5 files changed, 117 insertions(+), 96 deletions(-) diff --git a/app/main.py b/app/main.py index 31459e95..1c374cae 100644 --- a/app/main.py +++ b/app/main.py @@ -417,7 +417,7 @@ class Main: raise web.HTTPBadRequest(reason='file is required.') return web.Response( - text=M3u8(url=f"{self.config.url_host}{self.config.url_prefix}").make_stream( + text=await M3u8(url=f"{self.config.url_host}{self.config.url_prefix}").make_stream( download_path=self.config.download_path, file=file ), diff --git a/app/player/M3u8.py b/app/player/M3u8.py index adcde458..8980e4d4 100644 --- a/app/player/M3u8.py +++ b/app/player/M3u8.py @@ -16,19 +16,20 @@ class M3u8: self.url = url self.segment_duration = float(segment_duration) - def make_stream(self, download_path: str, file: str): + async def make_stream(self, download_path: str, file: str): realFile: str = calcDownloadPath(basePath=download_path, folder=file, createPath=False) if not os.path.exists(realFile): - raise Exception(f"File {realFile} does not exist.") + raise Exception(f"File '{realFile}' does not exist.") try: ffprobe = FFProbe(realFile) + await ffprobe.run() except UnicodeDecodeError as e: pass if not 'duration' in ffprobe.metadata: - raise Exception(f"Unable to get {realFile} duration.") + raise Exception(f"Unable to get '{realFile}' duration.") duration: float = float(ffprobe.metadata.get('duration')) diff --git a/app/player/Segments.py b/app/player/Segments.py index 01e65d57..b862717c 100644 --- a/app/player/Segments.py +++ b/app/player/Segments.py @@ -1,3 +1,4 @@ +import asyncio import hashlib import logging import os @@ -37,69 +38,80 @@ class Segments: else: startTime: float = '{:.6f}'.format((self.segment_duration * self.segment_index)) - ffmpegCmd = [] - ffmpegCmd.append('ffmpeg') - ffmpegCmd.append('-xerror') - ffmpegCmd.append('-hide_banner') - ffmpegCmd.append('-loglevel') - ffmpegCmd.append('error') + fargs = [] + fargs.append('-xerror') + fargs.append('-hide_banner') + fargs.append('-loglevel') + fargs.append('error') - ffmpegCmd.append('-ss') - ffmpegCmd.append(str(startTime if startTime else '0.00000')) - ffmpegCmd.append('-t') - ffmpegCmd.append(str('{:.6f}'.format(self.segment_duration))) + fargs.append('-ss') + fargs.append(str(startTime if startTime else '0.00000')) + fargs.append('-t') + fargs.append(str('{:.6f}'.format(self.segment_duration))) - ffmpegCmd.append('-copyts') + fargs.append('-copyts') - ffmpegCmd.append('-i') - ffmpegCmd.append(f'file:{tmpFile}') - ffmpegCmd.append('-map_metadata') - ffmpegCmd.append('-1') + fargs.append('-i') + fargs.append(f'file:{tmpFile}') + fargs.append('-map_metadata') + fargs.append('-1') - ffmpegCmd.append('-pix_fmt') - ffmpegCmd.append('yuv420p') - ffmpegCmd.append('-g') - ffmpegCmd.append('52') + fargs.append('-pix_fmt') + fargs.append('yuv420p') + fargs.append('-g') + fargs.append('52') - ffmpegCmd.append('-map') - ffmpegCmd.append('0:v:0') - ffmpegCmd.append('-strict') - ffmpegCmd.append('-2') + fargs.append('-map') + fargs.append('0:v:0') + fargs.append('-strict') + fargs.append('-2') - ffmpegCmd.append('-codec:v') - ffmpegCmd.append('libx264' if self.vconvert else 'copy') + fargs.append('-codec:v') + fargs.append('libx264' if self.vconvert else 'copy') if self.vconvert: - ffmpegCmd.append('-crf') - ffmpegCmd.append('23') - ffmpegCmd.append('-preset:v') - ffmpegCmd.append('fast') - ffmpegCmd.append('-level') - ffmpegCmd.append('4.1') - ffmpegCmd.append('-profile:v') - ffmpegCmd.append('baseline') + fargs.append('-crf') + fargs.append('23') + fargs.append('-preset:v') + fargs.append('fast') + fargs.append('-level') + fargs.append('4.1') + fargs.append('-profile:v') + fargs.append('baseline') # audio section. - ffmpegCmd.append('-map') - ffmpegCmd.append('0:a:0') - ffmpegCmd.append('-codec:a') - ffmpegCmd.append('aac' if self.aconvert else 'copy') + fargs.append('-map') + fargs.append('0:a:0') + fargs.append('-codec:a') + fargs.append('aac' if self.aconvert else 'copy') if self.aconvert: - ffmpegCmd.append('-b:a') - ffmpegCmd.append('192k') - ffmpegCmd.append('-ar') - ffmpegCmd.append('22050') - ffmpegCmd.append('-ac') - ffmpegCmd.append('2') + fargs.append('-b:a') + fargs.append('192k') + fargs.append('-ar') + fargs.append('22050') + fargs.append('-ac') + fargs.append('2') - ffmpegCmd.append('-sn') + fargs.append('-sn') - ffmpegCmd.append('-muxdelay') - ffmpegCmd.append('0') - ffmpegCmd.append('-f') - ffmpegCmd.append('mpegts') - ffmpegCmd.append('pipe:1') + fargs.append('-muxdelay') + fargs.append('0') + fargs.append('-f') + fargs.append('mpegts') + fargs.append('pipe:1') - LOG.debug(f'Streaming {realFile} segment {self.segment_index}.' + ' '.join(ffmpegCmd)) - proc = subprocess.run(ffmpegCmd, stdout=subprocess.PIPE) + LOG.debug(f"Streaming '{realFile}' segment '{self.segment_index}'. " + " ".join(fargs)) - return proc.stdout + proc = await asyncio.subprocess.create_subprocess_exec( + 'ffmpeg', *fargs, + stdin=asyncio.subprocess.DEVNULL, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE + ) + + data, err = await proc.communicate() + + if 0 != proc.returncode: + LOG.error(f'Failed to stream {realFile} segment {self.segment_index}. {err.decode("utf-8")}') + raise Exception(f'Failed to stream {realFile} segment {self.segment_index}.') + + return data diff --git a/app/player/ffprobe.py b/app/player/ffprobe.py index ac52afb8..e40c7796 100644 --- a/app/player/ffprobe.py +++ b/app/player/ffprobe.py @@ -1,8 +1,10 @@ """ Python wrapper for ffprobe command line tool. ffprobe must exist in the path. """ +import asyncio import functools import json +import logging import operator import os import pipes @@ -178,55 +180,60 @@ class FFProbe: def __init__(self, path_to_video): self.path_to_video = path_to_video + async def run(self): try: with open(os.devnull, 'w') as tempf: - subprocess.check_call(["ffprobe", "-h"], stdout=tempf, stderr=tempf) + await asyncio.create_subprocess_exec( + "ffprobe", "-h", stdout=tempf, stderr=tempf + ) except FileNotFoundError: raise IOError('ffprobe not found.') - if os.path.isfile(self.path_to_video): - cmd: list = [ - "ffprobe -v quiet -of json -show_format -show_streams " + - pipes.quote(self.path_to_video) - ] + if not os.path.isfile(self.path_to_video): + raise IOError(f"No such media file '{self.path_to_video}'.") - p = subprocess.Popen( - args=cmd, - stdout=subprocess.PIPE, - stderr=subprocess.PIPE, - shell=True - ) - p.wait() - data, err = p.communicate() - if p.returncode == 0: - parsed: dict = json.loads(data.decode('utf-8')) - else: - raise FFProbeError(f"FFProbe error: {err}") + args = [ + '-v', 'quiet', + '-of', 'json', + '-show_streams', + '-show_format', + self.path_to_video, + ] + p = await asyncio.create_subprocess_exec( + 'ffprobe', *args, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + ) - stream = False - self.streams = [] - self.video = [] - self.audio = [] - self.subtitle = [] - self.attachment = [] + exitCode = await p.wait() - for stream in parsed['streams'] if 'streams' in parsed else []: - self.streams.append(FFStream(stream)) - - self.metadata = parsed['format'] if 'format' in parsed else {} - - for stream in self.streams: - if stream.is_audio(): - self.audio.append(stream) - elif stream.is_video(): - self.video.append(stream) - elif stream.is_subtitle(): - self.subtitle.append(stream) - elif stream.is_attachment(): - self.attachment.append(stream) + data, err = await p.communicate() + if 0 == exitCode: + parsed: dict = json.loads(data.decode('utf-8')) else: - raise IOError('No such media file ' + self.path_to_video) + raise FFProbeError(f"FFProbe return with non-0 exit code. '{err.decode('utf-8')}'") + + stream = False + self.streams = [] + self.video = [] + self.audio = [] + self.subtitle = [] + self.attachment = [] + + for stream in parsed.get('streams', []): + self.streams.append(FFStream(stream)) + + self.metadata = parsed.get('format', {}) + + for stream in self.streams: + if stream.is_audio(): + self.audio.append(stream) + elif stream.is_video(): + self.video.append(stream) + elif stream.is_subtitle(): + self.subtitle.append(stream) + elif stream.is_attachment(): + self.attachment.append(stream) def __repr__(self): return "".format(**vars(self)) - diff --git a/frontend/src/components/Video-Player.vue b/frontend/src/components/Video-Player.vue index 30194ff1..0acfc8ad 100644 --- a/frontend/src/components/Video-Player.vue +++ b/frontend/src/components/Video-Player.vue @@ -72,6 +72,7 @@ const prepareVideoPlayer = () => { enableWorker: true, lowLatencyMode: true, backBufferLength: 90, + fragLoadingTimeOut: 200000, }); hls.value.loadSource(props.link)