diff --git a/Pipfile.lock b/Pipfile.lock
index 952058dd..6ae296de 100644
--- a/Pipfile.lock
+++ b/Pipfile.lock
@@ -1299,12 +1299,12 @@
},
"yt-dlp": {
"hashes": [
- "sha256:3ed218eaeece55e9d715afd41abc450dc406ee63bf79355169dfde312d38fdb8",
- "sha256:f33ca76df2e4db31880f2fe408d44f5058d9f135015b13e50610dfbe78245bea"
+ "sha256:5bcf47b2897254ea3816935a8dde47d243bff556782cced6b16a2b85e6b682ba",
+ "sha256:80d5ce15f9223e0c27020b861a4c5b72c6ba5d6c957c1b8fd2a022a69783f482"
],
"index": "pypi",
"markers": "python_version >= '3.9'",
- "version": "==2025.2.19"
+ "version": "==2025.3.21"
}
},
"develop": {}
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 c6f491ce..d41271ef 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 .EventsSubscriber 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, data=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 8348ab70..00000000
--- a/app/library/Emitter.py
+++ /dev/null
@@ -1,132 +0,0 @@
-import asyncio
-import logging
-from collections.abc import Awaitable
-
-from .EventsSubscriber 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
new file mode 100644
index 00000000..281edac4
--- /dev/null
+++ b/app/library/Events.py
@@ -0,0 +1,364 @@
+import asyncio
+import datetime
+import logging
+import uuid
+from collections.abc import Awaitable
+from dataclasses import dataclass, field
+
+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"
+ UPDATED = "updated"
+ COMPLETED = "completed"
+ CANCELLED = "cancelled"
+ CLEARED = "cleared"
+ ERROR = "error"
+ LOG_INFO = "log_info"
+ LOG_SUCCESS = "log_success"
+
+ INITIAL_DATA = "initial_data"
+ YTDLP_CONVERT = "ytdlp_convert"
+ ITEM_DELETE = "item_delete"
+ ITEM_CANCEL = "item_cancel"
+ STATUS = "status"
+ CLI_CLOSE = "cli_close"
+ CLI_OUTPUT = "cli_output"
+ UPDATE = "update"
+ TEST = "test"
+ ADD_URL = "add_url"
+
+ CLI_POST = "cli_post"
+ PAUSED = "paused"
+
+ TASKS_ADD = "task_add"
+ TASK_DISPATCHED = "task_dispatched"
+ TASK_FINISHED = "task_finished"
+ TASK_ERROR = "task_error"
+
+ PRESETS_ADD = "presets_add"
+ 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.INITIAL_DATA,
+ Events.ADDED,
+ Events.ERROR,
+ Events.LOG_INFO,
+ Events.LOG_SUCCESS,
+ Events.COMPLETED,
+ Events.CANCELLED,
+ Events.CLEARED,
+ Events.UPDATED,
+ Events.UPDATE,
+ Events.PAUSED,
+ Events.PRESETS_UPDATE,
+ Events.STATUS,
+ Events.CLI_CLOSE,
+ Events.CLI_OUTPUT,
+ ]
+
+
+@dataclass(kw_only=True)
+class Event:
+ """
+ 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_at: 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_at": self.created_at, "event": self.event, "data": self.data}
+
+ def __repr__(self):
+ return f"Event(id={self.id}, created_at={self.created_at}, 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_at={self.created_at}, 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.
+ """
+
+ _instance = None
+ """the instance of the EventsSubscriber"""
+
+ _listeners: dict[str, list[str, EventListener]] = {}
+ """The listeners for the events."""
+
+ def __init__(self):
+ EventBus._instance = self
+
+ @staticmethod
+ def get_instance() -> "EventBus":
+ """
+ Get the instance of the EventsSubscriber.
+
+ Returns:
+ EventsSubscriber: The instance of the EventsSubscriber
+
+ """
+ if not EventBus._instance:
+ EventBus._instance = EventBus()
+
+ 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.
+ name (str|None): The name of the subscriber, if None a random uuid will be generated.
+ callback(Event, name, **kwargs) (Awaitable): The function to call. Must be a coroutine.
+
+ Returns:
+ EventsSubscriber: The instance of the EventsSubscriber
+
+ """
+ all_events = Events.get_all()
+
+ if isinstance(event, str):
+ 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] = EventListener(name, callback)
+
+ return self
+
+ def unsubscribe(self, event: str | list | tuple, name: str) -> "EventBus":
+ """
+ Unsubscribe from an event.
+
+ Args:
+ event (str): The event to unsubscribe from.
+ name (str): The name of the subscriber.
+
+ Returns:
+ EventsSubscriber: The instance of the EventsSubscriber
+
+ """
+ if isinstance(event, str):
+ event = [event]
+
+ for e in event:
+ if e in self._listeners and name in self._listeners[e]:
+ del self._listeners[e][name]
+
+ return self
+
+ def sync_emit(self, event: str, data: any, **kwargs) -> list:
+ """
+ Emit an event synchronously.
+
+ Args:
+ event (str): The event to emit.
+ data (any): The data to pass to the event.
+ **kwargs: The keyword arguments to pass to the event.
+
+ Returns:
+ list: The results are the return values of the coroutines. If the coroutine raises an exception,
+ the exception is caught and logged. If event does not exist, an empty list is returned.
+
+ """
+ if event not in self._listeners:
+ return []
+
+ ev = Event(event=event, data=data)
+ LOG.debug(f"Emitting event '{ev.id}: {ev.event}'.", extra={"data": data})
+
+ results = []
+ for handler in self._listeners[event].items():
+ try:
+ results.append(asyncio.get_event_loop().run_until_complete(handler.handle(ev, **kwargs)))
+ except Exception as e:
+ LOG.exception(e)
+ LOG.error(f"Failed to emit event '{event}' to '{handler.name}'. Error message '{e!s}'.")
+
+ return results
+
+ async def emit(self, event: str, data: any, **kwargs) -> Awaitable:
+ """
+ Emit an event.
+
+ Args:
+ event (str): The event to emit.
+ data (any): The data to pass to the event.
+ **kwargs: The keyword arguments to pass to the event.
+
+ Returns:
+ Awaitable: The task that was created to run the event.
+
+ """
+ if event not in self._listeners:
+ return []
+
+ ev = Event(event=event, data=data)
+ LOG.debug(f"Emitting event '{ev.id}: {ev.event}'.", extra={"data": data})
+
+ tasks = []
+ for handler in self._listeners[event].values():
+ try:
+ tasks.append(handler.handle(ev, **kwargs))
+ except Exception as e:
+ LOG.exception(e)
+ 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/EventsSubscriber.py b/app/library/EventsSubscriber.py
deleted file mode 100644
index 755fd471..00000000
--- a/app/library/EventsSubscriber.py
+++ /dev/null
@@ -1,193 +0,0 @@
-import asyncio
-import logging
-from collections.abc import Awaitable
-from dataclasses import dataclass
-
-from .Singleton import Singleton
-
-LOG = logging.getLogger("EventsSubscriber")
-
-
-class Events:
- """
- The events that can be emitted.
- """
-
- STARTUP = "startup"
- SHUTDOWN = "shutdown"
-
- ADDED = "added"
- UPDATED = "updated"
- COMPLETED = "completed"
- CANCELLED = "cancelled"
- CLEARED = "cleared"
- ERROR = "error"
- LOG_INFO = "log_info"
- LOG_SUCCESS = "log_success"
-
- INITIAL_DATA = "initial_data"
- YTDLP_CONVERT = "ytdlp_convert"
- ITEM_DELETE = "item_delete"
- ITEM_CANCEL = "item_cancel"
- STATUS = "status"
- CLI_CLOSE = "cli_close"
- CLI_OUTPUT = "cli_output"
- UPDATE = "update"
- TEST = "test"
- ADD_URL = "add_url"
-
- CLI_POST = "cli_post"
- PAUSED = "paused"
-
- TASKS_ADD = "task_add"
- TASK_DISPATCHED = "task_dispatched"
- TASK_FINISHED = "task_finished"
- TASK_ERROR = "task_error"
-
- PRESETS_ADD = "presets_add"
- PRESETS_UPDATE = "presets_update"
- SCHEDULE_ADD = "schedule_add"
-
-
-@dataclass(kw_only=True)
-class Event:
- id: str
- data: dict
-
-
-class EventsSubscriber(metaclass=Singleton):
- """
- This class is used to subscribe to and emit events to the registered listeners.
- """
-
- _instance = None
- """the instance of the EventsSubscriber"""
-
- _listeners: dict[str, list[str, Awaitable]] = {}
- """The listeners for the events."""
-
- def __init__(self):
- EventsSubscriber._instance = self
-
- @staticmethod
- def get_instance() -> "EventsSubscriber":
- """
- Get the instance of the EventsSubscriber.
-
- Returns:
- EventsSubscriber: The instance of the EventsSubscriber
-
- """
- if not EventsSubscriber._instance:
- EventsSubscriber._instance = EventsSubscriber()
- return EventsSubscriber._instance
-
- def subscribe(self, event: str | list | tuple, id: str, callback: Awaitable) -> "EventsSubscriber":
- """
- 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.
-
- Returns:
- EventsSubscriber: The instance of the EventsSubscriber
-
- """
- if isinstance(event, str):
- event = [event]
-
- for e in event:
- if e not in self._listeners:
- self._listeners[e] = {}
-
- self._listeners[e][id] = callback
-
- return self
-
- def unsubscribe(self, event: str | list | tuple, id: str) -> "EventsSubscriber":
- """
- Unsubscribe from an event.
-
- Args:
- event (str): The event to unsubscribe from.
- id (str): The id of the subscriber.
-
- Returns:
- EventsSubscriber: The instance of the EventsSubscriber
-
- """
- if isinstance(event, str):
- event = [event]
-
- for e in event:
- if e in self._listeners and id in self._listeners[e]:
- del self._listeners[e][id]
-
- return self
-
- def emit_sync(self, event: str, *args, **kwargs):
- """
- Emit an event synchronously.
-
- Args:
- event (str): The event to emit.
- *args: The arguments to pass to the event.
- **kwargs: The keyword arguments to pass to the event.
-
- Returns:
- list: The results are the return values of the coroutines. If the coroutine raises an exception,
- the exception is caught and logged. If event does not exist, an empty list is returned.
-
- """
- if 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"]
-
- results.append(asyncio.get_event_loop().run_until_complete(callback(event, data)))
- except Exception as e:
- LOG.exception(e)
- LOG.error(f"Failed to emit event '{event}' to '{id}'. Error message '{e!s}'.")
-
- return results
-
- def emit(self, event: str, *args, **kwargs):
- """
- Emit an event.
-
- Args:
- event (str): The event to emit.
- *args: The arguments to pass to the event.
- **kwargs: The keyword arguments to pass to the event.
-
- Returns:
- Awaitable: The task that was created to run the event.
-
- """
- 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})
-
- tasks.append(asyncio.create_task(callback(event, data)))
- except Exception as e:
- LOG.exception(e)
- LOG.error(f"Failed to emit event '{event}' to '{id}'. Error message '{e!s}'.")
-
- return asyncio.gather(*tasks)
diff --git a/app/library/HttpAPI.py b/app/library/HttpAPI.py
index d8454a85..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 .EventsSubscriber 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 30ca13fc..e266d308 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 .EventsSubscriber import Event, Events, EventsSubscriber
+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,13 +77,11 @@ 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)
+ self._notify.subscribe(
+ Events.ADD_URL,
+ lambda data, _, **kwargs: self.add(**data.data), # noqa: ARG005
+ f"{__class__.__name__}.socket_add_url",
+ )
# register the shutdown event.
app.on_shutdown.append(self.on_shutdown)
@@ -94,11 +89,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:
@@ -149,18 +144,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:
@@ -177,58 +172,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):
@@ -279,36 +274,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 f5d0304e..f372fbb5 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 .EventsSubscriber 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
@@ -204,13 +223,42 @@ 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}'")
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:
"""
@@ -275,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",
@@ -316,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','??')}'."
@@ -337,43 +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 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
+ return []
- if not NotificationEvents.is_valid(event):
- return False
+ if not NotificationEvents.is_valid(e.event):
+ return []
- return self.send(event, 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 d9aa29e2..39ae334e 100644
--- a/app/library/Presets.py
+++ b/app/library/Presets.py
@@ -8,9 +8,8 @@ from typing import Any
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")
@@ -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:
@@ -81,13 +79,14 @@ 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, _, **kwargs: self.add(**data.data), # noqa: ARG005
+ f"{__class__.__name__}.save",
+ )
@staticmethod
def get_instance() -> "Presets":
diff --git a/app/library/Scheduler.py b/app/library/Scheduler.py
index 215bb1d8..489fa892 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,11 @@ 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, _, **kwargs: self.add(**data.data), # noqa: ARG005
+ f"{__class__.__name__}.add",
+ )
@staticmethod
def get_instance() -> "Scheduler":
@@ -76,8 +77,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..b3ac7a33 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 .EventsSubscriber import Event, Events, EventsSubscriber
+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,11 +80,6 @@ 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)
-
@staticmethod
def get_instance() -> "Tasks":
"""
@@ -113,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."""
@@ -299,23 +297,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)
@@ -325,8 +322,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/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/app/main.py b/app/main.py
index 8262428f..a49957bc 100644
--- a/app/main.py
+++ b/app/main.py
@@ -11,8 +11,7 @@ import magic
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
@@ -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(
- [EventsSubscriber().emit], local=True
- )
-
self._app.on_cleanup.append(_close_connection)
def _check_folders(self):
@@ -94,7 +89,7 @@ class Main:
"""
Start the application.
"""
- EventsSubscriber.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)
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 @@