From b512437ddefb93f052b4f616ac89f9a9a837e663 Mon Sep 17 00:00:00 2001 From: ArabCoders Date: Sun, 7 Jan 2024 16:44:29 +0300 Subject: [PATCH] Added Webhook support --- README.md | 29 +++++++++++++++++++++- app/Config.py | 2 +- app/DownloadQueue.py | 1 - app/Utils.py | 24 ++++++++++++++++-- app/Webhooks.py | 58 ++++++++++++++++++++++++++++++++++++++++++++ app/main.py | 11 ++++++--- 6 files changed, 116 insertions(+), 9 deletions(-) create mode 100644 app/Webhooks.py diff --git a/README.md b/README.md index 98093040..209a39eb 100644 --- a/README.md +++ b/README.md @@ -13,7 +13,8 @@ YTPTube started as a fork of [meTube](https://github.com/alexta69/metube) projec * Switched out of binary file storage in favor of SQLite. * Handle live streams. * Support per link, `yt-dlp config` and `cookies`. and `output format` -* Tasks Runner. It allow you to queue channels for downloading using simple `json` file for configuration. +* Tasks Runner. It allow you to queue channels for downloading using simple `json` file. +* Webhook sender. It allow you to add webhook endpoints that receive events related to downloads using simple `json` file. ### Tips Your `yt-dlp` config should include the following options for optimal working conditions. @@ -247,6 +248,32 @@ The task runner is doing what you are doing when you click the add button on the **WARNING**: We strongly advice turning on `YTP_KEEP_ARCHIVE` option. Otherwise, you will keep re-downloading the items, and you will eventually get banned from the source or or you will waste space, bandwidth re-downloading content over and over. +### 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", + // (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" + } + } + ... +] +``` + # Donation If you feel like donating and appreciate my work, you can do so by donating to children charity. For example [Make-A-Wish](https://worldwish.org). diff --git a/app/Config.py b/app/Config.py index c0bf668f..4b76444d 100644 --- a/app/Config.py +++ b/app/Config.py @@ -28,7 +28,7 @@ class Config: host: str = '0.0.0.0' port: int = 8081 - keep_archive: bool = False + keep_archive: bool = True base_path: str = '' diff --git a/app/DownloadQueue.py b/app/DownloadQueue.py index ad118220..a2d2bc0c 100644 --- a/app/DownloadQueue.py +++ b/app/DownloadQueue.py @@ -1,6 +1,5 @@ import asyncio from email.utils import formatdate -import json import logging import os import time diff --git a/app/Utils.py b/app/Utils.py index ebe268aa..f2436c2f 100644 --- a/app/Utils.py +++ b/app/Utils.py @@ -1,3 +1,4 @@ +import asyncio import copy from datetime import datetime, timezone import json @@ -6,6 +7,7 @@ import os from typing import Any import yt_dlp from socketio import AsyncServer +from Webhooks import Webhooks log = logging.getLogger('Utils') @@ -294,9 +296,21 @@ class Notifier: This class is used to send events to the frontend. ''' - def __init__(self, sio: AsyncServer, serializer: ObjectSerializer): + sio: AsyncServer = None + "SocketIO server instance" + serializer: ObjectSerializer = None + "Serializer used to serialize objects to JSON" + webhooks: Webhooks = None + "Send webhooks events." + webhooks_allowed_events: tuple = ( + 'added', 'completed', 'error', 'not_live' + ) + "Events that are allowed to be sent to webhooks." + + def __init__(self, sio: AsyncServer, serializer: ObjectSerializer, webhooks: Webhooks = None): self.sio = sio self.serializer = serializer + self.webhooks = webhooks async def added(self, dl: dict): await self.emit('added', dl) @@ -317,4 +331,10 @@ class Notifier: await self.emit('error', (dl, message)) async def emit(self, event: str, data): - await self.sio.emit(event, self.serializer.encode(data)) + tasks = [] + tasks.append(self.sio.emit(event, self.serializer.encode(data))) + + if self.webhooks and event in self.webhooks_allowed_events: + tasks.append(self.webhooks.send(event, data)) + + await asyncio.gather(*tasks) diff --git a/app/Webhooks.py b/app/Webhooks.py new file mode 100644 index 00000000..989ef45b --- /dev/null +++ b/app/Webhooks.py @@ -0,0 +1,58 @@ +import asyncio +import json +import logging +import os +from ItemDTO import ItemDTO +from aiohttp import client + +log = logging.getLogger('Webhooks') + + +class Webhooks: + targets: list[dict] = {} + + def __init__(self, file: str): + if os.path.exists(file): + try: + log.info(f'Loading webhooks from {file}') + with open(file, 'r') as f: + self.targets = json.load(f) + except Exception as e: + pass + + async def send(self, event: str, item: ItemDTO) -> list[dict]: + if len(self.targets) == 0: + return [] + + if not isinstance(item, ItemDTO): + 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: + req: dict = target.get('request') + try: + log.info(f"Sending {event=} {item.id=} to [{target.get('name')}]") + async with client.ClientSession() as session: + headers = req.get('headers', {}) if 'headers' in req else {} + async with session.request(method=req.get('method', 'POST'), url=req.get('url'), json=item.__dict__, headers=headers) as response: + return { + 'url': req.get('url'), + 'status': response.status, + 'text': await response.text() + } + except Exception as e: + return { + 'url': req.get('url'), + 'status': 500, + 'text': str(e) + } diff --git a/app/main.py b/app/main.py index 4e41cc6f..6e7c4418 100644 --- a/app/main.py +++ b/app/main.py @@ -10,6 +10,7 @@ from DownloadQueue import DownloadQueue from Utils import ObjectSerializer, Notifier from aiohttp import web from aiohttp.web import Request, Response +from Webhooks import Webhooks from player.M3u8 import M3u8 from player.Segments import Segments import socketio @@ -77,7 +78,8 @@ class Main: caribou.upgrade(self.config.db_file, os.path.join( os.path.realpath(os.path.dirname(__file__)), 'migrations')) - + self.webhooks = Webhooks(os.path.join( + self.config.config_path, 'webhooks.json')) self.loop = asyncio.get_event_loop() self.serializer = ObjectSerializer() self.app = web.Application() @@ -87,10 +89,11 @@ class Main: self.routes = web.RouteTableDef() self.connection = sqlite3.connect(self.config.db_file) self.dqueue = DownloadQueue( - self.config, - Notifier(sio=self.sio, serializer=self.serializer), + config=self.config, + notifier=Notifier( + sio=self.sio, serializer=self.serializer, webhooks=self.webhooks), connection=self.connection, - serializer=self.serializer + serializer=self.serializer, ) self.app.on_startup.append(lambda _: self.dqueue.initialize()) self.addRoutes()