From 07fc86565aada43e52fa47b8e1ce1bf407d293ec Mon Sep 17 00:00:00 2001 From: ArabCoders Date: Fri, 21 Mar 2025 23:56:02 +0300 Subject: [PATCH] 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)