Merge pull request #55 from arabcoders/dev

Better Post-Live Manifestless error mode.

.vscode/launch.json

@@ -34,9 +34,9 @@
         "YTP_TEMP_PATH": "${workspaceFolder}/var/tmp",
         "YTP_YTDL_OPTIONS_FILE": "${workspaceFolder}/var/config/ytdlp.json",
         "YTP_URL_HOST": "http://localhost:8081",
-        "YTP_YTDL_DEBUG": "true",
+        "YTP_YTDL_DEBUG": "false",
         "YTP_TEMP_KEEP": "true",
-        "YTP_KEEP_ARCHIVE": "true",
+        "YTP_KEEP_ARCHIVE": "false",
       }
     }
   ]

Dockerfile

@@ -28,8 +28,12 @@ ENV YTP_CONFIG_PATH=/config
 ENV YTP_TEMP_PATH=/tmp
 ENV YTP_DOWNLOAD_PATH=/downloads
 
+# removed ffmpeg as 6.1.0 is broken with DASH protocal downloads
+COPY --from=mwader/static-ffmpeg:6.1.1 /ffmpeg /usr/bin/
+COPY --from=mwader/static-ffmpeg:6.1.1 /ffprobe /usr/bin/
+
 RUN mkdir /config /downloads && ln -snf /usr/share/zoneinfo/${TZ} /etc/localtime && echo ${TZ} > /etc/timezone && \
-  apk add --update --no-cache bash ffmpeg mkvtoolnix patch aria2 coreutils curl shadow sqlite tzdata && \
+  apk add --update --no-cache bash mkvtoolnix patch aria2 coreutils curl shadow sqlite tzdata && \
   useradd -u ${USER_ID:-1000} -U -d /app -s /bin/bash app && \
   rm -rf /var/cache/apk/*
 

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/AsyncPool.py

@@ -32,7 +32,7 @@ class AsyncPool:
         @param expected_total: (optional) expected total number of jobs (used for `log_event_n` logging)
         @return: instance of AsyncWorkerPool
         """
-        loop = loop or asyncio.get_event_loop()
+        loop = loop if loop else asyncio.get_event_loop()
         self._loop = loop
         self._num_workers = num_workers
         self._logger = logger
@@ -65,7 +65,7 @@ class AsyncPool:
                 future, args, kwargs = item
                 # the wait_for will cancel the task (task sees CancelledError) and raises a TimeoutError from here
                 # so be wary of catching TimeoutErrors in this loop
-                result = await asyncio.wait_for(self._worker_co(*args, **kwargs), self._max_task_time, loop=self._loop)
+                result = await asyncio.wait_for(self._worker_co(*args, **kwargs), self._max_task_time)
 
                 if future:
                     future.set_result(result)
@@ -94,6 +94,12 @@ class AsyncPool:
     def total_queued(self):
         return self._total_queued
 
+    def has_open_workers(self) -> bool:
+        """
+        :return: True if there are open workers.
+        """
+        return self._queue.qsize() < self._num_workers
+
     async def __aenter__(self):
         self.start()
         return self
@@ -144,7 +150,7 @@ class AsyncPool:
             await self._queue.put(Terminator())
 
         try:
-            await asyncio.gather(*self._workers, loop=self._loop)
+            await asyncio.gather(*self._workers)
             self._workers = None
         except:
             self._logger.exception('Exception joining {}'.format(self._name))

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 = ''
 
@@ -36,6 +36,8 @@ class Config:
 
     allow_manifestless: bool = False
 
+    max_workers: int = 0
+
     version: str = APP_VERSION
 
     _boolean_vars: tuple = (

app/DataStore.py

@@ -114,6 +114,16 @@ class DataStore:
     def empty(self):
         return not bool(self.dict)
 
+    def hasDownloads(self):
+        if 0 == len(self.dict):
+            return False
+
+        for key in self.dict:
+            if self.dict[key].started() is False:
+                return True
+
+        return False
+
     def _updateStoreItem(self, type: str, item: ItemDTO) -> None:
         sqlStatement = """
         INSERT INTO "history" ("id", "type", "url", "data")

app/Download.py

@@ -1,9 +1,9 @@
-
 import asyncio
 import json
 import logging
 import multiprocessing
 import os
+import queue
 import re
 import shutil
 import yt_dlp
@@ -15,6 +15,7 @@ log = logging.getLogger('download')
 
 
 class Download:
+    id: str = None
     manager = None
     download_dir: str = None
     temp_dir: str = None
@@ -57,6 +58,7 @@ class Download:
         self.ytdl_opts = get_opts(
             info.format, info.quality, info.ytdlp_config if info.ytdlp_config else {})
         self.info = info
+        self.id = info._id
         self.default_ytdl_opts = default_ytdl_opts
         self.debug = debug
 
@@ -66,13 +68,15 @@ class Download:
         self.proc = None
         self.loop = None
         self.notifier = None
-        self.tempKeep = bool(Config.get_instance().temp_keep)
+        config = Config.get_instance()
+        self.max_workers = int(config.max_workers)
+        self.tempKeep = bool(config.temp_keep)
 
     def _download(self):
         try:
             def put_status(st):
-                self.status_queue.put(
+                dicto = {k: v for k, v in st.items() if k in self._ytdlp_fields}
-                    {k: v for k, v in st.items() if k in self._ytdlp_fields})
+                self.status_queue.put({'id': self.id, **dicto})
 
             def put_status_postprocessor(d):
                 if d['postprocessor'] == 'MoveFiles' and d['status'] == 'finished':
@@ -83,12 +87,12 @@ class Download:
                         )
                     else:
                         filename = d['info_dict']['filepath']
-                    self.status_queue.put(
+
-                        {
+                    self.status_queue.put({
-                            'status': 'finished',
+                        'id': self.id,
-                            'filename': filename
+                        'status': 'finished',
-                        }
+                        'filename': filename
-                    )
+                    })
 
             # Create temp dir for each download.
             self.tempPath = os.path.join(
@@ -136,13 +140,16 @@ class Download:
 
             log.info(
                 f'Downloading id="{self.info.id}" title="{self.info.title}".')
+
             ret = yt_dlp.YoutubeDL(params=params).download([self.info.url])
 
-            self.status_queue.put(
+            self.status_queue.put({
-                {'status': 'finished' if ret == 0 else 'error'}
+                'id': self.id,
-            )
+                'status': 'finished' if ret == 0 else 'error'
+            })
         except Exception as exc:
             self.status_queue.put({
+                'id': self.id,
                 'status': 'error',
                 'msg': str(exc),
                 'error': str(exc)
@@ -179,15 +186,16 @@ class Download:
     def close(self):
         if self.started():
             self.proc.close()
-            self.status_queue.put(None)
+            if self.max_workers < 1:
+                self.status_queue.put(None)
 
-    def running(self):
+    def running(self) -> bool:
         try:
             return self.proc is not None and self.proc.is_alive()
         except ValueError:
             return False
 
-    def started(self):
+    def started(self) -> bool:
         return self.proc is not None
 
     async def progress_update(self):
@@ -196,7 +204,7 @@ class Download:
         """
         while True:
             status = await self.loop.run_in_executor(None, self.status_queue.get)
-            if not status:
+            if not status or status.get('id') != self.id:
                 return
 
             if self.debug:

app/DownloadQueue.py

@@ -1,6 +1,5 @@
 import asyncio
 from email.utils import formatdate
-import json
 import logging
 import os
 import time
@@ -11,8 +10,11 @@ from Download import Download
 from ItemDTO import ItemDTO
 from DataStore import DataStore
 from Utils import Notifier, ObjectSerializer, calcDownloadPath, ExtractInfo, isDownloaded, mergeConfig
+from AsyncPool import AsyncPool
+from concurrent.futures import ProcessPoolExecutor
 
 log = logging.getLogger('DownloadQueue')
+dl_queue = asyncio.Queue()
 
 
 class DownloadQueue:
@@ -36,7 +38,15 @@ class DownloadQueue:
 
     async def initialize(self):
         self.event = asyncio.Event()
-        asyncio.create_task(self.__download())
+        if self.config.max_workers > 0:
+            log.info(
+                f'Using {self.config.max_workers} workers for downloading.')
+        else:
+            log.info('Using single threaded download.')
+
+        asyncio.create_task(
+            self.__download_pool() if self.config.max_workers > 0 else self.__download()
+        )
 
     async def __add_entry(
         self,
@@ -156,11 +166,18 @@ class DownloadQueue:
             if dlInfo.info.live_in:
                 dlInfo.info.status = 'not_live'
                 itemDownload = self.done.put(dlInfo)
+                NotifiyEvent = 'completed'
+            elif self.config.allow_manifestless is False and 'live_status' in entry and 'post_live' == entry.get('live_status'):
+                dlInfo.info.status = 'error'
+                dlInfo.info.error = 'Video is in Post-Live Manifestless mode.'
+                itemDownload = self.done.put(dlInfo)
+                NotifiyEvent = 'completed'
             else:
+                NotifiyEvent = 'added'
                 itemDownload = self.queue.put(dlInfo)
                 self.event.set()
 
-            await self.notifier.emit('completed' if itemDownload.info.live_in else 'added', itemDownload.info)
+            await self.notifier.emit(NotifiyEvent, itemDownload.info)
 
             return {
                 'status': 'ok'
@@ -221,18 +238,12 @@ class DownloadQueue:
             if self.isDownloaded(entry):
                 raise yt_dlp.utils.ExistingVideoReached()
 
-            if self.config.allow_manifestless is False and 'live_status' in entry and 'post_live' == entry.get('live_status'):
-                raise yt_dlp.utils.YoutubeDLError(
-                    'Video is in Post-Live Manifestless mode.')
-
             log.debug(f'entry: extract info says: {entry}')
         except yt_dlp.utils.ExistingVideoReached:
             return {'status': 'error', 'msg': 'Video has been downloaded already and recorded in archive.log file.'}
         except yt_dlp.utils.YoutubeDLError as exc:
             return {'status': 'error', 'msg': str(exc)}
 
-        #
-
         return await self.__add_entry(
             entry=entry,
             quality=quality,
@@ -252,9 +263,9 @@ class DownloadQueue:
 
             item = self.queue.get(key=id)
 
-            if self.queue.get(id).started():
+            if item.started():
                 log.info(f'Canceling {id=} {item.info.title=}')
-                self.queue.get(id).cancel()
+                item.cancel()
             else:
                 self.queue.delete(id)
                 log.info(f'Deleting {id=} {item.info.title=}')
@@ -293,6 +304,59 @@ class DownloadQueue:
 
         return items
 
+    async def __download_pool(self):
+        async with AsyncPool(
+            loop=asyncio.get_running_loop(),
+            num_workers=self.config.max_workers,
+            worker_co=self.__downloadFile,
+            name='WorkerPool',
+            logger=logging.getLogger('WorkerPool')
+        ) as executor:
+            while True:
+                while not self.queue.hasDownloads():
+                    log.info('waiting for item to download.')
+                    await self.event.wait()
+                    self.event.clear()
+
+                while True:
+                    if executor.has_open_workers() is False:
+                        log.info(
+                            f'Waiting for workers to finish. {executor.total_queued} items in queue.')
+                        await asyncio.sleep(1)
+                    else:
+                        break
+
+                for id in self.queue.dict:
+                    entry = self.queue.get(key=id)
+                    if entry.started() is False and entry.canceled is False:
+                        await executor.push(id=id, entry=entry)
+                        await asyncio.sleep(1)
+
+    async def __downloadFile(self, id: str, entry: Download):
+        log.info(
+            f'Queuing {id=} - {entry.info.title=} - {entry.info.url=} - {entry.info.folder=}.')
+
+        await entry.start(self.notifier)
+
+        if entry.info.status != 'finished':
+            if entry.tmpfilename and os.path.isfile(entry.tmpfilename):
+                try:
+                    os.remove(entry.tmpfilename)
+                except:
+                    pass
+
+            entry.info.status = 'error'
+
+        entry.close()
+
+        if self.queue.exists(id):
+            self.queue.delete(id)
+            if entry.canceled:
+                await self.notifier.canceled(id)
+            else:
+                self.done.put(entry)
+                await self.notifier.completed(entry.info)
+
     async def __download(self):
         while True:
             while self.queue.empty():
@@ -301,29 +365,7 @@ class DownloadQueue:
                 self.event.clear()
 
             id, entry = self.queue.next()
-            log.info(
+            await self.__downloadFile(id, entry)
-                f'Queuing {id=} - {entry.info.title=} - {entry.info.url=} - {entry.info.folder=}.')
-
-            await entry.start(self.notifier)
-
-            if entry.info.status != 'finished':
-                if entry.tmpfilename and os.path.isfile(entry.tmpfilename):
-                    try:
-                        os.remove(entry.tmpfilename)
-                    except:
-                        pass
-
-                entry.info.status = 'error'
-
-            entry.close()
-
-            if self.queue.exists(id):
-                self.queue.delete(id)
-                if entry.canceled:
-                    await self.notifier.canceled(id)
-                else:
-                    self.done.put(entry)
-                    await self.notifier.completed(entry.info)
 
     def isDownloaded(self, info: dict) -> bool:
         return self.config.keep_archive and isDownloaded(self.config.ytdl_options.get('download_archive', None), info)

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,62 @@
+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:
+                if os.path.getsize(file) < 2:
+                    raise Exception(f'file is empty.')
+
+                log.info(f'Loading webhooks from {file}')
+                with open(file, 'r') as f:
+                    self.targets = json.load(f)
+            except Exception as e:
+                log.error(f'Error loading webhooks from {file}: {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,
+            config=self.config,
-            Notifier(sio=self.sio, serializer=self.serializer),
+            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()

frontend/vue.config.js

@@ -1,5 +1,15 @@
 const { defineConfig } = require('@vue/cli-service')
 module.exports = defineConfig({
   publicPath: './',
-  transpileDependencies: true
+  transpileDependencies: true,
+  chainWebpack: (config) => {
+    config.plugin('define').tap((definitions) => {
+      Object.assign(definitions[0], {
+        __VUE_OPTIONS_API__: 'true',
+        __VUE_PROD_DEVTOOLS__: 'false',
+        __VUE_PROD_HYDRATION_MISMATCH_DETAILS__: 'false'
+      })
+      return definitions
+    })
+  }
 })