danvics/ytptube
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1

Automatically re-queue upcoming live streams in history.

Browse source
This commit is contained in:
arabcoders 2025-04-22 16:52:52 +03:00
parent cdcb5d61be
commit 0b79b6f6aa
4 changed files with 101 additions and 13 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
.vscode/settings.json vendored
View file

@ -52,6 +52,7 @@
"tmpfilename",
"upgrader",
"urandom",
"usegmt",
"vcodec",
"vconvert",
"writedescription",

75
app/library/DownloadQueue.py
View file

@ -5,8 +5,8 @@ import logging
import os
import time
import uuid
from datetime import UTC, datetime
from email.utils import formatdate
from datetime import UTC, datetime, timedelta
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):
@ -243,8 +248,8 @@ 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"):
live_in = formatdate(entry.get("release_timestamp"))
if entry.get("release_timestamp"):
live_in = formatdate(entry.get("release_timestamp"), usegmt=True)
else:
error = "Live stream not yet started. And no date is set."
else:
@ -752,7 +757,7 @@ class DownloadQueue(metaclass=Singleton):
"""
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 +772,59 @@ class DownloadQueue(metaclass=Singleton):
await self.cancel([id])
except Exception as e:
LOG.error(f"Failed to cancel staled item '{item_ref}'. {e!s}")
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 at '{dt_delta(starts_in-time_now)}'.")
continue
LOG.info(f"Re-queuing item '{item_ref}' 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}")

32
app/library/Utils.py
View file

@ -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(),
"ctime": datetime.datetime.fromtimestamp(stat.st_ctime, tz=datetime.UTC).isoformat(),
"mtime": datetime.fromtimestamp(stat.st_mtime, tz=UTC).isoformat(),
"ctime": datetime.fromtimestamp(stat.st_ctime, tz=UTC).isoformat(),
"is_dir": file.is_dir(),
"is_file": file.is_file(),
}
@ -1089,3 +1089,27 @@ def get_archive_id(url: str) -> tuple[bool, dict[str | None, str | None, str | N
break
return idDict
def dt_delta(delta: timedelta) -> str:
"""
Turn a timedelta into “1d 20h 5min” style output.
"""
total_secs = int(delta.total_seconds())
days, rem = divmod(total_secs, 86400)
hours, rem = divmod(rem, 3600)
minutes = rem // 60
parts = []
if days:
parts.append(f"{days}d")
if hours:
parts.append(f"{hours}h")
if minutes:
parts.append(f"{minutes}min")
# if it's under one minute:
if not parts:
parts.append("0min")
return " ".join(parts)

6
ui/components/History.vue
View file

@ -131,8 +131,9 @@
</td>
<td class="is-vcentered has-text-centered is-unselectable"
v-if="item.live_in && 'not_live' === item.status">
<span class="icon"><i class="fa-solid fa-spin fa-rotate-right" /></span>
<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() }}
</span>
</td>
@ -258,8 +259,9 @@
</div>
<div class="column is-half-mobile has-text-centered is-text-overflow is-unselectable"
v-if="item.live_in && 'not_live' === item.status">
<span class="icon"><i class="fa-solid fa-spin fa-rotate-right" /></span>
<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() }}
</span>
</div>