Merge pull request #53 from arabcoders/dev

Added sort option for completed downloads & changed dark mode progress bar color.

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 <sigh>
+        return datetime.utcnow().replace(tzinfo=timezone.utc)

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),

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)}
 

frontend/src/assets/css/bulma-dark.css

@@ -8763,7 +8763,7 @@
 
   #progress {
     z-index: 1;
-    background-color: #00d1b2;
+    background-color: #087363;
   }
 
 }

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;
     }

frontend/src/components/Page-Completed.vue

@@ -63,11 +63,21 @@
           </span>
         </button>
       </div>
+      <div class="column is-1">
+        <button type="button" class="button is-fullwidth" @click="direction = direction === 'desc' ? 'asc' : 'desc'">
+          <span class="icon-text">
+            <span class="icon">
+              <font-awesome-icon
+                :icon="direction === 'desc' ? 'fa-solid fa-arrow-down-a-z' : 'fa-solid fa-arrow-up-a-z'" />
+            </span>
+          </span>
+        </button>
+      </div>
     </div>
 
     <div class="columns is-multiline">
-      <div class="column is-6" v-for="item in completed" :key="item._id">
-        <div class="card" :class="{ 'is-bordered-danger': hasError(item) }">
+      <div class="column is-6" v-for="item in sortCompleted" :key="item._id">
+        <div class="card" :class="{ 'is-bordered-danger': hasError(item), 'is-bordered-info': item.live_in }">
           <header class="card-header has-tooltip" :data-tooltip="item.title">
             <div class="card-header-title has-text-centered is-text-overflow is-block">
               <a v-if="item.filename" referrerpolicy="no-referrer" :href="makeDownload(config, item, 'm3u8')"
@@ -79,7 +89,7 @@
           </header>
           <div class="card-content">
             <div class="columns is-multiline">
-              <div class="column is-12" v-if="item.status === 'error' && item.live_in">
+              <div class="column is-12" v-if="item.live_in">
                 <span class="has-text-info">
                   LIVE stream is scheduled to start at {{ moment(item.datetime).format() }}
                 </span>
@@ -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);

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);