diff --git a/app/AsyncPool.py b/app/AsyncPool.py new file mode 100644 index 00000000..f8e2fafc --- /dev/null +++ b/app/AsyncPool.py @@ -0,0 +1,160 @@ +import asyncio +from datetime import datetime, timezone + + +class Terminator: + pass + + +class AsyncPool: + def __init__(self, loop, num_workers: int, name: str, logger, worker_co, load_factor: int = 1, + job_accept_duration: int = None, max_task_time: int = None, return_futures: bool = False, + raise_on_join: bool = False, log_every_n: int = None, expected_total=None): + """ + This class will create `num_workers` asyncio tasks to work against a queue of + `num_workers * load_factor` items of back-pressure (IOW we will block after such + number of items of work is in the queue). `worker_co` will be called + against each item retrieved from the queue. If any exceptions are raised out of + worker_co, self.exceptions will be set to True. + @param loop: asyncio loop to use + @param num_workers: number of async tasks which will pull from the internal queue + @param name: name of the worker pool (used for logging) + @param logger: logger to use + @param worker_co: async coroutine to call when an item is retrieved from the queue + @param load_factor: multiplier used for number of items in queue + @param job_accept_duration: maximum number of seconds from first push to last push before a TimeoutError will be thrown. + Set to None for no limit. Note this does not get reset on aenter/aexit. + @param max_task_time: maximum time allowed for each task before a CancelledError is raised in the task. + Set to None for no limit. + @param return_futures: set to reture to return a future for each `push` (imposes CPU overhead) + @param raise_on_join: raise on join if any exceptions have occurred, default is False + @param log_every_n: (optional) set to number of `push`s each time a log statement should be printed (default does not print every-n pushes) + @param expected_total: (optional) expected total number of jobs (used for `log_event_n` logging) + @return: instance of AsyncWorkerPool + """ + loop = loop or asyncio.get_event_loop() + self._loop = loop + self._num_workers = num_workers + self._logger = logger + self._queue = asyncio.Queue(num_workers * load_factor) + self._workers = None + self._exceptions = False + self._job_accept_duration = job_accept_duration + self._first_push_dt = None + self._max_task_time = max_task_time + self._return_futures = return_futures + self._raise_on_join = raise_on_join + self._name = name + self._worker_co = worker_co + self._total_queued = 0 + self._log_every_n = log_every_n + self._expected_total = expected_total + + async def _worker_loop(self): + while True: + got_obj = False + future = None + + try: + item = await self._queue.get() + got_obj = True + + if item.__class__ is Terminator: + break + + future, args, kwargs = item + # the wait_for will cancel the task (task sees CancelledError) and raises a TimeoutError from here + # so be wary of catching TimeoutErrors in this loop + result = await asyncio.wait_for(self._worker_co(*args, **kwargs), self._max_task_time, loop=self._loop) + + if future: + future.set_result(result) + except (KeyboardInterrupt, MemoryError, SystemExit) as e: + if future: + future.set_exception(e) + self._exceptions = True + raise + except BaseException as e: + self._exceptions = True + + if future: + # don't log the failure when the client is receiving the future + future.set_exception(e) + else: + self._logger.exception('Worker call failed') + finally: + if got_obj: + self._queue.task_done() + + @property + def exceptions(self): + return self._exceptions + + @property + def total_queued(self): + return self._total_queued + + async def __aenter__(self): + self.start() + return self + + async def __aexit__(self, exc_type, exc_val, exc_tb): + await self.join() + + async def push(self, *args, **kwargs) -> asyncio.Future: + """ Method to push work to `worker_co` passed to `__init__`. + :param args: position arguments to be passed to `worker_co` + :param kwargs: keyword arguments to be passed to `worker_co` + :return: future of result """ + if self._first_push_dt is None: + self._first_push_dt = self._time() + + if self._job_accept_duration is not None and (self._time() - self._first_push_dt) > self._job_accept_duration: + raise TimeoutError("Maximum lifetime of {} seconds of AsyncWorkerPool: {} exceeded".format( + self._job_accept_duration, self._name)) + + future = asyncio.futures.Future( + loop=self._loop) if self._return_futures else None + await self._queue.put((future, args, kwargs)) + self._total_queued += 1 + + if self._log_every_n is not None and (self._total_queued % self._log_every_n) == 0: + self._logger.info("pushed {}/{} items to {} AsyncWorkerPool".format( + self._total_queued, self._expected_total, self._name)) + + return future + + def start(self): + """ Will start up worker pool and reset exception state """ + assert self._workers is None + self._exceptions = False + + self._workers = [asyncio.ensure_future( + self._worker_loop(), loop=self._loop) for _ in range(self._num_workers)] + + async def join(self): + # no-op if workers aren't running + if not self._workers: + return + + self._logger.debug('Joining {}'.format(self._name)) + # The Terminators will kick each worker from being blocked against the _queue.get() and allow + # each one to exit + for _ in range(self._num_workers): + await self._queue.put(Terminator()) + + try: + await asyncio.gather(*self._workers, loop=self._loop) + self._workers = None + except: + self._logger.exception('Exception joining {}'.format(self._name)) + raise + finally: + self._logger.debug('Completed {}'.format(self._name)) + + if self._exceptions and self._raise_on_join: + raise Exception("Exception occurred in pool {}".format(self._name)) + + def _time(self): + # utcnow returns a naive datetime, so we have to set the timezone manually + return datetime.utcnow().replace(tzinfo=timezone.utc) diff --git a/app/Download.py b/app/Download.py index 09e28a23..bc7afad7 100644 --- a/app/Download.py +++ b/app/Download.py @@ -103,7 +103,9 @@ class Download: 'default': self.output_template, 'chapter': self.output_template_chapter }, - 'socket_timeout': 30, + 'socket_timeout': 60, + 'break_per_url': True, + 'ignoreerrors': False, 'progress_hooks': [put_status], 'postprocessor_hooks': [put_status_postprocessor], **mergeConfig(self.default_ytdl_opts, self.ytdl_opts), diff --git a/app/DownloadQueue.py b/app/DownloadQueue.py index ff8d9a19..081ddab5 100644 --- a/app/DownloadQueue.py +++ b/app/DownloadQueue.py @@ -10,7 +10,6 @@ from Download import Download from ItemDTO import ItemDTO from DataStore import DataStore from Utils import Notifier, ObjectSerializer, calcDownloadPath, ExtractInfo, mergeConfig -from datetime import datetime, timezone log = logging.getLogger('DownloadQueue') @@ -55,14 +54,6 @@ class DownloadQueue: error: str = None live_in: str = None - if 'live_status' in entry and entry.get('live_status') == 'is_upcoming': - if 'release_timestamp' in entry and entry.get('release_timestamp'): - live_in = formatdate(entry.get('release_timestamp')) - else: - error = 'Live stream not yet started. And no date is set.' - else: - error = entry.get('msg', None) - etype = entry.get('_type') or 'video' if etype == 'playlist': entries = entry['entries'] @@ -95,6 +86,15 @@ class DownloadQueue: return {'status': 'ok'} elif (etype == 'video' or etype.startswith('url')) and 'id' in entry and 'title' in entry: + # check if the video is live stream. + if 'live_status' in entry and entry.get('live_status') == 'is_upcoming': + if 'release_timestamp' in entry and entry.get('release_timestamp'): + live_in = formatdate(entry.get('release_timestamp')) + else: + error = 'Live stream not yet started. And no date is set.' + else: + error = entry.get('msg', None) + log.debug( f"entry: {entry.get('id', None)} - {entry.get('webpage_url', None)} - {entry.get('url', None) }") @@ -143,17 +143,23 @@ class DownloadQueue: dl.output_template = dl.output_template.replace( f"%({property})s", str(value)) - itemDownload = self.queue.put(Download( + dlInfo: Download = Download( info=dl, download_dir=dldirectory, temp_dir=self.config.temp_path, output_template_chapter=output_chapter, default_ytdl_opts=self.config.ytdl_options, debug=bool(self.config.ytdl_debug) - )) + ) - self.event.set() - await self.notifier.added(itemDownload.info) + if dlInfo.info.live_in: + dlInfo.info.status = 'not_live' + itemDownload = self.done.put(dlInfo) + else: + itemDownload = self.queue.put(dlInfo) + self.event.set() + + await self.notifier.emit('completed' if itemDownload.info.live_in else 'added', itemDownload.info) return { 'status': 'ok' @@ -211,6 +217,8 @@ class DownloadQueue: 'msg': 'No metadata, most likely video has been downloaded before.' if self.config.keep_archive else 'Unable to extract info check logs.' } log.debug(f'entry: extract info says: {entry}') + except yt_dlp.utils.ExistingVideoReached: + return {'status': 'error', 'msg': 'Video has been downloaded already and recorded in archive.log file.'} except yt_dlp.utils.YoutubeDLError as exc: return {'status': 'error', 'msg': str(exc)} diff --git a/frontend/src/assets/css/bulma-dark.css b/frontend/src/assets/css/bulma-dark.css index de0281ad..9b908028 100644 --- a/frontend/src/assets/css/bulma-dark.css +++ b/frontend/src/assets/css/bulma-dark.css @@ -8763,7 +8763,7 @@ #progress { z-index: 1; - background-color: #00d1b2; + background-color: #087363; } } diff --git a/frontend/src/assets/css/style.css b/frontend/src/assets/css/style.css index fb1b6c00..5d6bb3cb 100644 --- a/frontend/src/assets/css/style.css +++ b/frontend/src/assets/css/style.css @@ -18,6 +18,10 @@ border: 1px solid red; } + .is-bordered-info { + border: 1px solid #3e8ed0; + } + html, body { background-color: #eaeaea; @@ -63,6 +67,10 @@ border: 1px solid red; } + .is-bordered-info { + border: 1px solid #3e8ed0; + } + hr { background-color: #fff; } diff --git a/frontend/src/components/Page-Completed.vue b/frontend/src/components/Page-Completed.vue index 75aba9ec..5a378343 100644 --- a/frontend/src/components/Page-Completed.vue +++ b/frontend/src/components/Page-Completed.vue @@ -63,11 +63,21 @@ +
+ +
-
-
+
+
-
+
LIVE stream is scheduled to start at {{ moment(item.datetime).format() }} @@ -223,6 +233,7 @@ const props = defineProps({ const selectedElms = ref([]); const masterSelectAll = ref(false); const showCompleted = useStorage('showCompleted', true) +const direction = useStorage('sortCompleted', 'desc') watch(masterSelectAll, (value) => { for (const key in props.completed) { @@ -235,6 +246,15 @@ watch(masterSelectAll, (value) => { } }) +const sortCompleted = computed(() => { + const thisDirection = direction.value; + return Object.values(props.completed).sort((a, b) => { + if (thisDirection === 'asc') { + return new Date(a.datetime) - new Date(b.datetime); + } + return new Date(b.datetime) - new Date(a.datetime); + }) +}) const hasSelected = computed(() => selectedElms.value.length > 0) const hasItems = computed(() => Object.keys(props.completed)?.length > 0) const getTotal = computed(() => Object.keys(props.completed)?.length); diff --git a/frontend/src/main.js b/frontend/src/main.js index 5b613676..f0ac61a4 100644 --- a/frontend/src/main.js +++ b/frontend/src/main.js @@ -5,7 +5,7 @@ import { library } from '@fortawesome/fontawesome-svg-core' import { FontAwesomeIcon } from '@fortawesome/vue-fontawesome' import { faCog, faTrash, faLink, faPlus, faTrashCan, faCircleXmark, faCircleCheck, faRotateRight, faDownload, faUpRightFromSquare, - faSpinner, faArrowUp, faArrowDown, faTasks, faCalendar + faSpinner, faArrowUp, faArrowDown, faTasks, faCalendar, faArrowUpAZ, faArrowDownAZ } from '@fortawesome/free-solid-svg-icons' import { faSquare, faSquareCheck } from '@fortawesome/free-regular-svg-icons' @@ -17,7 +17,7 @@ import './assets/css/style.css' import '@creativebulma/bulma-tooltip/dist/bulma-tooltip.min.css'; library.add(faCog, faTrash, faLink, faPlus, faTrashCan, faCircleXmark, faCircleCheck, faRotateRight, faDownload, faUpRightFromSquare, - faSquare, faSquareCheck, faSpinner, faArrowUp, faArrowDown, faTasks, faCalendar) + faSquare, faSquareCheck, faSpinner, faArrowUp, faArrowDown, faTasks, faCalendar, faArrowUpAZ, faArrowDownAZ) const app = createApp(App); app.config.globalProperties.capitalize = s => s && s[0].toUpperCase() + s.slice(1);