From a973ce8bcf0c045dcfa34b397c2beebd72e10973 Mon Sep 17 00:00:00 2001 From: ArabCoders Date: Fri, 21 Mar 2025 21:54:28 +0300 Subject: [PATCH 1/4] Finalize EventBus design --- app/library/DownloadQueue.py | 2 +- app/library/Emitter.py | 2 +- .../{EventsSubscriber.py => Events.py} | 122 ++++++++++++------ app/library/HttpAPI.py | 2 +- app/library/HttpSocket.py | 12 +- app/library/Notifications.py | 2 +- app/library/Presets.py | 9 +- app/library/Scheduler.py | 14 +- app/library/Tasks.py | 7 +- app/main.py | 6 +- 10 files changed, 105 insertions(+), 73 deletions(-) rename app/library/{EventsSubscriber.py => Events.py} (55%) diff --git a/app/library/DownloadQueue.py b/app/library/DownloadQueue.py index c6f491ce..611bb520 100644 --- a/app/library/DownloadQueue.py +++ b/app/library/DownloadQueue.py @@ -18,7 +18,7 @@ from .config import Config from .DataStore import DataStore from .Download import Download from .Emitter import Emitter -from .EventsSubscriber import Events +from .Events import Events from .ItemDTO import ItemDTO from .Presets import Presets from .Singleton import Singleton diff --git a/app/library/Emitter.py b/app/library/Emitter.py index 8348ab70..6405606a 100644 --- a/app/library/Emitter.py +++ b/app/library/Emitter.py @@ -2,7 +2,7 @@ import asyncio import logging from collections.abc import Awaitable -from .EventsSubscriber import Events +from .Events import Events from .Singleton import Singleton LOG = logging.getLogger("Emitter") diff --git a/app/library/EventsSubscriber.py b/app/library/Events.py similarity index 55% rename from app/library/EventsSubscriber.py rename to app/library/Events.py index 755fd471..c00d6354 100644 --- a/app/library/EventsSubscriber.py +++ b/app/library/Events.py @@ -1,7 +1,9 @@ import asyncio +import datetime import logging +import uuid from collections.abc import Awaitable -from dataclasses import dataclass +from dataclasses import dataclass, field from .Singleton import Singleton @@ -51,11 +53,50 @@ class Events: @dataclass(kw_only=True) class Event: - id: str - data: dict + """ + Event is a data transfer object that represents an event that was emitted. + """ + + id: str = field(default_factory=lambda: str(uuid.uuid4()), init=False) + """The id of the event.""" + + created: str = field(default_factory=lambda: str(datetime.datetime.now(tz=datetime.timezone.UTC).isoformat())) + """The time the event was created.""" + + event: str + """The event that was emitted.""" + + data: any + """The data that was passed to the event.""" + + def serialize(self) -> dict: + """ + Serialize the event. + + Returns: + dict: The serialized event. + + """ + return {"id": self.id, "created": self.created, "event": self.event, "data": self.data} + + def __repr__(self): + return f"Event(id={self.id}, created={self.created}, event={self.event}, data={self.data})" + + def datatype(self) -> str: + """ + Get the datatype of the data. + + Returns: + str: The datatype of the data. + + """ + return type(self.data).__name__ + + def __str__(self): + return f"Event(id={self.id}, created={self.created}, event={self.event})" -class EventsSubscriber(metaclass=Singleton): +class EventBus(metaclass=Singleton): """ This class is used to subscribe to and emit events to the registered listeners. """ @@ -67,10 +108,10 @@ class EventsSubscriber(metaclass=Singleton): """The listeners for the events.""" def __init__(self): - EventsSubscriber._instance = self + EventBus._instance = self @staticmethod - def get_instance() -> "EventsSubscriber": + def get_instance() -> "EventBus": """ Get the instance of the EventsSubscriber. @@ -78,18 +119,19 @@ class EventsSubscriber(metaclass=Singleton): EventsSubscriber: The instance of the EventsSubscriber """ - if not EventsSubscriber._instance: - EventsSubscriber._instance = EventsSubscriber() - return EventsSubscriber._instance + if not EventBus._instance: + EventBus._instance = EventBus() - def subscribe(self, event: str | list | tuple, id: str, callback: Awaitable) -> "EventsSubscriber": + return EventBus._instance + + def subscribe(self, event: str | list | tuple, callback: Awaitable, name: str | None = None) -> "EventBus": """ Subscribe to an event. Args: event (str): The event to subscribe to. - id (str|None): The id of the subscriber, if None a random uuid will be generated. - callback (Awaitable): The function to call. Must be a coroutine. + name (str|None): The name of the subscriber, if None a random uuid will be generated. + callback(Event) (Awaitable): The function to call. Must be a coroutine. Returns: EventsSubscriber: The instance of the EventsSubscriber @@ -98,21 +140,24 @@ class EventsSubscriber(metaclass=Singleton): if isinstance(event, str): event = [event] + if not name: + name = str(uuid.uuid4()) + for e in event: if e not in self._listeners: self._listeners[e] = {} - self._listeners[e][id] = callback + self._listeners[e][name] = callback return self - def unsubscribe(self, event: str | list | tuple, id: str) -> "EventsSubscriber": + def unsubscribe(self, event: str | list | tuple, name: str) -> "EventBus": """ Unsubscribe from an event. Args: event (str): The event to unsubscribe from. - id (str): The id of the subscriber. + name (str): The name of the subscriber. Returns: EventsSubscriber: The instance of the EventsSubscriber @@ -122,18 +167,18 @@ class EventsSubscriber(metaclass=Singleton): event = [event] for e in event: - if e in self._listeners and id in self._listeners[e]: - del self._listeners[e][id] + if e in self._listeners and name in self._listeners[e]: + del self._listeners[e][name] return self - def emit_sync(self, event: str, *args, **kwargs): + def emit_sync(self, event: str, data: any, **kwargs) -> list: """ Emit an event synchronously. Args: event (str): The event to emit. - *args: The arguments to pass to the event. + data (any): The data to pass to the event. **kwargs: The keyword arguments to pass to the event. Returns: @@ -144,28 +189,26 @@ class EventsSubscriber(metaclass=Singleton): if event not in self._listeners: return [] - results = [] - for id, callback in self._listeners[event].items(): - try: - if "data" not in kwargs or not isinstance(kwargs["data"], Event): - data = Event(id=id, data={"args": args if args else [], **kwargs}) - else: - data = kwargs["data"] + ev = Event(event=event, data=data) + LOG.debug(f"Emitting event '{ev}'.") - results.append(asyncio.get_event_loop().run_until_complete(callback(event, data))) + results = [] + for name, callback in self._listeners[event].items(): + try: + results.append(asyncio.get_event_loop().run_until_complete(callback(ev, name, **kwargs))) except Exception as e: LOG.exception(e) - LOG.error(f"Failed to emit event '{event}' to '{id}'. Error message '{e!s}'.") + LOG.error(f"Failed to emit event '{event}' to '{name}'. Error message '{e!s}'.") return results - def emit(self, event: str, *args, **kwargs): + def emit(self, event: str, data: any, **kwargs) -> Awaitable: """ Emit an event. Args: event (str): The event to emit. - *args: The arguments to pass to the event. + data (any): The data to pass to the event. **kwargs: The keyword arguments to pass to the event. Returns: @@ -175,19 +218,16 @@ class EventsSubscriber(metaclass=Singleton): if event not in self._listeners: return None - tasks = [] - for id, callback in self._listeners[event].items(): - try: - if args and isinstance(args[0], Event): - data = args[0] - elif "data" in kwargs and isinstance(kwargs["data"], Event): - data = kwargs["data"] - else: - data = Event(id=id, data={"args": args if args else [], **kwargs}) + ev = Event(event=event, data=data) - tasks.append(asyncio.create_task(callback(event, data))) + LOG.debug(f"Emitting event '{ev}'.") + + tasks = [] + for name, callback in self._listeners[event].items(): + try: + tasks.append(asyncio.create_task(callback(ev, name, **kwargs))) except Exception as e: LOG.exception(e) - LOG.error(f"Failed to emit event '{event}' to '{id}'. Error message '{e!s}'.") + LOG.error(f"Failed to emit event '{event}' to '{name}'. Error message '{e!s}'.") return asyncio.gather(*tasks) diff --git a/app/library/HttpAPI.py b/app/library/HttpAPI.py index d8454a85..a39c9f47 100644 --- a/app/library/HttpAPI.py +++ b/app/library/HttpAPI.py @@ -26,7 +26,7 @@ from .config import Config from .DownloadQueue import DownloadQueue from .Emitter import Emitter from .encoder import Encoder -from .EventsSubscriber import Events +from .Events import Events from .ffprobe import ffprobe from .M3u8 import M3u8 from .Notifications import Notification, NotificationEvents diff --git a/app/library/HttpSocket.py b/app/library/HttpSocket.py index 30ca13fc..d55c32c1 100644 --- a/app/library/HttpSocket.py +++ b/app/library/HttpSocket.py @@ -18,7 +18,7 @@ from .config import Config from .DownloadQueue import DownloadQueue from .Emitter import Emitter from .encoder import Encoder -from .EventsSubscriber import Event, Events, EventsSubscriber +from .Events import EventBus, Events from .Presets import Presets from .Utils import arg_converter, is_downloaded @@ -80,13 +80,9 @@ class HttpSocket(Common): if hasattr(method, "_ws_event") and self.sio: self.sio.on(method._ws_event)(method) # type: ignore - # self.sio.on("*", es.emit) - - async def handle_event(_: str, data: Event): - LOG.debug(f"Event received. '{data}'") - await self.add(**data.data) - - EventsSubscriber.get_instance().subscribe(Events.ADD_URL, "socket_add_url", handle_event) + EventBus.get_instance().subscribe( + Events.ADD_URL, lambda data, _: self.add(**data.data), "socket_add_url" + ) # register the shutdown event. app.on_shutdown.append(self.on_shutdown) diff --git a/app/library/Notifications.py b/app/library/Notifications.py index f5d0304e..794d7175 100644 --- a/app/library/Notifications.py +++ b/app/library/Notifications.py @@ -10,7 +10,7 @@ import httpx from .config import Config from .encoder import Encoder -from .EventsSubscriber import Events +from .Events import Events from .ItemDTO import ItemDTO from .Singleton import Singleton from .Utils import ag, validate_uuid diff --git a/app/library/Presets.py b/app/library/Presets.py index d9aa29e2..2e0d54d4 100644 --- a/app/library/Presets.py +++ b/app/library/Presets.py @@ -10,7 +10,7 @@ from aiohttp import web from .config import Config from .Emitter import Emitter from .encoder import Encoder -from .EventsSubscriber import Event, Events, EventsSubscriber +from .Events import EventBus, Events from .Singleton import Singleton LOG = logging.getLogger("presets") @@ -81,13 +81,12 @@ class Presets(metaclass=Singleton): except Exception: pass - def handle_event(_, e: Event): - self.save(**e.data) - with open(os.path.join(os.path.dirname(__file__), "presets.json")) as f: self._default_presets = [Preset(**preset) for preset in json.load(f)] - EventsSubscriber.get_instance().subscribe(Events.PRESETS_ADD, f"{__class__}.save", handle_event) + EventBus.get_instance().subscribe( + Events.PRESETS_ADD, lambda data, _: self.add(**data.data), f"{__class__}.save" + ) @staticmethod def get_instance() -> "Presets": diff --git a/app/library/Scheduler.py b/app/library/Scheduler.py index 215bb1d8..7b6756de 100644 --- a/app/library/Scheduler.py +++ b/app/library/Scheduler.py @@ -4,7 +4,7 @@ import logging from aiocron import Cron from aiohttp import web -from .EventsSubscriber import Event, Events, EventsSubscriber +from .Events import EventBus, Events from .Singleton import Singleton LOG = logging.getLogger("scheduler") @@ -26,10 +26,9 @@ class Scheduler(metaclass=Singleton): self._loop = loop or asyncio.get_event_loop() - def handle_event(_, e: Event): - self.add(**e.data) - - EventsSubscriber.get_instance().subscribe(Events.SCHEDULE_ADD, f"{__class__}.add", handle_event) + EventBus.get_instance().subscribe( + Events.SCHEDULE_ADD, lambda data, _: self.add(**data.data), f"{__class__}.add" + ) @staticmethod def get_instance() -> "Scheduler": @@ -76,8 +75,9 @@ class Scheduler(metaclass=Singleton): """Return the job by id.""" return self._jobs.get(id) - def add(self, timer: str, func: callable, args: tuple = (), - kwargs: dict | None = None, id: str | None = None) -> str: + def add( + self, timer: str, func: callable, args: tuple = (), kwargs: dict | None = None, id: str | None = None + ) -> str: """ Add a job to the schedule. diff --git a/app/library/Tasks.py b/app/library/Tasks.py index 0550590b..9d69112d 100644 --- a/app/library/Tasks.py +++ b/app/library/Tasks.py @@ -13,7 +13,7 @@ from aiohttp import web from .config import Config from .Emitter import Emitter from .encoder import Encoder -from .EventsSubscriber import Event, Events, EventsSubscriber +from .Events import Event, EventBus, Events from .Scheduler import Scheduler from .Singleton import Singleton @@ -82,10 +82,7 @@ class Tasks(metaclass=Singleton): except Exception: pass - def handle_event(_, e: Event): - self.save(**e.data) - - EventsSubscriber.get_instance().subscribe(Events.TASKS_ADD, f"{__class__}.save", handle_event) + EventBus.get_instance().subscribe(Events.TASKS_ADD, lambda data, _: self.add(**data.data), f"{__class__}.save") @staticmethod def get_instance() -> "Tasks": diff --git a/app/main.py b/app/main.py index 8262428f..bbe54962 100644 --- a/app/main.py +++ b/app/main.py @@ -12,7 +12,7 @@ from aiohttp import web from library.config import Config from library.DownloadQueue import DownloadQueue from library.Emitter import Emitter -from library.EventsSubscriber import Events, EventsSubscriber +from library.Events import EventBus, Events from library.HttpAPI import HttpAPI from library.HttpSocket import HttpSocket from library.Notifications import Notification @@ -61,7 +61,7 @@ class Main: self._socket = HttpSocket(queue=self._queue) Emitter.get_instance().add_emitter([Notification().emit], local=False).add_emitter( - [EventsSubscriber().emit], local=True + [EventBus().emit], local=True ) self._app.on_cleanup.append(_close_connection) @@ -94,7 +94,7 @@ class Main: """ Start the application. """ - EventsSubscriber.get_instance().emit(Events.STARTUP, data={"app": self._app}) + EventBus.get_instance().emit(Events.STARTUP, data={"app": self._app}) self._socket.attach(self._app) self._http.attach(self._app) From 07fc86565aada43e52fa47b8e1ce1bf407d293ec Mon Sep 17 00:00:00 2001 From: ArabCoders Date: Fri, 21 Mar 2025 23:56:02 +0300 Subject: [PATCH 2/4] updated the app to use EventBus --- app/library/Download.py | 18 ++-- app/library/DownloadQueue.py | 24 +++-- app/library/Emitter.py | 132 ---------------------------- app/library/Events.py | 165 +++++++++++++++++++++++++++++++---- app/library/HttpAPI.py | 15 ++-- app/library/HttpSocket.py | 67 +++++++------- app/library/Notifications.py | 85 +++++++++++------- app/library/Presets.py | 6 +- app/library/Scheduler.py | 2 +- app/library/Tasks.py | 48 +++++----- app/main.py | 10 +-- 11 files changed, 290 insertions(+), 282 deletions(-) delete mode 100644 app/library/Emitter.py diff --git a/app/library/Download.py b/app/library/Download.py index 14186fe7..8be556c2 100644 --- a/app/library/Download.py +++ b/app/library/Download.py @@ -12,7 +12,7 @@ import yt_dlp from .AsyncPool import Terminator from .config import Config -from .Emitter import Emitter +from .Events import EventBus, Events from .ffprobe import ffprobe from .ItemDTO import ItemDTO from .YTDLPOpts import YTDLPOpts @@ -52,7 +52,6 @@ class Download: default_ytdl_opts: dict = None debug: bool = False temp_path: str = None - emitter: Emitter = None cancelled: bool = False is_live: bool = False info_dict: dict = None @@ -102,7 +101,7 @@ class Download: self.tmpfilename = None self.status_queue = None self.proc = None - self.emitter = None + self._notify = EventBus.get_instance() self.max_workers = int(config.max_workers) self.temp_keep = bool(config.temp_keep) self.is_live = bool(info.is_live) or info.live_in is not None @@ -222,8 +221,7 @@ class Download: self.logger.info(f'Task id="{self.info.id}" PID="{os.getpid()}" title="{self.info.title}" completed.') - async def start(self, emitter: Emitter): - self.emitter = emitter + async def start(self): self.status_queue = Config.get_manager().Queue() # Create temp dir for each download. @@ -236,7 +234,7 @@ class Download: self.proc.start() self.info.status = "preparing" - asyncio.create_task(self.emitter.updated(dl=self.info), name=f"emitter-{self.id}") + await self._notify.emit(Events.UPDATED, data=self.info) asyncio.create_task(self.progress_update(), name=f"update-{self.id}") return await asyncio.get_running_loop().run_in_executor(None, self.proc.join) @@ -365,7 +363,7 @@ class Download: self.logger.debug(f"Status Update: {self.info._id=} {status=}") if isinstance(status, str): - asyncio.create_task(self.emitter.updated(dl=self.info), name=f"emitter-u-{self.id}") + await self._notify.emit(Events.UPDATED, data=self.info) continue self.tmpfilename = status.get("tmpfilename") @@ -384,9 +382,7 @@ class Download: if "error" == self.info.status and "error" in status: self.info.error = status.get("error") - asyncio.create_task( - self.emitter.error(message=self.info.error, data=self.info), name=f"emitter-e-{self.id}" - ) + await self._notify.emit(Events.ERROR, data={"message": self.info.error, "data": self.info}) if "downloaded_bytes" in status: total = status.get("total_bytes") or status.get("total_bytes_estimate") @@ -422,4 +418,4 @@ class Download: self.logger.exception(e) self.logger.error(f"Failed to ffprobe: {status.get}. {e}") - asyncio.create_task(self.emitter.updated(dl=self.info), name=f"emitter-u-{self.id}") + await self._notify.emit(Events.UPDATED, data=self.info) diff --git a/app/library/DownloadQueue.py b/app/library/DownloadQueue.py index 611bb520..9bfaa235 100644 --- a/app/library/DownloadQueue.py +++ b/app/library/DownloadQueue.py @@ -17,8 +17,7 @@ from .AsyncPool import AsyncPool from .config import Config from .DataStore import DataStore from .Download import Download -from .Emitter import Emitter -from .Events import Events +from .Events import EventBus, Events from .ItemDTO import ItemDTO from .Presets import Presets from .Singleton import Singleton @@ -53,11 +52,11 @@ class DownloadQueue(metaclass=Singleton): _instance = None """Instance of the DownloadQueue.""" - def __init__(self, connection: Connection, emitter: Emitter | None = None, config: Config | None = None): + def __init__(self, connection: Connection, config: Config | None = None): DownloadQueue._instance = self self.config = config or Config.get_instance() - self.emitter = emitter or Emitter.get_instance() + self._notify = EventBus.get_instance() self.done = DataStore(type=DownloadQueue.TYPE_DONE, connection=connection) self.queue = DataStore(type=DownloadQueue.TYPE_QUEUE, connection=connection) self.done.load() @@ -326,9 +325,7 @@ class DownloadQueue(metaclass=Singleton): itemDownload = self.queue.put(dlInfo) self.event.set() - asyncio.create_task( - self.emitter.emit(NotifyEvent, itemDownload.info), name=f"notifier-{NotifyEvent}-{itemDownload.info.id}" - ) + await self._notify.emit(NotifyEvent, data=itemDownload.info.serialize()) return {"status": "ok"} @@ -505,11 +502,11 @@ class DownloadQueue(metaclass=Singleton): await item.close() LOG.debug(f"Deleting from queue {item_ref}") self.queue.delete(id) - asyncio.create_task(self.emitter.cancelled(dl=item.info.serialize()), name=f"notifier-c-{id}") + await self._notify.emit(Events.CANCELLED, info=item.info.serialize()) item.info.status = "cancelled" item.info.error = "Cancelled by user." self.done.put(item) - asyncio.create_task(self.emitter.completed(dl=item.info.serialize()), name=f"notifier-d-{id}") + await self._notify.emit(Events.COMPLETED, data=item.info.serialize()) LOG.info(f"Deleted from queue {item_ref}") status[id] = "ok" @@ -563,7 +560,8 @@ class DownloadQueue(metaclass=Singleton): LOG.error(f"Unable to remove '{itemRef}' local file '{filename}'. {e!s}") self.done.delete(id) - asyncio.create_task(self.emitter.cleared(dl=item.info.serialize()), name=f"notifier-c-{id}") + await self._notify.emit(Events.CLEARED, data=item.info.serialize()) + msg = f"Deleted completed download '{itemRef}'." if fileDeleted and filename: msg += f" and removed local file '{filename}'." @@ -672,7 +670,7 @@ class DownloadQueue(metaclass=Singleton): try: self._active_downloads[entry.info._id] = entry - await entry.start(self.emitter) + await entry.start() if "finished" != entry.info.status: if entry.tmpfilename and os.path.isfile(entry.tmpfilename): @@ -694,12 +692,12 @@ class DownloadQueue(metaclass=Singleton): self.queue.delete(key=id) if entry.is_cancelled() is True: - asyncio.create_task(self.emitter.cancelled(dl=entry.info.serialize()), name=f"notifier-c-{id}") + await self._notify.emit(Events.CANCELLED, data=entry.info.serialize()) entry.info.status = "cancelled" entry.info.error = "Cancelled by user." self.done.put(value=entry) - asyncio.create_task(self.emitter.completed(dl=entry.info.serialize()), name=f"notifier-d-{id}") + await self._notify.emit(Events.COMPLETED, data=entry.info.serialize()) else: LOG.warning(f"Download '{id}' not found in queue.") diff --git a/app/library/Emitter.py b/app/library/Emitter.py deleted file mode 100644 index 6405606a..00000000 --- a/app/library/Emitter.py +++ /dev/null @@ -1,132 +0,0 @@ -import asyncio -import logging -from collections.abc import Awaitable - -from .Events import Events -from .Singleton import Singleton - -LOG = logging.getLogger("Emitter") - - -class Emitter(metaclass=Singleton): - """ - This class is used to emit events to the registered emitters. - """ - - emitters: list[(str, Awaitable)] = [] - """The emitters for the events.""" - - _instance = None - - def __init__(self): - Emitter._instance = self - - @staticmethod - def get_instance(): - """ - Get the instance of the Emitter. - - Returns: - Emitter: The instance of the Emitter - - """ - if not Emitter._instance: - Emitter._instance = Emitter() - return Emitter._instance - - def add_emitter(self, emitter: list[Awaitable] | Awaitable, local: bool = False) -> "Emitter": - """ - Add an emitter to the list of emitters. - - Args: - emitter (Awaitable|list[Awaitable]): The emitter function. The function must return a coroutine or None. - local (bool): Mark the emitter as target for local events. - - Returns: - Emitter: The instance of the Emitter - - """ - if not isinstance(emitter, list): - emitter = [emitter] - - for e in emitter: - self.emitters.append((local, e)) - - return self - - async def added(self, dl: dict, local: bool = False, **kwargs): - await self.emit(Events.ADDED, data=dl, local=local, **kwargs) - - async def updated(self, dl: dict, local: bool = False, **kwargs): - await self.emit(Events.UPDATED, data=dl, local=local, **kwargs) - - async def completed(self, dl: dict, local: bool = False, **kwargs): - await self.emit(Events.COMPLETED, data=dl, local=local, **kwargs) - - async def cancelled(self, dl: dict, local: bool = False, **kwargs): - await self.emit(Events.CANCELLED, data=dl, local=local, **kwargs) - - async def cleared(self, dl: dict | None = None, local: bool = False, **kwargs): - await self.emit(Events.CLEARED, data=dl, local=local, **kwargs) - - async def error(self, message: str, data: dict|None = None, local: bool = False, **kwargs): - msg = {"type": "error", "message": message, "data": {}} - if data: - msg.update({"data": data}) - await self.emit(Events.ERROR, data=msg, local=local, **kwargs) - - async def info(self, message: str, data: dict|None = None, local: bool = False, **kwargs): - msg = {"type": "info", "message": message, "data": {}} - if data: - msg.update({"data": data}) - await self.emit(Events.LOG_INFO, data=msg, local=local, **kwargs) - - async def success(self, message: str, data: dict|None = None, local: bool = False, **kwargs): - msg = {"type": "success", "message": message, "data": {}} - if data: - msg.update({"data": data}) - await self.emit(Events.LOG_SUCCESS, data=msg, local=local, **kwargs) - - async def emit(self, event: str, data, local: bool = False, **kwargs): - """ - Emit an event. - - Args: - event (str): The event to emit. - data (dict): The data to send with the event. - local (bool): If the event should be sent to the local emitters only. - kwargs: Additional arguments to pass to the emitters. - - Returns: - None - - """ - tasks = [] - - for emitter in self.emitters: - try: - isLocal, callback = emitter - if local and not isLocal: - continue - - _ret = callback(event, data, **kwargs) - if _ret: - if isinstance(_ret, list): - tasks.extend(_ret) - else: - tasks.append(_ret) - except Exception as e: - LOG.error(f"Emitter '{callback}' failed with error '{e}'.") - - if len(tasks) < 1: - return - - try: - await asyncio.wait_for(asyncio.gather(*tasks), timeout=60) - except asyncio.CancelledError: - LOG.error(f"Cancelled sending event '{event}'.") - except TimeoutError: - LOG.error(f"Timed out sending event '{event}'.") - except Exception as e: - LOG.exception(e) - LOG.error(f"Failed to send event '{event}'. '{e!s}'.") diff --git a/app/library/Events.py b/app/library/Events.py index c00d6354..d4f0eece 100644 --- a/app/library/Events.py +++ b/app/library/Events.py @@ -10,12 +10,74 @@ from .Singleton import Singleton LOG = logging.getLogger("EventsSubscriber") +def error(msg: str, data: dict | None = None) -> dict: + """ + Create an error message. + + Args: + msg (str): The message. + data (dict|None): The data to include in the message. + + Returns: + dict : The message wrapped in a dictionary. + + """ + return message("error", msg, data) + + +def info(msg: str, data: dict | None = None) -> dict: + """ + Create an info message. + + Args: + msg (str): The message. + data (dict|None): The data to include in the message. + + Returns: + dict : The message wrapped in a dictionary. + + """ + return message("info", msg, data) + + +def success(msg: str, data: dict | None = None) -> dict: + """ + Create a success message. + + Args: + msg (str): The message. + data (dict|None): The data to include in the message. + + Returns: + dict : The message wrapped in a dictionary. + + """ + return message("success", msg, data) + + +def message(type: str, message: str, data: dict | None = None) -> dict: + """ + Create a message. + + Args: + type (str): The type of the message. + message (str): The message. + data (dict|None): The data to include in the message. + + Returns: + dict : The message wrapped in a dictionary. + + """ + return {"type": type, "message": message, "data": data if data else {}} + + class Events: """ The events that can be emitted. """ STARTUP = "startup" + LOADED = "loaded" SHUTDOWN = "shutdown" ADDED = "added" @@ -50,6 +112,46 @@ class Events: PRESETS_UPDATE = "presets_update" SCHEDULE_ADD = "schedule_add" + def get_all() -> list: + """ + Get all the events. + + Returns: + list: The list of events. + + """ + return [ + getattr(Events, ev) for ev in dir(Events) if not ev.startswith("_") and not callable(getattr(Events, ev)) + ] + + def frontend() -> list: + """ + Get the frontend events. + + Returns: + list: The list of frontend events. + + """ + return [ + Events.ADDED, + Events.UPDATED, + Events.COMPLETED, + Events.CANCELLED, + Events.CLEARED, + Events.ERROR, + Events.LOG_INFO, + Events.LOG_SUCCESS, + Events.INITIAL_DATA, + Events.YTDLP_CONVERT, + Events.ITEM_DELETE, + Events.ITEM_CANCEL, + Events.STATUS, + Events.CLI_CLOSE, + Events.CLI_OUTPUT, + Events.UPDATE, + Events.PAUSED, + ] + @dataclass(kw_only=True) class Event: @@ -60,7 +162,7 @@ class Event: id: str = field(default_factory=lambda: str(uuid.uuid4()), init=False) """The id of the event.""" - created: str = field(default_factory=lambda: str(datetime.datetime.now(tz=datetime.timezone.UTC).isoformat())) + created: str = field(default_factory=lambda: str(datetime.datetime.now(tz=datetime.timezone.utc).isoformat())) """The time the event was created.""" event: str @@ -96,6 +198,23 @@ class Event: return f"Event(id={self.id}, created={self.created}, event={self.event})" +class EventListener: + name: str + is_coroutine: bool = False + call_back: callable + + def __init__(self, name: str, callback: callable): + self.name = name + self.call_back = callback + self.is_coroutine = asyncio.iscoroutinefunction(callback) + + async def handle(self, event: Event, **kwargs): + if self.is_coroutine: + return self.call_back(event, self.name, **kwargs) + else: + return asyncio.create_task(self.call_back(event, self.name, **kwargs)) + + class EventBus(metaclass=Singleton): """ This class is used to subscribe to and emit events to the registered listeners. @@ -104,7 +223,7 @@ class EventBus(metaclass=Singleton): _instance = None """the instance of the EventsSubscriber""" - _listeners: dict[str, list[str, Awaitable]] = {} + _listeners: dict[str, list[str, EventListener]] = {} """The listeners for the events.""" def __init__(self): @@ -137,17 +256,32 @@ class EventBus(metaclass=Singleton): EventsSubscriber: The instance of the EventsSubscriber """ + all_events = Events.get_all() + if isinstance(event, str): - event = [event] + if "*" == event: + event = all_events + elif "frontend" == event: + event = Events.frontend() + else: + if event not in all_events: + LOG.error(f"'{name}' attempted to listen on '{event}' which does not exist.") + return self + + event = [event] if not name: name = str(uuid.uuid4()) for e in event: + if e not in all_events: + LOG.error(f"'{name}' attempted to listen on '{e}' which does not exist.") + continue + if e not in self._listeners: self._listeners[e] = {} - self._listeners[e][name] = callback + self._listeners[e][name] = EventListener(name, callback) return self @@ -172,7 +306,7 @@ class EventBus(metaclass=Singleton): return self - def emit_sync(self, event: str, data: any, **kwargs) -> list: + def sync_emit(self, event: str, data: any, **kwargs) -> list: """ Emit an event synchronously. @@ -190,19 +324,19 @@ class EventBus(metaclass=Singleton): return [] ev = Event(event=event, data=data) - LOG.debug(f"Emitting event '{ev}'.") + LOG.debug(f"Emitting event '{ev.id}: {ev.event}'.", extra={"data": data}) results = [] - for name, callback in self._listeners[event].items(): + for handler in self._listeners[event].items(): try: - results.append(asyncio.get_event_loop().run_until_complete(callback(ev, name, **kwargs))) + 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 '{name}'. Error message '{e!s}'.") + LOG.error(f"Failed to emit event '{event}' to '{handler.name}'. Error message '{e!s}'.") return results - def emit(self, event: str, data: any, **kwargs) -> Awaitable: + async def emit(self, event: str, data: any, **kwargs) -> Awaitable: """ Emit an event. @@ -216,18 +350,17 @@ class EventBus(metaclass=Singleton): """ if event not in self._listeners: - return None + return [] ev = Event(event=event, data=data) - - LOG.debug(f"Emitting event '{ev}'.") + LOG.debug(f"Emitting event '{ev.id}: {ev.event}'.", extra={"data": data}) tasks = [] - for name, callback in self._listeners[event].items(): + for handler in self._listeners[event].values(): try: - tasks.append(asyncio.create_task(callback(ev, name, **kwargs))) + tasks.append(handler.handle(ev, **kwargs)) except Exception as e: LOG.exception(e) - LOG.error(f"Failed to emit event '{event}' to '{name}'. Error message '{e!s}'.") + LOG.error(f"Failed to emit event '{ev.event}' to '{handler.name}'. Error message '{e!s}'.") return asyncio.gather(*tasks) diff --git a/app/library/HttpAPI.py b/app/library/HttpAPI.py index a39c9f47..9d1ba7aa 100644 --- a/app/library/HttpAPI.py +++ b/app/library/HttpAPI.py @@ -24,9 +24,8 @@ from .cache import Cache from .common import Common from .config import Config from .DownloadQueue import DownloadQueue -from .Emitter import Emitter from .encoder import Encoder -from .Events import Events +from .Events import EventBus, Events, message from .ffprobe import ffprobe from .M3u8 import M3u8 from .Notifications import Notification, NotificationEvents @@ -70,14 +69,13 @@ class HttpAPI(Common): def __init__( self, queue: DownloadQueue | None = None, - emitter: Emitter | None = None, encoder: Encoder | None = None, config: Config | None = None, ): self.queue = queue or DownloadQueue.get_instance() - self.emitter = emitter or Emitter.get_instance() self.encoder = encoder or Encoder() self.config = config or Config.get_instance() + self._notify = EventBus.get_instance() self.rootPath = str(Path(__file__).parent.parent.parent) self.routes = web.RouteTableDef() @@ -793,7 +791,7 @@ class HttpAPI(Common): status=web.HTTPInternalServerError.status_code, ) - await self.emitter.emit(Events.PRESETS_UPDATE, presets) + await self._notify.emit(Events.PRESETS_UPDATE, data=presets) return web.json_response(data=presets, status=web.HTTPOk.status_code, dumps=self.encoder.encode) @route("GET", "api/tasks") @@ -951,7 +949,7 @@ class HttpAPI(Common): if updated: self.queue.done.put(item) - await self.emitter.emit(Events.UPDATE, item.info) + await self._notify.emit(Events.UPDATE, data=item.info) return web.json_response( data=item.info, @@ -1621,7 +1619,8 @@ class HttpAPI(Common): Response: The response object. """ - data = {"type": "test", "message": "This is a test notification."} - await self.emitter.emit(Events.TEST, data) + data = message("test", "This is a test notification.") + + await self._notify.emit(Events.TEST, data=data) return web.json_response(data=data, status=web.HTTPOk.status_code, dumps=self.encoder.encode) diff --git a/app/library/HttpSocket.py b/app/library/HttpSocket.py index d55c32c1..a1db7190 100644 --- a/app/library/HttpSocket.py +++ b/app/library/HttpSocket.py @@ -7,7 +7,6 @@ import pty import shlex import time from datetime import UTC, datetime -from typing import Any import anyio import socketio @@ -16,9 +15,8 @@ from aiohttp import web from .common import Common from .config import Config from .DownloadQueue import DownloadQueue -from .Emitter import Emitter from .encoder import Encoder -from .Events import EventBus, Events +from .Events import Event, EventBus, Events, error from .Presets import Presets from .Utils import arg_converter, is_downloaded @@ -33,26 +31,25 @@ class HttpSocket(Common): config: Config sio: socketio.AsyncServer queue: DownloadQueue - emitter: Emitter def __init__( self, queue: DownloadQueue | None = None, - emitter: Emitter | None = None, encoder: Encoder | None = None, config: Config | None = None, sio: socketio.AsyncServer | None = None, ): self.config = config or Config.get_instance() self.queue = queue or DownloadQueue.get_instance() - self.emitter = emitter or Emitter.get_instance() + self._notify = EventBus.get_instance() + self.sio = sio or socketio.AsyncServer(cors_allowed_origins="*") encoder = encoder or Encoder() - def emit(event: str, data: Any, **kwargs): - return self.sio.emit(event=event, data=encoder.encode(data), **kwargs) + def emit(e: Event, _, **kwargs): + return self.sio.emit(event=e.event, data=encoder.encode(e.data), **kwargs) - self.emitter.add_emitter([emit], local=False) + self._notify.subscribe("frontend", emit, f"{__class__.__name__}.socket_api") super().__init__(queue=queue, encoder=encoder, config=config) @@ -80,8 +77,8 @@ class HttpSocket(Common): if hasattr(method, "_ws_event") and self.sio: self.sio.on(method._ws_event)(method) # type: ignore - EventBus.get_instance().subscribe( - Events.ADD_URL, lambda data, _: self.add(**data.data), "socket_add_url" + self._notify.subscribe( + Events.ADD_URL, lambda data, _: self.add(**data.data), f"{__class__.__name__}.socket_add_url" ) # register the shutdown event. @@ -90,11 +87,11 @@ class HttpSocket(Common): @ws_event async def cli_post(self, sid: str, data): if not self.config.console_enabled: - await self.emitter.error("Console is disabled.", to=sid) + await self._notify.emit(Events.ERROR, data=error("Console is disabled."), to=sid) return if not data: - await self.emitter.emit(Events.CLI_CLOSE, {"exitcode": 0}, to=sid) + await self._notify.emit(Events.CLI_CLOSE, data={"exitcode": 0}, to=sid) return try: @@ -145,18 +142,18 @@ class HttpSocket(Common): # No more output if buffer: # Emit any remaining partial line - await self.emitter.emit( + await self._notify.emit( Events.CLI_OUTPUT, - {"type": "stdout", "line": buffer.decode("utf-8", errors="replace")}, + data={"type": "stdout", "line": buffer.decode("utf-8", errors="replace")}, to=sid, ) break buffer += chunk *lines, buffer = buffer.split(b"\n") for line in lines: - await self.emitter.emit( + await self._notify.emit( Events.CLI_OUTPUT, - {"type": "stdout", "line": line.decode("utf-8", errors="replace")}, + data={"type": "stdout", "line": line.decode("utf-8", errors="replace")}, to=sid, ) try: @@ -173,58 +170,58 @@ class HttpSocket(Common): # Ensure reading is done await read_task - await self.emitter.emit(Events.CLI_CLOSE, {"exitcode": returncode}, to=sid) + await self._notify.emit(Events.CLI_CLOSE, data={"exitcode": returncode}, to=sid) except Exception as e: LOG.error(f"CLI execute exception was thrown for client '{sid}'.") LOG.exception(e) - await self.emitter.emit(Events.CLI_OUTPUT, {"type": "stderr", "line": str(e)}, to=sid) - await self.emitter.emit(Events.CLI_CLOSE, {"exitcode": -1}, to=sid) + await self._notify.emit(Events.CLI_OUTPUT, data={"type": "stderr", "line": str(e)}, to=sid) + await self._notify.emit(Events.CLI_CLOSE, data={"exitcode": -1}, to=sid) @ws_event async def add_url(self, sid: str, data: dict): url: str | None = data.get("url") if not url: - await self.emitter.error("No URL provided.", data={"unlock": True}, to=sid) + await self._notify.emit(Events.ERROR, data=error("No URL provided.", data={"unlock": True}), to=sid) return try: item = self.format_item(data) except ValueError as e: - await self.emitter.error(str(e), to=sid) + await self._notify.emit(Events.ERROR, data=error(str(e)), to=sid) return status = await self.add(**item) - await self.emitter.emit(event=Events.STATUS, data=status, to=sid) + await self._notify.emit(Events.STATUS, data=status, to=sid) @ws_event async def item_cancel(self, sid: str, id: str): if not id: - await self.emitter.error("Invalid request.", to=sid) + await self._notify.emit(Events.ERROR, data=error("Invalid request."), to=sid) return status: dict[str, str] = {} status = await self.queue.cancel([id]) status.update({"identifier": id}) - await self.emitter.emit(event=Events.ITEM_CANCEL, data=status) + await self._notify.emit(Events.ITEM_CANCEL, data=status) @ws_event async def item_delete(self, sid: str, data: dict): if not data: - await self.emitter.error("Invalid request.", to=sid) + await self._notify.emit(Events.ERROR, data=error("Invalid request."), to=sid) return id: str | None = data.get("id") if not id: - await self.emitter.error("Invalid request.", to=sid) + await self._notify.emit(Events.ERROR, data=error("Invalid request."), to=sid) return status: dict[str, str] = {} status = await self.queue.clear([id], remove_file=bool(data.get("remove_file", False))) status.update({"identifier": id}) - await self.emitter.emit(event=Events.ITEM_DELETE, data=status) + await self._notify.emit(Events.ITEM_DELETE, data=status) @ws_event async def archive_item(self, _: str, data: dict): @@ -275,36 +272,36 @@ class HttpSocket(Common): downloadPath: str = self.config.download_path data["folders"] = [name for name in os.listdir(downloadPath) if os.path.isdir(os.path.join(downloadPath, name))] - await self.emitter.emit(event=Events.INITIAL_DATA, data=data, to=sid) + await self._notify.emit(Events.INITIAL_DATA, data=data, to=sid) @ws_event async def pause(self, *_, **__): self.queue.pause() - await self.emitter.emit(event=Events.PAUSED, data={"paused": True, "at": time.time()}) + await self._notify.emit(Events.PAUSED, data={"paused": True, "at": time.time()}) @ws_event async def resume(self, *_, **__): self.queue.resume() - await self.emitter.emit(event=Events.PAUSED, data={"paused": False, "at": time.time()}) + await self._notify.emit(Events.PAUSED, data={"paused": False, "at": time.time()}) @ws_event async def ytdlp_convert(self, sid: str, data: dict): if not isinstance(data, dict) or "args" not in data: - await self.emitter.error("Invalid request or no options were given.", to=sid) + await self._notify.emit(Events.ERROR, data=error("Invalid request or no options were given."), to=sid) return args: str | None = data.get("args") if not args: - await self.emitter.error("no options were given.", to=sid) + await self._notify.emit(Events.ERROR, data=error("no options were given."), to=sid) return try: - await self.emitter.emit(event=Events.YTDLP_CONVERT, data=arg_converter(args), to=sid) + await self._notify.emit(Events.YTDLP_CONVERT, data=arg_converter(args), to=sid) except Exception as e: err = str(e).strip() err = err.split("\n")[-1] if "\n" in err else err LOG.error(f"Failed to convert args. '{err}'.") - await self.emitter.error(f"Failed to convert options. '{err}'.", to=sid) + await self._notify.emit(Events.ERROR, data=error(f"Failed to convert options. '{err}'."), to=sid) return diff --git a/app/library/Notifications.py b/app/library/Notifications.py index 794d7175..9ddeabf2 100644 --- a/app/library/Notifications.py +++ b/app/library/Notifications.py @@ -7,10 +7,11 @@ from datetime import UTC, datetime from typing import Any import httpx +from aiohttp import web from .config import Config from .encoder import Encoder -from .Events import Events +from .Events import EventBus, Event, Events from .ItemDTO import ItemDTO from .Singleton import Singleton from .Utils import ag, validate_uuid @@ -98,6 +99,13 @@ class NotificationEvents: def get_events() -> dict[str, str]: return {k: v for k, v in vars(NotificationEvents).items() if not k.startswith("__") and not callable(v)} + def events() -> list: + return [ + getattr(NotificationEvents, ev) + for ev in dir(NotificationEvents) + if not ev.startswith("_") and not callable(getattr(NotificationEvents, ev)) + ] + @staticmethod def is_valid(event: str) -> bool: return event in NotificationEvents.get_events().values() @@ -142,6 +150,17 @@ class Notification(metaclass=Singleton): return Notification._instance + def attach(self, _: web.Application): + """ + Attach the class to the application. + + Args: + _ (web.Application): The aiohttp application. + + """ + self.load() + EventBus.get_instance().subscribe(NotificationEvents.events(), self.emit, f"{__class__.__name__}.emit") + def get_targets(self) -> list[Target]: """Get the list of notification targets.""" return self._targets @@ -211,6 +230,35 @@ class Notification(metaclass=Singleton): return self + def make_target(self, target: dict) -> Target: + """ + Make a notification target from a dictionary. + + Args: + target (dict): The target details. + + Returns: + Target: The notification target. + + """ + return Target( + id=target.get("id"), + name=target.get("name"), + on=target.get("on", []), + request=TargetRequest( + type=target.get("request", {}).get("type", "json"), + method=target.get("request", {}).get("method", "POST"), + url=target.get("request", {}).get("url"), + headers=[ + TargetRequestHeader( + key=str(h.get("key", "")).strip(), + value=str(h.get("value", "")).strip(), + ) + for h in target.get("request", {}).get("headers", []) + ], + ), + ) + @staticmethod def validate(target: Target | dict) -> bool: """ @@ -340,40 +388,11 @@ class Notification(metaclass=Singleton): LOG.error(f"Error sending Notification event '{event}' id '{itemId}' to '{target.name}'. '{e}'.") return {"url": target.request.url, "status": 500, "text": str(e)} - def make_target(self, target: dict) -> Target: - """ - Make a notification target from a dictionary. - - Args: - target (dict): The target details. - - Returns: - Target: The notification target. - - """ - return Target( - id=target.get("id"), - name=target.get("name"), - on=target.get("on", []), - request=TargetRequest( - type=target.get("request", {}).get("type", "json"), - method=target.get("request", {}).get("method", "POST"), - url=target.get("request", {}).get("url"), - headers=[ - TargetRequestHeader( - key=str(h.get("key", "")).strip(), - value=str(h.get("value", "")).strip(), - ) - for h in target.get("request", {}).get("headers", []) - ], - ), - ) - - def emit(self, event, data, **kwargs): # noqa: ARG002 + def emit(self, e: Event, _, **kwargs): # noqa: ARG002 if len(self._targets) < 1: return False - if not NotificationEvents.is_valid(event): + if not NotificationEvents.is_valid(e.event): return False - return self.send(event, data) + return self.send(e.event, e.data) diff --git a/app/library/Presets.py b/app/library/Presets.py index 2e0d54d4..83ded605 100644 --- a/app/library/Presets.py +++ b/app/library/Presets.py @@ -8,7 +8,6 @@ from typing import Any from aiohttp import web from .config import Config -from .Emitter import Emitter from .encoder import Encoder from .Events import EventBus, Events from .Singleton import Singleton @@ -67,13 +66,12 @@ class Presets(metaclass=Singleton): _default_presets: list[Preset] = [] - def __init__(self, file: str | None = None, emitter: Emitter | None = None, config: Config | None = None): + def __init__(self, file: str | None = None, config: Config | None = None): Presets._instance = self config = config or Config.get_instance() self._file: str = file or os.path.join(config.config_path, "presets.json") - self._emitter: Emitter = emitter or Emitter.get_instance() if os.path.exists(self._file) and "600" != oct(os.stat(self._file).st_mode)[-3:]: try: @@ -85,7 +83,7 @@ class Presets(metaclass=Singleton): self._default_presets = [Preset(**preset) for preset in json.load(f)] EventBus.get_instance().subscribe( - Events.PRESETS_ADD, lambda data, _: self.add(**data.data), f"{__class__}.save" + Events.PRESETS_ADD, lambda data, _: self.add(**data.data), f"{__class__.__name__}.save" ) @staticmethod diff --git a/app/library/Scheduler.py b/app/library/Scheduler.py index 7b6756de..89397b01 100644 --- a/app/library/Scheduler.py +++ b/app/library/Scheduler.py @@ -27,7 +27,7 @@ class Scheduler(metaclass=Singleton): self._loop = loop or asyncio.get_event_loop() EventBus.get_instance().subscribe( - Events.SCHEDULE_ADD, lambda data, _: self.add(**data.data), f"{__class__}.add" + Events.SCHEDULE_ADD, lambda data, _: self.add(**data.data), f"{__class__.__name__}.add" ) @staticmethod diff --git a/app/library/Tasks.py b/app/library/Tasks.py index 9d69112d..5966feed 100644 --- a/app/library/Tasks.py +++ b/app/library/Tasks.py @@ -11,9 +11,8 @@ import httpx from aiohttp import web from .config import Config -from .Emitter import Emitter from .encoder import Encoder -from .Events import Event, EventBus, Events +from .Events import EventBus, Events, error, info, success from .Scheduler import Scheduler from .Singleton import Singleton @@ -56,7 +55,6 @@ class Tasks(metaclass=Singleton): def __init__( self, file: str | None = None, - emitter: Emitter | None = None, loop: asyncio.AbstractEventLoop | None = None, config: Config | None = None, encoder: Encoder | None = None, @@ -73,8 +71,8 @@ class Tasks(metaclass=Singleton): self._client = client or httpx.AsyncClient() self._encoder = encoder or Encoder() self._loop = loop or asyncio.get_event_loop() - self._emitter = emitter or Emitter.get_instance() self._scheduler = scheduler or Scheduler.get_instance() + self._notify = EventBus.get_instance() if os.path.exists(self._file) and "600" != oct(os.stat(self._file).st_mode)[-3:]: try: @@ -82,7 +80,7 @@ class Tasks(metaclass=Singleton): except Exception: pass - EventBus.get_instance().subscribe(Events.TASKS_ADD, lambda data, _: self.add(**data.data), f"{__class__}.save") + self._notify.subscribe(Events.TASKS_ADD, lambda data, _: self.add(**data.data), f"{__class__.__name__}.save") @staticmethod def get_instance() -> "Tasks": @@ -296,23 +294,22 @@ class Tasks(metaclass=Singleton): LOG.info(f"Task '{task.id}: {task.name}' dispatched at '{timeNow}'.") tasks = [] - tasks.append(self._emitter.info(f"Task '{task.name}' dispatched at '{timeNow}'.")) tasks.append( - self._emitter.emit( - event=Events.ADD_URL, - data=Event( - id=task.id, - data={ - "url": task.url, - "preset": preset, - "folder": folder, - "cookies": cookies, - "config": config, - "template": template, - }, - ), - local=True, - ) + self._notify.emit(Events.LOG_INFO, data=info(f"Task '{task.name}' dispatched at '{timeNow}'.")) + ) + tasks.append( + self._notify.emit( + Events.ADD_URL, + data={ + "url": task.url, + "preset": preset, + "folder": folder, + "cookies": cookies, + "config": config, + "template": template, + }, + id=task.id, + ), ) await asyncio.wait_for(asyncio.gather(*tasks), timeout=None) @@ -322,8 +319,13 @@ class Tasks(metaclass=Singleton): ended = time.time() LOG.info(f"Task '{task.id}: {task.name}' completed at '{timeNow}' took '{ended - started:.2f}' seconds.") - await self._emitter.success(f"Task '{task.name}' completed in '{ended - started:.2f}' seconds.") + await self._notify.emit( + Events.LOG_SUCCESS, + data=success(f"Task '{task.name}' completed in '{ended - started:.2f}' seconds."), + ) except Exception as e: timeNow = datetime.now(UTC).isoformat() LOG.error(f"Task '{task.id}: {task.name}' has failed to execute at '{timeNow}'. '{e!s}'.") - await self._emitter.error(f"Task '{task.name}' failed to execute at '{timeNow}'. '{e!s}'.") + await self._notify.emit( + Events.ERROR, data=error(f"Task '{task.name}' failed to execute at '{timeNow}'. '{e!s}'.") + ) diff --git a/app/main.py b/app/main.py index bbe54962..a49957bc 100644 --- a/app/main.py +++ b/app/main.py @@ -11,7 +11,6 @@ import magic from aiohttp import web from library.config import Config from library.DownloadQueue import DownloadQueue -from library.Emitter import Emitter from library.Events import EventBus, Events from library.HttpAPI import HttpAPI from library.HttpSocket import HttpSocket @@ -60,10 +59,6 @@ class Main: self._http = HttpAPI(queue=self._queue) self._socket = HttpSocket(queue=self._queue) - Emitter.get_instance().add_emitter([Notification().emit], local=False).add_emitter( - [EventBus().emit], local=True - ) - self._app.on_cleanup.append(_close_connection) def _check_folders(self): @@ -94,7 +89,7 @@ class Main: """ Start the application. """ - EventBus.get_instance().emit(Events.STARTUP, data={"app": self._app}) + EventBus.get_instance().sync_emit(Events.STARTUP, data={"app": self._app}) self._socket.attach(self._app) self._http.attach(self._app) @@ -102,6 +97,9 @@ class Main: Scheduler.get_instance().attach(self._app) Tasks.get_instance().attach(self._app) Presets.get_instance().attach(self._app) + Notification.get_instance().attach(self._app) + + EventBus.get_instance().sync_emit(Events.LOADED, data={"app": self._app}) def started(_): LOG.info("=" * 40) From 4ed7c26bb45ebc07a306ba8cb596dde4c6885af6 Mon Sep 17 00:00:00 2001 From: ArabCoders Date: Sat, 22 Mar 2025 02:05:52 +0300 Subject: [PATCH 3/4] migrated all our emitter usage to EventBus --- app/library/DownloadQueue.py | 2 +- app/library/Events.py | 28 ++++++++---------- app/library/HttpSocket.py | 4 ++- app/library/Notifications.py | 56 ++++++++++++++++++++--------------- app/library/Presets.py | 4 ++- app/library/Scheduler.py | 4 ++- app/library/Tasks.py | 7 +++-- app/library/encoder.py | 5 ++++ ui/components/History.vue | 2 +- ui/components/Queue.vue | 57 +++++++++++++++++++++++++++++------- ui/pages/console.vue | 8 +++++ ui/pages/notifications.vue | 6 ++++ ui/stores/SocketStore.js | 2 +- 13 files changed, 128 insertions(+), 57 deletions(-) diff --git a/app/library/DownloadQueue.py b/app/library/DownloadQueue.py index 9bfaa235..d41271ef 100644 --- a/app/library/DownloadQueue.py +++ b/app/library/DownloadQueue.py @@ -502,7 +502,7 @@ class DownloadQueue(metaclass=Singleton): await item.close() LOG.debug(f"Deleting from queue {item_ref}") self.queue.delete(id) - await self._notify.emit(Events.CANCELLED, info=item.info.serialize()) + await self._notify.emit(Events.CANCELLED, data=item.info.serialize()) item.info.status = "cancelled" item.info.error = "Cancelled by user." self.done.put(item) diff --git a/app/library/Events.py b/app/library/Events.py index d4f0eece..281edac4 100644 --- a/app/library/Events.py +++ b/app/library/Events.py @@ -133,23 +133,21 @@ class Events: """ return [ + Events.INITIAL_DATA, Events.ADDED, - Events.UPDATED, - Events.COMPLETED, - Events.CANCELLED, - Events.CLEARED, Events.ERROR, Events.LOG_INFO, Events.LOG_SUCCESS, - Events.INITIAL_DATA, - Events.YTDLP_CONVERT, - Events.ITEM_DELETE, - Events.ITEM_CANCEL, + Events.COMPLETED, + Events.CANCELLED, + Events.CLEARED, + Events.UPDATED, + Events.UPDATE, + Events.PAUSED, + Events.PRESETS_UPDATE, Events.STATUS, Events.CLI_CLOSE, Events.CLI_OUTPUT, - Events.UPDATE, - Events.PAUSED, ] @@ -162,7 +160,7 @@ class Event: id: str = field(default_factory=lambda: str(uuid.uuid4()), init=False) """The id of the event.""" - created: str = field(default_factory=lambda: str(datetime.datetime.now(tz=datetime.timezone.utc).isoformat())) + created_at: str = field(default_factory=lambda: str(datetime.datetime.now(tz=datetime.timezone.utc).isoformat())) """The time the event was created.""" event: str @@ -179,10 +177,10 @@ class Event: dict: The serialized event. """ - return {"id": self.id, "created": self.created, "event": self.event, "data": self.data} + return {"id": self.id, "created_at": self.created_at, "event": self.event, "data": self.data} def __repr__(self): - return f"Event(id={self.id}, created={self.created}, event={self.event}, data={self.data})" + return f"Event(id={self.id}, created_at={self.created_at}, event={self.event}, data={self.data})" def datatype(self) -> str: """ @@ -195,7 +193,7 @@ class Event: return type(self.data).__name__ def __str__(self): - return f"Event(id={self.id}, created={self.created}, event={self.event})" + return f"Event(id={self.id}, created_at={self.created_at}, event={self.event})" class EventListener: @@ -250,7 +248,7 @@ class EventBus(metaclass=Singleton): Args: event (str): The event to subscribe to. name (str|None): The name of the subscriber, if None a random uuid will be generated. - callback(Event) (Awaitable): The function to call. Must be a coroutine. + callback(Event, name, **kwargs) (Awaitable): The function to call. Must be a coroutine. Returns: EventsSubscriber: The instance of the EventsSubscriber diff --git a/app/library/HttpSocket.py b/app/library/HttpSocket.py index a1db7190..e266d308 100644 --- a/app/library/HttpSocket.py +++ b/app/library/HttpSocket.py @@ -78,7 +78,9 @@ class HttpSocket(Common): self.sio.on(method._ws_event)(method) # type: ignore self._notify.subscribe( - Events.ADD_URL, lambda data, _: self.add(**data.data), f"{__class__.__name__}.socket_add_url" + Events.ADD_URL, + lambda data, _, **kwargs: self.add(**data.data), # noqa: ARG005 + f"{__class__.__name__}.socket_add_url", ) # register the shutdown event. diff --git a/app/library/Notifications.py b/app/library/Notifications.py index 9ddeabf2..f372fbb5 100644 --- a/app/library/Notifications.py +++ b/app/library/Notifications.py @@ -223,7 +223,7 @@ class Notification(metaclass=Singleton): self._targets.append(target) LOG.info( - f"Will send '{target.on if len(target.on) > 0 else 'all'}' as {target.request.type} notification events to '{target.name}'." + f"Will send {target.request.type} request on '{', '.join(target.on) if len(target.on) > 0 else 'all events'}' to '{target.name}'." ) except Exception as e: LOG.error(f"Error loading notification target '{target}'. '{e!s}'") @@ -323,37 +323,35 @@ class Notification(metaclass=Singleton): return True - async def send(self, event: str, item: ItemDTO | dict) -> list[dict]: + async def send(self, ev: Event) -> list[dict]: if len(self._targets) < 1: return [] - if not isinstance(item, ItemDTO) and not isinstance(item, dict): - LOG.debug(f"Received invalid item type '{type(item)}' with event '{event}'.") + if not isinstance(ev.data, ItemDTO) and not isinstance(ev.data, dict): + LOG.debug(f"Received invalid item type '{type(ev.data)}' with event '{ev.event}'.") return [] tasks = [] for target in self._targets: - if len(target.on) > 0 and event not in target.on and "test" != event: + if len(target.on) > 0 and ev.event not in target.on and "test" != ev.event: continue - tasks.append(self._send(event, target, item)) + tasks.append(self._send(target, ev)) return await asyncio.gather(*tasks) - async def _send(self, event: str, target: Target, item: ItemDTO | dict) -> dict: + async def _send(self, target: Target, ev: Event) -> dict: try: - itemId = item.get("id", item.get("_id", "??")) - except Exception: - itemId = "??" + LOG.info(f"Sending Notification event '{ev.event}: {ev.id}' to '{target.name}'.") - try: - LOG.info(f"Sending Notification event '{event}' id '{itemId}' to '{target.name}'.") reqBody = { "method": target.request.method.upper(), "url": target.request.url, "headers": { "User-Agent": f"YTPTube/{APP_VERSION}", + "X-Event-Id": ev.id, + "X-Event": ev.event, "Content-Type": "application/json" if "json" == target.request.type.lower() else "application/x-www-form-urlencoded", @@ -364,20 +362,16 @@ class Notification(metaclass=Singleton): for h in target.request.headers: reqBody["headers"][h.key] = h.value - reqBody["json" if "json" == target.request.type.lower() else "data"] = { - "event": event, - "created_at": datetime.now(tz=UTC).isoformat(), - "payload": item.__dict__ if isinstance(item, ItemDTO) else item, - } + reqBody["json" if "json" == target.request.type.lower() else "data"] = self._deep_unpack(ev.serialize()) if "form" == target.request.type.lower(): - reqBody["data"]["payload"] = self._encoder.encode(reqBody["data"]["payload"]) + reqBody["data"]["data"] = self._encoder.encode(reqBody["data"]["data"]) response = await self._client.request(**reqBody) respData = {"url": target.request.url, "status": response.status_code, "text": response.text} - msg = f"Notification target '{target.name}' Responded to event '{event}' id '{itemId}' with status '{response.status_code}'." + msg = f"Notification target '{target.name}' Responded to event '{ev.event}: {ev.id}' with status '{response.status_code}'." if self._debug and respData.get("text"): msg += f" body '{respData.get('text','??')}'." @@ -385,14 +379,28 @@ class Notification(metaclass=Singleton): return respData except Exception as e: - LOG.error(f"Error sending Notification event '{event}' id '{itemId}' to '{target.name}'. '{e}'.") - return {"url": target.request.url, "status": 500, "text": str(e)} + LOG.exception(e) + LOG.error(f"Error sending Notification event '{ev.event}: {ev.id}' to '{target.name}'. '{e!s}'.") + return {"url": target.request.url, "status": 500, "text": str(ev)} def emit(self, e: Event, _, **kwargs): # noqa: ARG002 if len(self._targets) < 1: - return False + return [] if not NotificationEvents.is_valid(e.event): - return False + return [] - return self.send(e.event, e.data) + return self.send(e) + + def _deep_unpack(self, data: dict) -> dict: + for k, v in data.items(): + if isinstance(v, dict): + data[k] = self._deep_unpack(v) + if isinstance(v, list): + data[k] = [self._deep_unpack(i) for i in v] + if isinstance(v, datetime): + data[k] = v.isoformat() + if isinstance(v, ItemDTO): + data[k] = v.serialize() + + return data diff --git a/app/library/Presets.py b/app/library/Presets.py index 83ded605..39ae334e 100644 --- a/app/library/Presets.py +++ b/app/library/Presets.py @@ -83,7 +83,9 @@ class Presets(metaclass=Singleton): self._default_presets = [Preset(**preset) for preset in json.load(f)] EventBus.get_instance().subscribe( - Events.PRESETS_ADD, lambda data, _: self.add(**data.data), f"{__class__.__name__}.save" + Events.PRESETS_ADD, + lambda data, _, **kwargs: self.add(**data.data), # noqa: ARG005 + f"{__class__.__name__}.save", ) @staticmethod diff --git a/app/library/Scheduler.py b/app/library/Scheduler.py index 89397b01..489fa892 100644 --- a/app/library/Scheduler.py +++ b/app/library/Scheduler.py @@ -27,7 +27,9 @@ class Scheduler(metaclass=Singleton): self._loop = loop or asyncio.get_event_loop() EventBus.get_instance().subscribe( - Events.SCHEDULE_ADD, lambda data, _: self.add(**data.data), f"{__class__.__name__}.add" + Events.SCHEDULE_ADD, + lambda data, _, **kwargs: self.add(**data.data), # noqa: ARG005 + f"{__class__.__name__}.add", ) @staticmethod diff --git a/app/library/Tasks.py b/app/library/Tasks.py index 5966feed..b3ac7a33 100644 --- a/app/library/Tasks.py +++ b/app/library/Tasks.py @@ -80,8 +80,6 @@ class Tasks(metaclass=Singleton): except Exception: pass - self._notify.subscribe(Events.TASKS_ADD, lambda data, _: self.add(**data.data), f"{__class__.__name__}.save") - @staticmethod def get_instance() -> "Tasks": """ @@ -108,6 +106,11 @@ class Tasks(metaclass=Singleton): """ self.load() + self._notify.subscribe( + Events.TASKS_ADD, + lambda data, _, **kwargs: self.add(**data.data), # noqa: ARG005 + f"{__class__.__name__}.save", + ) def get_all(self) -> list[Task]: """Return the tasks.""" diff --git a/app/library/encoder.py b/app/library/encoder.py index ad3d154f..5f292d64 100644 --- a/app/library/encoder.py +++ b/app/library/encoder.py @@ -5,6 +5,8 @@ from pathlib import Path from yt_dlp.networking.impersonate import ImpersonateTarget from yt_dlp.utils import DateRange +from .ItemDTO import ItemDTO + class Encoder(json.JSONEncoder): """ @@ -26,6 +28,9 @@ class Encoder(json.JSONEncoder): if isinstance(o, ImpersonateTarget): return str(o) + if isinstance(o, ItemDTO): + return o.serialize() + if isinstance(o, object): if hasattr(o, "serialize"): return o.serialize() diff --git a/ui/components/History.vue b/ui/components/History.vue index 36e1f0f4..ddc5d985 100644 --- a/ui/components/History.vue +++ b/ui/components/History.vue @@ -411,7 +411,7 @@ const setIcon = item => { } const setIconColor = item => { - if (item.status === 'finished') { + if ('finished' === item.status) { return 'has-text-success' } diff --git a/ui/components/Queue.vue b/ui/components/Queue.vue index e35d2bea..d5b68b33 100644 --- a/ui/components/Queue.vue +++ b/ui/components/Queue.vue @@ -76,11 +76,10 @@
- - + + - Live Streaming - {{ ucFirst(item.status) }} +
@@ -153,7 +152,7 @@