diff --git a/app/library/BackgroundWorker.py b/app/library/BackgroundWorker.py
new file mode 100644
index 00000000..108d78a7
--- /dev/null
+++ b/app/library/BackgroundWorker.py
@@ -0,0 +1,61 @@
+import asyncio
+import inspect
+import logging
+import threading
+from queue import Empty, Queue
+
+from .Singleton import Singleton
+
+LOG = logging.getLogger(__name__)
+
+
+class BackgroundWorker(metaclass=Singleton):
+ _instance = None
+ """The instance of the Notification class."""
+
+ def __init__(self):
+ self.queue = Queue()
+ self.running = True
+ self.thread = threading.Thread(target=self._run, daemon=True)
+ self.thread.start()
+
+ @staticmethod
+ def get_instance() -> "BackgroundWorker":
+ if BackgroundWorker._instance is None:
+ BackgroundWorker._instance = BackgroundWorker()
+
+ return BackgroundWorker._instance
+
+ def _run(self):
+ asyncio.set_event_loop(asyncio.new_event_loop())
+ loop = asyncio.get_event_loop()
+
+ # Start loop in a background thread
+ def _loop_runner():
+ try:
+ loop.run_forever()
+ except Exception as e:
+ LOG.exception("Loop error: %s", e)
+
+ threading.Thread(target=_loop_runner, daemon=True).start()
+
+ while self.running:
+ try:
+ fn, args, kwargs = self.queue.get(timeout=1)
+ try:
+ LOG.debug("Running background task: %s", fn.__name__)
+ result = fn(*args, **kwargs)
+ if inspect.iscoroutine(result):
+ loop.call_soon_threadsafe(loop.create_task, result)
+ except Exception as e:
+ LOG.exception(e)
+ LOG.error(f"Error in background task: {fn.__name__}")
+ except Empty:
+ continue
+
+ def submit(self, fn, *args, **kwargs):
+ self.queue.put((fn, args, kwargs))
+
+ def shutdown(self):
+ self.running = False
+ self.thread.join()
diff --git a/app/library/Download.py b/app/library/Download.py
index 0cae6fbe..5e568abb 100644
--- a/app/library/Download.py
+++ b/app/library/Download.py
@@ -491,7 +491,12 @@ class Download:
if "error" == self.info.status and "error" in status:
self.info.error = status.get("error")
- await self._notify.emit(Events.ERROR, data={"message": self.info.error, "data": self.info})
+ await self._notify.emit(
+ Events.ERROR,
+ data={"message": self.info.error, "data": self.info},
+ title="Download Error",
+ message=f"Error in download task '{self.info.title}': {self.info.error}",
+ )
if "downloaded_bytes" in status and status.get("downloaded_bytes") > 0:
self.info.downloaded_bytes = status.get("downloaded_bytes")
diff --git a/app/library/DownloadQueue.py b/app/library/DownloadQueue.py
index c89ed0db..559b08e8 100644
--- a/app/library/DownloadQueue.py
+++ b/app/library/DownloadQueue.py
@@ -461,7 +461,12 @@ class DownloadQueue(metaclass=Singleton):
NotifyEvent = Events.COMPLETED
dlInfo.info.status = "not_live"
dlInfo.info.msg = f"{'Premiere video' if is_premiere else 'Stream' } is not available yet." + text_logs
- await self._notify.emit(Events.LOG_INFO, data=event_info(dlInfo.info.msg, {"lowPriority": True}))
+ await self._notify.emit(
+ Events.LOG_INFO,
+ data=event_info(dlInfo.info.msg, {"lowPriority": True}),
+ title="Premiere video" if is_premiere else "Live Stream",
+ message=f"Item '{dlInfo.info.title}' is not available yet. {dlInfo.info.msg}",
+ )
itemDownload: Download = self.done.put(dlInfo)
elif len(entry.get("formats", [])) < 1:
availability: str = entry.get("availability", "public")
@@ -473,7 +478,12 @@ class DownloadQueue(metaclass=Singleton):
dlInfo.info.status = "error"
itemDownload = self.done.put(dlInfo)
NotifyEvent = Events.COMPLETED
- await self._notify.emit(Events.LOG_WARNING, data=event_warning(f"No formats found for '{dl.title}'."))
+ await self._notify.emit(
+ Events.LOG_WARNING,
+ data=event_warning(f"No formats found for '{dl.title}'."),
+ title="No Formats Found",
+ message=f"No formats found for '{dl.title}'. {dlInfo.info.error}",
+ )
elif is_premiere and self.config.prevent_live_premiere:
dlInfo.info.error = "Premiering right now."
@@ -507,6 +517,8 @@ class DownloadQueue(metaclass=Singleton):
await self._notify.emit(
Events.LOG_INFO,
data=event_info(f"'{dl.title}' is {dlInfo.info.error}.", {"lowPriority": True}),
+ title="Item Not Live",
+ message=f"Item '{dl.title}' is not live. {dlInfo.info.error}",
)
else:
NotifyEvent = Events.ADDED
@@ -516,7 +528,12 @@ class DownloadQueue(metaclass=Singleton):
else:
LOG.debug(f"Item {itemDownload.info.name()} is not set to auto-start.")
- await self._notify.emit(NotifyEvent, data=itemDownload.info.serialize())
+ await self._notify.emit(
+ NotifyEvent,
+ data=itemDownload.info.serialize(),
+ title="Item Added" if Events.ADDED == NotifyEvent else None,
+ message=f"Added '{itemDownload.info.title}'." if NotifyEvent == Events.ADDED else None,
+ )
return {"status": "ok"}
except Exception as e:
@@ -721,7 +738,12 @@ 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, data=item.info.serialize())
+ await self._notify.emit(
+ Events.CANCELLED,
+ data=item.info.serialize(),
+ title="Download Cancelled",
+ message=f"Download '{item.info.title}' has been cancelled.",
+ )
item.info.status = "cancelled"
# item.info.error = "Cancelled by user."
self.done.put(item)
@@ -793,7 +815,12 @@ class DownloadQueue(metaclass=Singleton):
LOG.error(f"Unable to remove '{itemRef}' local file '{filename}'. {e!s}")
self.done.delete(id)
- await self._notify.emit(Events.CLEARED, data=item.info.serialize())
+ await self._notify.emit(
+ Events.CLEARED,
+ data=item.info.serialize(),
+ title="Download Cleared",
+ message=f"Cleared download '{item.info.title}' from history.",
+ )
msg = f"Deleted completed download '{itemRef}'."
if removed_files > 0:
@@ -905,12 +932,24 @@ class DownloadQueue(metaclass=Singleton):
self.queue.delete(key=id)
if entry.is_cancelled() is True:
- await self._notify.emit(Events.CANCELLED, data=entry.info.serialize())
+ await self._notify.emit(
+ Events.CANCELLED,
+ data=entry.info.serialize(),
+ title="Download Cancelled",
+ message=f"Download '{entry.info.title}' has been cancelled.",
+ )
entry.info.status = "cancelled"
# entry.info.error = "Cancelled by user."
self.done.put(value=entry)
- await self._notify.emit(Events.COMPLETED, data=entry.info.serialize())
+ await self._notify.emit(
+ Events.COMPLETED,
+ data=entry.info.serialize(),
+ title="Download Completed" if entry.info.status == "finished" else None,
+ message=f"Download '{entry.info.title}' has been completed."
+ if entry.info.status == "finished"
+ else None,
+ )
else:
LOG.warning(f"Download '{id}' not found in queue.")
diff --git a/app/library/Events.py b/app/library/Events.py
index 4025c156..cbeb62f9 100644
--- a/app/library/Events.py
+++ b/app/library/Events.py
@@ -203,6 +203,12 @@ class Event:
event: str
"""The event that was emitted."""
+ title: str | None = None
+ """The title of the event, if any."""
+
+ message: str | None = None
+ """The message of the event, if any."""
+
data: any
"""The data that was passed to the event."""
@@ -214,10 +220,17 @@ class Event:
dict: The serialized event.
"""
- return {"id": self.id, "created_at": self.created_at, "event": self.event, "data": self.data}
+ return {
+ "id": self.id,
+ "created_at": self.created_at,
+ "event": self.event,
+ "title": self.title,
+ "message": self.message,
+ "data": self.data,
+ }
def __repr__(self):
- return f"Event(id={self.id}, created_at={self.created_at}, event={self.event}, data={self.data})"
+ return f"Event(id={self.id}, created_at={self.created_at}, event={self.event}, title={self.title}, message={self.message} data={self.data})"
def datatype(self) -> str:
"""
@@ -230,7 +243,7 @@ class Event:
return type(self.data).__name__
def __str__(self):
- return f"Event(id={self.id}, created_at={self.created_at}, event={self.event})"
+ return f"Event(id={self.id}, created_at={self.created_at}, event={self.event}, title={self.title}, message={self.message})"
class EventListener:
@@ -414,13 +427,17 @@ class EventBus(metaclass=Singleton):
fut = asyncio.run_coroutine_threadsafe(emit_all(), loop)
return fut.result() if wait else fut
- async def emit(self, event: str, data: Any, **kwargs) -> Awaitable:
+ async def emit(
+ self, event: str, data: Any, title: str | None = None, message: str | None = None, **kwargs
+ ) -> Awaitable:
"""
Emit an event.
Args:
event (str): The event to emit.
data (Any): The data to pass to the event.
+ title (str | None): The title of the event, if any.
+ message (str | None): The message of the event, if any.
**kwargs: The keyword arguments to pass to the event.
Returns:
@@ -430,7 +447,7 @@ class EventBus(metaclass=Singleton):
if event not in self._listeners:
return []
- ev = Event(event=event, data=data)
+ ev = Event(event=event, data=data, title=title, message=message)
if self.debug or event not in Events.only_debug():
LOG.debug(f"Emitting event '{ev.id}: {ev.event}'.", extra={"data": data})
diff --git a/app/library/Notifications.py b/app/library/Notifications.py
index 39a9578b..295daa85 100644
--- a/app/library/Notifications.py
+++ b/app/library/Notifications.py
@@ -11,6 +11,7 @@ import httpx
from aiohttp import web
from .ag_utils import ag
+from .BackgroundWorker import BackgroundWorker
from .config import Config
from .encoder import Encoder
from .Events import Event, EventBus, Events
@@ -129,6 +130,7 @@ class Notification(metaclass=Singleton):
client: httpx.AsyncClient | None = None,
encoder: Encoder | None = None,
config: Config | None = None,
+ background_worker: BackgroundWorker | None = None,
):
Notification._instance = self
config: Config = config or Config.get_instance()
@@ -138,6 +140,7 @@ class Notification(metaclass=Singleton):
self._client: httpx.AsyncClient = client or httpx.AsyncClient()
self._encoder: Encoder = encoder or Encoder()
self._version = config.app_version
+ self._offload: BackgroundWorker = background_worker or BackgroundWorker.get_instance()
if self._file.exists() and "600" != self._file.stat().st_mode:
try:
@@ -335,17 +338,54 @@ class Notification(metaclass=Singleton):
tasks = []
+ apprise_targets: list[Target] = []
+
for target in self._targets:
if len(target.on) > 0 and ev.event not in target.on and "test" != ev.event:
continue
- tasks.append(self._send(target, ev))
+ if not target.request.url.startswith("http"):
+ apprise_targets.append(target)
+ else:
+ tasks.append(self._send(target, ev))
+
+ if len(apprise_targets) > 0:
+ tasks.append(self._apprise(apprise_targets, ev))
if wait:
return await asyncio.gather(*tasks)
return tasks
+ async def _apprise(self, target: list[Target], ev: Event) -> dict:
+ if not target or not isinstance(target, list):
+ return {}
+
+ import apprise
+
+ try:
+ notify = apprise.Apprise()
+ apr_config = Path(Config.get_instance().apprise_config)
+ if apr_config.exists():
+ apprise_config = notify.AppriseConfig()
+ apprise_config.add(apr_config)
+ notify.add(apprise_config)
+
+ for t in target:
+ notify.add(t.request.url)
+
+ notify.notify(
+ body=ev.message or json.dumps(ev.serialize(), sort_keys=False, ensure_ascii=False),
+ title=ev.title or f"YTPTube Event: {ev.event}",
+ notify_type=ev.event,
+ )
+ except Exception as e:
+ LOG.exception(e)
+ LOG.error(f"Error sending Apprise notification: {e!s}")
+ return {"error": str(e), "event": ev.event, "id": ev.id, "targets": [t.name for t in target]}
+
+ return {}
+
async def _send(self, target: Target, ev: Event) -> dict:
try:
LOG.info(f"Sending Notification event '{ev.event}: {ev.id}' to '{target.name}'.")
@@ -397,11 +437,13 @@ class Notification(metaclass=Singleton):
LOG.error(f"Error sending Notification event '{ev.event}: {ev.id}' to '{target.name}'. '{err_msg!s}'.")
return {"url": target.request.url, "status": 500, "text": str(ev)}
- def emit(self, e: Event, _, **kwargs): # noqa: ARG002
+ def emit(self, e: Event, _, **__):
if len(self._targets) < 1 or not NotificationEvents.is_valid(e.event):
- return asyncio.sleep(0)
+ return self.noop()
- return self.send(e)
+ self._offload.submit(self.send, e)
+
+ return self.noop()
def _deep_unpack(self, data: dict) -> dict:
for k, v in data.items():
@@ -415,3 +457,6 @@ class Notification(metaclass=Singleton):
data[k] = v.serialize()
return data
+
+ async def noop(self) -> None:
+ return None
diff --git a/app/library/Tasks.py b/app/library/Tasks.py
index 25e7dedd..896ec42f 100644
--- a/app/library/Tasks.py
+++ b/app/library/Tasks.py
@@ -325,6 +325,8 @@ class Tasks(metaclass=Singleton):
"template": template,
"cli": cli,
},
+ title=f"Task '{task.name}' started",
+ message=f"Task '{task.name}' started at '{timeNow}'",
id=task.id,
)
@@ -338,11 +340,17 @@ class Tasks(metaclass=Singleton):
data=success(
f"Task '{task.name}' completed in '{ended - started:.2f}' seconds.", data={"lowPriority": True}
),
+ title=f"Task '{task.name}' completed",
+ message=f"Task '{task.name}' completed at '{timeNow}'",
+ id=task.id,
)
except Exception as e:
LOG.error(f"Failed to execute '{task.name}' at '{timeNow}'. '{e!s}'.")
await self._notify.emit(
- Events.ERROR, data=error(f"Failed to execute '{task.name}' at '{timeNow}'. '{e!s}'.")
+ Events.ERROR,
+ data=error(f"Failed to execute '{task.name}' at '{timeNow}'. '{e!s}'."),
+ title=f"Task '{task.name}' failed",
+ message=str(e),
)
diff --git a/app/library/Utils.py b/app/library/Utils.py
index 8f9a8cdc..d592ea04 100644
--- a/app/library/Utils.py
+++ b/app/library/Utils.py
@@ -164,7 +164,7 @@ def extract_info(
archive_id = f".{idDict['id']}" if idDict.get("id") else None
log_wrapper.add_target(
- target=logging.getLogger(f"yt-dlp{archive_id}"),
+ target=logging.getLogger(f"yt-dlp{archive_id if archive_id else '.extract_info'}"),
level=logging.DEBUG if debug else logging.WARNING,
)
diff --git a/app/library/config.py b/app/library/config.py
index c71047e4..57407530 100644
--- a/app/library/config.py
+++ b/app/library/config.py
@@ -103,6 +103,9 @@ class Config:
manual_archive: str = "{config_path}/archive.manual.log"
"""The path to the manual archive file."""
+ apprise_config: str = "{config_path}/apprise.yml"
+ """The path to the Apprise configuration file."""
+
ui_update_title: bool = True
"""Update the title of the browser tab with the current status."""
@@ -454,8 +457,9 @@ class Config:
self.started = time.time()
- logging.getLogger("httpx").setLevel(logging.WARNING)
logging.getLogger("httpcore").setLevel(logging.INFO)
+ for _tool in ("httpx", "urllib3.connectionpool", "apprise"):
+ logging.getLogger(_tool).setLevel(logging.WARNING)
# check env
if self.app_env not in ("production", "development"):
diff --git a/app/main.py b/app/main.py
index 702082ec..0999d9cf 100644
--- a/app/main.py
+++ b/app/main.py
@@ -16,6 +16,7 @@ import caribou
import magic
from aiohttp import web
+from app.library.BackgroundWorker import BackgroundWorker
from app.library.conditions import Conditions
from app.library.config import Config
from app.library.DownloadQueue import DownloadQueue
@@ -39,8 +40,10 @@ class Main:
self._config = Config.get_instance(is_native=is_native)
self._app = web.Application()
self._app.on_shutdown.append(self.on_shutdown)
+ self._background_worker = BackgroundWorker()
Services.get_instance().add("app", self._app)
+ Services.get_instance().add("background_worker", self._background_worker)
self._check_folders()
@@ -94,7 +97,12 @@ class Main:
raise
async def on_shutdown(self, _: web.Application):
- await EventBus.get_instance().emit(Events.SHUTDOWN, data={"app": self._app})
+ await EventBus.get_instance().emit(
+ Events.SHUTDOWN,
+ data={"app": self._app},
+ title="Application Shutdown",
+ message="The application is shutting down.",
+ )
def start(self, host: str | None = None, port: int | None = None, cb=None):
"""
diff --git a/app/routes/api/notifications.py b/app/routes/api/notifications.py
index 9d912002..81b975ee 100644
--- a/app/routes/api/notifications.py
+++ b/app/routes/api/notifications.py
@@ -106,6 +106,6 @@ async def notification_test(encoder: Encoder, notify: EventBus) -> Response:
"""
data = message("test", "This is a test notification.")
- await notify.emit(Events.TEST, data=data)
+ await notify.emit(Events.TEST, data=data, title="Test Notification", message="This is a test notification.")
return web.json_response(data=data, status=web.HTTPOk.status_code, dumps=encoder.encode)
diff --git a/app/routes/socket/history.py b/app/routes/socket/history.py
index 1868a0f8..b969e5b7 100644
--- a/app/routes/socket/history.py
+++ b/app/routes/socket/history.py
@@ -19,13 +19,23 @@ LOG: logging.Logger = logging.getLogger(__name__)
@route(RouteType.SOCKET, "pause", "pause_downloads")
async def pause(notify: EventBus, queue: DownloadQueue):
queue.pause()
- await notify.emit(Events.PAUSED, data={"paused": True, "at": time.time()})
+ await notify.emit(
+ Events.PAUSED,
+ data={"paused": True, "at": time.time()},
+ title="Downloads Paused",
+ message="Download pool has been paused.",
+ )
@route(RouteType.SOCKET, "resume", "resume_downloads")
async def resume(notify: EventBus, queue: DownloadQueue):
queue.resume()
- await notify.emit(Events.PAUSED, data={"paused": False, "at": time.time()})
+ await notify.emit(
+ Events.PAUSED,
+ data={"paused": False, "at": time.time()},
+ title="Downloads Resumed",
+ message="Download pool has been resumed.",
+ )
@route(RouteType.SOCKET, "add_url", "add_url")
@@ -76,7 +86,9 @@ async def item_delete(queue: DownloadQueue, notify: EventBus, sid: str, data: di
status = await queue.clear([id], remove_file=bool(data.get("remove_file", False)))
status.update({"identifier": id})
- await notify.emit(Events.ITEM_DELETE, data=status)
+ await notify.emit(
+ Events.ITEM_DELETE, data=status, title="Item Deleted", message=f"Item with ID '{id}' has been deleted."
+ )
@route(RouteType.SOCKET, "archive_item", "archive_item")
diff --git a/pyproject.toml b/pyproject.toml
index 52421121..48a2205c 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -37,6 +37,7 @@ dependencies = [
"dateparser>=1.2.1",
"defusedxml>=0.7.1",
"zipstream-ng>=1.8.0",
+ "apprise>=1.9.3",
]
[tool.ruff]
diff --git a/ui/components/NotificationForm.vue b/ui/components/NotificationForm.vue
index 15599088..e4d8da7c 100644
--- a/ui/components/NotificationForm.vue
+++ b/ui/components/NotificationForm.vue
@@ -83,12 +83,13 @@
- The URL to send the notification to.
+ The URL to send the notification to. It can be regular http/https endpoint.
+ or