From 4ed7c26bb45ebc07a306ba8cb596dde4c6885af6 Mon Sep 17 00:00:00 2001
From: ArabCoders <admin@arabcoders.org>
Date: Sat, 22 Mar 2025 02:05:52 +0300
Subject: [PATCH] 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 @@
               </div>
               <div class="column is-half-mobile has-text-centered is-text-overflow">
                 <span class="icon-text">
-                  <span class="icon" :class="{ 'has-text-success': item.status == 'downloading' }">
-                    <i class="fas" :class="setIcon(item)" />
+                  <span class="icon" :class="setIconColor(item)">
+                    <i class="fas fa-solid" :class="setIcon(item)" />
                   </span>
-                  <span v-if="item.status == 'downloading' && item.is_live">Live Streaming</span>
-                  <span v-else>{{ ucFirst(item.status) }}</span>
+                  <span v-text="setStatus(item)" />
                 </span>
               </div>
               <div class="column is-half-mobile has-text-centered is-text-overflow">
@@ -153,7 +152,7 @@
 
 <script setup>
 import moment from 'moment'
-import { useStorage } from '@vueuse/core'
+import { set, useStorage } from '@vueuse/core'
 import { ucFirst } from '~/utils/index'
 import { isEmbedable, getEmbedable } from '~/utils/embedable'
 
@@ -184,24 +183,62 @@ const hasQueuedItems = computed(() => stateStore.count('queue') > 0)
 
 const setIcon = item => {
   if ('downloading' === item.status && item.is_live) {
-    return 'fa-solid fa-globe';
+    return 'fa-globe fa-spin';
   }
 
   if ('downloading' === item.status) {
-    return 'fa-solid fa-circle-check';
+    return 'fa-download';
   }
 
   if ('postprocessing' === item.status) {
-    return 'fa-solid fa-cog fa-spin';
+    return 'fa-cog fa-spin';
   }
 
   if (null === item.status && true === config.paused) {
-    return 'fa-solid fa-pause-circle';
+    return 'fa-pause-circle';
   }
 
-  return 'fa-solid fa-spinner fa-spin';
+  if (!item.status) {
+    return 'fa-question';
+  }
+
+  return 'fa-spinner fa-spin';
 }
 
+const setStatus = item => {
+  if (null === item.status && true === config.paused) {
+    return 'Paused';
+  }
+
+  if ('downloading' === item.status && item.is_live) {
+    return 'Live Streaming';
+  }
+
+  if (!item.status) {
+    return 'Unknown..';
+  }
+
+  return ucFirst(item.status)
+}
+
+const setIconColor = item => {
+  if (item.status === 'downloading') {
+    return 'has-text-success'
+  }
+
+  if ('postprocessing' === item.status) {
+    return 'has-text-info'
+  }
+
+  if (null === item.status && true === config.paused) {
+    return 'has-text-warning'
+  }
+
+  return ''
+}
+
+
+
 const ETAPipe = value => {
   if (value === null || 0 === value) {
     return 'Live';
diff --git a/ui/pages/console.vue b/ui/pages/console.vue
index 05b3df2c..0bb9cbd5 100644
--- a/ui/pages/console.vue
+++ b/ui/pages/console.vue
@@ -103,6 +103,14 @@ const runCommand = async () => {
     return
   }
 
+  if (command.value.startsWith('yt-dlp')) {
+    command.value = command.value.replace(/^yt-dlp/, '').trim()
+    await nextTick()
+    if ('' === command.value) {
+      return
+    }
+  }
+
   if (!terminal.value) {
     terminal.value = new Terminal({
       fontSize: 14,
diff --git a/ui/pages/notifications.vue b/ui/pages/notifications.vue
index 5bb6574d..559089ef 100644
--- a/ui/pages/notifications.vue
+++ b/ui/pages/notifications.vue
@@ -124,6 +124,12 @@ div.is-centered {
             However this might not be enough to remove credentials from the exported data. it's your responsibility
             to ensure that the exported data does not contain any sensitive information for sharing.
           </li>
+          <li>
+            When you set the request type as <code>Form</code>, the event data will be JSON encoded and sent as the
+            and sent as <code>...&data=json_string</code>, only the <code>data</code> field will be JSON encoded. The
+            other keys <code>id</code>, <code>event</code> and <code>created_at</code> will be sent as they are.
+          </li>
+          <li>We also send two special headers <code>X-Event-ID</code> and <code>X-Event</code> with the request.</li>
         </ul>
       </Message>
     </div>
diff --git a/ui/stores/SocketStore.js b/ui/stores/SocketStore.js
index bcde4715..dfba3d5e 100644
--- a/ui/stores/SocketStore.js
+++ b/ui/stores/SocketStore.js
@@ -70,7 +70,7 @@ export const useSocketStore = defineStore('socket', () => {
         return
       }
 
-      toast.info(`Download cancelled: ${ag(stateStore.get('queue', id, {}), id)}`);
+      toast.warning(`Download cancelled: ${ag(stateStore.get('queue', id, {}), 'title')}`);
 
       if (true === stateStore.has('queue', id)) {
         stateStore.remove('queue', id);