danvics/ytptube
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1

migrated all our emitter usage to EventBus

Browse source
This commit is contained in:
ArabCoders 2025-03-22 02:05:52 +03:00
parent 07fc86565a
commit 4ed7c26bb4
13 changed files with 128 additions and 57 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
app/library/DownloadQueue.py
View file

@ -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)

28
app/library/Events.py
View file

@ -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

4
app/library/HttpSocket.py
View file

@ -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.

56
app/library/Notifications.py
View file

@ -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

4
app/library/Presets.py
View file

@ -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

4
app/library/Scheduler.py
View file

@ -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

7
app/library/Tasks.py
View file

@ -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."""

5
app/library/encoder.py
View file

@ -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()

2
ui/components/History.vue
View file

@ -411,7 +411,7 @@ const setIcon = item => {
}
const setIconColor = item => {
if (item.status === 'finished') {
if ('finished' === item.status) {
return 'has-text-success'
}

57
ui/components/Queue.vue
View file

@ -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';

8
ui/pages/console.vue
View file

@ -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,

6
ui/pages/notifications.vue
View file

@ -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>

2
ui/stores/SocketStore.js
View file

@ -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);