From ddc216ccb668b42755f126d5ffcf07f834879574 Mon Sep 17 00:00:00 2001 From: ArabCoders Date: Sat, 25 Jan 2025 22:24:13 +0300 Subject: [PATCH] revamped the notification system --- README.md | 42 +--- app/library/HttpAPI.py | 69 ++++++- app/library/Notifications.py | 310 +++++++++++++++++++++++++++++ app/library/Singleton.py | 32 +++ app/library/Utils.py | 17 ++ app/library/Webhooks.py | 114 ----------- app/library/cache.py | 25 +-- app/main.py | 6 +- ui/components/NotificationForm.vue | 268 +++++++++++++++++++++++++ ui/components/TaskForm.vue | 2 +- ui/layouts/default.vue | 15 +- ui/pages/notifications.vue | 300 ++++++++++++++++++++++++++++ ui/pages/tasks.vue | 11 +- 13 files changed, 1024 insertions(+), 187 deletions(-) create mode 100644 app/library/Notifications.py create mode 100644 app/library/Singleton.py delete mode 100644 app/library/Webhooks.py create mode 100644 ui/components/NotificationForm.vue create mode 100644 ui/pages/notifications.vue diff --git a/README.md b/README.md index c315e925..3a3c1101 100644 --- a/README.md +++ b/README.md @@ -10,13 +10,13 @@ YTPTube started as a fork of [meTube](https://github.com/alexta69/metube), Since * Multi-downloads support. * Handle live streams. * Schedule Channels or Playlists to be downloaded automatically at a specific time. +* Send notification to targets based on specified events. * Support per link `yt-dlp JSON config or cli options`, `cookies` & `output format`. * Queue multiple URLs separated by comma. * A built in video player that can play any video file regardless of the format. **With support for sidecar external subtitles**. * New `/api/add_batch` endpoint that allow multiple links to be sent. * Completely redesigned the frontend UI. * Switched out of binary file storage in favor of SQLite. -* Webhook sender. It allow you to add webhook endpoints that receive events related to downloads using simple `json` file. * Basic Authentication support. * Support for curl_cffi, see [yt-dlp documentation](https://github.com/yt-dlp/yt-dlp?tab=readme-ov-file#impersonation) * Support for both advanced and basic mode for WebUI. @@ -235,34 +235,6 @@ The `config/ytdlp.json`, is a json file which can be used to alter the default ` } ``` -### webhooks.json File - -The `config/webhooks.json`, is a json file, which can be used to add webhook endpoints that would receive events related to the downloads. - -```json5 -[ - { - // (name: string) - REQUIRED - The webhook name. - "name": "my very smart webhook receiver", - // (on: array) - OPTIONAL - List of accepted events, if left empty it will send all events. - // Allowed events ["added", "completed", "error", "not_live" ] you can choose one or all of them. - "on": [ "added", "completed", "error", "not_live" ], - "request": { - // (url: string) - REQUIRED- The webhook url - "url": "https://mysecert.webhook.com/endpoint", - // (type: string) - OPTIONAL - The request type, it can be json or form. - "type": "json", - // (method: string) - OPTIONAL - The request method, it can be POST or PUT - "method": "POST", - // (headers: dictionary) - OPTIONAL - Extra headers to include. - "headers": { - "Authorization": "Bearer my_secret_token" - } - } - ... -] -``` - ### presets.json File The `config/presets.json`, is a json file, which can be used to add custom presets for selection in WebUI. @@ -312,20 +284,20 @@ What does the basic mode do? it hides the the following features from the WebUI. ### Header -It disables the `Check cookies`, `Console`, `Tasks` and `Add` buttons. +It disables everything except the `theme switcher` and `reload` button. ### Add form -Disables everything except the `URL` and `Add` button. the default preset `YTP_DEFAULT_PRESET` will be used. The folder will be -the root download path `YTP_DOWNLOAD_PATH`. - -The add form will always be visible and un-collapsible. +* The form will always be visible and un-collapsible. +* Everything except the `URL` and `Add` button will be disabled and hidden. +* The preset will be the default preset, which can be specified via `YTP_DEFAULT_PRESET` environment variable. +* The output template will be the default template which can be specified via `YTP_OUTPUT_TEMPLATE` environment variable. +* The download path will be the default download path which can be specified via `YTP_DOWNLOAD_PATH` environment variable. ### Queue & History Disables the `Information` button. - # Social contact If you have short or quick questions, you are free to join my [discord server](https://discord.gg/G3GpVR8xpb) and ask diff --git a/app/library/HttpAPI.py b/app/library/HttpAPI.py index 863614cd..1c95d483 100644 --- a/app/library/HttpAPI.py +++ b/app/library/HttpAPI.py @@ -26,7 +26,8 @@ from .M3u8 import M3u8 from .Playlist import Playlist from .Segments import Segments from .Subtitle import Subtitle -from .Utils import StreamingError, arg_converter, calcDownloadPath, getVideoInfo, validate_url +from .Utils import StreamingError, arg_converter, calcDownloadPath, getVideoInfo, validate_url, validateUUID +from .Notifications import Notification LOG = logging.getLogger("http_api") MIME = magic.Magic(mime=True) @@ -830,3 +831,69 @@ class HttpAPI(common): data={"message": f"Failed to request website. {str(e)}"}, status=web.HTTPInternalServerError.status_code, ) + + @route("GET", "api/notifications") + async def notifications(self, _: Request) -> Response: + data = { + "notifications": Notification.get_instance().getTargets(), + "allowedTypes": Notification.allowed_events, + } + + return web.json_response(data=data, status=web.HTTPOk.status_code, dumps=self.encoder.encode) + + @route("POST", "api/notifications/test") + async def test_notifications(self, _: Request) -> Response: + data = {"type": "test", "message": "This is a test notification."} + await self.emitter.emit("test", data) + + return web.json_response(data=data, status=web.HTTPOk.status_code, dumps=self.encoder.encode) + + @route("PUT", "api/notifications") + async def add_notifications(self, request: Request) -> Response: + post = await request.json() + if not isinstance(post, list): + return web.json_response( + {"error": "Invalid request body expecting list with dicts."}, + status=web.HTTPBadRequest.status_code, + ) + + targets: list = [] + + for item in post: + if not isinstance(item, dict): + return web.json_response( + {"error": "Invalid request body expecting list with dicts."}, + status=web.HTTPBadRequest.status_code, + ) + + if not item.get("id", None) or validateUUID(item.get("id"), version=4): + item["id"] = str(uuid.uuid4()) + + try: + Notification.validate(item) + except ValueError as e: + return web.json_response( + {"error": f"Invalid notification target settings. {str(e)}", "data": item}, + status=web.HTTPBadRequest.status_code, + ) + + targets.append(item) + + ins = Notification.get_instance() + + try: + if len(targets) < 1: + ins.clearTargets() + + ins.save(targets=targets) + ins.load() + except Exception as e: + LOG.exception(e) + return web.json_response({"error": "Failed to save tasks."}, status=web.HTTPInternalServerError.status_code) + + data = { + "notifications": targets, + "allowedTypes": Notification.allowed_events, + } + + return web.json_response(data=data, status=web.HTTPOk.status_code, dumps=self.encoder.encode) diff --git a/app/library/Notifications.py b/app/library/Notifications.py new file mode 100644 index 00000000..d5619f7e --- /dev/null +++ b/app/library/Notifications.py @@ -0,0 +1,310 @@ +import asyncio +from datetime import datetime, timezone +import json +import logging +import os +from typing import List, TypedDict +import uuid + +import httpx + +from .config import Config +from .ItemDTO import ItemDTO +from .Singleton import Singleton +from .Utils import ag, validateUUID +from .version import APP_VERSION +from .encoder import Encoder + +LOG = logging.getLogger("notifications") + + +class targetRequestHeader(TypedDict): + """Request header details for a notification target.""" + + key: str + value: str + + +class targetRequest(TypedDict): + """Request details for a notification target.""" + + type: str + method: str + url: str + headers: list[targetRequestHeader] = [] + + +class Target(TypedDict): + """Notification target details.""" + + id: str + name: str + on: List[str] + request: targetRequest + + +class Notification(metaclass=Singleton): + targets: list[Target] = [] + """Notification targets to send events to.""" + + allowed_events: tuple = ( + "added", + "completed", + "error", + "not_live", + "canceled", + "cleared", + "log_info", + "log_success", + ) + """Events that can be sent to the targets.""" + + _instance = None + + def __init__(self): + Notification._instance = self + + config = Config.get_instance() + + self._debug = config.debug + self.file: str = os.path.join(config.config_path, "notifications.json") + self._client: httpx.AsyncClient = httpx.AsyncClient() + self._encoder: Encoder = Encoder() + + if os.path.exists(self.file) and os.path.getsize(self.file) > 10: + self.load() + + @staticmethod + def get_instance() -> "Notification": + if Notification._instance is None: + Notification._instance = Notification() + + return Notification._instance + + def getTargets(self) -> list[Target]: + """Get the list of notification targets.""" + return self.targets + + def clearTargets(self) -> "Notification": + """Clear the list of notification targets.""" + self.targets.clear() + return self + + def save(self, targets: list[Target] | None = None) -> "Notification": + """ + Save notification targets to the file. + + Args: + targets (list[Target]|None): The list of targets to save. + + Returns: + Notification: The Notification instance. + """ + LOG.info(f"Saving notification targets to '{self.file}'.") + try: + with open(self.file, "w") as f: + f.write(json.dumps(targets or self.targets, indent=4)) + except Exception as e: + LOG.error(f"Error saving notification targets to '{self.file}'. '{e}'") + pass + + return self + + def load(self): + """Load or reload notification targets from the file.""" + if len(self.targets) > 0: + LOG.info("Clearing existing notification targets.") + self.targets.clear() + + modified = False + + targets = [] + + if not os.path.exists(self.file) or os.path.getsize(self.file) < 10: + return + + LOG.info(f"Loading notification targets from '{self.file}'.") + + try: + with open(self.file, "r") as f: + targets = json.load(f) + except Exception as e: + LOG.error(f"Error loading notification targets from '{self.file}'. '{e}'") + pass + + for target in targets: + try: + try: + Notification.validate(target) + except ValueError as e: + LOG.error(f"Invalid notification target '{target}'. '{e}'") + continue + + if "on" not in target: + target["on"] = [] + modified = True + + if "type" not in target["request"]: + target["request"]["type"] = "json" + modified = True + + if "method" not in target["request"]: + target["request"]["method"] = "POST" + modified = True + + if "id" not in target or validateUUID(target["id"], version=4) is False: + target["id"] = str(uuid.uuid4()) + modified = True + + target = 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=h.get("key"), value=h.get("value")) + for h in target.get("request", {}).get("headers", []) + ], + ), + ) + + 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']}'." + ) + except Exception as e: + LOG.error(f"Error loading notification target '{target}'. '{e}'") + pass + + if modified: + self.save() + + @staticmethod + def validate(target: Target | dict) -> bool: + """ + Validate a notification target. + + Args: + target (Target|dict): The target to validate. + + Returns: + bool: True if the target is valid, False otherwise. + """ + + if "id" not in target or validateUUID(target["id"], version=4) is False: + raise ValueError("Invalid notification target. No ID found.") + + if "name" not in target: + raise ValueError("Invalid notification target. No name found.") + + if "request" not in target: + raise ValueError("Invalid notification target. No request details found.") + + if "url" not in target["request"]: + raise ValueError("Invalid notification target. No URL found.") + + if "method" in target["request"] and target["request"]["method"].upper() not in ["POST", "PUT"]: + raise ValueError("Invalid notification target. Invalid method found.") + + if "type" in target["request"] and target["request"]["type"].lower() not in ["json", "form"]: + raise ValueError("Invalid notification target. Invalid type found.") + + if "on" in target: + if not isinstance(target["on"], list): + raise ValueError("Invalid notification target. Invalid 'on' event list found.") + + for e in target["on"]: + if e not in Notification.allowed_events: + raise ValueError(f"Invalid notification target. Invalid event '{e}' found.") + + if "headers" in target["request"]: + if not isinstance(target["request"]["headers"], list): + raise ValueError("Invalid notification target. Invalid headers list found.") + + for h in target["request"]["headers"]: + if "key" not in h: + raise ValueError("Invalid notification target. No header key found.") + if "value" not in h: + raise ValueError("Invalid notification target. No header value found.") + + return True + + async def send(self, event: str, item: ItemDTO | dict) -> 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}'.") + return [] + + tasks = [] + + for target in self.targets: + if len(target["on"]) > 0 and event not in target["on"] and "test" != event: + continue + + tasks.append(self._send(event, target, item)) + + return await asyncio.gather(*tasks) + + async def _send(self, event: str, target: Target, item: ItemDTO | dict) -> dict: + req: targetRequest = target["request"] + + try: + itemId = ag(item, ["id", "_id"], "??") + except Exception: + itemId = "??" + + try: + LOG.info(f"Sending Notification event '{event}' id '{itemId}' to '{target.get('name')}'.") + reqBody = { + "method": req["method"].upper(), + "url": req["url"], + "headers": { + "User-Agent": f"YTPTube/{APP_VERSION}", + "Content-Type": "application/json" + if "json" == req["type"].lower() + else "application/x-www-form-urlencoded", + }, + } + + if len(req["headers"]) > 0: + for h in req["headers"]: + reqBody["headers"][h["key"]] = h["value"] + + reqBody["json" if "json" == req["type"].lower() else "data"] = { + "event": event, + "created_at": datetime.now(tz=timezone.utc).isoformat(), + "payload": item.__dict__ if isinstance(item, ItemDTO) else item, + } + + if "form" == req["type"].lower(): + reqBody["data"]["payload"] = self._encoder.encode(reqBody["data"]["payload"]) + + response = await self._client.request(**reqBody) + + respData = {"url": req["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}'." + if self._debug and respData.get("text"): + msg += f" body '{respData.get('text','??')}'." + + LOG.info(msg) + + return respData + except Exception as e: + LOG.error(f"Error sending Notification event '{event}' id '{itemId}' to '{target['name']}'. '{e}'.") + return {"url": req["url"], "status": 500, "text": str(e)} + + def emit(self, event, data, **kwargs): + if len(self.targets) < 1: + return False + + if event not in self.allowed_events and "test" != event: + return False + + return self.send(event, data) diff --git a/app/library/Singleton.py b/app/library/Singleton.py new file mode 100644 index 00000000..d786bbf4 --- /dev/null +++ b/app/library/Singleton.py @@ -0,0 +1,32 @@ +from typing import Any, Dict +import threading + + +class Singleton(type): + """ + A metaclass that creates a Singleton base class when called. + """ + + _instances: Dict[type, Any] = {} + + def __call__(cls, *args: Any, **kwargs: Any) -> Any: + if cls not in cls._instances: + instance = super().__call__(*args, **kwargs) + cls._instances[cls] = instance + return cls._instances[cls] + + +class threadSafe(type): + """ + A metaclass that creates a Singleton base class when called. + """ + + _instances: Dict[type, Any] = {} + _lock = threading.Lock() + + def __call__(cls, *args: Any, **kwargs: Any) -> Any: + with cls._lock: + if cls not in cls._instances: + instance = super().__call__(*args, **kwargs) + cls._instances[cls] = instance + return cls._instances[cls] diff --git a/app/library/Utils.py b/app/library/Utils.py index 8fe3e06c..a663cd49 100644 --- a/app/library/Utils.py +++ b/app/library/Utils.py @@ -535,3 +535,20 @@ def arg_converter(args: str) -> dict: diff["postprocessors"] = [pp for pp in diff["postprocessors"] if pp not in default_opts["postprocessors"]] return json.loads(json.dumps(diff)) + + +def validateUUID(uuid_str: str, version: int = 4) -> bool: + """ + Validate if the UUID is valid. + + Args: + uuid_str (str): UUID to validate. + + Returns: + bool: True if the UUID is valid. + """ + try: + uuid.UUID(uuid_str, version=version) + return True + except ValueError: + return False diff --git a/app/library/Webhooks.py b/app/library/Webhooks.py deleted file mode 100644 index 631cc9e1..00000000 --- a/app/library/Webhooks.py +++ /dev/null @@ -1,114 +0,0 @@ -import asyncio -import logging -import os -from .ItemDTO import ItemDTO -import httpx -from .version import APP_VERSION - -LOG = logging.getLogger("Webhooks") - - -class Webhooks: - targets: list[dict] = [] - allowed_events: tuple = ( - "added", - "completed", - "error", - "not_live", - ) - - def __init__(self, file: str): - if os.path.exists(file): - self.load(file) - - def load(self, file: str): - try: - if os.path.getsize(file) < 3: - raise Exception("file is empty.") - - LOG.info(f"Loading webhooks from '{file}'.") - from .Utils import load_file - - (targets, status, error) = load_file(file, list) - if not status: - raise Exception(f"{error}") - - for target in targets: - LOG.info(f"Will send '{target.get('on',['all'])}' events to '{target.get('name')}'.") - - self.targets = targets - except Exception as e: - LOG.error(f"Error loading webhooks from '{file}'. '{e}'") - pass - - async def send(self, event: str, item: ItemDTO | dict) -> list[dict]: - if len(self.targets) == 0: - return [] - - if not isinstance(item, ItemDTO) and not isinstance(item, dict): - LOG.debug(f"Received invalid item type '{type(item)}' with event '{event}'.") - return [] - - tasks = [] - - for target in self.targets: - allowed_events = target.get("on") if "on" in target else [] - if len(allowed_events) > 0 and event not in allowed_events: - continue - - tasks.append(self.__send(event, target, item)) - - return await asyncio.gather(*tasks) - - async def __send(self, event: str, target: dict, item: ItemDTO | dict) -> dict: - from .config import Config - - req: dict = target.get("request") - - try: - itemId = item.get("id", item.get("_id", "??")) - except Exception: - itemId = "??" - - try: - LOG.info(f"Sending event '{event}' id '{itemId}' to '{target.get('name')}'.") - async with httpx.AsyncClient() as client: - request_type = req.get("type", "json") - - reqBody = { - "method": req.get("method", "POST"), - "url": req.get("url"), - "headers": {"User-Agent": f"YTPTube/{APP_VERSION}"}, - } - - if req.get("headers", None): - reqBody["headers"].update(req.get("headers")) - - match request_type: - case "json": - reqBody["json"] = item.__dict__ if isinstance(item, ItemDTO) else item - reqBody["headers"]["Content-Type"] = "application/json" - case _: - reqBody["data"] = item.__dict__ if isinstance(item, ItemDTO) else item - reqBody["headers"]["Content-Type"] = "application/x-www-form-urlencoded" - - response = await client.request(**reqBody) - - respData = {"url": req.get("url"), "status": response.status_code, "text": response.text} - - msg = f"Webhook target '{target.get('name')}' Responded to event '{event}' id '{itemId}' with status '{response.status_code}'." - if Config.get_instance().debug and respData.get("text"): - msg += f" body '{respData.get('text','??')}'." - - LOG.info(msg) - - return respData - except Exception as e: - LOG.error(f"Error sending event '{event}' id '{itemId}' to '{target.get('name')}'. '{e}'") - return {"url": req.get("url"), "status": 500, "text": str(e)} - - def emit(self, event, data, **kwargs): - if len(self.targets) < 1 or event not in self.allowed_events: - return False - - return self.send(event, data) diff --git a/app/library/cache.py b/app/library/cache.py index a8a2e5d5..b2e33f02 100644 --- a/app/library/cache.py +++ b/app/library/cache.py @@ -1,27 +1,12 @@ import hashlib -import time import threading -from typing import Any, Optional, Dict, Tuple +import time +from typing import Any, Dict, Optional, Tuple + +from .Singleton import threadSafe -class SingletonMeta(type): - """ - A metaclass that creates a Singleton base class when called. - """ - - _instances: Dict[type, Any] = {} - _lock = threading.Lock() # Ensures thread-safe singleton creation - - def __call__(cls, *args: Any, **kwargs: Any) -> Any: - # Thread-safe instance creation - with cls._lock: - if cls not in cls._instances: - instance = super().__call__(*args, **kwargs) - cls._instances[cls] = instance - return cls._instances[cls] - - -class Cache(metaclass=SingletonMeta): +class Cache(metaclass=threadSafe): def __init__(self) -> None: """ Initialize the Cache. diff --git a/app/main.py b/app/main.py index d67bc192..c91f857f 100644 --- a/app/main.py +++ b/app/main.py @@ -23,7 +23,7 @@ from library.Emitter import Emitter from library.encoder import Encoder from library.HttpAPI import HttpAPI, LOG as http_logger from library.HttpSocket import HttpSocket -from library.Webhooks import Webhooks +from library.Notifications import Notification from library.PackageInstaller import PackageInstaller LOG = logging.getLogger("app") @@ -69,10 +69,8 @@ class Main: self.http = HttpAPI(queue=queue, emitter=self.emitter, encoder=self.encoder, load_tasks=self.load_tasks) self.socket = HttpSocket(queue=queue, emitter=self.emitter, encoder=self.encoder) + self.emitter.add_emitter(Notification().emit) - WebhookFile = os.path.join(self.config.config_path, "webhooks.json") - if os.path.exists(WebhookFile): - self.emitter.add_emitter(Webhooks(WebhookFile).emit) def checkFolders(self) -> None: try: diff --git a/ui/components/NotificationForm.vue b/ui/components/NotificationForm.vue new file mode 100644 index 00000000..1cf3f6eb --- /dev/null +++ b/ui/components/NotificationForm.vue @@ -0,0 +1,268 @@ + + + diff --git a/ui/components/TaskForm.vue b/ui/components/TaskForm.vue index 90b0c1e1..e2b1a1ee 100644 --- a/ui/components/TaskForm.vue +++ b/ui/components/TaskForm.vue @@ -270,7 +270,7 @@ const checkInfo = async () => { } } - //addInProgress.value = true; + addInProgress.value = true; emitter('submit', { reference: toRaw(props.reference), task: toRaw(form) }); } diff --git a/ui/layouts/default.vue b/ui/layouts/default.vue index 6dc97b06..23344906 100644 --- a/ui/layouts/default.vue +++ b/ui/layouts/default.vue @@ -4,14 +4,13 @@
- - + + YTPTube
-
- +
+
+ + + +
+
-
+
diff --git a/ui/pages/notifications.vue b/ui/pages/notifications.vue new file mode 100644 index 00000000..7f528a79 --- /dev/null +++ b/ui/pages/notifications.vue @@ -0,0 +1,300 @@ + + + + + diff --git a/ui/pages/tasks.vue b/ui/pages/tasks.vue index e350eb31..3cfa063e 100644 --- a/ui/pages/tasks.vue +++ b/ui/pages/tasks.vue @@ -25,7 +25,7 @@ div.is-centered {

-

@@ -46,7 +46,7 @@ div.is-centered {
- +
@@ -175,7 +175,7 @@ const reloadContent = async (fromMounted = false) => { } } -const resetTask = (closeForm = false) => { +const resetForm = (closeForm = false) => { task.value = {} taskRef.value = null if (closeForm) { @@ -243,13 +243,11 @@ const updateItem = async ({ reference, task }) => { } toast.success('Task updated') - resetTask(true) - toggleForm.value = false + resetForm(true) } const filterItem = item => { - // -- remove the raw key from the item. before showing it to user const { raw, ...rest } = item return JSON.stringify(rest, null, 2) } @@ -269,5 +267,4 @@ const calcPath = path => { } onMounted(async () => socket.isConnected ? await reloadContent(true) : '') -