Merge pull request #262 from arabcoders/dev

Automatically re-queue upcoming live streams in history.

.vscode/settings.json

@@ -48,10 +48,13 @@
     "preferredquality",
     "printtraffic",
     "quicktime",
+    "rtime",
+    "smhd",
     "timespec",
     "tmpfilename",
     "upgrader",
     "urandom",
+    "usegmt",
     "vcodec",
     "vconvert",
     "writedescription",

README.md

@@ -25,6 +25,7 @@ Web GUI for [yt-dlp](https://github.com/yt-dlp/yt-dlp) with playlist & channel s
 * Support for curl_cffi, see [yt-dlp documentation](https://github.com/yt-dlp/yt-dlp?tab=readme-ov-file#impersonation)
 * Support for both advanced and basic mode for WebUI.
 * Bundled tools in container: curl-cffi, ffmpeg, ffprobe, aria2, rtmpdump, mkvtoolsnix, mp4box.
+* Automatic upcoming live stream re-queue.
 
 # Run using docker command
 

app/library/DownloadQueue.py

@@ -5,8 +5,8 @@ import logging
 import os
 import time
 import uuid
-from datetime import UTC, datetime
+from datetime import UTC, datetime, timedelta
-from email.utils import formatdate
+from email.utils import formatdate, parsedate_to_datetime
 from pathlib import Path
 from sqlite3 import Connection
 
@@ -24,7 +24,7 @@ from .ItemDTO import Item, ItemDTO
 from .Presets import Presets
 from .Scheduler import Scheduler
 from .Singleton import Singleton
-from .Utils import arg_converter, calc_download_path, extract_info, is_downloaded, load_cookies
+from .Utils import arg_converter, calc_download_path, dt_delta, extract_info, is_downloaded, load_cookies
 from .YTDLPOpts import YTDLPOpts
 
 LOG = logging.getLogger("DownloadQueue")
@@ -95,9 +95,14 @@ class DownloadQueue(metaclass=Singleton):
         Scheduler.get_instance().add(
             timer="* * * * *",
             func=self.monitor_queue_for_stale_items,
-            id=f"{__class__.__name__}.monitor_queue_for_stale_items",
+            id=f"{__class__.__name__}.{__class__.monitor_queue_for_stale_items.__name__}",
         )
 
+        Scheduler.get_instance().add(
+            timer="* * * * *",
+            func=self.monitor_queue_live,
+            id=f"{__class__.__name__}.{__class__.monitor_queue_live.__name__}",
+        )
         # app.on_shutdown.append(self.on_shutdown)
 
         # async def close_pool(_: web.Application):
@@ -195,6 +200,8 @@ class DownloadQueue(metaclass=Singleton):
             for key in item.extras.copy():
                 if not self.keep_extra_key(key):
                     item.extras.pop(key)
+        else:
+            item.extras = {}
 
         if not entry:
             return {"status": "error", "msg": "Invalid/empty data was given."}
@@ -243,8 +250,9 @@ class DownloadQueue(metaclass=Singleton):
         if ("video" == eventType or eventType.startswith("url")) and "id" in entry and "title" in entry:
             # check if the video is live stream.
             if "live_status" in entry and "is_upcoming" == entry.get("live_status"):
-                if "release_timestamp" in entry and entry.get("release_timestamp"):
+                if entry.get("release_timestamp"):
-                    live_in = formatdate(entry.get("release_timestamp"))
+                    live_in = formatdate(entry.get("release_timestamp"), usegmt=True)
+                    item.extras.update({"live_in": live_in})
                 else:
                     error = "Live stream not yet started. And no date is set."
             else:
@@ -304,7 +312,7 @@ class DownloadQueue(metaclass=Singleton):
                 datetime=formatdate(time.time()),
                 error=error,
                 is_live=is_live,
-                live_in=live_in,
+                live_in=live_in if live_in else item.extras.get("live_in", None),
                 options=options,
                 cli=item.cli,
                 extras=item.extras,
@@ -313,7 +321,7 @@ class DownloadQueue(metaclass=Singleton):
             try:
                 dlInfo: Download = Download(info=dl, info_dict=entry)
 
-                if dlInfo.info.live_in or "is_upcoming" == entry.get("live_status"):
+                if live_in or "is_upcoming" == entry.get("live_status"):
                     NotifyEvent = Events.COMPLETED
                     dlInfo.info.status = "not_live"
                     dlInfo.info.msg = "Stream is not live yet."
@@ -745,14 +753,14 @@ class DownloadQueue(metaclass=Singleton):
             bool: True if the extra key should be kept, False otherwise.
 
         """
-        keys = ("playlist", "external_downloader")
+        keys = ("playlist", "external_downloader", "live_in")
         return any(key == k or key.startswith(k) for k in keys)
 
     async def monitor_queue_for_stale_items(self):
         """
         Monitor the queue for stale items and cancel them if needed.
         """
-        if self.is_paused() or not self.queue.has_downloads():
+        if self.is_paused() or self.queue.empty():
             return
 
         LOG.debug("Checking for stale items in the download queue.")
@@ -767,3 +775,60 @@ class DownloadQueue(metaclass=Singleton):
                 await self.cancel([id])
             except Exception as e:
                 LOG.error(f"Failed to cancel staled item '{item_ref}'. {e!s}")
+                LOG.exception(e)
+
+    async def monitor_queue_live(self):
+        """
+        Monitor the queue for items marked as live events and queue them when time is reached.
+        """
+        if self.is_paused() or self.done.empty():
+            return
+
+        LOG.debug("Checking for live stream items in the history queue.")
+
+        time_now = datetime.now(tz=UTC)
+
+        status = ["not_live", "is_upcoming", "is_live"]
+
+        for id, item in list(self.done.items()):
+            if item.info.status not in status:
+                continue
+
+            item_ref = f"{id=} {item.info.id=} {item.info.title=}"
+            if not item.is_live:
+                LOG.debug(f"Item '{item_ref}' is not a live stream.")
+                continue
+
+            if not item.info.live_in:
+                LOG.debug(f"Item '{item_ref}' marked as live stream, but no date is set.")
+                continue
+
+            starts_in = parsedate_to_datetime(item.info.live_in)
+
+            if time_now < (starts_in + timedelta(minutes=1)):
+                LOG.debug(f"Item '{item_ref}' is not yet live. will start in '{dt_delta(starts_in-time_now)}'.")
+                continue
+
+            LOG.info(f"Re-queuing item '{item_ref} {item.info.extras=}' for download.")
+
+            try:
+                await self.clear([item.info._id], remove_file=False)
+            except Exception as e:
+                LOG.error(f"Failed to clear item '{item_ref}'. {e!s}")
+                continue
+
+            try:
+                info = item.info
+                new_queue = Item(
+                    url=info.url,
+                    preset=info.preset,
+                    folder=info.folder,
+                    cookies=info.cookies,
+                    template=info.template,
+                    cli=item.info.cli,
+                    extras=item.info.extras,
+                )
+                await self.add(item=new_queue)
+            except Exception as e:
+                LOG.error(f"Failed to re-queue item '{item_ref}'. {e!s}")
+                LOG.exception(e)

app/library/Utils.py

@@ -1,6 +1,5 @@
 import base64
 import copy
-import datetime
 import glob
 import ipaddress
 import json
@@ -11,6 +10,7 @@ import re
 import shlex
 import socket
 import uuid
+from datetime import UTC, datetime, timedelta
 from functools import lru_cache
 from http.cookiejar import MozillaCookieJar
 from typing import Any
@@ -68,7 +68,7 @@ class StreamingError(Exception):
 
 class FileLogFormatter(logging.Formatter):
     def formatTime(self, record, datefmt=None):  # noqa: ARG002, N802
-        return datetime.datetime.fromtimestamp(record.created).astimezone().isoformat(timespec="milliseconds")
+        return datetime.fromtimestamp(record.created).astimezone().isoformat(timespec="milliseconds")
 
 
 def calc_download_path(base_path: str, folder: str | None = None, create_path: bool = True) -> str:
@@ -844,8 +844,8 @@ def get_files(base_path: str, dir: str | None = None):
                 "path": str(file).replace(base_path, "").strip("/"),
                 "size": stat.st_size,
                 "mime": get_mime_type({}, file) if file.is_file() else "directory",
-                "mtime": datetime.datetime.fromtimestamp(stat.st_mtime, tz=datetime.UTC).isoformat(),
+                "mtime": datetime.fromtimestamp(stat.st_mtime, tz=UTC).isoformat(),
-                "ctime": datetime.datetime.fromtimestamp(stat.st_ctime, tz=datetime.UTC).isoformat(),
+                "ctime": datetime.fromtimestamp(stat.st_ctime, tz=UTC).isoformat(),
                 "is_dir": file.is_dir(),
                 "is_file": file.is_file(),
             }
@@ -1089,3 +1089,36 @@ def get_archive_id(url: str) -> tuple[bool, dict[str | None, str | None, str | N
         break
 
     return idDict
+
+
+def dt_delta(delta: timedelta) -> str:
+    """
+    Convert a timedelta object to a human-readable string.
+
+    Args:
+        delta (timedelta): The timedelta object.
+
+    Returns:
+        str: A human-readable string representing the timedelta.
+
+    """
+    total_secs = int(delta.total_seconds())
+    days, rem = divmod(total_secs, 86400)
+    hours, rem = divmod(rem, 3600)
+    minutes, secs = divmod(rem, 60)
+
+    parts = []
+    if days:
+        parts.append(f"{days}d")
+    if hours:
+        parts.append(f"{hours}h")
+    if minutes:
+        parts.append(f"{minutes}m")
+    if secs:
+        parts.append(f"{secs}s")
+
+    # if it's under one second
+    if not parts:
+        parts.append("<1s")
+
+    return " ".join(parts)

ui/components/History.vue

@@ -125,16 +125,13 @@
                 </td>
                 <td class="is-vcentered has-text-centered is-unselectable">
                   <span class="user-hint" :date-datetime="item.datetime"
-                    v-tooltip="moment(item.datetime).format('YYYY-M-DD H:mm Z')">
+                    v-tooltip="moment(item.datetime).format('YYYY-M-DD H:mm Z')" v-rtime="item.datetime" />
-                    {{ moment(item.datetime).fromNow() }}
-                  </span>
                 </td>
                 <td class="is-vcentered has-text-centered is-unselectable"
                   v-if="item.live_in && 'not_live' === item.status">
                   <span :date-datetime="item.live_in" class="user-hint"
-                    v-tooltip="'Starts at: ' + moment(item.live_in).format('YYYY-M-DD H:mm Z')">
+                    v-tooltip="'Will automatically be requeued at: ' + moment(item.live_in).format('YYYY-M-DD H:mm Z')"
-                    {{ moment(item.live_in).fromNow() }}
+                    v-rtime="item.live_in" />
-                  </span>
                 </td>
                 <td class="is-vcentered has-text-centered is-unselectable" v-else>
                   {{ item.file_size ? formatBytes(item.file_size) : '-' }}
@@ -259,15 +256,12 @@
               <div class="column is-half-mobile has-text-centered is-text-overflow is-unselectable"
                 v-if="item.live_in && 'not_live' === item.status">
                 <span :date-datetime="item.live_in" class="user-hint"
-                  v-tooltip="'Starts at: ' + moment(item.live_in).format('YYYY-M-DD H:mm Z')">
+                  v-tooltip="'Will automatically be requeued at: ' + moment(item.live_in).format('YYYY-M-DD H:mm Z')"
-                  {{ moment(item.live_in).fromNow() }}
+                  v-rtime="item.live_in" />
-                </span>
               </div>
               <div class="column is-half-mobile has-text-centered is-text-overflow is-unselectable">
                 <span class="user-hint" :date-datetime="item.datetime"
-                  v-tooltip="moment(item.datetime).format('YYYY-M-DD H:mm Z')">
+                  v-tooltip="moment(item.datetime).format('YYYY-M-DD H:mm Z')" v-rtime="item.datetime" />
-                  {{ moment(item.datetime).fromNow() }}
-                </span>
               </div>
               <div class="column is-half-mobile has-text-centered is-text-overflow is-unselectable"
                 v-if="item.file_size">
@@ -624,16 +618,6 @@ const removeItem = item => {
 }
 
 const reQueueItem = (item, event = null) => {
-  let extras = {}
-
-  if (item?.extras) {
-    Object.keys(item.extras).forEach(k => {
-      if (k && true === k.startsWith('playlist')) {
-        extras[k] = item.extras[k]
-      }
-    })
-  }
-
   const item_req = {
     url: item.url,
     preset: item.preset,
@@ -641,7 +625,7 @@ const reQueueItem = (item, event = null) => {
     cookies: item.cookies,
     template: item.template,
     cli: item?.cli,
-    extras: extras
+    extras: item?.extras && Object.keys(item.extras) > 0 ? item.extras : {},
   };
 
   socket.emit('item_delete', { id: item._id, remove_file: false })

ui/components/Queue.vue

@@ -83,10 +83,8 @@
                   </div>
                 </td>
                 <td class="has-text-centered is-text-overflow is-unselectable">
-                  <span :data-datetime="item.datetime"
+                  <span v-tooltip="moment(item.datetime).format('MMMM Do YYYY, h:mm:ss a')"
-                    v-tooltip="moment(item.datetime).format('MMMM Do YYYY, h:mm:ss a')">
+                    :data-datetime="item.datetime" v-rtime="item.datetime" />
-                    {{ moment(item.datetime).fromNow() }}
-                  </span>
                 </td>
                 <td class="is-vcentered is-items-center">
                   <div class="field is-grouped is-grouped-centered">
@@ -167,10 +165,8 @@
                 </span>
               </div>
               <div class="column is-half-mobile has-text-centered is-text-overflow is-unselectable">
-                <span :data-datetime="item.datetime"
+                <span v-tooltip="moment(item.datetime).format('MMMM Do YYYY, h:mm:ss a')" :data-datetime="item.datetime"
-                  v-tooltip="moment(item.datetime).format('MMMM Do YYYY, h:mm:ss a')">
+                  v-rtime="item.datetime" />
-                  {{ moment(item.datetime).fromNow() }}
-                </span>
               </div>
               <div class="column is-half-mobile has-text-centered is-text-overflow is-unselectable">
                 <label class="checkbox is-block">

ui/plugins/rtime.js

@@ -0,0 +1,49 @@
+import { defineNuxtPlugin } from '#app'
+import moment from 'moment'
+
+const parseInterval = arg => {
+  const match = arg.match(/^(\d+)([smhd])$/)
+  if (!match) {
+    return 60 * 1000
+  }
+
+  const [, numStr, unit] = match
+  const num = parseInt(numStr, 10)
+
+  switch (unit) {
+    case 'd': return num * 24 * 3600 * 1000
+    case 'h': return num * 3600 * 1000
+    case 'm': return num * 60 * 1000
+    case 's': return num * 1000
+    default: return 60 * 1000
+  }
+}
+
+export default defineNuxtPlugin(nuxtApp => nuxtApp.vueApp.directive('rtime', {
+  mounted(el, binding) {
+    const intervalMs = binding.arg ? parseInterval(binding.arg) : 60 * 1000
+    const update = () => el.textContent = moment(binding.value).fromNow()
+
+    update()
+    el._next_timer = window.setInterval(update, intervalMs)
+  },
+  updated(el, binding) {
+    if (binding.oldValue !== binding.value) {
+      if (null != el._next_timer) {
+        clearInterval(el._next_timer)
+      }
+
+      const intervalMs = binding.arg ? parseInterval(binding.arg) : 60 * 1000
+
+      const update = () => el.textContent = moment(binding.value).fromNow()
+      update()
+
+      el._next_timer = window.setInterval(update, intervalMs)
+    }
+  },
+  beforeUnmount(el) {
+    if (null != el._next_timer) {
+      clearInterval(el._next_timer)
+    }
+  }
+}))