diff --git a/.github/workflows/main.yml b/.github/workflows/main.yml index 1e01a61b..c4fd07cb 100644 --- a/.github/workflows/main.yml +++ b/.github/workflows/main.yml @@ -131,17 +131,20 @@ jobs: VERSION="${GITHUB_REF##*/}" SHA=$(git rev-parse HEAD) 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_SHA=${SHA}" >> "$GITHUB_ENV" echo "APP_DATE=${DATE}" >> "$GITHUB_ENV" + echo "APP_BRANCH=${BRANCH}" >> "$GITHUB_ENV" sed -i \ -e "s/^APP_VERSION = \".*\"/APP_VERSION = \"${VERSION}\"/" \ -e "s/^APP_COMMIT_SHA = \".*\"/APP_COMMIT_SHA = \"${SHA}\"/" \ -e "s/^APP_BUILD_DATE = \".*\"/APP_BUILD_DATE = \"${DATE}\"/" \ + -e "s/^APP_BRANCH = \".*\"/APP_BRANCH = \"${BRANCH}\"/" \ app/library/version.py - name: Set up QEMU diff --git a/app/library/AsyncPool.py b/app/library/AsyncPool.py deleted file mode 100644 index 2c0581ad..00000000 --- a/app/library/AsyncPool.py +++ /dev/null @@ -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) diff --git a/app/library/Download.py b/app/library/Download.py index 958de882..29474493 100644 --- a/app/library/Download.py +++ b/app/library/Download.py @@ -11,7 +11,6 @@ from pathlib import Path import yt_dlp -from .AsyncPool import Terminator from .config import Config from .Events import EventBus, Events from .ffprobe import ffprobe @@ -20,6 +19,10 @@ from .Utils import delete_dir, extract_info, extract_ytdlp_logs, load_cookies from .YTDLPOpts import YTDLPOpts +class Terminator: + pass + + class NestedLogger: debug_messages = ["[debug] ", "[download] "] @@ -120,6 +123,15 @@ class Download: self.logs = logs if logs else [] 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} 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]) 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: self.logger.exception(exc) + self.logger.error(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.') diff --git a/app/library/DownloadQueue.py b/app/library/DownloadQueue.py index 60e357c6..03ef709e 100644 --- a/app/library/DownloadQueue.py +++ b/app/library/DownloadQueue.py @@ -15,7 +15,6 @@ from aiohttp import web from app.library.ag_utils import ag -from .AsyncPool import AsyncPool from .conditions import Conditions from .config import Config from .DataStore import DataStore @@ -62,9 +61,6 @@ class DownloadQueue(metaclass=Singleton): event: asyncio.Event """Event to signal the download queue to start downloading.""" - pool: AsyncPool | None = None - """Pool of workers to download the files.""" - _active: dict[str, Download] = {} """Dictionary of active downloads.""" @@ -77,6 +73,12 @@ class DownloadQueue(metaclass=Singleton): done: DataStore """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): DownloadQueue._instance = self @@ -89,6 +91,8 @@ class DownloadQueue(metaclass=Singleton): self.paused = asyncio.Event() self.paused.set() self.event = asyncio.Event() + self.workers = asyncio.Semaphore(self.config.max_workers) + self.processors = asyncio.Semaphore(self.config.playlist_items_concurrency) @staticmethod def get_instance(): @@ -104,15 +108,17 @@ class DownloadQueue(metaclass=Singleton): return DownloadQueue._instance - def attach(self, app: web.Application): + def attach(self, _: web.Application): """ Attach the download queue to the application. 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( timer="* * * * *", @@ -127,14 +133,6 @@ class DownloadQueue(metaclass=Singleton): ) # 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: """ Test the datastore connection to the database. @@ -206,81 +204,56 @@ class DownloadQueue(metaclass=Singleton): LOG.error(f"Failed to cancel downloads. {e!s}") 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", []) + + LOG.info(f"Processing '{entry.get('id')}: {entry.get('title')}' Playlist.") + playlistCount = int(entry.get("playlist_count", len(entries))) 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 = { - "playlist": entry.get("id"), - "playlist_index": f"{{0:0{len(str(playlistCount))}d}}".format(i), - "playlist_autonumber": i, - } + extras = { + "playlist": entry.get("title") or entry.get("id"), + "playlist_count": playlistCount, + "playlist_id": entry.get("id"), + "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"): - if property in entry: - extras[f"playlist_{property}"] = entry.get(property) + for property in ("id", "title", "uploader", "uploader_id"): + if property in entry: + 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( item=item.new_with(url=etr.get("url") or etr.get("webpage_url"), extras=extras), already=already, ) + finally: + self.processors.release() - tasks = [process_entry(i, etr) for i, etr in enumerate(entries, start=1)] - 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 = [] - + tasks: list[asyncio.Task] = [] for i, etr in enumerate(entries, start=1): - extras = { - "playlist": entry.get("id"), - "playlist_index": f"{{0:0{len(str(playlistCount))}d}}".format(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 = asyncio.create_task( + playlist_processor(i, etr), + name=f"playlist_processor_{etr.get('id')}_{i}", ) + task.add_done_callback(self._handle_task_exception) + tasks.append(task) + + results: list[dict] = await asyncio.gather(*tasks) LOG.info( 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) 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." - LOG.info(message) - await self._notify.emit(Events.LOG_WARNING, data=event_warning(message)) + message = f"'{id_dict.get('id')}': The URL '{item.url}' is already downloaded and recorded in archive." + LOG.error(message) return {"status": "error", "msg": message} started: float = time.perf_counter() @@ -576,6 +548,7 @@ class DownloadQueue(metaclass=Singleton): ) 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)} 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: """ 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: - 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(): - LOG.info(f"Waiting for item to download. '{self.pool.get_available_workers()}' free workers.") - if self.event: - await self.event.wait() - self.event.clear() - LOG.debug("Cleared wait event.") + LOG.info("Waiting for item to download.") + await self.event.wait() + self.event.clear() - if self.paused and isinstance(self.paused, asyncio.Event) and self.is_paused(): + if self.is_paused(): LOG.info("Download pool is paused.") await self.paused.wait() LOG.info("Download pool resumed downloading.") - entry = self.queue.get_next_download() - await asyncio.sleep(0.2) + for _id, entry in list(self.queue.items()): + if entry.started() or entry.is_cancelled(): + continue - if entry is None: - continue + if entry.is_live: + 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: - await self.pool.push(is_temp=entry.is_live, id=entry.info._id, entry=entry) - LOG.debug(f"Pushed {entry=} to executor.") - await asyncio.sleep(1) + def _release_semaphore(t: asyncio.Task): + self.workers.release() + self._handle_task_exception(t) + + 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: """ @@ -822,7 +784,7 @@ class DownloadQueue(metaclass=Singleton): self._active[entry.info._id] = entry await entry.start() - if "finished" != entry.info.status: + if entry.info.status not in ("finished", "skip"): entry.info.status = "error" finally: 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.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): """ 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 - 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: await self.clear([item.info._id], remove_file=False) @@ -1002,4 +918,11 @@ class DownloadQueue(metaclass=Singleton): except Exception as e: self.done.put(item) 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}") diff --git a/app/library/Events.py b/app/library/Events.py index 937b743e..4025c156 100644 --- a/app/library/Events.py +++ b/app/library/Events.py @@ -4,6 +4,7 @@ import logging import uuid from collections.abc import Awaitable from dataclasses import dataclass, field +from typing import Any from .Singleton import Singleton @@ -93,6 +94,7 @@ class Events: STARTUP = "startup" LOADED = "loaded" + STARTED = "started" SHUTDOWN = "shutdown" ADDED = "added" @@ -245,7 +247,7 @@ class EventListener: if self.is_coroutine: 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): @@ -323,6 +325,8 @@ class EventBus(metaclass=Singleton): self._listeners[e][name] = EventListener(name, callback) + LOG.debug(f"'{name}' subscribed to '{event}'.") + return self def unsubscribe(self, event: str | list | tuple, name: str) -> "EventBus": @@ -340,50 +344,83 @@ class EventBus(metaclass=Singleton): if isinstance(event, str): event = [event] + events = [] for e in event: if e in self._listeners and name in self._listeners[e]: + events.append(e) del self._listeners[e][name] + if len(events) > 0: + LOG.debug(f"'{name}' unsubscribed from '{events}'.") + 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: event (str): The event to emit. - data (any): The data to pass to the event. - **kwargs: The keyword arguments to pass to the event. + data (Any): The data 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: 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. + 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: return [] - ev = Event(event=event, data=data) - LOG.debug(f"Emitting event '{ev.id}: {ev.event}'.", extra={"data": data}) + async def emit_all(): + 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(): - try: - results.append(asyncio.get_event_loop().run_until_complete(handler.handle(ev, **kwargs))) - except Exception as e: - LOG.exception(e) - LOG.error(f"Failed to emit event '{event}' to '{handler.name}'. Error message '{e!s}'.") + for h in self._listeners[event].values(): + try: + res.append(await h.handle(ev, **kwargs)) + except Exception as e: + LOG.exception(e) + 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. Args: 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. Returns: diff --git a/app/library/HttpAPI.py b/app/library/HttpAPI.py index ab073695..df1e883c 100644 --- a/app/library/HttpAPI.py +++ b/app/library/HttpAPI.py @@ -153,7 +153,6 @@ class HttpAPI: Args: username (str): The username. password (str): The password. - this (HttpAPI): The instance of the HttpAPI. Returns: 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 try: @@ -226,7 +226,7 @@ class HttpAPI: key=Config.get_instance().secret_key, ), 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, 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) if contentType and "/" != base_path and contentType.startswith("text/html"): @@ -272,11 +272,8 @@ class HttpAPI: .replace('baseURL:""', f'baseURL:"{rewrite_path}/"') ) - return web.Response( - body=content.encode("utf-8"), - status=response.status, - headers=response.headers, - ) + response.body = content.encode("utf-8") + return response if request.path.startswith(static_path) and isinstance(response, web.FileResponse): try: diff --git a/app/library/ItemDTO.py b/app/library/ItemDTO.py index d6075072..ea8f7c8e 100644 --- a/app/library/ItemDTO.py +++ b/app/library/ItemDTO.py @@ -39,7 +39,7 @@ class Item: """Extra data to be added to the download.""" 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: return self.__dict__.copy() diff --git a/app/library/Notifications.py b/app/library/Notifications.py index e0ca7f91..39a9578b 100644 --- a/app/library/Notifications.py +++ b/app/library/Notifications.py @@ -137,7 +137,7 @@ class Notification(metaclass=Singleton): self._file: Path = Path(file) if file else Path(config.config_path).joinpath("notifications.json") self._client: httpx.AsyncClient = client or httpx.AsyncClient() 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: try: diff --git a/app/library/Scheduler.py b/app/library/Scheduler.py index e26d49fe..4b03b509 100644 --- a/app/library/Scheduler.py +++ b/app/library/Scheduler.py @@ -74,9 +74,31 @@ class Scheduler(metaclass=Singleton): return self._jobs 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) + 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( self, timer: str, func: callable, args: tuple = (), kwargs: dict | None = None, id: str | None = None ) -> str: diff --git a/app/library/Tasks.py b/app/library/Tasks.py index 00a83c03..6b063916 100644 --- a/app/library/Tasks.py +++ b/app/library/Tasks.py @@ -15,10 +15,10 @@ from app.library.Services import Services from .config import Config from .encoder import Encoder -from .Events import EventBus, Events, error, info, success +from .Events import EventBus, Events, error, success from .Scheduler import Scheduler from .Singleton import Singleton -from .Utils import init_class +from .Utils import init_class, validate_url LOG: logging.Logger = logging.getLogger("tasks") @@ -162,11 +162,12 @@ class Tasks(metaclass=Singleton): from cronsim import CronSim cs = CronSim(task.timer, datetime.now(UTC)) - schedule_time = cs.explain() + schedule_time: str = cs.explain() except Exception: 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: LOG.exception(e) LOG.error(f"Failed to queue '{i}: {task.name}'. '{e!s}'.") @@ -185,8 +186,11 @@ class Tasks(metaclass=Singleton): return self for task in self._tasks: + if not self._scheduler.has(task.id): + continue + try: - LOG.info(f"Stopping '{task.name}'.") + LOG.debug(f"Stopping '{task.name}'.") self._scheduler.remove(task.id) except Exception as e: if not shutdown: @@ -220,15 +224,25 @@ class Tasks(metaclass=Singleton): msg = "No name found." raise ValueError(msg) + task["name"] = task["name"].strip() + if not task.get("url"): msg = "No URL found." 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"): try: from cronsim import CronSim CronSim(task.get("timer"), datetime.now(UTC)) + task["timer"] = str(task["timer"]).strip() except Exception as e: msg = f"Invalid timer format. '{e!s}'." raise ValueError(msg) from e @@ -238,6 +252,7 @@ class Tasks(metaclass=Singleton): from .Utils import arg_converter arg_converter(args=task.get("cli")) + task["cli"] = str(task["cli"]).strip() except Exception as e: msg = f"Invalid command options for yt-dlp. '{e!s}'." raise ValueError(msg) from e @@ -300,26 +315,17 @@ class Tasks(metaclass=Singleton): template: str = task.template if task.template else "" cli: str = task.cli if task.cli else "" - LOG.info(f"Dispatched '{task.name}' at '{timeNow}'.") - - tasks: list = [ - self._notify.emit( - Events.LOG_INFO, data=info(f"Dispatched '{task.name}' at '{timeNow}'.", data={"lowPriority": True}) - ), - self._notify.emit( - Events.ADD_URL, - data={ - "url": task.url, - "preset": preset, - "folder": folder, - "template": template, - "cli": cli, - }, - id=task.id, - ), - ] - - await asyncio.wait_for(asyncio.gather(*tasks), timeout=None) + await self._notify.emit( + Events.ADD_URL, + data={ + "url": task.url, + "preset": preset, + "folder": folder, + "template": template, + "cli": cli, + }, + id=task.id, + ) timeNow = datetime.now(UTC).isoformat() @@ -374,10 +380,19 @@ class HandleTask: if handler is None: 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: 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: for cls in self._handlers: try: diff --git a/app/library/Utils.py b/app/library/Utils.py index a5182c78..43fa2c0b 100644 --- a/app/library/Utils.py +++ b/app/library/Utils.py @@ -398,12 +398,13 @@ def is_private_address(hostname: str) -> bool: 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. Args: url (str): URL to validate. + allow_internal (bool): If True, allow internal URLs or private networks. Returns: 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." raise ValueError(msg) - hostname = parsed_url.host - if not hostname or is_private_address(hostname): + hostname: str | None = parsed_url.host + if allow_internal is False and (not hostname or is_private_address(hostname)): msg = "Access to internal urls or private networks is not allowed." raise ValueError(msg) diff --git a/app/library/conditions.py b/app/library/conditions.py index f9ee2ee4..76eed61d 100644 --- a/app/library/conditions.py +++ b/app/library/conditions.py @@ -28,7 +28,7 @@ class Condition: """The filter to run on info dict.""" 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: return self.__dict__ diff --git a/app/library/config.py b/app/library/config.py index ce2986be..92e8acbc 100644 --- a/app/library/config.py +++ b/app/library/config.py @@ -12,7 +12,7 @@ import coloredlogs from dotenv import load_dotenv 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: @@ -109,10 +109,6 @@ class Config: pip_ignore_updates: bool = False """Ignore pip package updates.""" - # immutable config vars. - version: str = APP_VERSION - "The version of the application." - app_version: str = APP_VERSION "The version of the application, same as `version`." @@ -122,6 +118,9 @@ class Config: app_build_date: str = APP_BUILD_DATE "The build date of the application." + app_branch: str = APP_BRANCH + "The branch of the application." + __instance = None "The instance of the class." @@ -191,7 +190,6 @@ class Config: "The variables that are set manually." _immutable: tuple = ( - "version", "__instance", "ytdl_options", "started", @@ -201,6 +199,7 @@ class Config: "app_version", "app_commit_sha", "app_build_date", + "app_branch", ) "The variables that are immutable." @@ -237,7 +236,6 @@ class Config: "download_path", "keep_archive", "output_template", - "version", "started", "remove_files", "ui_update_title", @@ -257,6 +255,7 @@ class Config: "app_version", "app_commit_sha", "app_build_date", + "app_branch", ) "The variables that are relevant to the frontend." @@ -452,6 +451,9 @@ class Config: ) raise ValueError(msg) + if "dev-master" == self.app_version: + self._version_via_git() + def _get_attributes(self) -> dict: attrs: dict = {} vClass: str = self.__class__ @@ -519,3 +521,66 @@ class Config: return YTDLP_VERSION except ImportError: 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}") diff --git a/app/library/version.py b/app/library/version.py index e10ec5de..4902396f 100644 --- a/app/library/version.py +++ b/app/library/version.py @@ -11,3 +11,6 @@ APP_COMMIT_SHA = "unknown" APP_BUILD_DATE = "unknown" "Build date of the application" + +APP_BRANCH = "unknown" +"Branch of the application" diff --git a/app/main.py b/app/main.py index ee7fcfd5..55be9872 100644 --- a/app/main.py +++ b/app/main.py @@ -38,14 +38,10 @@ class Main: def __init__(self, is_native: bool = False): self._config = Config.get_instance(is_native=is_native) self._app = web.Application() + self._app.on_shutdown.append(self.on_shutdown) 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() 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}") 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): """ Start the application. @@ -120,8 +119,17 @@ class Main: def started(_): 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) + + 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: cb() diff --git a/app/routes/api/dev.py b/app/routes/api/dev.py index 06dc4e81..c53fe397 100644 --- a/app/routes/api/dev.py +++ b/app/routes/api/dev.py @@ -1,11 +1,9 @@ import asyncio -import json import logging 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.encoder import Encoder from app.library.router import route @@ -42,139 +40,3 @@ async def debug_asyncio(config: Config, encoder: Encoder) -> Response: 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"<>"), - ) - - -@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) diff --git a/app/routes/api/images.py b/app/routes/api/images.py index ac794814..4eb17460 100644 --- a/app/routes/api/images.py +++ b/app/routes/api/images.py @@ -47,7 +47,7 @@ async def get_thumbnail(request: Request, config: Config) -> Response: "proxy": ytdlp_args.get("proxy", None), "headers": { "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 = { "proxy": ytdlp_args.get("proxy", None), "headers": { - "User-Agent": ytdlp_args.get("user_agent", f"YTPTube/{config.version}"), + "User-Agent": ytdlp_args.get("user_agent", f"YTPTube/{config.app_version}"), }, } diff --git a/app/routes/api/tasks.py b/app/routes/api/tasks.py index c326433b..92ecdea4 100644 --- a/app/routes/api/tasks.py +++ b/app/routes/api/tasks.py @@ -75,7 +75,7 @@ async def tasks_add(request: Request, encoder: Encoder) -> Response: Tasks.validate(item) except ValueError as e: 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, ) diff --git a/app/routes/socket/connection.py b/app/routes/socket/connection.py index 7977fe06..192a719d 100644 --- a/app/routes/socket/connection.py +++ b/app/routes/socket/connection.py @@ -77,7 +77,7 @@ async def subscribe(config: Config, notify: EventBus, sio: socketio.AsyncServer, await sio.emit(Events.SUBSCRIBED, data={"event": data}, to=sid) 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: log_file = Path(config.config_path) / "logs" / "app.log" diff --git a/app/routes/socket/terminal.py b/app/routes/socket/terminal.py index 911223a0..4ff563ce 100644 --- a/app/routes/socket/terminal.py +++ b/app/routes/socket/terminal.py @@ -119,7 +119,7 @@ async def cli_post(config: Config, notify: EventBus, sid: str, data: str): except Exception as e: 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() diff --git a/ui/@types/changelogs.d.ts b/ui/@types/changelogs.d.ts new file mode 100644 index 00000000..976ffaf8 --- /dev/null +++ b/ui/@types/changelogs.d.ts @@ -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} diff --git a/ui/@types/config.d.ts b/ui/@types/config.d.ts new file mode 100644 index 00000000..5f33c982 --- /dev/null +++ b/ui/@types/config.d.ts @@ -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 + /** 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 } diff --git a/ui/components/History.vue b/ui/components/History.vue index bb51733d..df558a1c 100644 --- a/ui/components/History.vue +++ b/ui/components/History.vue @@ -55,10 +55,10 @@
-
@@ -145,7 +145,7 @@ @@ -154,8 +154,8 @@
-
@@ -200,7 +200,7 @@