Added Webhook support

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

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 = ''
 

app/DownloadQueue.py

@@ -1,6 +1,5 @@
 import asyncio
 from email.utils import formatdate
-import json
 import logging
 import os
 import time

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)

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)
+            }

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