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

Merge pull request #311 from arabcoders/dev

Browse source 
Use Asyncio.Semaphore for managing concurrency.
This commit is contained in:
Abdulmohsen 2025-06-24 22:26:10 +03:00 committed by GitHub
commit 1ce9acd316
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
28 changed files with 608 additions and 879 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

5
.github/workflows/main.yml vendored
View file

@ -131,17 +131,20 @@ jobs:
VERSION="${GITHUB_REF##*/}" VERSION="${GITHUB_REF##*/}"
SHA=$(git rev-parse HEAD) SHA=$(git rev-parse HEAD)
DATE=$(date -u +"%Y%m%d") DATE=$(date -u +"%Y%m%d")
BRANCH=$(echo "${GITHUB_REF#refs/heads/}" | sed 's/\//-/g')
echo "Current version: ${VERSION}, SHA: ${SHA}, Date: ${DATE}" echo "Current version: ${VERSION}, Branch: ${BRANCH}, SHA: ${SHA}, Date: ${DATE}"
echo "APP_VERSION=${VERSION}" >> $"$GITHUB_ENV" echo "APP_VERSION=${VERSION}" >> $"$GITHUB_ENV"
echo "APP_SHA=${SHA}" >> "$GITHUB_ENV" echo "APP_SHA=${SHA}" >> "$GITHUB_ENV"
echo "APP_DATE=${DATE}" >> "$GITHUB_ENV" echo "APP_DATE=${DATE}" >> "$GITHUB_ENV"
echo "APP_BRANCH=${BRANCH}" >> "$GITHUB_ENV"
sed -i \ sed -i \
-e "s/^APP_VERSION = \".*\"/APP_VERSION = \"${VERSION}\"/" \ -e "s/^APP_VERSION = \".*\"/APP_VERSION = \"${VERSION}\"/" \
-e "s/^APP_COMMIT_SHA = \".*\"/APP_COMMIT_SHA = \"${SHA}\"/" \ -e "s/^APP_COMMIT_SHA = \".*\"/APP_COMMIT_SHA = \"${SHA}\"/" \
-e "s/^APP_BUILD_DATE = \".*\"/APP_BUILD_DATE = \"${DATE}\"/" \ -e "s/^APP_BUILD_DATE = \".*\"/APP_BUILD_DATE = \"${DATE}\"/" \
-e "s/^APP_BRANCH = \".*\"/APP_BRANCH = \"${BRANCH}\"/" \
app/library/version.py app/library/version.py
- name: Set up QEMU - name: Set up QEMU

320
app/library/AsyncPool.py
View file

@ -1,320 +0,0 @@
import asyncio
import logging
import uuid
from datetime import UTC, datetime
class Terminator:
pass
class AsyncPool:
def __init__(
self,
num_workers: int,
worker_co,
name: str,
logger: logging.Logger,
loop: asyncio.AbstractEventLoop | None = None,
load_factor: int = 1,
max_task_time: int | None = None,
return_futures: bool = False,
raise_on_join: bool = False,
):
"""
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.
Args:
num_workers (int): number of async tasks which will pull from the internal queue
worker_co (coroutine): async coroutine to call when an item is retrieved from the queue
name (str): name of the worker pool (used for logging)
logger (logging.Logger): logger to use
loop (asyncio.AbstractEventLoop|None): asyncio loop to use
load_factor (int): multiplier used for number of items in queue
max_task_time (int|None): maximum time allowed for each task before a CancelledError is raised in the task.
return_futures (bool): set to reture to return a future for each `push` (imposes CPU overhead)
raise_on_join (bool): raise on join if any exceptions have occurred, default is False
Returns:
AsyncPool: instance of AsyncPool
"""
self._loop = loop if loop else asyncio.get_event_loop()
self._num_workers = num_workers
self._logger = logger if logger else logging.getLogger(__name__)
self._queue = asyncio.Queue(num_workers * load_factor)
self._workers: dict[str, asyncio.Future] = {}
self._exceptions = False
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._status: dict[str, dict | None] = {}
@property
def exceptions(self):
return self._exceptions
async def _worker_loop(self, worker_id: str):
"""
The main persistent worker loop that continuously processes jobs from the shared queue.
Args:
worker_id (str): the id of the worker processing this job.
"""
while True:
item = await self._queue.get()
should_continue = await self._process_item(worker_id, item, from_queue=True)
if not should_continue:
break
async def _process_item(self, worker_id: str, item: tuple | Terminator, from_queue: bool = True) -> bool:
"""
Processes a single job item.
Args:
worker_id (str): the id of the worker processing this job.
item (tuple|Terminator): the job item (tuple of (future, args, kwargs)) or a Terminator.
from_queue (bool): indicates whether the job came from the shared queue; if True, task_done() is called.
Returns:
bool: False if the item is a Terminator (indicating termination), True otherwise.
"""
future = None
is_terminator = isinstance(item, Terminator)
try:
if is_terminator:
return False
future, args, kwargs = item
self._status[worker_id] = {
"started": self._time().isoformat(),
"data": kwargs.get("entry", {"info": {}}).info.__dict__,
}
result = await asyncio.wait_for(self._worker_co(*args, **kwargs), timeout=self._max_task_time)
if future:
future.set_result(result)
except (KeyboardInterrupt, MemoryError, SystemExit) as e:
if future:
future.set_exception(e)
self._exceptions = True
raise
except Exception as e:
self._exceptions = True
if future:
future.set_exception(e)
else:
self._logger.exception(f"Worker call failed. {e!s}")
finally:
self._status[worker_id] = None
if not is_terminator and from_queue:
self._queue.task_done()
return True
def has_open_workers(self) -> bool:
"""
Check if there are open workers.
Returns:
bool: True if there are open workers, False otherwise.
"""
return self.get_available_workers() > 0
def get_available_workers(self) -> int:
"""
Get the number of available workers.
Returns:
int: number of available workers.
"""
return sum(1 for worker_status in self._status.values() if worker_status is None)
def get_workers_status(self) -> dict[str, datetime | None]:
"""
Get the status of all workers.
Returns:
dict: dictionary of worker status.
"""
return self._status
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, is_temp: bool = False, **kwargs) -> asyncio.Future:
"""
Push work to the worker_co. If is_temp is True, a temporary worker is spawned
Args:
args: position arguments to be passed to `worker_co`
is_temp (bool): flag to indicate if the job is temporary, default is False
kwargs (dict): keyword arguments to be passed to `worker_co`
Returns:
asyncio.Future: future of result.
"""
future = asyncio.futures.Future(loop=self._loop) if self._return_futures else None
if is_temp:
temp_worker_id = f"temp_worker_{uuid.uuid4()!s}"
self._logger.info(f"Creating temporary worker '{temp_worker_id}'.")
# Spawn a temporary worker that processes this one job.
task = asyncio.ensure_future(
self._process_item(temp_worker_id, (future, args, kwargs), from_queue=False), loop=self._loop
)
self._workers[temp_worker_id] = task
# Attach a callback to clean up temporary worker entries when done.
def cleanup_temp_worker(_):
self._workers.pop(temp_worker_id, None)
self._status.pop(temp_worker_id, None)
self._logger.info(f"Temporary worker '{temp_worker_id}' has terminated.")
task.add_done_callback(cleanup_temp_worker)
return future
await self._queue.put((future, args, kwargs))
self._logger.debug(f"'{self._name}' pool has received a new job. {args} {kwargs}")
return future
def start(self):
"""
Will start up worker pool and reset exception state
"""
self._exceptions = False
for worker_number in range(self._num_workers):
worker_id = f"worker_{worker_number+1}"
self._create_worker(worker_id)
async def restart(self, worker_id: str, msg: str | None = None) -> bool:
"""
Will restart the worker pool
Args:
worker_id (str): worker id to restart
msg (str|None): message to send to the worker, default is None
Returns:
bool: True if worker was restarted.
"""
if worker_id not in self._workers:
self._logger.warning(f"Worker {worker_id} does not exist.")
return False
try:
self._workers[worker_id].cancel(msg)
await self._workers[worker_id]
except asyncio.exceptions.CancelledError as e:
self._logger.warning(f"Worker {worker_id} restarted. {e!s}")
if worker_id in self._status:
self._status.pop(worker_id)
if worker_id in self._workers:
self._workers.pop(worker_id)
self._create_worker(worker_id)
return True
async def on_shutdown(self, _) -> None:
"""
Will shutdown the worker pool
"""
try:
await asyncio.wait_for(asyncio.gather(*[self.stop(worker_id) for worker_id in self._workers]), timeout=10)
except Exception:
self._logger.error(f"Exception shutting down {self._name}")
async def stop(self, worker_id: str, msg: str | None = None) -> bool:
"""
Will stop the worker
Args:
worker_id (str): worker id to stop
msg (str|None): message to send to the worker, default is None
Returns:
bool: True if worker was stopped.
"""
if worker_id not in self._workers:
self._logger.warning(f"Worker {worker_id} does not exist.")
return False
if self._workers[worker_id].cancel(msg):
try:
await self._workers[worker_id]
except asyncio.exceptions.CancelledError as e:
self._logger.warning(f"Worker {worker_id} stopped. {e!s}")
if worker_id in self._status:
self._status.pop(worker_id)
if worker_id in self._workers:
self._workers.pop(worker_id)
return True
async def join(self) -> None:
# no-op if workers aren't running
if len(self._workers) < 1:
return
self._logger.info(f"Joining {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(*list(self._workers))
self._workers = {}
except:
self._logger.exception(f"Exception joining {self._name}")
raise
finally:
self._logger.info(f"Completed {self._name}")
if self._exceptions and self._raise_on_join:
msg = f"Exception occurred in {self._name} pool"
raise Exception(msg)
def _create_worker(self, worker_id: str) -> asyncio.Future:
if worker_id in self._workers:
self._logger.debug(f"Worker {worker_id} already exists.")
return self._workers[worker_id]
self._status[worker_id] = None
self._workers[worker_id] = asyncio.ensure_future(
coro_or_future=self._worker_loop(worker_id=worker_id), loop=self._loop
)
self._logger.debug(f"Created {worker_id}")
return self._workers[worker_id]
def _time(self) -> datetime:
return datetime.now(tz=UTC)

20
app/library/Download.py
View file

@ -11,7 +11,6 @@ from pathlib import Path
import yt_dlp import yt_dlp
from .AsyncPool import Terminator
from .config import Config from .config import Config
from .Events import EventBus, Events from .Events import EventBus, Events
from .ffprobe import ffprobe from .ffprobe import ffprobe
@ -20,6 +19,10 @@ from .Utils import delete_dir, extract_info, extract_ytdlp_logs, load_cookies
from .YTDLPOpts import YTDLPOpts from .YTDLPOpts import YTDLPOpts
class Terminator:
pass
class NestedLogger: class NestedLogger:
debug_messages = ["[debug] ", "[download] "] debug_messages = ["[debug] ", "[download] "]
@ -120,6 +123,15 @@ class Download:
self.logs = logs if logs else [] self.logs = logs if logs else []
def _progress_hook(self, data: dict): def _progress_hook(self, data: dict):
if self.debug:
from copy import deepcopy
d_copy = deepcopy(data)
for k in ["formats", "thumbnails", "description", "tags", "_format_sort_fields"]:
d_copy["info_dict"].pop(k, None)
self.logger.debug(f"Progress hook: {d_copy}")
dataDict = {k: v for k, v in data.items() if k in self._ytdlp_fields} dataDict = {k: v for k, v in data.items() if k in self._ytdlp_fields}
if "finished" == data.get("status") and data.get("info_dict", {}).get("filename", None): if "finished" == data.get("status") and data.get("info_dict", {}).get("filename", None):
@ -269,9 +281,15 @@ class Download:
ret = cls.download(url_list=[self.info.url]) ret = cls.download(url_list=[self.info.url])
self.status_queue.put({"id": self.id, "status": "finished" if ret == 0 else "error"}) self.status_queue.put({"id": self.id, "status": "finished" if ret == 0 else "error"})
except yt_dlp.utils.ExistingVideoReached as exc:
self.logger.error(exc)
self.status_queue.put({"id": self.id, "status": "skip", "msg": "Item has already been downloaded."})
except Exception as exc: except Exception as exc:
self.logger.exception(exc) self.logger.exception(exc)
self.logger.error(exc)
self.status_queue.put({"id": self.id, "status": "error", "msg": str(exc), "error": str(exc)}) self.status_queue.put({"id": self.id, "status": "error", "msg": str(exc), "error": str(exc)})
finally:
self.status_queue.put(Terminator())
self.logger.info(f'Task id="{self.info.id}" PID="{os.getpid()}" title="{self.info.title}" completed.') self.logger.info(f'Task id="{self.info.id}" PID="{os.getpid()}" title="{self.info.title}" completed.')

255
app/library/DownloadQueue.py
View file

@ -15,7 +15,6 @@ from aiohttp import web
from app.library.ag_utils import ag from app.library.ag_utils import ag
from .AsyncPool import AsyncPool
from .conditions import Conditions from .conditions import Conditions
from .config import Config from .config import Config
from .DataStore import DataStore from .DataStore import DataStore
@ -62,9 +61,6 @@ class DownloadQueue(metaclass=Singleton):
event: asyncio.Event event: asyncio.Event
"""Event to signal the download queue to start downloading.""" """Event to signal the download queue to start downloading."""
pool: AsyncPool | None = None
"""Pool of workers to download the files."""
_active: dict[str, Download] = {} _active: dict[str, Download] = {}
"""Dictionary of active downloads.""" """Dictionary of active downloads."""
@ -77,6 +73,12 @@ class DownloadQueue(metaclass=Singleton):
done: DataStore done: DataStore
"""DataStore for the completed downloads.""" """DataStore for the completed downloads."""
workers: asyncio.Semaphore
"""Semaphore to limit the number of concurrent downloads."""
processors: asyncio.Semaphore
"""Semaphore to limit the number of concurrent processors."""
def __init__(self, connection: Connection, config: Config | None = None): def __init__(self, connection: Connection, config: Config | None = None):
DownloadQueue._instance = self DownloadQueue._instance = self
@ -89,6 +91,8 @@ class DownloadQueue(metaclass=Singleton):
self.paused = asyncio.Event() self.paused = asyncio.Event()
self.paused.set() self.paused.set()
self.event = asyncio.Event() self.event = asyncio.Event()
self.workers = asyncio.Semaphore(self.config.max_workers)
self.processors = asyncio.Semaphore(self.config.playlist_items_concurrency)
@staticmethod @staticmethod
def get_instance(): def get_instance():
@ -104,15 +108,17 @@ class DownloadQueue(metaclass=Singleton):
return DownloadQueue._instance return DownloadQueue._instance
def attach(self, app: web.Application): def attach(self, _: web.Application):
""" """
Attach the download queue to the application. Attach the download queue to the application.
Args: Args:
app (web.Application): The application to attach the download queue to. _ (web.Application): The application to attach the download queue to.
""" """
app.on_startup.append(lambda _: self.initialize()) self._notify.subscribe(
Events.STARTED, lambda _, __: self.initialize(), f"{__class__.__name__}.{__class__.initialize.__name__}"
)
Scheduler.get_instance().add( Scheduler.get_instance().add(
timer="* * * * *", timer="* * * * *",
@ -127,14 +133,6 @@ class DownloadQueue(metaclass=Singleton):
) )
# app.on_shutdown.append(self.on_shutdown) # app.on_shutdown.append(self.on_shutdown)
# async def close_pool(_: web.Application):
# try:
# await self.pool.on_shutdown(_)
# except Exception as e:
# LOG.error(f"Failed to cleanup download pool. {e!s}")
# app.on_cleanup.append(close_pool)
async def test(self) -> bool: async def test(self) -> bool:
""" """
Test the datastore connection to the database. Test the datastore connection to the database.
@ -206,81 +204,56 @@ class DownloadQueue(metaclass=Singleton):
LOG.error(f"Failed to cancel downloads. {e!s}") LOG.error(f"Failed to cancel downloads. {e!s}")
async def _process_playlist(self, entry: dict, item: Item, already=None): async def _process_playlist(self, entry: dict, item: Item, already=None):
if 1 == self.config.playlist_items_concurrency:
return await self._process_playlist_old(entry=entry, item=item, already=already)
LOG.info(f"Playlist '{entry.get('id')}: {entry.get('title')}' processing.")
entries = entry.get("entries", []) entries = entry.get("entries", [])
LOG.info(f"Processing '{entry.get('id')}: {entry.get('title')}' Playlist.")
playlistCount = int(entry.get("playlist_count", len(entries))) playlistCount = int(entry.get("playlist_count", len(entries)))
results = [] results = []
semaphore = asyncio.Semaphore(self.config.playlist_items_concurrency) async def playlist_processor(i: int, etr: dict):
await self.processors.acquire()
try:
LOG.debug(f"Processing entry {i}/{playlistCount} - ID: {etr.get('id')} - Title: {etr.get('title')}")
async def process_entry(i, etr): extras = {
extras = { "playlist": entry.get("title") or entry.get("id"),
"playlist": entry.get("id"), "playlist_count": playlistCount,
"playlist_index": f"{{0:0{len(str(playlistCount))}d}}".format(i), "playlist_id": entry.get("id"),
"playlist_autonumber": i, "playlist_title": entry.get("title"),
} "playlist_uploader": entry.get("uploader"),
"playlist_uploader_id": entry.get("uploader_id"),
"playlist_channel": entry.get("channel"),
"playlist_channel_id": entry.get("channel_id"),
"playlist_webpage_url": entry.get("webpage_url"),
"playlist_index": f"{{0:0{len(str(playlistCount))}d}}".format(i),
"playlist_autonumber": i + 1,
}
for property in ("id", "title", "uploader", "uploader_id"): for property in ("id", "title", "uploader", "uploader_id"):
if property in entry: if property in entry:
extras[f"playlist_{property}"] = entry.get(property) extras[f"playlist_{property}"] = entry.get(property)
LOG.debug(f"Processing entry {i}/{playlistCount} - ID: {etr.get('id')} - Title: {etr.get('title')}") if "thumbnail" not in etr and "youtube:" in entry.get("extractor", ""):
extras["thumbnail"] = f"https://img.youtube.com/vi/{etr['id']}/maxresdefault.jpg"
if "thumbnail" not in etr and "youtube:" in entry.get("extractor", ""):
extras["thumbnail"] = f"https://img.youtube.com/vi/{etr['id']}/maxresdefault.jpg"
async with semaphore:
return await self.add( return await self.add(
item=item.new_with(url=etr.get("url") or etr.get("webpage_url"), extras=extras), item=item.new_with(url=etr.get("url") or etr.get("webpage_url"), extras=extras),
already=already, already=already,
) )
finally:
self.processors.release()
tasks = [process_entry(i, etr) for i, etr in enumerate(entries, start=1)] tasks: list[asyncio.Task] = []
results = await asyncio.gather(*tasks)
LOG.info(
f"Playlist '{entry.get('id')}: {entry.get('title')}' processing completed with '{len(results)}' entries."
)
if any("error" == res["status"] for res in results):
return {
"status": "error",
"msg": ", ".join(res["msg"] for res in results if res["status"] == "error" and "msg" in res),
}
return {"status": "ok"}
async def _process_playlist_old(self, entry: dict, item: Item, already=None):
LOG.info(f"Playlist '{entry.get('id')}: {entry.get('title')}' processing.")
entries = entry.get("entries", [])
playlistCount = int(entry.get("playlist_count", len(entries)))
results = []
for i, etr in enumerate(entries, start=1): for i, etr in enumerate(entries, start=1):
extras = { task = asyncio.create_task(
"playlist": entry.get("id"), playlist_processor(i, etr),
"playlist_index": f"{{0:0{len(str(playlistCount))}d}}".format(i), name=f"playlist_processor_{etr.get('id')}_{i}",
"playlist_autonumber": i,
}
for property in ("id", "title", "uploader", "uploader_id"):
if property in entry:
extras[f"playlist_{property}"] = entry.get(property)
if "thumbnail" not in etr and "youtube:" in entry.get("extractor", ""):
extras["thumbnail"] = f"https://img.youtube.com/vi/{etr['id']}/maxresdefault.jpg"
LOG.debug(f"Processing entry {i}/{playlistCount} - ID: {etr.get('id')} - Title: {etr.get('title')}")
results.append(
await self.add(
item=item.new_with(url=etr.get("url") or etr.get("webpage_url"), extras=extras),
already=already,
)
) )
task.add_done_callback(self._handle_task_exception)
tasks.append(task)
results: list[dict] = await asyncio.gather(*tasks)
LOG.info( LOG.info(
f"Playlist '{entry.get('id')}: {entry.get('title')}' processing completed with '{len(results)}' entries." f"Playlist '{entry.get('id')}: {entry.get('title')}' processing completed with '{len(results)}' entries."
@ -541,9 +514,8 @@ class DownloadQueue(metaclass=Singleton):
downloaded, id_dict = self._is_downloaded(file=yt_conf.get("download_archive", None), url=item.url) downloaded, id_dict = self._is_downloaded(file=yt_conf.get("download_archive", None), url=item.url)
if downloaded is True and id_dict: if downloaded is True and id_dict:
message = f"This url with ID '{id_dict.get('id')}' has been downloaded already and recorded in archive." message = f"'{id_dict.get('id')}': The URL '{item.url}' is already downloaded and recorded in archive."
LOG.info(message) LOG.error(message)
await self._notify.emit(Events.LOG_WARNING, data=event_warning(message))
return {"status": "error", "msg": message} return {"status": "error", "msg": message}
started: float = time.perf_counter() started: float = time.perf_counter()
@ -576,6 +548,7 @@ class DownloadQueue(metaclass=Singleton):
) )
if not entry: if not entry:
LOG.error(f"Unable to extract info for '{item.url}'. Logs: {logs}")
return {"status": "error", "msg": "Unable to extract info." + "\n".join(logs)} return {"status": "error", "msg": "Unable to extract info." + "\n".join(logs)}
if not item.requeued and (condition := Conditions.get_instance().match(info=entry)): if not item.requeued and (condition := Conditions.get_instance().match(info=entry)):
@ -752,56 +725,45 @@ class DownloadQueue(metaclass=Singleton):
async def _download_pool(self) -> None: async def _download_pool(self) -> None:
""" """
Create a pool of workers to download the files. Create a pool of workers to download the files.
Returns:
None
""" """
self.pool = AsyncPool(
loop=asyncio.get_running_loop(),
num_workers=self.config.max_workers,
worker_co=self._download_file,
name="download_pool",
logger=logging.getLogger("WorkerPool"),
)
self.pool.start()
lastLog = time.time()
while True: while True:
while True:
if self.pool.has_open_workers() is True:
break
if self.config.max_workers > 1 and time.time() - lastLog > 600:
lastLog = time.time()
LOG.info("Waiting for worker to be free.", extra={"workers": self.pool.get_available_workers()})
await asyncio.sleep(1)
while not self.queue.has_downloads(): while not self.queue.has_downloads():
LOG.info(f"Waiting for item to download. '{self.pool.get_available_workers()}' free workers.") LOG.info("Waiting for item to download.")
if self.event: await self.event.wait()
await self.event.wait() self.event.clear()
self.event.clear()
LOG.debug("Cleared wait event.")
if self.paused and isinstance(self.paused, asyncio.Event) and self.is_paused(): if self.is_paused():
LOG.info("Download pool is paused.") LOG.info("Download pool is paused.")
await self.paused.wait() await self.paused.wait()
LOG.info("Download pool resumed downloading.") LOG.info("Download pool resumed downloading.")
entry = self.queue.get_next_download() for _id, entry in list(self.queue.items()):
await asyncio.sleep(0.2) if entry.started() or entry.is_cancelled():
continue
if entry is None: if entry.is_live:
continue task = asyncio.create_task(self._download_live(_id, entry), name=f"download_live_{_id}")
task.add_done_callback(self._handle_task_exception)
else:
await self.workers.acquire()
LOG.debug(f"Pushing {entry=} to executor.") task = asyncio.create_task(self._download_file(_id, entry), name=f"download_file_{_id}")
if entry.started() is False and entry.is_cancelled() is False: def _release_semaphore(t: asyncio.Task):
await self.pool.push(is_temp=entry.is_live, id=entry.info._id, entry=entry) self.workers.release()
LOG.debug(f"Pushed {entry=} to executor.") self._handle_task_exception(t)
await asyncio.sleep(1)
task.add_done_callback(_release_semaphore)
await asyncio.sleep(0.5)
async def _download_live(self, _id: str, entry: Download) -> None:
LOG.info(f"Creating temporary worker for entry '{entry.info.name()}'.")
try:
await self._download_file(_id, entry)
finally:
LOG.info(f"Temporary worker for '{entry.info.name()}' completed.")
async def _download_file(self, id: str, entry: Download) -> None: async def _download_file(self, id: str, entry: Download) -> None:
""" """
@ -822,7 +784,7 @@ class DownloadQueue(metaclass=Singleton):
self._active[entry.info._id] = entry self._active[entry.info._id] = entry
await entry.start() await entry.start()
if "finished" != entry.info.status: if entry.info.status not in ("finished", "skip"):
entry.info.status = "error" entry.info.status = "error"
finally: finally:
if entry.info._id in self._active: if entry.info._id in self._active:
@ -886,52 +848,6 @@ class DownloadQueue(metaclass=Singleton):
LOG.error(f"Failed to cancel staled item '{item_ref}'. {e!s}") LOG.error(f"Failed to cancel staled item '{item_ref}'. {e!s}")
LOG.exception(e) LOG.exception(e)
if self.pool:
time_now = datetime.now(tz=UTC)
workers = self.pool.get_workers_status()
if len(workers) > 0:
LOG.debug(f"Checking for stale workers. {len(workers)} workers found.")
for worker_id in workers:
worker = workers.get(worker_id, {})
if not worker:
continue
started = worker.get("started", None)
if not started:
LOG.debug(f"Worker '{worker_id}' not working yet.")
continue
started = datetime.fromisoformat(started)
if time_now - started < timedelta(minutes=5):
LOG.debug(f"Worker '{worker_id}' is not consider stale yet.")
continue
data = worker.get("data", {})
status = data.get("status", None)
if "preparing" != status:
LOG.debug(f"Worker '{worker_id}' not stuck. Status '{status}'.")
continue
_id = data.get("data._id", None)
if not _id:
LOG.debug(f"Worker '{worker_id}' has no id.")
continue
id = data.get("id", None)
title = data.get("title", None)
item_ref = f"{_id=} {id=} {title=}"
LOG.warning(f"Cancelling staled item '{item_ref}' from worker pool.")
try:
await self.cancel([_id])
except Exception as e:
LOG.error(f"Failed to cancel staled item '{item_ref}' from worker pool. {e!s}")
LOG.exception(e)
async def _check_live(self): async def _check_live(self):
""" """
Monitor the queue for items marked as live events and queue them when time is reached. Monitor the queue for items marked as live events and queue them when time is reached.
@ -979,7 +895,7 @@ class DownloadQueue(metaclass=Singleton):
) )
continue continue
LOG.info(f"Re-queuing item '{item_ref} {item.info.extras=}' for download.") LOG.info(f"Retrying item '{item_ref} {item.info.extras=}' for download.")
try: try:
await self.clear([item.info._id], remove_file=False) await self.clear([item.info._id], remove_file=False)
@ -1002,4 +918,11 @@ class DownloadQueue(metaclass=Singleton):
except Exception as e: except Exception as e:
self.done.put(item) self.done.put(item)
LOG.exception(e) LOG.exception(e)
LOG.error(f"Failed to re-queue item '{item_ref}'. {e!s}") LOG.error(f"Failed to retry item '{item_ref}'. {e!s}")
def _handle_task_exception(self, task: asyncio.Task):
if task.cancelled():
return
if exc := task.exception():
LOG.error(f"Unhandled exception in background task: {exc!s}")

71
app/library/Events.py
View file

@ -4,6 +4,7 @@ import logging
import uuid import uuid
from collections.abc import Awaitable from collections.abc import Awaitable
from dataclasses import dataclass, field from dataclasses import dataclass, field
from typing import Any
from .Singleton import Singleton from .Singleton import Singleton
@ -93,6 +94,7 @@ class Events:
STARTUP = "startup" STARTUP = "startup"
LOADED = "loaded" LOADED = "loaded"
STARTED = "started"
SHUTDOWN = "shutdown" SHUTDOWN = "shutdown"
ADDED = "added" ADDED = "added"
@ -245,7 +247,7 @@ class EventListener:
if self.is_coroutine: if self.is_coroutine:
return self.call_back(event, self.name, **kwargs) return self.call_back(event, self.name, **kwargs)
return asyncio.create_task(self.call_back(event, self.name, **kwargs)) return asyncio.create_task(self.call_back(event, self.name, **kwargs), name=f"EL-{self.name}-{event.id}")
class EventBus(metaclass=Singleton): class EventBus(metaclass=Singleton):
@ -323,6 +325,8 @@ class EventBus(metaclass=Singleton):
self._listeners[e][name] = EventListener(name, callback) self._listeners[e][name] = EventListener(name, callback)
LOG.debug(f"'{name}' subscribed to '{event}'.")
return self return self
def unsubscribe(self, event: str | list | tuple, name: str) -> "EventBus": def unsubscribe(self, event: str | list | tuple, name: str) -> "EventBus":
@ -340,50 +344,83 @@ class EventBus(metaclass=Singleton):
if isinstance(event, str): if isinstance(event, str):
event = [event] event = [event]
events = []
for e in event: for e in event:
if e in self._listeners and name in self._listeners[e]: if e in self._listeners and name in self._listeners[e]:
events.append(e)
del self._listeners[e][name] del self._listeners[e][name]
if len(events) > 0:
LOG.debug(f"'{name}' unsubscribed from '{events}'.")
return self return self
def sync_emit(self, event: str, data: any, **kwargs) -> list: def sync_emit(self, event: str, data: Any, loop=None, wait: bool = True, **kwargs):
""" """
Emit an event synchronously. Emit event and (optionally) wait for results.
Args: Args:
event (str): The event to emit. event (str): The event to emit.
data (any): The data to pass to the event. data (Any): The data to pass to the event.
**kwargs: The keyword arguments to pass to the event. loop (asyncio.AbstractEventLoop | None): The event loop to use. If None, the current running loop is used.
wait (bool): Whether to wait for the results of the event handlers. Defaults to True.
**kwargs: Additional keyword arguments to pass to the event
Returns: Returns:
list: The results are the return values of the coroutines. If the coroutine raises an exception, list: The results are the return values of the coroutines. If the coroutine raises an exception,
the exception is caught and logged. If event does not exist, an empty list is returned. the exception is caught and logged. If event does not exist, an empty list is returned.
If wait is False, a list of asyncio.Task objects is returned instead if we are in a running event loop,
or the result of the coroutine if we are not in a running event loop.
""" """
if event not in self._listeners: if event not in self._listeners:
return [] return []
ev = Event(event=event, data=data) async def emit_all():
LOG.debug(f"Emitting event '{ev.id}: {ev.event}'.", extra={"data": data}) ev = Event(event=event, data=data)
LOG.debug(f"Emitting event '{ev.id}: {ev.event}'.", extra={"data": data})
results = [] res: list = []
for handler in self._listeners[event].values(): for h in self._listeners[event].values():
try: try:
results.append(asyncio.get_event_loop().run_until_complete(handler.handle(ev, **kwargs))) res.append(await h.handle(ev, **kwargs))
except Exception as e: except Exception as e:
LOG.exception(e) LOG.exception(e)
LOG.error(f"Failed to emit event '{event}' to '{handler.name}'. Error message '{e!s}'.") LOG.error(f"Failed to emit event '{event}' to '{h.name}'. Error message '{e!s}'.")
return results return res
async def emit(self, event: str, data: any, **kwargs) -> Awaitable: try:
loop = loop or asyncio.get_running_loop()
in_same_loop: bool = asyncio.get_running_loop() is loop
except RuntimeError:
loop = None
in_same_loop = False
if loop is None or not loop.is_running():
return asyncio.run(emit_all())
if in_same_loop:
if wait:
msg = (
"Calling EventsBus.sync_emit(...,wait=True) from within the running event loop would cause dead-lock. "
"Use `await EventsBus.emit(...)` or `EventsBus.sync_emit(..., wait=False)`."
)
raise RuntimeError(msg)
return loop.create_task(emit_all())
fut = asyncio.run_coroutine_threadsafe(emit_all(), loop)
return fut.result() if wait else fut
async def emit(self, event: str, data: Any, **kwargs) -> Awaitable:
""" """
Emit an event. Emit an event.
Args: Args:
event (str): The event to emit. event (str): The event to emit.
data (any): The data to pass to the event. data (Any): The data to pass to the event.
**kwargs: The keyword arguments to pass to the event. **kwargs: The keyword arguments to pass to the event.
Returns: Returns:

17
app/library/HttpAPI.py
View file

@ -153,7 +153,6 @@ class HttpAPI:
Args: Args:
username (str): The username. username (str): The username.
password (str): The password. password (str): The password.
this (HttpAPI): The instance of the HttpAPI.
Returns: Returns:
Awaitable: The middleware handler. Awaitable: The middleware handler.
@ -213,9 +212,10 @@ class HttpAPI:
}, },
) )
response = await handler(request) response: Response = await handler(request)
if request.path != "/": contentType: str | None = response.headers.get("content-type", None)
if contentType and not contentType.startswith("text/html"):
return response return response
try: try:
@ -226,7 +226,7 @@ class HttpAPI:
key=Config.get_instance().secret_key, key=Config.get_instance().secret_key,
), ),
max_age=60 * 60 * 24 * 7, max_age=60 * 60 * 24 * 7,
expires=datetime.now(UTC) + timedelta(days=7), expires=(datetime.now(UTC) + timedelta(days=7)).strftime("%a, %d %b %Y %H:%M:%S GMT"),
httponly=True, httponly=True,
samesite="Strict", samesite="Strict",
) )
@ -259,7 +259,7 @@ class HttpAPI:
}, },
) )
response = await handler(request) response: Response = await handler(request)
contentType: str | None = response.headers.get("content-type", None) contentType: str | None = response.headers.get("content-type", None)
if contentType and "/" != base_path and contentType.startswith("text/html"): if contentType and "/" != base_path and contentType.startswith("text/html"):
@ -272,11 +272,8 @@ class HttpAPI:
.replace('baseURL:""', f'baseURL:"{rewrite_path}/"') .replace('baseURL:""', f'baseURL:"{rewrite_path}/"')
) )
return web.Response( response.body = content.encode("utf-8")
body=content.encode("utf-8"), return response
status=response.status,
headers=response.headers,
)
if request.path.startswith(static_path) and isinstance(response, web.FileResponse): if request.path.startswith(static_path) and isinstance(response, web.FileResponse):
try: try:

2
app/library/ItemDTO.py
View file

@ -39,7 +39,7 @@ class Item:
"""Extra data to be added to the download.""" """Extra data to be added to the download."""
requeued: bool = False requeued: bool = False
"""If the item has been re-queued already via conditions.""" """If the item has been retried already via conditions."""
def serialize(self) -> dict: def serialize(self) -> dict:
return self.__dict__.copy() return self.__dict__.copy()

2
app/library/Notifications.py
View file

@ -137,7 +137,7 @@ class Notification(metaclass=Singleton):
self._file: Path = Path(file) if file else Path(config.config_path).joinpath("notifications.json") self._file: Path = Path(file) if file else Path(config.config_path).joinpath("notifications.json")
self._client: httpx.AsyncClient = client or httpx.AsyncClient() self._client: httpx.AsyncClient = client or httpx.AsyncClient()
self._encoder: Encoder = encoder or Encoder() self._encoder: Encoder = encoder or Encoder()
self._version = config.version self._version = config.app_version
if self._file.exists() and "600" != self._file.stat().st_mode: if self._file.exists() and "600" != self._file.stat().st_mode:
try: try:

24
app/library/Scheduler.py
View file

@ -74,9 +74,31 @@ class Scheduler(metaclass=Singleton):
return self._jobs return self._jobs
def get(self, id: str) -> Cron | None: def get(self, id: str) -> Cron | None:
"""Return the job by id.""" """
Get a job by its id.
Args:
id (str): The id of the job.
Returns:
Cron | None: The job if it exists, None otherwise
"""
return self._jobs.get(id) return self._jobs.get(id)
def has(self, id: str) -> bool:
"""
Check if a job with the given id exists.
Args:
id (str): The id of the job.
Returns:
bool: True if the job exists, False otherwise
"""
return id in self._jobs
def add( def add(
self, timer: str, func: callable, args: tuple = (), kwargs: dict | None = None, id: str | None = None self, timer: str, func: callable, args: tuple = (), kwargs: dict | None = None, id: str | None = None
) -> str: ) -> str:

67
app/library/Tasks.py
View file

@ -15,10 +15,10 @@ from app.library.Services import Services
from .config import Config from .config import Config
from .encoder import Encoder from .encoder import Encoder
from .Events import EventBus, Events, error, info, success from .Events import EventBus, Events, error, success
from .Scheduler import Scheduler from .Scheduler import Scheduler
from .Singleton import Singleton from .Singleton import Singleton
from .Utils import init_class from .Utils import init_class, validate_url
LOG: logging.Logger = logging.getLogger("tasks") LOG: logging.Logger = logging.getLogger("tasks")
@ -162,11 +162,12 @@ class Tasks(metaclass=Singleton):
from cronsim import CronSim from cronsim import CronSim
cs = CronSim(task.timer, datetime.now(UTC)) cs = CronSim(task.timer, datetime.now(UTC))
schedule_time = cs.explain() schedule_time: str = cs.explain()
except Exception: except Exception:
schedule_time = task.timer schedule_time = task.timer
LOG.info(f"Task '{task.name}' queued to be executed '{schedule_time}'.") if not has_tasks:
LOG.info(f"Task '{task.name}' queued to be executed '{schedule_time}'.")
except Exception as e: except Exception as e:
LOG.exception(e) LOG.exception(e)
LOG.error(f"Failed to queue '{i}: {task.name}'. '{e!s}'.") LOG.error(f"Failed to queue '{i}: {task.name}'. '{e!s}'.")
@ -185,8 +186,11 @@ class Tasks(metaclass=Singleton):
return self return self
for task in self._tasks: for task in self._tasks:
if not self._scheduler.has(task.id):
continue
try: try:
LOG.info(f"Stopping '{task.name}'.") LOG.debug(f"Stopping '{task.name}'.")
self._scheduler.remove(task.id) self._scheduler.remove(task.id)
except Exception as e: except Exception as e:
if not shutdown: if not shutdown:
@ -220,15 +224,25 @@ class Tasks(metaclass=Singleton):
msg = "No name found." msg = "No name found."
raise ValueError(msg) raise ValueError(msg)
task["name"] = task["name"].strip()
if not task.get("url"): if not task.get("url"):
msg = "No URL found." msg = "No URL found."
raise ValueError(msg) raise ValueError(msg)
task["url"] = task["url"].strip()
try:
validate_url(task["url"], allow_internal=True)
except ValueError as e:
msg = f"Invalid URL format. '{e!s}'."
raise ValueError(msg) from e
if task.get("timer"): if task.get("timer"):
try: try:
from cronsim import CronSim from cronsim import CronSim
CronSim(task.get("timer"), datetime.now(UTC)) CronSim(task.get("timer"), datetime.now(UTC))
task["timer"] = str(task["timer"]).strip()
except Exception as e: except Exception as e:
msg = f"Invalid timer format. '{e!s}'." msg = f"Invalid timer format. '{e!s}'."
raise ValueError(msg) from e raise ValueError(msg) from e
@ -238,6 +252,7 @@ class Tasks(metaclass=Singleton):
from .Utils import arg_converter from .Utils import arg_converter
arg_converter(args=task.get("cli")) arg_converter(args=task.get("cli"))
task["cli"] = str(task["cli"]).strip()
except Exception as e: except Exception as e:
msg = f"Invalid command options for yt-dlp. '{e!s}'." msg = f"Invalid command options for yt-dlp. '{e!s}'."
raise ValueError(msg) from e raise ValueError(msg) from e
@ -300,26 +315,17 @@ class Tasks(metaclass=Singleton):
template: str = task.template if task.template else "" template: str = task.template if task.template else ""
cli: str = task.cli if task.cli else "" cli: str = task.cli if task.cli else ""
LOG.info(f"Dispatched '{task.name}' at '{timeNow}'.") await self._notify.emit(
Events.ADD_URL,
tasks: list = [ data={
self._notify.emit( "url": task.url,
Events.LOG_INFO, data=info(f"Dispatched '{task.name}' at '{timeNow}'.", data={"lowPriority": True}) "preset": preset,
), "folder": folder,
self._notify.emit( "template": template,
Events.ADD_URL, "cli": cli,
data={ },
"url": task.url, id=task.id,
"preset": preset, )
"folder": folder,
"template": template,
"cli": cli,
},
id=task.id,
),
]
await asyncio.wait_for(asyncio.gather(*tasks), timeout=None)
timeNow = datetime.now(UTC).isoformat() timeNow = datetime.now(UTC).isoformat()
@ -374,10 +380,19 @@ class HandleTask:
if handler is None: if handler is None:
continue continue
asyncio.create_task(self.dispatch(task, handler=handler), name=f"task-{task.id}") coro = self.dispatch(task, handler=handler)
t = asyncio.create_task(coro, name=f"task-{task.id}")
t.add_done_callback(lambda fut, t=task: self._handle_exception(fut, t))
except Exception as e: except Exception as e:
LOG.error(f"Failed to handle task '{task.name}'. '{e!s}'.") LOG.error(f"Failed to handle task '{task.name}'. '{e!s}'.")
def _handle_exception(self, fut: asyncio.Task, task: Task) -> None:
if fut.cancelled():
return
if exc := fut.exception():
LOG.error(f"Exception while handling task '{task.name}': {exc}")
def _find_handler(self, task: Task) -> type | None: def _find_handler(self, task: Task) -> type | None:
for cls in self._handlers: for cls in self._handlers:
try: try:

7
app/library/Utils.py
View file

@ -398,12 +398,13 @@ def is_private_address(hostname: str) -> bool:
return True return True
def validate_url(url: str) -> bool: def validate_url(url: str, allow_internal: bool = False) -> bool:
""" """
Validate if the url is valid and allowed. Validate if the url is valid and allowed.
Args: Args:
url (str): URL to validate. url (str): URL to validate.
allow_internal (bool): If True, allow internal URLs or private networks.
Returns: Returns:
bool: True if the URL is valid and allowed. bool: True if the URL is valid and allowed.
@ -429,8 +430,8 @@ def validate_url(url: str) -> bool:
msg = "Invalid scheme usage. Only HTTP or HTTPS allowed." msg = "Invalid scheme usage. Only HTTP or HTTPS allowed."
raise ValueError(msg) raise ValueError(msg)
hostname = parsed_url.host hostname: str | None = parsed_url.host
if not hostname or is_private_address(hostname): if allow_internal is False and (not hostname or is_private_address(hostname)):
msg = "Access to internal urls or private networks is not allowed." msg = "Access to internal urls or private networks is not allowed."
raise ValueError(msg) raise ValueError(msg)

2
app/library/conditions.py
View file

@ -28,7 +28,7 @@ class Condition:
"""The filter to run on info dict.""" """The filter to run on info dict."""
cli: str cli: str
"""If matched append this to the download request and re-queue.""" """If matched append this to the download request and retry."""
def serialize(self) -> dict: def serialize(self) -> dict:
return self.__dict__ return self.__dict__

79
app/library/config.py
View file

@ -12,7 +12,7 @@ import coloredlogs
from dotenv import load_dotenv from dotenv import load_dotenv
from .Utils import FileLogFormatter, arg_converter from .Utils import FileLogFormatter, arg_converter
from .version import APP_BUILD_DATE, APP_COMMIT_SHA, APP_VERSION from .version import APP_BRANCH, APP_BUILD_DATE, APP_COMMIT_SHA, APP_VERSION
class Config: class Config:
@ -109,10 +109,6 @@ class Config:
pip_ignore_updates: bool = False pip_ignore_updates: bool = False
"""Ignore pip package updates.""" """Ignore pip package updates."""
# immutable config vars.
version: str = APP_VERSION
"The version of the application."
app_version: str = APP_VERSION app_version: str = APP_VERSION
"The version of the application, same as `version`." "The version of the application, same as `version`."
@ -122,6 +118,9 @@ class Config:
app_build_date: str = APP_BUILD_DATE app_build_date: str = APP_BUILD_DATE
"The build date of the application." "The build date of the application."
app_branch: str = APP_BRANCH
"The branch of the application."
__instance = None __instance = None
"The instance of the class." "The instance of the class."
@ -191,7 +190,6 @@ class Config:
"The variables that are set manually." "The variables that are set manually."
_immutable: tuple = ( _immutable: tuple = (
"version",
"__instance", "__instance",
"ytdl_options", "ytdl_options",
"started", "started",
@ -201,6 +199,7 @@ class Config:
"app_version", "app_version",
"app_commit_sha", "app_commit_sha",
"app_build_date", "app_build_date",
"app_branch",
) )
"The variables that are immutable." "The variables that are immutable."
@ -237,7 +236,6 @@ class Config:
"download_path", "download_path",
"keep_archive", "keep_archive",
"output_template", "output_template",
"version",
"started", "started",
"remove_files", "remove_files",
"ui_update_title", "ui_update_title",
@ -257,6 +255,7 @@ class Config:
"app_version", "app_version",
"app_commit_sha", "app_commit_sha",
"app_build_date", "app_build_date",
"app_branch",
) )
"The variables that are relevant to the frontend." "The variables that are relevant to the frontend."
@ -452,6 +451,9 @@ class Config:
) )
raise ValueError(msg) raise ValueError(msg)
if "dev-master" == self.app_version:
self._version_via_git()
def _get_attributes(self) -> dict: def _get_attributes(self) -> dict:
attrs: dict = {} attrs: dict = {}
vClass: str = self.__class__ vClass: str = self.__class__
@ -519,3 +521,66 @@ class Config:
return YTDLP_VERSION return YTDLP_VERSION
except ImportError: except ImportError:
return "0.0.0" return "0.0.0"
def _version_via_git(self):
"""
Updates the version of the application using git tags.
This is used to set the version to the latest git tag.
"""
git_path: str = Path(__file__).parent / ".." / ".." / ".git"
if not git_path.exists():
return
try:
import subprocess
branch_result = subprocess.run( # noqa: S603
["git", "rev-parse", "--abbrev-ref", "HEAD"], # noqa: S607
cwd=os.path.dirname(git_path),
capture_output=True,
text=True,
check=False,
)
if 0 != branch_result.returncode:
logging.error(f"Git rev-parse failed: {branch_result.stderr.strip()}")
return
branch_name: str = branch_result.stdout.strip()
if not branch_name:
logging.warning("Git branch name is empty.")
return
commit_result = subprocess.run( # noqa: S603
["git", "log", "-1", "--format=%ct_%H"], # noqa: S607
cwd=os.path.dirname(git_path),
capture_output=True,
text=True,
check=False,
)
if 0 != commit_result.returncode:
logging.error(f"Git log failed: {commit_result.stderr.strip()}")
return
commit_info: str = commit_result.stdout.strip()
if not commit_info:
logging.warning("Git commit info is empty.")
return
commit_date, commit_sha = commit_info.split("_", 1)
commit_date = time.strftime("%Y%m%d", time.localtime(int(commit_date)))
self.app_version = f"{branch_name}-{commit_date}-{commit_sha[:8]}"
self.app_branch = branch_name
self.app_commit_sha = commit_sha
self.app_build_date = commit_date
version_data = {
"version": self.app_version,
"branch": self.app_branch,
"commit": self.app_commit_sha,
"build_date": self.app_build_date,
}
logging.info(f"Application version info set to '{version_data}'")
except Exception as e:
logging.error(f"Error while getting git version: {e!s}")

3
app/library/version.py
View file

@ -11,3 +11,6 @@ APP_COMMIT_SHA = "unknown"
APP_BUILD_DATE = "unknown" APP_BUILD_DATE = "unknown"
"Build date of the application" "Build date of the application"
APP_BRANCH = "unknown"
"Branch of the application"

20
app/main.py
View file

@ -38,14 +38,10 @@ class Main:
def __init__(self, is_native: bool = False): def __init__(self, is_native: bool = False):
self._config = Config.get_instance(is_native=is_native) self._config = Config.get_instance(is_native=is_native)
self._app = web.Application() self._app = web.Application()
self._app.on_shutdown.append(self.on_shutdown)
Services.get_instance().add("app", self._app) Services.get_instance().add("app", self._app)
if self._config.debug:
loop = asyncio.get_event_loop()
loop.set_debug(True)
loop.slow_callback_duration = 0.05
self._check_folders() self._check_folders()
caribou.upgrade(self._config.db_file, ROOT_PATH / "migrations") caribou.upgrade(self._config.db_file, ROOT_PATH / "migrations")
@ -97,6 +93,9 @@ class Main:
LOG.error(f"Could not create database file at '{self._config.db_file}'. {e!s}") LOG.error(f"Could not create database file at '{self._config.db_file}'. {e!s}")
raise raise
async def on_shutdown(self, _: web.Application):
await EventBus.get_instance().emit(Events.SHUTDOWN, data={"app": self._app})
def start(self, host: str | None = None, port: int | None = None, cb=None): def start(self, host: str | None = None, port: int | None = None, cb=None):
""" """
Start the application. Start the application.
@ -120,8 +119,17 @@ class Main:
def started(_): def started(_):
LOG.info("=" * 40) LOG.info("=" * 40)
LOG.info(f"YTPTube {self._config.version} - started on http://{host}:{port}{self._config.base_path}") LOG.info(f"YTPTube {self._config.app_version} - started on http://{host}:{port}{self._config.base_path}")
LOG.info("=" * 40) LOG.info("=" * 40)
loop = asyncio.get_event_loop()
EventBus.get_instance().sync_emit(Events.STARTED, data={"app": self._app}, loop=loop, wait=False)
if loop and self._config.debug:
loop.set_debug(True)
loop.slow_callback_duration = 0.05
if cb: if cb:
cb() cb()

140
app/routes/api/dev.py
View file

@ -1,11 +1,9 @@
import asyncio import asyncio
import json
import logging import logging
from aiohttp import web from aiohttp import web
from aiohttp.web import Request, Response from aiohttp.web import Response
from app.library import DownloadQueue
from app.library.config import Config from app.library.config import Config
from app.library.encoder import Encoder from app.library.encoder import Encoder
from app.library.router import route from app.library.router import route
@ -42,139 +40,3 @@ async def debug_asyncio(config: Config, encoder: Encoder) -> Response:
dumps=encoder.encode, dumps=encoder.encode,
) )
@route("GET", "api/dev/workers/", "pool_list")
async def pool_list(config: Config, queue: DownloadQueue) -> Response:
"""
Get the workers status.
Args:
config (Config): The configuration object.
queue (DownloadQueue): The download queue object.
Returns:
Response: The response object.
"""
if not config.is_dev():
return web.json_response(
{"error": "This endpoint is only available in development mode."},
status=web.HTTPNotFound.status_code,
)
if queue.pool is None:
return web.json_response({"error": "Worker pool not initialized."}, status=web.HTTPNotFound.status_code)
status = queue.pool.get_workers_status()
data = []
for worker in status:
worker_status = status.get(worker)
data.append(
{
"id": worker,
"data": {"status": "Waiting for download."} if worker_status is None else worker_status,
}
)
return web.json_response(
data={
"open": queue.pool.has_open_workers(),
"count": queue.pool.get_available_workers(),
"workers": data,
},
status=web.HTTPOk.status_code,
dumps=lambda obj: json.dumps(obj, default=lambda o: f"<<non-serializable: {type(o).__qualname__}>>"),
)
@route("POST", "api/dev/workers/", "pool_start")
async def pool_restart(config: Config, queue: DownloadQueue) -> Response:
"""
Restart the workers pool.
Args:
config (Config): The configuration object.
queue (DownloadQueue): The download queue object.
Returns:
Response: The response object.
"""
if not config.is_dev():
return web.json_response(
{"error": "This endpoint is only available in development mode."},
status=web.HTTPNotFound.status_code,
)
if queue.pool is None:
return web.json_response({"error": "Worker pool not initialized."}, status=web.HTTPNotFound.status_code)
queue.pool.start()
return web.json_response({"message": "Workers pool being restarted."}, status=web.HTTPOk.status_code)
@route("PATCH", "api/dev/workers/{id}", "worker_restart")
async def worker_restart(request: Request, config: Config, queue: DownloadQueue) -> Response:
"""
Restart a worker.
Args:
request (Request): The request object.
config (Config): The configuration object.
queue (DownloadQueue): The download queue object.
Returns:
Response: The response object
"""
if not config.is_dev():
return web.json_response(
{"error": "This endpoint is only available in development mode."},
status=web.HTTPNotFound.status_code,
)
worker_id: str = request.match_info.get("id")
if not worker_id:
return web.json_response({"error": "worker id is required."}, status=web.HTTPBadRequest.status_code)
if queue.pool is None:
return web.json_response({"error": "Worker pool not initialized."}, status=web.HTTPNotFound.status_code)
status = await queue.pool.restart(worker_id, "requested by user.")
return web.json_response({"status": "restarted" if status else "in_error_state"}, status=web.HTTPOk.status_code)
@route("DELETE", "api/dev/workers/{id}", "worker_stop")
async def worker_stop(request: Request, config: Config, queue: DownloadQueue) -> Response:
"""
Stop a worker.
Args:
request (Request): The request object.
config (Config): The configuration object.
queue (DownloadQueue): The download queue object.
Returns:
Response: The response object.
"""
if not config.is_dev():
return web.json_response(
{"error": "This endpoint is only available in development mode."},
status=web.HTTPNotFound.status_code,
)
worker_id: str = request.match_info.get("id")
if not worker_id:
raise web.HTTPBadRequest(text="worker id is required.")
if queue.pool is None:
return web.json_response({"error": "Worker pool not initialized."}, status=web.HTTPNotFound.status_code)
status = await queue.pool.stop(worker_id, "requested by user.")
return web.json_response({"status": "stopped" if status else "in_error_state"}, status=web.HTTPOk.status_code)

4
app/routes/api/images.py
View file

@ -47,7 +47,7 @@ async def get_thumbnail(request: Request, config: Config) -> Response:
"proxy": ytdlp_args.get("proxy", None), "proxy": ytdlp_args.get("proxy", None),
"headers": { "headers": {
"User-Agent": ytdlp_args.get( "User-Agent": ytdlp_args.get(
"user_agent", request.headers.get("User-Agent", f"YTPTube/{config.version}") "user_agent", request.headers.get("User-Agent", f"YTPTube/{config.app_version}")
), ),
}, },
} }
@ -118,7 +118,7 @@ async def get_background(request: Request, config: Config, cache: Cache) -> Resp
opts = { opts = {
"proxy": ytdlp_args.get("proxy", None), "proxy": ytdlp_args.get("proxy", None),
"headers": { "headers": {
"User-Agent": ytdlp_args.get("user_agent", f"YTPTube/{config.version}"), "User-Agent": ytdlp_args.get("user_agent", f"YTPTube/{config.app_version}"),
}, },
} }

2
app/routes/api/tasks.py
View file

@ -75,7 +75,7 @@ async def tasks_add(request: Request, encoder: Encoder) -> Response:
Tasks.validate(item) Tasks.validate(item)
except ValueError as e: except ValueError as e:
return web.json_response( return web.json_response(
{"error": f"Failed to validate task '{item.get('name')}'. '{e!s}'"}, {"error": f"Failed to validate task '{item.get('name', '??')}'. '{e!s}'"},
status=web.HTTPBadRequest.status_code, status=web.HTTPBadRequest.status_code,
) )

2
app/routes/socket/connection.py
View file

@ -77,7 +77,7 @@ async def subscribe(config: Config, notify: EventBus, sio: socketio.AsyncServer,
await sio.emit(Events.SUBSCRIBED, data={"event": data}, to=sid) await sio.emit(Events.SUBSCRIBED, data={"event": data}, to=sid)
async def emit_logs(data: dict): async def emit_logs(data: dict):
await subscribe_emit(sio=sio, event=data, data=data) await subscribe_emit(sio=sio, event="log_lines", data=data)
if "log_lines" == data and _Data.log_task is None: if "log_lines" == data and _Data.log_task is None:
log_file = Path(config.config_path) / "logs" / "app.log" log_file = Path(config.config_path) / "logs" / "app.log"

2
app/routes/socket/terminal.py
View file

@ -119,7 +119,7 @@ async def cli_post(config: Config, notify: EventBus, sid: str, data: str):
except Exception as e: except Exception as e:
LOG.error(f"Error closing PTY. '{e!s}'.") LOG.error(f"Error closing PTY. '{e!s}'.")
read_task: Task = asyncio.create_task(reader(sid=sid)) read_task: Task = asyncio.create_task(reader(sid=sid), name=f"cli_reader_{sid}")
returncode: int = await proc.wait() returncode: int = await proc.wait()

18
ui/@types/changelogs.d.ts vendored Normal file
View file

@ -0,0 +1,18 @@
type changelogs = changeset[]
type changeset = {
tag: string
full_sha: string
date: string
commits: Commit[]
}
type Commit = {
sha: string
full_sha: string
message: string
author: string
date: string
}
export type {changelogs, changeset, Commit}

79
ui/@types/config.d.ts vendored Normal file
View file

@ -0,0 +1,79 @@
type AppConfig = {
/** Path where downloaded files will be saved */
download_path: string
/** Indicates if the app should keep an archive of downloaded files */
keep_archive: boolean
/** Indicates if files should be removed after download */
remove_files: boolean
/** Indicates if the UI should update the title with the current download status */
ui_update_title: boolean
/** Output template for yt-dlp, e.g. "%(title)s.%(ext)s" */
output_template: string
/** yt-dlp version, e.g. "2023.10.01" */
ytdlp_version: string
/** Maximum number of concurrent download workers */
max_workers: number
/** Indicates if the app is in basic mode, which may limit some features */
basic_mode: boolean
/** Default preset name, e.g. "default" */
default_preset: string
/** Instance title for the app, null if not set */
instance_title: string | null
/** Sentry DSN for error tracking, null if not configured */
sentry_dsn: string | null
/** Indicates if the console is enabled */
console_enabled: boolean
/** Indicates if the file browser is enabled */
browser_enabled: boolean
/** Command options for yt-dlp */
ytdlp_cli: string
/** Indicates if file logging is enabled */
file_logging: boolean
/** Indicates if the app is running in a native environment (e.g., Electron) */
is_native: boolean
/** App version in format "1.0.0" */
app_version: string
/** App version in semantic versioning format, e.g. "1.0.0" */
app_commit_sha: string
/** App build date in ISO format, e.g. "2023-10-01T12:00:00Z" */
app_build_date: string
/** App branch name, e.g. "main" or "develop" */
app_branch: string
}
type Preset = {
/** Unique identifier for the preset */
id?: string
/** Preset name, e.g. "default" */
name: string
/** Optional description for the preset */
description: string
/** Folder where files will be saved, e.g. "/downloads" */
folder: string
/** Output template for the preset, e.g. "%(title)s.%(ext)s" */
template: string
/** Cookies for the preset, e.g. "cookies.txt" */
cookies: string
/** Additional command line options for yt-dlp */
cli: string
/** Indicates if this is the default preset */
default: boolean
}
type ConfigState = {
/** Show or hide the download form */
showForm: RemovableRef<boolean>
/** Application configuration */
app: AppConfig
/** List of presets */
presets: Preset[]
/** List of folders where files can be saved */
folders: string[]
/** Indicates if downloads are currently paused */
paused: boolean
/** Indicates if the configuration has been loaded */
is_loaded: boolean
}
export type { AppConfig, Preset, ConfigState }

40
ui/components/History.vue
View file

@ -55,10 +55,10 @@
</button> </button>
</div> </div>
<div class="column is-half-mobile" v-if="hasIncomplete"> <div class="column is-half-mobile" v-if="hasIncomplete">
<button type="button" class="button is-fullwidth is-warning is-inverted" @click="requeueIncomplete"> <button type="button" class="button is-fullwidth is-warning is-inverted" @click="retryIncomplete">
<span class="icon-text is-block"> <span class="icon-text is-block">
<span class="icon"><i class="fa-solid fa-rotate-right" /></span> <span class="icon"><i class="fa-solid fa-rotate-right" /></span>
<span>Re-queue Incomplete</span> <span>Retry Incomplete</span>
</span> </span>
</button> </button>
</div> </div>
@ -145,7 +145,7 @@
<td class="is-vcentered has-text-centered is-unselectable" <td class="is-vcentered has-text-centered is-unselectable"
v-if="item.live_in && 'not_live' === item.status"> v-if="item.live_in && 'not_live' === item.status">
<span :date-datetime="item.live_in" class="user-hint" <span :date-datetime="item.live_in" class="user-hint"
v-tooltip="'Will automatically be requeued at: ' + moment(item.live_in).format('YYYY-M-DD H:mm Z')" v-tooltip="'Will automatically be retried at: ' + moment(item.live_in).format('YYYY-M-DD H:mm Z')"
v-rtime="item.live_in" /> v-rtime="item.live_in" />
</td> </td>
<td class="is-vcentered has-text-centered is-unselectable" v-else> <td class="is-vcentered has-text-centered is-unselectable" v-else>
@ -154,8 +154,8 @@
<td class="is-vcentered is-items-center"> <td class="is-vcentered is-items-center">
<div class="field is-grouped is-grouped-centered"> <div class="field is-grouped is-grouped-centered">
<div class="control" v-if="item.status != 'finished' || !item.filename"> <div class="control" v-if="item.status != 'finished' || !item.filename">
<button class="button is-warning is-fullwidth is-small" v-tooltip="'Re-queue video'" <button class="button is-warning is-fullwidth is-small" v-tooltip="'Retry download'"
@click="(event) => reQueueItem(item, event)"> @click="(event) => retryItem(item, event)">
<span class="icon"><i class="fa-solid fa-rotate-right" /></span> <span class="icon"><i class="fa-solid fa-rotate-right" /></span>
</button> </button>
</div> </div>
@ -200,7 +200,7 @@
<template v-if="item.status != 'finished' || !item.filename"> <template v-if="item.status != 'finished' || !item.filename">
<hr class="dropdown-divider" /> <hr class="dropdown-divider" />
<NuxtLink class="dropdown-item" @click="reQueueItem(item, true)"> <NuxtLink class="dropdown-item" @click="retryItem(item, true)">
<span class="icon"><i class="fa-solid fa-rotate-right" /></span> <span class="icon"><i class="fa-solid fa-rotate-right" /></span>
<span>Add to download form</span> <span>Add to download form</span>
</NuxtLink> </NuxtLink>
@ -312,10 +312,10 @@
</div> </div>
<div class="columns is-mobile is-multiline"> <div class="columns is-mobile is-multiline">
<div class="column is-half-mobile" v-if="item.status != 'finished' || !item.filename"> <div class="column is-half-mobile" v-if="item.status != 'finished' || !item.filename">
<a class="button is-warning is-fullwidth" @click="(event) => reQueueItem(item, event)"> <a class="button is-warning is-fullwidth" @click="(event) => retryItem(item, event)">
<span class="icon-text is-block"> <span class="icon-text is-block">
<span class="icon"><i class="fa-solid fa-rotate-right" /></span> <span class="icon"><i class="fa-solid fa-rotate-right" /></span>
<span>Re-queue</span> <span>Retry</span>
</span> </span>
</a> </a>
</div> </div>
@ -370,7 +370,7 @@
<template v-if="item.status != 'finished' || !item.filename"> <template v-if="item.status != 'finished' || !item.filename">
<hr class="dropdown-divider" /> <hr class="dropdown-divider" />
<NuxtLink class="dropdown-item" @click="reQueueItem(item, true)"> <NuxtLink class="dropdown-item" @click="retryItem(item, true)">
<span class="icon"><i class="fa-solid fa-rotate-right" /></span> <span class="icon"><i class="fa-solid fa-rotate-right" /></span>
<span>Add to download form</span> <span>Add to download form</span>
</NuxtLink> </NuxtLink>
@ -666,6 +666,10 @@ const setIcon = item => {
return item.extras?.is_premiere ? 'fa-solid fa-star' : 'fa-solid fa-headset' return item.extras?.is_premiere ? 'fa-solid fa-star' : 'fa-solid fa-headset'
} }
if ('skip' === item.status) {
return 'fa-solid fa-ban'
}
return 'fa-solid fa-circle' return 'fa-solid fa-circle'
} }
@ -681,7 +685,7 @@ const setIconColor = item => {
return 'has-text-info' return 'has-text-info'
} }
if ('cancelled' === item.status) { if ('cancelled' === item.status || "skipped" === item.status) {
return 'has-text-warning' return 'has-text-warning'
} }
@ -690,10 +694,6 @@ const setIconColor = item => {
const setStatus = item => { const setStatus = item => {
if ('finished' === item.status) { if ('finished' === item.status) {
if (!item.filename) {
return 'Skipped?'
}
if (item.extras?.is_premiere) { if (item.extras?.is_premiere) {
return 'Premiered' return 'Premiered'
} }
@ -716,11 +716,15 @@ const setStatus = item => {
return display_style.value === 'cards' ? 'Stream' : 'Live' return display_style.value === 'cards' ? 'Stream' : 'Live'
} }
if ('skip' === item.status) {
return 'Skipped'
}
return item.status return item.status
} }
const requeueIncomplete = () => { const retryIncomplete = () => {
if (false === box.confirm('Are you sure you want to re-queue all incomplete downloads?')) { if (false === box.confirm('Are you sure you want to retry all incomplete downloads?')) {
return false return false
} }
@ -729,7 +733,7 @@ const requeueIncomplete = () => {
if ('finished' === item.status) { if ('finished' === item.status) {
continue continue
} }
reQueueItem(item) retryItem(item)
} }
} }
@ -784,7 +788,7 @@ const removeItem = item => {
}) })
} }
const reQueueItem = (item, re_add = false) => { const retryItem = (item, re_add = false) => {
const item_req = { const item_req = {
url: item.url, url: item.url,
preset: item.preset, preset: item.preset,

10
ui/layouts/default.vue
View file

@ -113,14 +113,14 @@
<div class="columns mt-3 is-mobile"> <div class="columns mt-3 is-mobile">
<div class="column is-8-mobile"> <div class="column is-8-mobile">
<div class="has-text-left" v-if="config.app?.version"> <div class="has-text-left" v-if="config.app?.app_version">
© {{ Year }} - <NuxtLink href="https://github.com/ArabCoders/ytptube" target="_blank">YTPTube</NuxtLink> © {{ Year }} - <NuxtLink href="https://github.com/ArabCoders/ytptube" target="_blank">YTPTube</NuxtLink>
<span class="is-hidden-mobile" <span class="is-hidden-mobile has-tooltip"
v-tooltip="`Build Date: ${config.app?.app_build_date}, commit: ${config.app?.app_commit_sha}`"> v-tooltip="`Build Date: ${config.app?.app_build_date}, Branch: ${config.app?.app_branch}, commit: ${config.app?.app_commit_sha}`">
&nbsp;({{ config?.app?.version || 'unknown' }})</span> &nbsp;({{ config?.app?.app_version || 'unknown' }})</span>
- <NuxtLink target="_blank" href="https://github.com/yt-dlp/yt-dlp">yt-dlp</NuxtLink> - <NuxtLink target="_blank" href="https://github.com/yt-dlp/yt-dlp">yt-dlp</NuxtLink>
<span class="is-hidden-mobile">&nbsp;({{ config?.app?.ytdlp_version || 'unknown' }})</span> <span class="is-hidden-mobile">&nbsp;({{ config?.app?.ytdlp_version || 'unknown' }})</span>
- <NuxtLink :to="`/changelog?version=${config?.app?.version || 'unknown'}`">CHANGELOG</NuxtLink> - <NuxtLink to="/changelog">CHANGELOG</NuxtLink>
</div> </div>
</div> </div>
<div class="column is-4-mobile" v-if="config.app?.started"> <div class="column is-4-mobile" v-if="config.app?.started">

106
ui/pages/changelog.vue
View file

@ -1,3 +1,18 @@
<style scoped>
.logs-container {
padding: 1rem;
min-width: 100%;
max-height: 73vh;
overflow-y: auto;
overflow-x: auto;
}
hr {
background-color: unset;
border-bottom: 1px solid var(--bulma-grey-light) !important
}
</style>
<template> <template>
<main> <main>
<div class="mt-1 columns is-multiline"> <div class="mt-1 columns is-multiline">
@ -32,13 +47,17 @@
<hr> <hr>
<ul> <ul>
<li v-for="commit in log.commits" :key="commit.sha"> <li v-for="commit in log.commits" :key="commit.sha">
<strong>{{ ucFirst(commit.message).replace(/\.$/, "") }}.</strong> - <strong>
{{ ucFirst(commit.message).replace(/\.$/, "") }}.
</strong> -
<small> <small>
<NuxtLink :to="`${REPO}/commit/${commit.full_sha}`" target="_blank"> <NuxtLink :to="`${REPO}/commit/${commit.full_sha}`" target="_blank">
<span class="has-tooltip" v-tooltip="`SHA: ${commit.full_sha} - Date: ${commit.date}`"> <span class="has-tooltip" v-tooltip="`SHA: ${commit.full_sha} - Date: ${commit.date}`">
{{ moment(commit.date).fromNow() }} {{ moment(commit.date).fromNow() }}
</span> </span>
</NuxtLink> </NuxtLink>
<span v-tooltip="'Code is at this commit.'" v-if="commit.full_sha === app_sha" class="icon has-text-success"><i
class="fas fa-check" /></span>
</small> </small>
</li> </li>
</ul> </ul>
@ -51,8 +70,9 @@
</main> </main>
</template> </template>
<script setup> <script setup lang="ts">
import moment from 'moment' import moment from 'moment'
import type { changelogs, changeset } from '~/@types/changelogs'
const toast = useNotification() const toast = useNotification()
const config = useConfigStore() const config = useConfigStore()
@ -63,63 +83,38 @@ const PROJECT = 'ytptube'
const REPO = `https://github.com/arabcoders/${PROJECT}` const REPO = `https://github.com/arabcoders/${PROJECT}`
const REPO_URL = `https://arabcoders.github.io/${PROJECT}/CHANGELOG.json?version={version}` const REPO_URL = `https://arabcoders.github.io/${PROJECT}/CHANGELOG.json?version={version}`
const logs = ref([]) const logs = ref<changelogs>([])
const api_version = ref('') const app_version = ref('')
const isLoading = ref(false) const app_branch = ref('')
const hashLength = ref(7) const app_sha = ref('')
const isLoading = ref(true)
const branch = computed(() => {
const branch = String(api_version.value).split('-')[0] ?? 'master'
return ['master', 'dev'].includes(branch) ? branch : 'master'
})
watch(() => config.app.version, async value => {
if (!value) {
return
}
api_version.value = value
hashLength.value = value.split('-').pop().length
await nextTick()
await loadContent()
}, { immediate: true })
const loadContent = async () => { const loadContent = async () => {
if ('' === api_version.value) { if ('' === app_version.value || logs.value.length > 0) {
return return
} }
isLoading.value = true
try { try {
try{ try {
logs.value = await (await request(uri('/CHANGELOG.json'), { method: 'GET' })).json() const changes = await fetch(REPO_URL.replace('{branch}', app_branch.value).replace('{version}', app_version.value))
}catch(e){
console.error(e)
const changes = await fetch(REPO_URL.replace('{branch}', branch.value).replace('{version}', api_version.value))
logs.value = await changes.json() logs.value = await changes.json()
} catch (e) {
console.error(e)
logs.value = await (await request(uri('/CHANGELOG.json'), { method: 'GET' })).json()
} }
await nextTick() await nextTick()
logs.value = logs.value.slice(0, 10)
logs.value = logs.value.slice(0, 10).map(log => { } catch (e: any) {
log.commits = log.commits.map(commit => {
commit.full_sha = commit.sha.padEnd(hashLength.value, '0')
return commit
})
log.full_sha = log.tag.padEnd(hashLength.value, '0')
return log
})
} catch (e) {
console.error(e) console.error(e)
toast.error('error', 'Error', `Failed to fetch changelog. ${e.message}`) toast.error(`Failed to fetch changelog. ${e.message}`)
} finally { } finally {
isLoading.value = false isLoading.value = false
} }
} }
const isInstalled = log => { const isInstalled = (log: changeset) => {
const installed = String(api_version.value).split('-').pop() const installed = String(app_sha.value)
if (log.full_sha.startsWith(installed)) { if (log.full_sha.startsWith(installed)) {
return true return true
@ -134,20 +129,11 @@ const isInstalled = log => {
return false return false
} }
onMounted(() => loadContent()) onMounted(async () => {
await awaiter(config.isLoaded)
app_branch.value = config.app.app_branch
app_version.value = config.app.app_version
app_sha.value = config.app.app_commit_sha
loadContent()
})
</script> </script>
<style scoped>
.logs-container {
padding: 1rem;
min-width: 100%;
max-height: 73vh;
overflow-y: auto;
overflow-x: auto;
}
hr {
background-color: unset;
border-bottom: 1px solid var(--bulma-grey-light) !important
}
</style>

11
ui/pages/logs.vue
View file

@ -123,16 +123,17 @@ const socket = useSocketStore()
const config = useConfigStore() const config = useConfigStore()
const route = useRoute() const route = useRoute()
const logContainer = useTemplateRef<HTMLDivElement>('logContainer')
const bottomMarker = useTemplateRef<HTMLDivElement>('bottomMarker')
const textWrap = useStorage<boolean>('logs_wrap', true)
const bg_enable = useStorage<boolean>('random_bg', true)
const bg_opacity = useStorage<number>('random_bg_opacity', 0.85)
const logs = ref<Array<log_line>>([]) const logs = ref<Array<log_line>>([])
const offset = ref<number>(0) const offset = ref<number>(0)
const loading = ref<boolean>(false) const loading = ref<boolean>(false)
const logContainer = useTemplateRef<HTMLDivElement>('logContainer')
const bottomMarker = useTemplateRef<HTMLDivElement>('bottomMarker')
const autoScroll = ref<boolean>(true) const autoScroll = ref<boolean>(true)
const textWrap = ref<boolean>(true)
const reachedEnd = ref<boolean>(false) const reachedEnd = ref<boolean>(false)
const bg_enable = useStorage<boolean>('random_bg', true)
const bg_opacity = useStorage<number>('random_bg_opacity', 0.85)
const query = ref<string>((() => { const query = ref<string>((() => {
const filter = route.query.filter ?? '' const filter = route.query.filter ?? ''

86
ui/stores/ConfigStore.js
View file

@ -1,86 +0,0 @@
import { useStorage } from '@vueuse/core'
const CONFIG_KEYS = {
showForm: useStorage('showForm', false),
app: {
download_path: '/downloads',
keep_archive: false,
remove_files: false,
ui_update_title: true,
output_template: '',
ytdlp_version: '',
max_workers: 1,
version: '',
basic_mode: true,
default_preset: 'default',
instance_title: null,
sentry_dsn: null,
console_enabled: false,
browser_enabled: false,
ytdlp_cli: '',
file_logging: false,
is_native: false,
app_version: '',
app_commit_sha: '',
app_build_date: '',
},
presets: [
{
'name': 'default',
'format': 'default',
'folder': '',
'template': '',
'cookies': '',
'cli': '',
'default': true
}
],
folders: [],
tasks: [],
paused: false,
};
export const useConfigStore = defineStore('config', () => {
const state = reactive(CONFIG_KEYS);
const actions = {
add(key, value) {
if (key.includes('.')) {
const [parentKey, subKey] = key.split('.');
state[parentKey][subKey] = value;
return;
}
state[key] = value;
},
get(key, defaultValue = null) {
if (key.includes('.')) {
const [parentKey, subKey] = key.split('.');
return state[parentKey][subKey] || defaultValue;
}
return state[key] || defaultValue;
},
update(key, value) {
if (key.includes('.')) {
const [parentKey, subKey] = key.split('.');
state[parentKey][subKey] = value;
return;
}
state[key] = value;
},
getAll() {
return state;
},
setAll(data) {
Object.keys(data).forEach((key) => {
if (key.includes('.')) {
const [parentKey, subKey] = key.split('.');
state[parentKey][subKey] = data[key];
return;
}
state[key] = data[key];
});
},
}
return { ...toRefs(state), ...actions };
});

93
ui/stores/ConfigStore.ts Normal file
View file

@ -0,0 +1,93 @@
import { useStorage } from '@vueuse/core'
import type { ConfigState } from '~/@types/config';
export const useConfigStore = defineStore('config', () => {
const state = reactive<ConfigState>({
showForm: useStorage('showForm', false),
app: {
download_path: '/downloads',
keep_archive: false,
remove_files: false,
ui_update_title: true,
output_template: '',
ytdlp_version: '',
max_workers: 1,
basic_mode: true,
default_preset: 'default',
instance_title: null,
sentry_dsn: null,
console_enabled: false,
browser_enabled: false,
ytdlp_cli: '',
file_logging: false,
is_native: false,
app_version: '',
app_commit_sha: '',
app_build_date: '',
app_branch: '',
},
presets: [
{
'name': 'default',
'description': 'Default preset for downloads',
'folder': '',
'template': '',
'cookies': '',
'cli': '',
'default': true
}
],
folders: [],
paused: false,
is_loaded: false,
});
const add = (key: string, value: any) => {
if (key.includes('.')) {
const [parentKey, subKey] = key.split('.') as [keyof ConfigState, string]
state[parentKey][subKey] = value;
return;
}
(state as any)[key] = value
}
const get = (key: string, defaultValue: any = null): any => {
if (key.includes('.')) {
const [parentKey, subKey] = key.split('.') as [keyof ConfigState, string]
const parent = state[parentKey] as any
return parent?.[subKey] ?? defaultValue
}
return (state as any)[key] ?? defaultValue
}
const update = add
const getAll = (): ConfigState => state
const setAll = (data: Record<string, any>) => {
Object.keys(data).forEach((key) => {
if (key.includes('.')) {
const [parentKey, subKey] = key.split('.') as [keyof ConfigState, string]
const parent = state[parentKey] as any
parent[subKey] = data[key]
return
}
(state as any)[key] = data[key]
})
state.is_loaded = true
}
const isLoaded = () => state.is_loaded
return {
...toRefs(state), add, get, update, getAll, setAll, isLoaded,
} as { [K in keyof ConfigState]: Ref<ConfigState[K]> } & {
add: typeof add
get: typeof get
update: typeof update
getAll: typeof getAll
setAll: typeof setAll
isLoaded: typeof isLoaded
}
});