From 2d70de20320b75f66538bd7b28037d1424f2fa25 Mon Sep 17 00:00:00 2001 From: ArabCoders Date: Fri, 20 Dec 2024 21:20:51 +0300 Subject: [PATCH] updating python code to be little more clean and implements more type checks. --- .vscode/launch.json | 1 + app/library/AsyncPool.py | 54 ++-- app/library/DataStore.py | 56 ++-- app/library/Download.py | 174 +++++----- app/library/DownloadQueue.py | 24 +- app/library/Emitter.py | 28 +- app/library/HttpAPI.py | 596 +++++++++++++++++++---------------- app/library/HttpSocket.py | 136 ++++---- app/library/ItemDTO.py | 53 ++-- app/library/M3u8.py | 49 +-- app/library/Playlist.py | 54 ++-- app/library/Segments.py | 96 +++--- app/library/Subtitle.py | 22 +- app/library/Utils.py | 218 ++++++++----- app/library/Webhooks.py | 59 ++-- app/library/common.py | 15 +- app/library/config.py | 9 +- app/library/encoder.py | 2 +- app/library/ffprobe.py | 84 ++--- app/library/version.py | 1 + app/main.py | 12 +- entrypoint.sh | 6 +- healthcheck.sh | 4 +- 23 files changed, 970 insertions(+), 783 deletions(-) diff --git a/.vscode/launch.json b/.vscode/launch.json index b043d06c..ef518c49 100644 --- a/.vscode/launch.json +++ b/.vscode/launch.json @@ -37,6 +37,7 @@ "YTP_LOG_LEVEL": "DEBUG", "YTP_DEBUG": "false", "YTP_YTDL_DEBUG": "false", + "YTP_MAX_WORKERS": "2", } }, { diff --git a/app/library/AsyncPool.py b/app/library/AsyncPool.py index 852659b3..81f72263 100644 --- a/app/library/AsyncPool.py +++ b/app/library/AsyncPool.py @@ -1,6 +1,6 @@ import asyncio -from datetime import datetime, timezone import logging +from datetime import datetime, timezone class Terminator: @@ -8,13 +8,18 @@ class Terminator: class AsyncPool: - def __init__(self, - num_workers: int, worker_co, - name: str, logger: logging.Logger, - loop: asyncio.AbstractEventLoop = None, - load_factor: int = 1, max_task_time: int = None, - return_futures: bool = False, raise_on_join: bool = False - ): + def __init__( + self, + num_workers: int, + worker_co, + name: str, + logger: logging.Logger, + loop: asyncio.AbstractEventLoop = None, + load_factor: int = 1, + max_task_time: int = None, + return_futures: bool = False, + raise_on_join: bool = False, + ): """ This class will create `num_workers` asyncio tasks to work against a queue of `num_workers * load_factor` items of back-pressure (IOW we will block after such @@ -62,8 +67,8 @@ class AsyncPool: future, args, kwargs = item self._status[worker_id] = { - 'started': self._time().isoformat(), - 'data': kwargs.get('entry', {'info': {}}).info.__dict__, + "started": self._time().isoformat(), + "data": kwargs.get("entry", {"info": {}}).info.__dict__, } # the wait_for will cancel the task (task sees CancelledError) and raises a TimeoutError from here @@ -87,7 +92,7 @@ class AsyncPool: # don't log the failure when the client is receiving the future future.set_exception(e) else: - self._logger.exception(f'Worker call failed. {str(e)}') + self._logger.exception(f"Worker call failed. {str(e)}") finally: self._status[worker_id] = None @@ -140,7 +145,7 @@ class AsyncPool: return future def start(self): - """ Will start up worker pool and reset exception state """ + """Will start up worker pool and reset exception state""" self._exceptions = False for worker_number in range(self._num_workers): @@ -148,16 +153,16 @@ class AsyncPool: self._createWorker(worker_id) async def restart(self, worker_id: str, msg: str = None) -> bool: - """ Will restart the worker pool """ + """Will restart the worker pool""" if worker_id not in self._workers: - self._logger.warning(f'Worker {worker_id} does not exist.') + self._logger.warning(f"Worker {worker_id} does not exist.") return False try: self._workers[worker_id].cancel(msg) await self._workers[worker_id] except asyncio.exceptions.CancelledError as e: - self._logger.warning(f'Worker {worker_id} restarted. {str(e)}') + self._logger.warning(f"Worker {worker_id} restarted. {str(e)}") if worker_id in self._status: self._status.pop(worker_id) @@ -169,16 +174,16 @@ class AsyncPool: return True async def stop(self, worker_id: str, msg: str = None) -> bool: - """ Will stop the worker """ + """Will stop the worker""" if worker_id not in self._workers: - self._logger.warning(f'Worker {worker_id} does not exist.') + self._logger.warning(f"Worker {worker_id} does not exist.") return False if self._workers[worker_id].cancel(msg): try: await self._workers[worker_id] except asyncio.exceptions.CancelledError as e: - self._logger.warning(f'Worker {worker_id} stopped. {str(e)}') + self._logger.warning(f"Worker {worker_id} stopped. {str(e)}") if worker_id in self._status: self._status.pop(worker_id) @@ -192,7 +197,7 @@ class AsyncPool: if len(self._workers) < 1: return - self._logger.info(f'Joining {self._name}') + self._logger.info(f"Joining {self._name}") # The Terminators will kick each worker from being blocked against the _queue.get() and allow # each one to exit. @@ -203,26 +208,25 @@ class AsyncPool: await asyncio.gather(*list(self._workers)) self._workers = {} except: - self._logger.exception(f'Exception joining {self._name}') + self._logger.exception(f"Exception joining {self._name}") raise finally: - self._logger.info(f'Completed {self._name}') + self._logger.info(f"Completed {self._name}") if self._exceptions and self._raise_on_join: raise Exception(f"Exception occurred in {self._name} pool") def _createWorker(self, worker_id: str) -> asyncio.Future: if worker_id in self._workers: - self._logger.debug(f'Worker {worker_id} already exists.') + self._logger.debug(f"Worker {worker_id} already exists.") return self._workers[worker_id] self._status[worker_id] = None self._workers[worker_id] = asyncio.ensure_future( - coro_or_future=self._worker_loop(worker_id=worker_id), - loop=self._loop + coro_or_future=self._worker_loop(worker_id=worker_id), loop=self._loop ) - self._logger.debug(f'Created {worker_id}') + self._logger.debug(f"Created {worker_id}") return self._workers[worker_id] diff --git a/app/library/DataStore.py b/app/library/DataStore.py index a0ea04a3..f3f92308 100644 --- a/app/library/DataStore.py +++ b/app/library/DataStore.py @@ -1,9 +1,10 @@ -from collections import OrderedDict import copy +import json +from collections import OrderedDict from datetime import datetime, timezone from email.utils import formatdate -import json from sqlite3 import Connection + from .config import Config from .Download import Download from .ItemDTO import ItemDTO @@ -13,6 +14,7 @@ class DataStore: """ Persistent queue. """ + type: str = None dict: OrderedDict[str, Download] = None config: Config = None @@ -31,7 +33,7 @@ class DataStore: def exists(self, key: str = None, url: str = None) -> bool: if not key and not url: - raise KeyError('key or url must be provided.') + raise KeyError("key or url must be provided.") if key and key in self.dict: return True @@ -44,13 +46,13 @@ class DataStore: def get(self, key: str, url: str = None) -> Download: if not key and not url: - raise KeyError('key or url must be provided.') + raise KeyError("key or url must be provided.") for i in self.dict: if (key and self.dict[i].info._id == key) or (url and self.dict[i].info.url == url): return self.dict[i] - raise KeyError(f'{key=} or {url=} not found.') + raise KeyError(f"{key=} or {url=} not found.") def getById(self, id: str) -> Download | None: return self.dict[id] if id in self.dict else None @@ -62,18 +64,17 @@ class DataStore: items: list[tuple[str, ItemDTO]] = [] cursor = self.connection.execute( - f'SELECT "id", "data", "created_at" FROM "history" WHERE "type" = ? ORDER BY "created_at" ASC', - (self.type,) + 'SELECT "id", "data", "created_at" FROM "history" WHERE "type" = ? ORDER BY "created_at" ASC', (self.type,) ) for row in cursor: - rowDate = datetime.strptime(row['created_at'], '%Y-%m-%d %H:%M:%S') - data: dict = json.loads(row['data']) - key: str = data.pop('_id') + rowDate = datetime.strptime(row["created_at"], "%Y-%m-%d %H:%M:%S") + data: dict = json.loads(row["data"]) + key: str = data.pop("_id") item: ItemDTO = ItemDTO(**data) item._id = key item.datetime = formatdate(rowDate.replace(tzinfo=timezone.utc).timestamp()) - items.append((row['id'], item)) + items.append((row["id"], item)) return items @@ -123,22 +124,37 @@ class DataStore: stored = copy.deepcopy(item) - if hasattr(stored, 'datetime'): + if hasattr(stored, "datetime"): try: - delattr(stored, 'datetime') + delattr(stored, "datetime") except AttributeError: pass - if hasattr(stored, 'live_in') and stored.status == 'finished': + if hasattr(stored, "live_in") and stored.status == "finished": try: - delattr(stored, 'live_in') + delattr(stored, "live_in") except AttributeError: pass - self.connection.execute(sqlStatement.strip(), ( - stored._id, type, stored.url, stored.json(), type, stored.url, - stored.json(), datetime.now(timezone.utc).strftime('%Y-%m-%d %H:%M:%S') - )) + self.connection.execute( + sqlStatement.strip(), + ( + stored._id, + type, + stored.url, + stored.json(), + type, + stored.url, + stored.json(), + datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S"), + ), + ) def _deleteStoreItem(self, key: str) -> None: - self.connection.execute('DELETE FROM "history" WHERE "type" = ? AND "id" = ?', (self.type, key,)) + self.connection.execute( + 'DELETE FROM "history" WHERE "type" = ? AND "id" = ?', + ( + self.type, + key, + ), + ) diff --git a/app/library/Download.py b/app/library/Download.py index be54bb61..7887ee96 100644 --- a/app/library/Download.py +++ b/app/library/Download.py @@ -1,25 +1,28 @@ import asyncio +import hashlib import json import logging import multiprocessing import os -import re import shutil +from multiprocessing.managers import SyncManager + import yt_dlp -import hashlib + from .AsyncPool import Terminator -from .Utils import get_opts, jsonCookie, mergeConfig -from .ItemDTO import ItemDTO from .config import Config from .Emitter import Emitter -from multiprocessing.managers import SyncManager -LOG = logging.getLogger('download') +from .ItemDTO import ItemDTO +from .Utils import get_opts, jsonCookie, mergeConfig + +LOG = logging.getLogger("download") class Download: """ Download task. """ + id: str = None manager = None download_dir: str = None @@ -47,15 +50,15 @@ class Download: "Bad yt-dlp options which are known to cause issues with live stream and post manifestless mode." _ytdlp_fields: tuple = ( - 'tmpfilename', - 'filename', - 'status', - 'msg', - 'total_bytes', - 'total_bytes_estimate', - 'downloaded_bytes', - 'speed', - 'eta', + "tmpfilename", + "filename", + "status", + "msg", + "total_bytes", + "total_bytes_estimate", + "downloaded_bytes", + "speed", + "eta", ) "Fields to be extracted from yt-dlp progress hook." @@ -83,69 +86,62 @@ class Download: self.max_workers = int(config.max_workers) self.tempKeep = bool(config.temp_keep) self.is_live = bool(info.is_live) or info.live_in is not None - self.is_manifestless = 'is_manifestless' in self.info.options and self.info.options['is_manifestless'] is True + self.is_manifestless = "is_manifestless" in self.info.options and self.info.options["is_manifestless"] is True self.info_dict = info_dict def _progress_hook(self, data: dict): dataDict = {k: v for k, v in data.items() if k in self._ytdlp_fields} - self.status_queue.put({'id': self.id, **dataDict}) + self.status_queue.put({"id": self.id, **dataDict}) def _postprocessor_hook(self, data: dict): - if data.get('postprocessor') != 'MoveFiles' or data.get('status') != 'finished': + if data.get("postprocessor") != "MoveFiles" or data.get("status") != "finished": return - if '__finaldir' in data['info_dict']: - filename = os.path.join(data['info_dict']['__finaldir'], os.path.basename(data['info_dict']['filepath'])) + if "__finaldir" in data["info_dict"]: + filename = os.path.join(data["info_dict"]["__finaldir"], os.path.basename(data["info_dict"]["filepath"])) else: - filename = data['info_dict']['filepath'] + filename = data["info_dict"]["filepath"] - self.status_queue.put({ - 'id': self.id, - 'status': 'finished', - 'filename': filename - }) + self.status_queue.put({"id": self.id, "status": "finished", "filename": filename}) def _download(self): try: params: dict = { - 'color': 'no_color', - 'paths': { - 'home': self.download_dir, - 'temp': self.tempPath, + "color": "no_color", + "paths": { + "home": self.download_dir, + "temp": self.tempPath, }, - 'outtmpl': { - 'default': self.output_template, - 'chapter': self.output_template_chapter - }, - 'noprogress': True, - 'break_on_existing': True, - 'progress_hooks': [self._progress_hook], - 'postprocessor_hooks': [self._postprocessor_hook], + "outtmpl": {"default": self.output_template, "chapter": self.output_template_chapter}, + "noprogress": True, + "break_on_existing": True, + "progress_hooks": [self._progress_hook], + "postprocessor_hooks": [self._postprocessor_hook], **get_opts(self.preset, mergeConfig(self.default_ytdl_opts, self.ytdl_opts)), } - if 'format' not in params and self.default_ytdl_opts.get('format', None): - params['format'] = self.default_ytdl_opts['format'] + if "format" not in params and self.default_ytdl_opts.get("format", None): + params["format"] = self.default_ytdl_opts["format"] - params['ignoreerrors'] = False + params["ignoreerrors"] = False if self.debug: - params['verbose'] = True - params['noprogress'] = False + params["verbose"] = True + params["noprogress"] = False if self.info.ytdlp_cookies: try: data = jsonCookie(json.loads(self.info.ytdlp_cookies)) if not data: LOG.warning( - f'The cookie string that was provided for {self.info.title} is empty or not in expected spec.') - with open(os.path.join(self.tempPath, f'cookie_{self.info._id}.txt'), 'w') as f: + f"The cookie string that was provided for {self.info.title} is empty or not in expected spec." + ) + with open(os.path.join(self.tempPath, f"cookie_{self.info._id}.txt"), "w") as f: f.write(data) - params['cookiefile'] = f.name + params["cookiefile"] = f.name except ValueError as e: - LOG.error( - f'Invalid cookies: was provided for {self.info.title} - {str(e)}') + LOG.error(f"Invalid cookies: was provided for {self.info.title} - {str(e)}") if self.is_live or self.is_manifestless: hasDeletedOptions = False @@ -158,29 +154,26 @@ class Download: if hasDeletedOptions: LOG.warning( - f"Live stream detected for '{self.info.title}', The following opts '{deletedOpts=}' have been deleted which are known to cause issues with live stream and post stream manifestless mode.") + f"Live stream detected for '{self.info.title}', The following opts '{deletedOpts=}' have been deleted which are known to cause issues with live stream and post stream manifestless mode." + ) LOG.info( - f'Downloading pid: {os.getpid()} id="{self.info.id}" title="{self.info.title}" preset="{self.preset}".') + f'Downloading pid: {os.getpid()} id="{self.info.id}" title="{self.info.title}" preset="{self.preset}".' + ) cls = yt_dlp.YoutubeDL(params=params) if isinstance(self.info_dict, dict) and len(self.info_dict) > 1: - LOG.debug(f'Downloading using pre-info.') + LOG.debug("Downloading using pre-info.") cls.process_ie_result(self.info_dict, download=True) ret = cls._download_retcode else: - LOG.debug(f'Downloading using url: {self.info.url}') + LOG.debug(f"Downloading using url: {self.info.url}") ret = cls.download([self.info.url]) - self.status_queue.put({'id': self.id, 'status': 'finished' if ret == 0 else 'error'}) + self.status_queue.put({"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) - }) + self.status_queue.put({"id": self.id, "status": "error", "msg": str(exc), "error": str(exc)}) LOG.info(f'Finished {os.getpid()=} id="{self.info.id}" title="{self.info.title}".') @@ -190,17 +183,14 @@ class Download: self.status_queue = self.manager.Queue() # Create temp dir for each download. - self.tempPath = os.path.join( - self.temp_dir, - hashlib.shake_256(f"D-{self.info.id}".encode('utf-8')).hexdigest(5) - ) + self.tempPath = os.path.join(self.temp_dir, hashlib.shake_256(f"D-{self.info.id}".encode("utf-8")).hexdigest(5)) if not os.path.exists(self.tempPath): os.makedirs(self.tempPath, exist_ok=True) self.proc = multiprocessing.Process(name=f"download-{self.id}", target=self._download) self.proc.start() - self.info.status = 'preparing' + self.info.status = "preparing" asyncio.create_task(self.emitter.updated(dl=self.info), name=f"emitter-{self.id}") asyncio.create_task(self.progress_update(), name=f"update-{self.id}") @@ -230,7 +220,7 @@ class Download: try: LOG.info(f"Closing download process: '{procId}'.") try: - if 'update_task' in self.__dict__ and self.update_task.done() is False: + if "update_task" in self.__dict__ and self.update_task.done() is False: self.update_task.cancel() except Exception as e: LOG.error(f"Failed to close status queue: '{procId}'. {e}") @@ -290,9 +280,10 @@ class Download: if self.tempKeep is True or not self.tempPath: return - if self.info.status != 'finished' and self.is_live: + if self.info.status != "finished" and self.is_live: LOG.warning( - f'Keeping live temp folder [{self.tempPath}], as the reported status is not finished [{self.info.status}].') + f"Keeping live temp folder [{self.tempPath}], as the reported status is not finished [{self.info.status}]." + ) return if not os.path.exists(self.tempPath): @@ -300,10 +291,11 @@ class Download: if self.tempPath == self.temp_dir: LOG.warning( - f'Attempted to delete video temp folder: {self.tempPath}, but it is the same as main temp folder.') + f"Attempted to delete video temp folder: {self.tempPath}, but it is the same as main temp folder." + ) return - LOG.info(f'Deleting Temp folder: {self.tempPath}') + LOG.info(f"Deleting Temp folder: {self.tempPath}") shutil.rmtree(self.tempPath, ignore_errors=True) async def progress_update(self): @@ -314,54 +306,54 @@ class Download: try: self.update_task = asyncio.get_running_loop().run_in_executor(None, self.status_queue.get) except asyncio.CancelledError: - LOG.debug(f'Closing progress update for: {self.info._id=}.') + LOG.debug(f"Closing progress update for: {self.info._id=}.") return status = await self.update_task if status is None or status.__class__ is Terminator: - LOG.debug(f'Closing progress update for: {self.info._id=}.') + LOG.debug(f"Closing progress update for: {self.info._id=}.") return - if status.get('id') != self.id or len(status) < 2: + if status.get("id") != self.id or len(status) < 2: continue if self.debug: - LOG.debug(f'Status Update: {self.info._id=} {status=}') + LOG.debug(f"Status Update: {self.info._id=} {status=}") if isinstance(status, str): asyncio.create_task(self.emitter.updated(dl=self.info), name=f"emitter-u-{self.id}") continue - self.tmpfilename = status.get('tmpfilename') + self.tmpfilename = status.get("tmpfilename") - if 'filename' in status: - self.info.filename = os.path.relpath(status.get('filename'), self.download_dir) + if "filename" in status: + self.info.filename = os.path.relpath(status.get("filename"), self.download_dir) - if os.path.exists(status.get('filename')): - self.info.file_size = os.path.getsize(status.get('filename')) + if os.path.exists(status.get("filename")): + self.info.file_size = os.path.getsize(status.get("filename")) - self.info.status = status.get('status', self.info.status) - self.info.msg = status.get('msg') + self.info.status = status.get("status", self.info.status) + self.info.msg = status.get("msg") - if self.info.status == 'error' and 'error' in status: - self.info.error = status.get('error') + if self.info.status == "error" and "error" in status: + self.info.error = status.get("error") asyncio.create_task( - self.emitter.error(message=self.info.error, data=self.info), - name=f"emitter-e-{self.id}") + self.emitter.error(message=self.info.error, data=self.info), name=f"emitter-e-{self.id}" + ) - if 'downloaded_bytes' in status: - total = status.get('total_bytes') or status.get('total_bytes_estimate') + if "downloaded_bytes" in status: + total = status.get("total_bytes") or status.get("total_bytes_estimate") if total: - self.info.percent = status['downloaded_bytes'] / total * 100 + self.info.percent = status["downloaded_bytes"] / total * 100 self.info.total_bytes = total - self.info.speed = status.get('speed') - self.info.eta = status.get('eta') + self.info.speed = status.get("speed") + self.info.eta = status.get("eta") - if self.info.status == 'finished' and 'filename' in status: + if self.info.status == "finished" and "filename" in status: try: - self.info.file_size = os.path.getsize(status.get('filename')) + self.info.file_size = os.path.getsize(status.get("filename")) except FileNotFoundError: LOG.warning(f'File not found: {status.get("filename")}') self.info.file_size = None diff --git a/app/library/DownloadQueue.py b/app/library/DownloadQueue.py index f64939d7..d6a2137d 100644 --- a/app/library/DownloadQueue.py +++ b/app/library/DownloadQueue.py @@ -1,18 +1,20 @@ import asyncio -from email.utils import formatdate import json import logging import os import time -import yt_dlp +from email.utils import formatdate from sqlite3 import Connection -from .config import Config -from .Download import Download -from .ItemDTO import ItemDTO -from .DataStore import DataStore -from .Utils import calcDownloadPath, ExtractInfo, isDownloaded, mergeConfig + +import yt_dlp + from .AsyncPool import AsyncPool +from .config import Config +from .DataStore import DataStore +from .Download import Download from .Emitter import Emitter +from .ItemDTO import ItemDTO +from .Utils import ExtractInfo, calcDownloadPath, isDownloaded, mergeConfig LOG = logging.getLogger("DownloadQueue") TYPE_DONE: str = "done" @@ -20,12 +22,8 @@ TYPE_QUEUE: str = "queue" class DownloadQueue: - config: Config = None - emitter: Emitter = None - queue: DataStore = None - done: DataStore = None - event: asyncio.Event = None - pool: AsyncPool = None + event: asyncio.Event | None = None + pool: AsyncPool | None = None def __init__(self, emitter: Emitter, connection: Connection): self.config = Config.get_instance() diff --git a/app/library/Emitter.py b/app/library/Emitter.py index 8bc42d15..d77a684c 100644 --- a/app/library/Emitter.py +++ b/app/library/Emitter.py @@ -1,48 +1,48 @@ - import asyncio import logging -LOG = logging.getLogger('Emitter') +LOG = logging.getLogger("Emitter") class Emitter: """ This class is used to emit events to the registered emitters. """ + emitters: list[callable] = [] def add_emitter(self, emitter: callable): """ - Add an emitter to the list of emitters. + Add an emitter to the list of emitters. - Args: - emitter (function): The emitter function. The function must return a coroutine or None. + Args: + emitter (function): The emitter function. The function must return a coroutine or None. """ self.emitters.append(emitter) async def added(self, dl: dict, **kwargs): - await self.emit('added', dl, **kwargs) + await self.emit("added", dl, **kwargs) async def updated(self, dl: dict, **kwargs): - await self.emit('updated', dl, **kwargs) + await self.emit("updated", dl, **kwargs) async def completed(self, dl: dict, **kwargs): - await self.emit('completed', dl, **kwargs) + await self.emit("completed", dl, **kwargs) async def canceled(self, id: str, dl: dict = None, **kwargs): - await self.emit('canceled', id, **kwargs) + await self.emit("canceled", id, **kwargs) async def cleared(self, id: str, dl: dict = None, **kwargs): - await self.emit('cleared', id, **kwargs) + await self.emit("cleared", id, **kwargs) async def error(self, message: str, data: dict = {}, **kwargs): - msg = {'status': 'error', 'message': message, 'data': {}} + msg = {"status": "error", "message": message, "data": {}} if data: - msg.update({'data': data}) - await self.emit('error', msg, **kwargs) + msg.update({"data": data}) + await self.emit("error", msg, **kwargs) async def warning(self, message: str, **kwargs): - await self.emit('error', message, **kwargs) + await self.emit("error", message, **kwargs) async def emit(self, event: str, data, **kwargs): tasks = [] diff --git a/app/library/HttpAPI.py b/app/library/HttpAPI.py index 3a38f20f..141c8d4b 100644 --- a/app/library/HttpAPI.py +++ b/app/library/HttpAPI.py @@ -1,44 +1,41 @@ import base64 -from datetime import datetime import functools +import hmac import json +import logging import os import time +from datetime import datetime +from pathlib import Path + import httpx +import magic +from aiohttp import web +from aiohttp.web import Request, RequestHandler, Response + +from .common import common from .config import Config from .DownloadQueue import DownloadQueue -from aiohttp import web -from aiohttp.web import Request, Response, RequestHandler -from .Playlist import Playlist +from .Emitter import Emitter +from .encoder import Encoder from .M3u8 import M3u8 +from .Playlist import Playlist from .Segments import Segments from .Subtitle import Subtitle -import logging -import magic -from .common import common -from pathlib import Path -from .encoder import Encoder -from .Emitter import Emitter -from .Utils import validate_url +from .Utils import validate_url, StreamingError -LOG = logging.getLogger('app') +LOG = logging.getLogger("http_api") MIME = magic.Magic(mime=True) class HttpAPI(common): - config: Config = None - encoder: Encoder = None - routes: web.RouteTableDef = None - queue: DownloadQueue = None - rootPath: str = None - staticHolder: dict = {} extToMime: dict = { - '.html': 'text/html', - '.css': 'text/css', - '.js': 'application/javascript', - '.json': 'application/json', - '.ico': 'image/x-icon', + ".html": "text/html", + ".css": "text/css", + ".js": "application/javascript", + ".json": "application/json", + ".ico": "image/x-icon", } def __init__(self, queue: DownloadQueue, emitter: Emitter, encoder: Encoder): @@ -57,13 +54,16 @@ class HttpAPI(common): """ Decorator to mark a method as an HTTP route handler. """ + def decorator(func): @functools.wraps(func) async def wrapper(*args, **kwargs): return await func(*args, **kwargs) + wrapper._http_method = method.upper() wrapper._http_path = path return wrapper + return decorator async def staticFile(self, req: Request) -> Response: @@ -72,22 +72,26 @@ class HttpAPI(common): """ path = req.path if req.path not in self.staticHolder: - return web.json_response({"error": "File not found.", "file": path}, status=404) + return web.json_response({"error": "File not found.", "file": path}, status=web.HTTPNotFound.status_code) item: dict = self.staticHolder[req.path] - return web.Response(body=item.get('content'), headers={ - 'Pragma': 'public', - 'Cache-Control': 'public, max-age=31536000', - 'Content-Type': item.get('content_type'), - 'X-Via': 'memory' if not item.get('file', None) else 'disk', - }) + return web.Response( + body=item.get("content"), + headers={ + "Pragma": "public", + "Cache-Control": "public, max-age=31536000", + "Content-Type": item.get("content_type"), + "X-Via": "memory" if not item.get("file", None) else "disk", + }, + status=web.HTTPOk.status_code, + ) def preloadStatic(self, app: web.Application): """ Preload static files from the ui/exported folder. """ - staticDir = os.path.join(self.rootPath, 'ui', 'exported') + staticDir = os.path.join(self.rootPath, "ui", "exported") if not os.path.exists(staticDir): raise ValueError(f"Could not find the frontend UI static assets. '{staticDir}'.") @@ -95,25 +99,25 @@ class HttpAPI(common): for root, _, files in os.walk(staticDir): for file in files: - if file.endswith('.map'): + if file.endswith(".map"): continue file = os.path.join(root, file) urlPath = f"{self.config.url_prefix}{file.replace(f'{staticDir}/', '')}" - content = open(file, 'rb').read() + content = open(file, "rb").read() contentType = self.extToMime.get(os.path.splitext(file)[1], MIME.from_file(file)) - self.staticHolder[urlPath] = {'content': content, 'content_type': contentType} + self.staticHolder[urlPath] = {"content": content, "content_type": contentType} LOG.debug(f"Preloading '{urlPath}'.") app.router.add_get(urlPath, self.staticFile) preloaded += 1 - if urlPath.endswith('/index.html') and urlPath != '/index.html': - parentSlash = urlPath.replace('/index.html', '/') - parentNoSlash = urlPath.replace('/index.html', '') - self.staticHolder[parentSlash] = {'content': content, 'content_type': contentType} - self.staticHolder[parentNoSlash] = {'content': content, 'content_type': contentType} + if urlPath.endswith("/index.html") and urlPath != "/index.html": + parentSlash = urlPath.replace("/index.html", "/") + parentNoSlash = urlPath.replace("/index.html", "") + self.staticHolder[parentSlash] = {"content": content, "content_type": contentType} + self.staticHolder[parentNoSlash] = {"content": content, "content_type": contentType} app.router.add_get(parentSlash, self.staticFile) app.router.add_get(parentNoSlash, self.staticFile) preloaded += 2 @@ -121,39 +125,37 @@ class HttpAPI(common): if preloaded < 1: raise ValueError(f"Could not find the frontend UI static assets. '{staticDir}'.") - LOG.info(f"Preloaded {preloaded} static files.") + LOG.info(f"Preloaded '{preloaded}' static files.") def attach(self, app: web.Application): if self.config.auth_username and self.config.auth_password: app.middlewares.append(HttpAPI.basic_auth(self.config.auth_username, self.config.auth_password)) self.add_routes(app) - pass def add_routes(self, app: web.Application): for attr_name in dir(self): method = getattr(self, attr_name) - if hasattr(method, '_http_method') and hasattr(method, '_http_path'): + if hasattr(method, "_http_method") and hasattr(method, "_http_path"): http_path = method._http_path - if http_path.startswith('/') and self.config.url_prefix.endswith('/'): + if http_path.startswith("/") and self.config.url_prefix.endswith("/"): http_path = method._http_path[1:] self.routes.route(method._http_method, self.config.url_prefix + http_path)(method) async def on_prepare(request: Request, response: Response): - if 'Server' in response.headers: - del response.headers['Server'] + if "Server" in response.headers: + del response.headers["Server"] - if 'Origin' in request.headers: - response.headers['Access-Control-Allow-Origin'] = request.headers['Origin'] - response.headers['Access-Control-Allow-Headers'] = 'Content-Type' - response.headers['Access-Control-Allow-Methods'] = 'PATCH, PUT, POST, DELETE' + if "Origin" in request.headers: + response.headers["Access-Control-Allow-Origin"] = request.headers["Origin"] + response.headers["Access-Control-Allow-Headers"] = "Content-Type" + response.headers["Access-Control-Allow-Methods"] = "PATCH, PUT, POST, DELETE" - if self.config.url_prefix != '/': - self.routes.route('GET', '/')(lambda _: web.HTTPFound(self.config.url_prefix)) + if self.config.url_prefix != "/": + self.routes.route("GET", "/")(lambda _: web.HTTPFound(self.config.url_prefix)) self.routes.get(self.config.url_prefix[:-1])(lambda _: web.HTTPFound(self.config.url_prefix)) - # add static files. self.routes.static(f"{self.config.url_prefix}download/", self.config.download_path) self.preloadStatic(app) @@ -161,54 +163,66 @@ class HttpAPI(common): app.add_routes(self.routes) app.on_response_prepare.append(on_prepare) except ValueError as e: - if 'ui/exported' in str(e): + if "ui/exported" in str(e): raise RuntimeError(f"Could not find the frontend UI static assets. '{e}'.") from e raise e def basic_auth(username: str, password: str): @web.middleware async def middleware_handler(request: Request, handler: RequestHandler) -> Response: - auth_header = request.headers.get('Authorization') + auth_header = request.headers.get("Authorization") if auth_header is None: - return web.Response(status=401, headers={ - 'WWW-Authenticate': 'Basic realm="Authorization Required."', - }, text='Unauthorized.') + return web.json_response( + status=web.HTTPUnauthorized.status_code, + headers={ + "WWW-Authenticate": 'Basic realm="Authorization Required."', + }, + data={"error": "Authorization Required."}, + ) - auth_type, encoded_credentials = auth_header.split(' ', 1) + auth_type, encoded_credentials = auth_header.split(" ", 1) - if 'basic' != auth_type.lower(): - return web.Response(status=401, text="Unsupported authentication method.") + if "basic" != auth_type.lower(): + return web.json_response( + data={"error": "Unsupported authentication method.", "method": auth_type}, + status=web.HTTPUnauthorized.status_code, + ) - decoded_credentials = base64.b64decode(encoded_credentials).decode('utf-8') - user_name, _, user_password = decoded_credentials.partition(':') + decoded_credentials = base64.b64decode(encoded_credentials).decode("utf-8") + user_name, _, user_password = decoded_credentials.partition(":") - if user_name != username or user_password != password: - return web.Response(status=401, text='Unauthorized (Invalid credentials).') + user_match = hmac.compare_digest(user_name, username) + pass_match = hmac.compare_digest(user_password, password) + + if not (user_match and pass_match): + return web.json_response( + data={"error": "Unauthorized (Invalid credentials)."}, status=web.HTTPUnauthorized.status_code + ) return await handler(request) return middleware_handler - @route('GET', 'ping') + @route("GET", "ping") async def ping(self, _) -> Response: await self.queue.test() - return web.Response(text='pong') + return web.json_response(data={"status": "pong"}, status=web.HTTPOk.status_code) - @route('POST', 'add') + @route("POST", "add") async def add_url(self, request: Request) -> Response: post = await request.json() - url: str = post.get('url') - preset: str = post.get('preset', 'default') + url: str = post.get("url") + preset: str = post.get("preset", "default") if not url: - raise web.HTTPBadRequest() + return web.json_response(data={"error": "url is required."}, status=web.HTTPBadRequest.status_code) - folder: str = post.get('folder') - ytdlp_cookies: str = post.get('ytdlp_cookies') - ytdlp_config: dict = post.get('ytdlp_config') - output_template: str = post.get('output_template') + folder: str = post.get("folder") + ytdlp_cookies: str = post.get("ytdlp_cookies") + ytdlp_config: dict | None = post.get("ytdlp_config") + output_template: str = post.get("output_template") if ytdlp_config is None: ytdlp_config = {} @@ -218,83 +232,92 @@ class HttpAPI(common): folder=folder, ytdlp_cookies=ytdlp_cookies, ytdlp_config=ytdlp_config, - output_template=output_template + output_template=output_template, ) - return web.Response(text=self.encoder.encode(status)) + return web.json_response(data=status, status=web.HTTPOk.status_code, dumps=self.encoder.encode) - @route('GET', 'tasks') + @route("GET", "tasks") async def tasks(self, _: Request) -> Response: - tasks_file: str = os.path.join(self.config.config_path, 'tasks.json') + tasks_file: str = os.path.join(self.config.config_path, "tasks.json") if not os.path.exists(tasks_file): - return web.json_response({"error": "No tasks defined."}, status=404) + return web.json_response({"error": "No tasks defined."}, status=web.HTTPNotFound.status_code) try: - with open(tasks_file, 'r') as f: + with open(tasks_file, "r") as f: tasks = json.load(f) except Exception as e: - return web.json_response({"error": str(e)}, status=500) + LOG.exception(e) + return web.json_response({"error": "Failed to load tasks."}, status=web.HTTPInternalServerError.status_code) - return web.json_response(tasks) + return web.json_response(data=tasks, status=web.HTTPOk.status_code, dumps=self.encoder.encode) - @route('POST', 'add_batch') + @route("POST", "add_batch") async def add_batch(self, request: Request) -> Response: status = {} post = await request.json() if not isinstance(post, list): - raise web.HTTPBadRequest(text='Invalid request body expecting list with dicts.') + return web.json_response( + data={"error": "Invalid request body expecting list with dicts."}, + status=web.HTTPBadRequest.status_code, + ) for item in post: if not isinstance(item, dict): - raise web.HTTPBadRequest(text='Invalid request body expecting list with dicts.') + return web.json_response( + data={"error": "Invalid request body expecting list with dicts."}, + status=web.HTTPBadRequest.status_code, + ) - if not item.get('url'): - raise web.HTTPBadRequest(text='url is required.') + if not item.get("url"): + return web.json_response( + data={"error": "url is required.", "data": post}, status=web.HTTPBadRequest.status_code + ) - status[item.get('url')] = await self.add( - url=item.get('url'), - preset=item.get('preset', 'default'), - folder=item.get('folder'), - ytdlp_cookies=item.get('ytdlp_cookies'), - ytdlp_config=item.get('ytdlp_config'), - output_template=item.get('output_template') + for item in post: + status[item.get("url")] = await self.add( + url=item.get("url"), + preset=item.get("preset", "default"), + folder=item.get("folder"), + ytdlp_cookies=item.get("ytdlp_cookies"), + ytdlp_config=item.get("ytdlp_config"), + output_template=item.get("output_template"), ) - return web.Response(text=self.encoder.encode(status)) + return web.json_response(data=status, status=web.HTTPOk.status_code, dumps=self.encoder.encode) - @route('DELETE', 'delete') + @route("DELETE", "delete") async def delete(self, request: Request) -> Response: post = await request.json() - ids = post.get('ids') - where = post.get('where') + ids = post.get("ids") + where = post.get("where") - if not ids or where not in ['queue', 'done']: - raise web.HTTPBadRequest() + if not ids or where not in ["queue", "done"]: + return web.json_response( + data={"error": "ids and where are required."}, status=web.HTTPBadRequest.status_code + ) - status: dict[str, str] = {} + return web.json_response( + data=await (self.queue.cancel(ids) if where == "queue" else self.queue.clear(ids)), + status=web.HTTPOk.status_code, + dumps=self.encoder.encode, + ) - status = await (self.queue.cancel(ids) if where == 'queue' else self.queue.clear(ids)) - - return web.Response(text=self.encoder.encode(status)) - - @route('POST', 'item/{id}') + @route("POST", "item/{id}") async def update_item(self, request: Request) -> Response: - id: str = request.match_info.get('id') + id: str = request.match_info.get("id") if not id: - raise web.HTTPBadRequest(text='id is required.') + return web.json_response(data={"error": "id is required."}, status=web.HTTPBadRequest.status_code) item = self.queue.done.getById(id) if not item: - raise web.HTTPNotFound(text='Item not found.') + return web.json_response(data={"error": "item not found."}, status=web.HTTPNotFound.status_code) - try: - post = await request.json() - if not post: - raise web.HTTPBadRequest(text='no data provided.') - except Exception as e: - raise web.HTTPBadRequest(text=str(e)) + post = await request.json() + if not post: + return web.json_response(data={"error": "no data provided."}, status=web.HTTPBadRequest.status_code) updated = False @@ -307,30 +330,33 @@ class HttpAPI(common): updated = True setattr(item.info, k, v) - LOG.info(f"Updated '{k}' to '{v}' for '{item.info.id}'") + LOG.debug(f"Updated '{k}' to '{v}' for '{item.info.id}'") - status = 200 if updated else 304 if updated: self.queue.done.put(item) - await self.notifier.emit('update', item.info) + await self.emitter.emit("update", item.info) - return web.Response(text=self.encoder.encode(item.info), status=status) + return web.json_response( + data=item.info, + status=web.HTTPOk.status_code if updated else web.HTTPNotModified.status_code, + dumps=self.encoder.encode, + ) - @route('GET', 'history') + @route("GET", "history") async def history(self, _: Request) -> Response: - history = {'done': [], 'queue': []} + data: dict = {"queue": [], "history": []} for _, v in self.queue.queue.saved_items(): - history['queue'].append(v) + data["queue"].append(v) for _, v in self.queue.done.saved_items(): - history['done'].append(v) + data["history"].append(v) - return web.Response(text=self.encoder.encode(history)) + return web.json_response(data=data, status=web.HTTPOk.status_code, dumps=self.encoder.encode) - @route('GET', 'workers') + @route("GET", "workers") async def workers(self, _) -> Response: if self.queue.pool is None: - return web.json_response({"error": "Worker pool not initialized."}, status=404) + return web.json_response({"error": "Worker pool not initialized."}, status=web.HTTPNotFound.status_code) status = self.queue.pool.get_workers_status() @@ -338,231 +364,271 @@ class HttpAPI(common): for worker in status: worker_status = status.get(worker) + data.append( + { + "id": worker, + "data": {"status": "Waiting for download."} if worker_status is None else worker_status, + } + ) - data.append({ - 'id': worker, - 'data': {"status": 'Waiting for download.'} if worker_status is None else worker_status, - }) + return web.json_response( + data={ + "open": self.queue.pool.has_open_workers(), + "count": self.queue.pool.get_available_workers(), + "workers": data, + }, + status=web.HTTPOk.status_code, + dumps=lambda obj: json.dumps(obj, default=lambda o: f"<>"), + ) - def safe_serialize(obj): - def default(o): return f"<>" - return json.dumps(obj, default=default) - - return web.Response(text=safe_serialize({ - 'open': self.queue.pool.has_open_workers(), - 'count': self.queue.pool.get_available_workers(), - 'workers': data, - }), headers={ - 'Content-Type': 'application/json', - }) - - @route('POST', 'workers') + @route("POST", "workers") async def restart_pool(self, _) -> Response: if self.queue.pool is None: - return web.json_response({"error": "Worker pool not initialized."}, status=404) + return web.json_response({"error": "Worker pool not initialized."}, status=web.HTTPNotFound.status_code) self.queue.pool.start() - return web.Response() + return web.json_response({"message": "Workers pool being restarted."}, status=web.HTTPOk.status_code) - @route('PATCH', 'workers/{id}') + @route("PATCH", "workers/{id}") async def restart_worker(self, request: Request) -> Response: - id: str = request.match_info.get('id') + id: str = request.match_info.get("id") if not id: - raise web.HTTPBadRequest(text='worker id is required.') + return web.json_response({"error": "worker id is required."}, status=web.HTTPBadRequest.status_code) if self.queue.pool is None: - return web.json_response({"error": "Worker pool not initialized."}, status=404) + return web.json_response({"error": "Worker pool not initialized."}, status=web.HTTPNotFound.status_code) - status = await self.queue.pool.restart(id, 'requested by user.') + status = await self.queue.pool.restart(id, "requested by user.") - return web.json_response({"status": "restarted" if status else "in_error_state"}) + return web.json_response({"status": "restarted" if status else "in_error_state"}, status=web.HTTPOk.status_code) - @route('delete', 'workers/{id}') + @route("delete", "workers/{id}") async def stop_worker(self, request: Request) -> Response: - id: str = request.match_info.get('id') + id: str = request.match_info.get("id") if not id: - raise web.HTTPBadRequest(text='worker id is required.') + raise web.HTTPBadRequest(text="worker id is required.") if self.queue.pool is None: - return web.json_response({"error": "Worker pool not initialized."}, status=404) + return web.json_response({"error": "Worker pool not initialized."}, status=web.HTTPNotFound.status_code) - status = await self.queue.pool.stop(id, 'requested by user.') + status = await self.queue.pool.stop(id, "requested by user.") - return web.json_response({"status": "stopped" if status else "in_error_state"}) + return web.json_response({"status": "stopped" if status else "in_error_state"}, status=web.HTTPOk.status_code) - @route('GET', 'player/playlist/{file:.*}.m3u8') + @route("GET", "player/playlist/{file:.*}.m3u8") async def playlist(self, request: Request) -> Response: - file: str = request.match_info.get('file') + file: str = request.match_info.get("file") if not file: - raise web.HTTPBadRequest(text='file is required.') + raise web.HTTPBadRequest(text="file is required.") try: text = await Playlist(url=f"{self.config.url_host}{self.config.url_prefix}").make( - download_path=self.config.download_path, - file=file + download_path=self.config.download_path, file=file ) if isinstance(text, Response): return text - except Exception as e: - return web.HTTPNotFound(text=str(e)) + except StreamingError as e: + return web.json_response(data={"error": str(e)}, status=web.HTTPNotFound.status_code) - return web.Response(text=text, headers={ - 'Content-Type': 'application/x-mpegURL', - 'Cache-Control': 'no-cache', - 'Access-Control-Max-Age': "300", - }) + return web.Response( + text=text, + headers={ + "Content-Type": "application/x-mpegURL", + "Cache-Control": "no-cache", + "Access-Control-Max-Age": "300", + }, + status=web.HTTPOk.status_code, + ) - @route('GET', 'player/m3u8/{mode}/{file:.*}.m3u8') + @route("GET", "player/m3u8/{mode}/{file:.*}.m3u8") async def m3u8(self, request: Request) -> Response: - file: str = request.match_info.get('file') - mode: str = request.match_info.get('mode') + file: str = request.match_info.get("file") + mode: str = request.match_info.get("mode") - if mode not in ['video', 'subtitle']: - raise web.HTTPBadRequest(text='Only video and subtitle modes are supported.') + if mode not in ["video", "subtitle"]: + return web.json_response( + data={"error": "Only video and subtitle modes are supported."}, status=web.HTTPBadRequest.status_code + ) if not file: - raise web.HTTPBadRequest(text='file is required.') + return web.json_response(data={"error": "file is required"}, status=web.HTTPBadRequest.status_code) - duration = request.query.get('duration', None) + duration = request.query.get("duration", None) - if 'subtitle' in mode: + if "subtitle" in mode: if not duration: - raise web.HTTPBadRequest(text='duration is required.') + return web.json_response(data={"error": "duration is required."}, status=web.HTTPBadRequest.status_code) duration = float(duration) try: cls = M3u8(f"{self.config.url_host}{self.config.url_prefix}") - if 'subtitle' in mode: + if "subtitle" in mode: text = await cls.make_subtitle(self.config.download_path, file, duration) else: text = await cls.make_stream(self.config.download_path, file) - except Exception as e: - return web.HTTPNotFound(text=str(e)) + except StreamingError as e: + LOG.exception(e) + return web.json_response(data={"error": str(e)}, status=web.HTTPNotFound.status_code) - return web.Response(text=text, headers={ - 'Content-Type': 'application/x-mpegURL', - 'Cache-Control': 'no-cache', - 'Access-Control-Max-Age': "300", - }) + return web.Response( + text=text, + headers={ + "Content-Type": "application/x-mpegURL", + "Cache-Control": "no-cache", + "Access-Control-Max-Age": "300", + }, + status=web.HTTPOk.status_code, + ) - @route('GET', r'player/segments/{segment:\d+}/{file:.*}.ts') + @route("GET", r"player/segments/{segment:\d+}/{file:.*}.ts") async def segments(self, request: Request) -> Response: - file: str = request.match_info.get('file') - segment: int = request.match_info.get('segment') - sd: int = request.query.get('sd') - vc: int = int(request.query.get('vc', 0)) - ac: int = int(request.query.get('ac', 0)) + file: str = request.match_info.get("file") + segment: int = request.match_info.get("segment") + sd: int = request.query.get("sd") + vc: int = int(request.query.get("vc", 0)) + ac: int = int(request.query.get("ac", 0)) file_path: str = os.path.normpath(os.path.join(self.config.download_path, file)) if not file_path.startswith(self.config.download_path): - raise web.HTTPBadRequest(text='Invalid file path.') + return web.json_response(data={"error": "Invalid file path."}, status=web.HTTPBadRequest.status_code) if request.if_modified_since: + lastMod = time.strftime( + "%a, %d %b %Y %H:%M:%S GMT", datetime.fromtimestamp(os.path.getmtime(file_path)).timetuple() + ) if os.path.exists(file_path) and request.if_modified_since.timestamp() == os.path.getmtime(file_path): - return web.Response(status=304) + return web.Response(status=web.HTTPNotModified.status_code, headers={"Last-Modified": lastMod}) if not file: - raise web.HTTPBadRequest(text='file is required') + return web.json_response(data={"error": "file is required"}, status=web.HTTPBadRequest.status_code) if not segment: - raise web.HTTPBadRequest(text='segment is required') + return web.json_response(data={"error": "segment id is required."}, status=web.HTTPBadRequest.status_code) segmenter = Segments( index=int(segment), - duration=float('{:.6f}'.format(float(sd if sd else M3u8.duration))), + duration=float("{:.6f}".format(float(sd if sd else M3u8.duration))), vconvert=True if vc == 1 else False, - aconvert=True if ac == 1 else False + aconvert=True if ac == 1 else False, ) - return web.Response(body=await segmenter.stream(path=self.config.download_path, file=file), headers={ - 'Content-Type': 'video/mpegts', - 'X-Accel-Buffering': 'no', - 'Access-Control-Allow-Origin': '*', - 'Pragma': 'public', - 'Cache-Control': f"public, max-age={time.time() + 31536000}", - 'Last-Modified': time.strftime('%a, %d %b %Y %H:%M:%S GMT', datetime.fromtimestamp(os.path.getmtime(file_path)).timetuple()), - 'Expires': time.strftime('%a, %d %b %Y %H:%M:%S GMT', datetime.fromtimestamp(time.time() + 31536000).timetuple()), - }) + return web.Response( + body=await segmenter.stream(path=self.config.download_path, file=file), + headers={ + "Content-Type": "video/mpegts", + "X-Accel-Buffering": "no", + "Access-Control-Allow-Origin": "*", + "Pragma": "public", + "Cache-Control": f"public, max-age={time.time() + 31536000}", + "Last-Modified": time.strftime( + "%a, %d %b %Y %H:%M:%S GMT", datetime.fromtimestamp(os.path.getmtime(file_path)).timetuple() + ), + "Expires": time.strftime( + "%a, %d %b %Y %H:%M:%S GMT", datetime.fromtimestamp(time.time() + 31536000).timetuple() + ), + }, + status=web.HTTPOk.status_code, + ) - @route('GET', 'player/subtitle/{file:.*}.vtt') + @route("GET", "player/subtitle/{file:.*}.vtt") async def subtitles(self, request: Request) -> Response: - file: str = request.match_info.get('file') + file: str = request.match_info.get("file") file_path: str = os.path.normpath(os.path.join(self.config.download_path, file)) if not file_path.startswith(self.config.download_path): - raise web.HTTPBadRequest(text='Invalid file path.') + return web.json_response(data={"error": "Invalid file path."}, status=web.HTTPBadRequest.status_code) if request.if_modified_since: - lastMod = time.strftime('%a, %d %b %Y %H:%M:%S GMT', datetime.fromtimestamp( - os.path.getmtime(file_path)).timetuple()) + lastMod = time.strftime( + "%a, %d %b %Y %H:%M:%S GMT", datetime.fromtimestamp(os.path.getmtime(file_path)).timetuple() + ) if os.path.exists(file_path) and request.if_modified_since.timestamp() == os.path.getmtime(file_path): - return web.Response(status=304, headers={'Last-Modified': lastMod}) + return web.Response(status=web.HTTPNotModified.status_code, headers={"Last-Modified": lastMod}) if not file: - raise web.HTTPBadRequest(text='file is required') + return web.json_response(data={"error": "file is required"}, status=web.HTTPBadRequest.status_code) - return web.Response(body=await Subtitle().make(path=self.config.download_path, file=file), headers={ - 'Content-Type': 'text/vtt; charset=UTF-8', - 'X-Accel-Buffering': 'no', - 'Access-Control-Allow-Origin': '*', - 'Pragma': 'public', - 'Cache-Control': f"public, max-age={time.time() + 31536000}", - 'Last-Modified': time.strftime('%a, %d %b %Y %H:%M:%S GMT', datetime.fromtimestamp(os.path.getmtime(file_path)).timetuple()), - 'Expires': time.strftime('%a, %d %b %Y %H:%M:%S GMT', datetime.fromtimestamp(time.time() + 31536000).timetuple()), - }) - - @route('OPTIONS', '/add') - async def add_cors(self, _: Request) -> Response: - return web.Response(text=self.encoder.encode({"status": "ok"})) - - @route('OPTIONS', '/delete') - async def delete_cors(self, _: Request) -> Response: - return web.Response(text=self.encoder.encode({"status": "ok"})) - - @route('GET', '/') - async def index(self, _) -> Response: - if '/index.html' not in self.staticHolder: - LOG.error("Static frontend files not found.") - return web.json_response({"error": "File not found.", "file": '/index.html'}, status=404) - - data = self.staticHolder['/index.html'] return web.Response( - body=data.get('content'), - content_type=data.get('content_type'), - charset='utf-8', - status=web.HTTPOk.status_code) + body=await Subtitle().make(path=self.config.download_path, file=file), + headers={ + "Content-Type": "text/vtt; charset=UTF-8", + "X-Accel-Buffering": "no", + "Access-Control-Allow-Origin": "*", + "Pragma": "public", + "Cache-Control": f"public, max-age={time.time() + 31536000}", + "Last-Modified": time.strftime( + "%a, %d %b %Y %H:%M:%S GMT", datetime.fromtimestamp(os.path.getmtime(file_path)).timetuple() + ), + "Expires": time.strftime( + "%a, %d %b %Y %H:%M:%S GMT", datetime.fromtimestamp(time.time() + 31536000).timetuple() + ), + }, + status=web.HTTPOk.status_code, + ) - @route('GET', '/thumbnail') - async def get_thumbnail(self, request: Request) -> dict: - url = request.query.get('url') + @route("OPTIONS", "/add") + async def add_cors(self, _: Request) -> Response: + return web.json_response(data={"status": "ok"}, status=web.HTTPOk.status_code) + + @route("OPTIONS", "/delete") + async def delete_cors(self, _: Request) -> Response: + return web.json_response(data={"status": "ok"}, status=web.HTTPOk.status_code) + + @route("GET", "/") + async def index(self, _) -> Response: + if "/index.html" not in self.staticHolder: + LOG.error("Static frontend files not found.") + return web.json_response( + data={"error": "File not found.", "file": "/index.html"}, status=web.HTTPNotFound.status_code + ) + + data = self.staticHolder["/index.html"] + return web.Response( + body=data.get("content"), + content_type=data.get("content_type"), + charset="utf-8", + status=web.HTTPOk.status_code, + ) + + @route("GET", "/thumbnail") + async def get_thumbnail(self, request: Request) -> Response: + url = request.query.get("url") if not url: - return web.json_response({"error": "URL is required."}, status=400) + return web.json_response(data={"error": "URL is required."}, status=web.HTTPForbidden.status_code) try: validate_url(url) - except Exception as e: - return web.json_response({"error": str(e)}, status=400) + except ValueError as e: + return web.json_response(data={"error": str(e)}, status=web.HTTPForbidden.status_code) try: opts = { - 'proxy': self.config.ytdl_options.get('proxy', None), - 'headers': { - 'User-Agent': self.config.ytdl_options.get( - 'user_agent', - request.headers.get('User-Agent', f"YTPTube/{self.config.version}")), + "proxy": self.config.ytdl_options.get("proxy", None), + "headers": { + "User-Agent": self.config.ytdl_options.get( + "user_agent", request.headers.get("User-Agent", f"YTPTube/{self.config.version}") + ), }, } async with httpx.AsyncClient(**opts) as client: - LOG.info(f"Fetching thumbnail from '{url}'.") - response = await client.request(method='GET', url=url) - return web.Response(body=response.content, - headers={'Content-Type': response.headers.get('Content-Type'), - 'Pragma': 'public', 'Access-Control-Allow-Origin': '*', - 'Cache-Control': f"public, max-age={time.time() + 31536000}", - 'Expires': time.strftime( - '%a, %d %b %Y %H:%M:%S GMT', datetime.fromtimestamp( - time.time() + 31536000).timetuple()), }) + LOG.debug(f"Fetching thumbnail from '{url}'.") + response = await client.request(method="GET", url=url) + return web.Response( + body=response.content, + headers={ + "Content-Type": response.headers.get("Content-Type"), + "Pragma": "public", + "Access-Control-Allow-Origin": "*", + "Cache-Control": f"public, max-age={time.time() + 31536000}", + "Expires": time.strftime( + "%a, %d %b %Y %H:%M:%S GMT", datetime.fromtimestamp(time.time() + 31536000).timetuple() + ), + }, + ) except Exception as e: - LOG.error(f"Error fetching thumbnail from '{url}'. '{e}'") - return web.json_response({"error": str(e)}, status=500) + LOG.error(f"Error fetching thumbnail from '{url}'. '{e}'.") + LOG.exception(e) + return web.json_response( + data={"error": "failed to retrieve the thumbnail."}, status=web.HTTPInternalServerError.status_code + ) diff --git a/app/library/HttpSocket.py b/app/library/HttpSocket.py index ef0d457b..46fc0a14 100644 --- a/app/library/HttpSocket.py +++ b/app/library/HttpSocket.py @@ -1,34 +1,37 @@ import asyncio -from datetime import datetime import errno import functools +import logging import os import pty import shlex -from aiohttp import web +from datetime import datetime + import socketio -import logging +from aiohttp import web + +from .common import common from .config import Config from .DownloadQueue import DownloadQueue -from .common import common +from .Emitter import Emitter from .encoder import Encoder from .Utils import isDownloaded -from .Emitter import Emitter -LOG = logging.getLogger('socket') +LOG = logging.getLogger("socket_api") class HttpSocket(common): """ This class is used to handle WebSocket events. """ + config: Config = None sio: socketio.AsyncServer = None def __init__(self, queue: DownloadQueue, emitter: Emitter, encoder: Encoder): super().__init__(queue=queue, encoder=encoder) - self.sio = socketio.AsyncServer(cors_allowed_origins='*') + self.sio = socketio.AsyncServer(cors_allowed_origins="*") emitter.add_emitter(lambda event, data, **kwargs: self.sio.emit(event, encoder.encode(data), **kwargs)) self.config = Config.get_instance() @@ -40,41 +43,45 @@ class HttpSocket(common): """ Decorator to mark a method as a socket event. """ + @functools.wraps(func) async def wrapper(*args, **kwargs): return await func(*args, **kwargs) + wrapper._ws_event = func.__name__ return wrapper def attach(self, app: web.Application): - self.sio.attach(app, socketio_path=self.config.url_prefix + 'socket.io') + self.sio.attach(app, socketio_path=self.config.url_prefix + "socket.io") for attr_name in dir(self): method = getattr(self, attr_name) - if hasattr(method, '_ws_event'): + if hasattr(method, "_ws_event"): event = method._ws_event self.sio.on(event)(method) @ws_event async def cli_post(self, sid: str, data): if not data: - await self.emitter.emit('cli_close', {'exitcode': 0}, to=sid) + await self.emitter.emit("cli_close", {"exitcode": 0}, to=sid) return try: LOG.info(f"Cli command from client '{sid}'. '{data}'") - args = ['yt-dlp'] + shlex.split(data) + args = ["yt-dlp"] + shlex.split(data) _env = os.environ.copy() - _env.update({ - "TERM": "xterm-256color", - "LANG": "en_US.UTF-8", - "SHELL": "/bin/bash", - "LC_ALL": "en_US.UTF-8", - "PWD": self.config.download_path, - "FORCE_COLOR": "1", - "PYTHONUNBUFFERED": "1", - }) + _env.update( + { + "TERM": "xterm-256color", + "LANG": "en_US.UTF-8", + "SHELL": "/bin/bash", + "LC_ALL": "en_US.UTF-8", + "PWD": self.config.download_path, + "FORCE_COLOR": "1", + "PYTHONUNBUFFERED": "1", + } + ) master_fd, slave_fd = pty.openpty() @@ -94,7 +101,7 @@ class HttpSocket(common): async def read_pty(): loop = asyncio.get_running_loop() - buffer = b'' + buffer = b"" while True: try: chunk = await loop.run_in_executor(None, lambda: os.read(master_fd, 1024)) @@ -108,18 +115,16 @@ class HttpSocket(common): # No more output if buffer: # Emit any remaining partial line - await self.emitter.emit('cli_output', { - 'type': 'stdout', - 'line': buffer.decode('utf-8', errors='replace') - }) + await self.emitter.emit( + "cli_output", {"type": "stdout", "line": buffer.decode("utf-8", errors="replace")} + ) break buffer += chunk - *lines, buffer = buffer.split(b'\n') + *lines, buffer = buffer.split(b"\n") for line in lines: - await self.emitter.emit('cli_output', { - 'type': 'stdout', - 'line': line.decode('utf-8', errors='replace') - }) + await self.emitter.emit( + "cli_output", {"type": "stdout", "line": line.decode("utf-8", errors="replace")} + ) try: os.close(master_fd) except Exception as e: @@ -134,28 +139,33 @@ class HttpSocket(common): # Ensure reading is done await read_task - await self.emitter.emit('cli_close', {'exitcode': returncode}, to=sid) + await self.emitter.emit("cli_close", {"exitcode": returncode}, to=sid) except Exception as e: - LOG.error(f"CLI execute exception was thrown for client '{sid}'. {str(e)}") - await self.emitter.emit('cli_out1put', { - 'type': 'stderr', - 'line': str(e), - }, to=sid) - await self.emitter.emit('cli_close', {'exitcode': -1}, to=sid) + LOG.error(f"CLI execute exception was thrown for client '{sid}'.") + LOG.exception(e) + await self.emitter.emit( + "cli_out1put", + { + "type": "stderr", + "line": str(e), + }, + to=sid, + ) + await self.emitter.emit("cli_close", {"exitcode": -1}, to=sid) @ws_event async def add_url(self, sid: str, data: dict): - url: str = data.get('url') + url: str = data.get("url") if not url: - self.emitter.warning('No URL provided.', to=sid) + self.emitter.warning("No URL provided.", to=sid) return - preset: str = data.get('preset', 'default') - folder: str = data.get('folder') - ytdlp_cookies: str = data.get('ytdlp_cookies') - ytdlp_config: dict = data.get('ytdlp_config') - output_template: str = data.get('output_template') + preset: str = data.get("preset", "default") + folder: str = data.get("folder") + ytdlp_cookies: str = data.get("ytdlp_cookies") + ytdlp_config: dict | None = data.get("ytdlp_config") + output_template: str = data.get("output_template") if ytdlp_config is None: ytdlp_config = {} @@ -165,64 +175,64 @@ class HttpSocket(common): folder=folder, ytdlp_cookies=ytdlp_cookies, ytdlp_config=ytdlp_config, - output_template=output_template + output_template=output_template, ) - await self.emitter.emit('status', status, to=sid) + await self.emitter.emit("status", status, to=sid) @ws_event async def item_cancel(self, sid: str, id: str): if not id: - await self.emitter.warning('Invalid request.', to=sid) + await self.emitter.warning("Invalid request.", to=sid) return status: dict[str, str] = {} status = await self.queue.cancel([id]) - status.update({'identifier': id}) + status.update({"identifier": id}) - await self.emitter.emit('item_cancel', status) + await self.emitter.emit("item_cancel", status) @ws_event async def item_delete(self, sid: str, id: str): if not id: - await self.emitter.warning('Invalid request.', to=sid) + await self.emitter.warning("Invalid request.", to=sid) return status: dict[str, str] = {} status = await self.queue.clear([id]) - status.update({'identifier': id}) + status.update({"identifier": id}) - await self.emitter.emit('item_delete', status) + await self.emitter.emit("item_delete", status) @ws_event async def archive_item(self, sid: str, data: dict): - if not isinstance(data, dict) or 'url' not in data or not self.config.keep_archive: + if not isinstance(data, dict) or "url" not in data or not self.config.keep_archive: return - file: str = self.config.ytdl_options.get('download_archive', None) + file: str = self.config.ytdl_options.get("download_archive", None) if not file: return - exists, idDict = isDownloaded(file, data['url']) - if exists or 'archive_id' not in idDict or idDict['archive_id'] is None: + exists, idDict = isDownloaded(file, data["url"]) + if exists or "archive_id" not in idDict or idDict["archive_id"] is None: return - with open(file, 'a') as f: + with open(file, "a") as f: f.write(f"{idDict['archive_id']}\n") manual_archive = self.config.manual_archive if manual_archive: previouslyArchived = False if os.path.exists(manual_archive): - with open(manual_archive, 'r') as f: + with open(manual_archive, "r") as f: for line in f.readlines(): - if idDict['archive_id'] in line: + if idDict["archive_id"] in line: previouslyArchived = True break if not previouslyArchived: - with open(manual_archive, 'a') as f: + with open(manual_archive, "a") as f: f.write(f"{idDict['archive_id']} - at: {datetime.now().isoformat()}\n") LOG.info(f"Archiving url '{data['url']}' with id '{idDict['archive_id']}'.") @@ -238,8 +248,6 @@ class HttpSocket(common): # get download folder listing downloadPath: str = self.config.download_path - data['folders'] = [ - name for name in os.listdir(downloadPath) if os.path.isdir(os.path.join(downloadPath, name)) - ] + data["folders"] = [name for name in os.listdir(downloadPath) if os.path.isdir(os.path.join(downloadPath, name))] - await self.emitter.emit('initial_data', data, to=sid) + await self.emitter.emit("initial_data", data, to=sid) diff --git a/app/library/ItemDTO.py b/app/library/ItemDTO.py index afdff6f2..f4aee17a 100644 --- a/app/library/ItemDTO.py +++ b/app/library/ItemDTO.py @@ -1,47 +1,52 @@ -from email.utils import formatdate import json import time -from dataclasses import dataclass, field import uuid +from dataclasses import dataclass, field +from email.utils import formatdate @dataclass(kw_only=True) class ItemDTO: - _id: int = field(default_factory=lambda: str(uuid.uuid4()), init=False) + """ + ItemDTO is a data transfer object that represents a single item in the queue. + It contains all the required information for both the frontend and the backend to process the item. + """ + + _id: str = field(default_factory=lambda: str(uuid.uuid4()), init=False) error: str = None id: str title: str url: str - quality: str = None - format: str = None - thumbnail: str = None + quality: str | None = None + format: str | None = None + thumbnail: str | None = None preset: str = "default" folder: str - download_dir: str = None - temp_dir: str = None - status: str = None - ytdlp_cookies: str = None + download_dir: str | None = None + temp_dir: str | None = None + status: str | None = None + ytdlp_cookies: str | None = None ytdlp_config: dict = field(default_factory=dict) - output_template: str = None - output_template_chapter: str = None + output_template: str | None = None + output_template_chapter: str | None = None timestamp: float = time.time_ns() - is_live: bool = None + is_live: bool | None = None datetime: str = field(default_factory=lambda: str(formatdate(time.time()))) - live_in: str = None - file_size: int = None + live_in: str | None = None + file_size: int | None = None options: dict = field(default_factory=dict) # yt-dlp injected fields. - tmpfilename: str = None - filename: str = None - total_bytes: int = None - total_bytes_estimate: int = None - downloaded_bytes: int = None - msg: str = None - percent: int = None - speed: str = None - eta: str = None + tmpfilename: str | None = None + filename: str | None = None + total_bytes: int | None = None + total_bytes_estimate: int | None = None + downloaded_bytes: int | None = None + msg: str | None = None + percent: int | None = None + speed: str | None = None + eta: str | None = None def json(self) -> str: return json.dumps(self, default=lambda o: o.__dict__, sort_keys=True, indent=4) diff --git a/app/library/M3u8.py b/app/library/M3u8.py index 180ea6df..3b6c95db 100644 --- a/app/library/M3u8.py +++ b/app/library/M3u8.py @@ -1,37 +1,44 @@ import math import os from urllib.parse import quote -from .Utils import calcDownloadPath +from .Utils import calcDownloadPath, StreamingError from .ffprobe import FFProbe class M3u8: - ok_vcodecs: tuple = ('h264', 'x264', 'avc',) - ok_acodecs: tuple = ('aac', 'm4a', 'mp3',) - - url: str = None duration: float = 6.000000 + ok_vcodecs: tuple = ( + "h264", + "x264", + "avc", + ) + ok_acodecs: tuple = ( + "aac", + "m4a", + "mp3", + ) + def __init__(self, url: str, segment_duration: float = None): self.url = url self.duration = float(segment_duration) if segment_duration is not None else self.duration - async def make_stream(self, download_path: str, file: str): + async def make_stream(self, download_path: str, file: str) -> str: realFile: str = calcDownloadPath(basePath=download_path, folder=file, createPath=False) if not os.path.exists(realFile): - raise Exception(f"File '{realFile}' does not exist.") + raise StreamingError(f"File '{realFile}' does not exist.") try: ffprobe = FFProbe(realFile) await ffprobe.run() - except UnicodeDecodeError as e: + except UnicodeDecodeError: pass - if not 'duration' in ffprobe.metadata: - raise Exception(f"Unable to get '{realFile}' duration.") + if "duration" not in ffprobe.metadata: + raise StreamingError(f"Unable to get '{realFile}' play duration.") - duration: float = float(ffprobe.metadata.get('duration')) + duration: float = float(ffprobe.metadata.get("duration")) m3u8 = [] @@ -41,7 +48,7 @@ class M3u8: m3u8.append("#EXT-X-MEDIA-SEQUENCE:0") m3u8.append("#EXT-X-PLAYLIST-TYPE:VOD") - segmentSize: float = '{:.6f}'.format(self.duration) + segmentSize: float = "{:.6f}".format(self.duration) splits: int = math.ceil(duration / self.duration) segmentParams: dict = {} @@ -49,33 +56,33 @@ class M3u8: for stream in ffprobe.streams: if stream.is_video(): if stream.codec_name not in self.ok_vcodecs: - segmentParams['vc'] = 1 + segmentParams["vc"] = 1 if stream.is_audio(): if stream.codec_name not in self.ok_acodecs: - segmentParams['ac'] = 1 + segmentParams["ac"] = 1 for i in range(splits): if (i + 1) == splits: - segmentSize = '{:.6f}'.format(duration - (i * self.duration)) - segmentParams.update({'sd': segmentSize}) + segmentSize = "{:.6f}".format(duration - (i * self.duration)) + segmentParams.update({"sd": segmentSize}) m3u8.append(f"#EXTINF:{segmentSize},") url = f"{self.url}player/segments/{i}/{quote(file)}.ts" if len(segmentParams) > 0: - url += '?'+'&'.join([f"{key}={value}" for key, value in segmentParams.items()]) + url += "?" + "&".join([f"{key}={value}" for key, value in segmentParams.items()]) m3u8.append(url) m3u8.append("#EXT-X-ENDLIST") - return '\n'.join(m3u8) + return "\n".join(m3u8) - async def make_subtitle(self, download_path: str, file: str, duration: float): + async def make_subtitle(self, download_path: str, file: str, duration: float) -> str: realFile: str = calcDownloadPath(basePath=download_path, folder=file, createPath=False) if not os.path.exists(realFile): - raise Exception(f"File '{realFile}' does not exist.") + raise StreamingError(f"File '{realFile}' does not exist.") m3u8 = [] @@ -88,4 +95,4 @@ class M3u8: m3u8.append(f"{self.url}player/subtitle/{quote(file)}.vtt") m3u8.append("#EXT-X-ENDLIST") - return '\n'.join(m3u8) + return "\n".join(m3u8) diff --git a/app/library/Playlist.py b/app/library/Playlist.py index 3112d176..27aa1c47 100644 --- a/app/library/Playlist.py +++ b/app/library/Playlist.py @@ -1,13 +1,15 @@ import glob -import pathlib import re +from pathlib import Path from urllib.parse import quote -from library.Utils import calcDownloadPath, checkId -import pathlib -from library.ffprobe import FFProbe -from .Subtitle import Subtitle + from aiohttp.web import Response +from .ffprobe import FFProbe +from .Subtitle import Subtitle +from .Utils import calcDownloadPath, checkId, StreamingError + + class Playlist: _url: str = None @@ -15,27 +17,30 @@ class Playlist: self.url = url async def make(self, download_path: str, file: str) -> str | Response: - rFile = pathlib.Path(calcDownloadPath(basePath=download_path, folder=file, createPath=False)) + rFile = Path(calcDownloadPath(basePath=download_path, folder=file, createPath=False)) if not rFile.exists(): possibleFile = checkId(download_path, rFile) if not possibleFile: - raise Exception(f"File '{rFile}' does not exist.") + raise StreamingError(f"File '{rFile}' does not exist.") - return Response(status=302, headers={ - 'Location': f"{self.url}player/playlist/{quote(str(possibleFile).replace(download_path, '').strip('/'))}.m3u8" - }) + return Response( + status=302, + headers={ + "Location": f"{self.url}player/playlist/{quote(str(possibleFile).replace(download_path, '').strip('/'))}.m3u8" + }, + ) try: ffprobe = FFProbe(rFile) await ffprobe.run() - except UnicodeDecodeError as e: + except UnicodeDecodeError: pass - if not 'duration' in ffprobe.metadata: - raise Exception(f"Unable to get '{rFile}' duration.") + if "duration" not in ffprobe.metadata: + raise StreamingError(f"Unable to get '{rFile}' duration.") - duration: float = float(ffprobe.metadata.get('duration')) + duration: float = float(ffprobe.metadata.get("duration")) playlist = [] playlist.append("#EXTM3U") @@ -44,28 +49,31 @@ class Playlist: index = 0 for item in self.getSideCarFiles(rFile): - if not item.suffix in Subtitle.allowedExtensions: + if item.suffix not in Subtitle.allowedExtensions: continue index += 1 lang: str = "und" - lg = re.search(r'\.(?P\w{2,3})\.\w{3}$', item.name) + lg = re.search(r"\.(?P\w{2,3})\.\w{3}$", item.name) if lg: - lang = lg.groupdict().get('lang') + lang = lg.groupdict().get("lang") subs = ',SUBTITLES="subs"' - url = f"{self.url}player/m3u8/subtitle/{quote(str(pathlib.Path(file).with_name(item.name)))}.m3u8?duration={duration}" + url = ( + f"{self.url}player/m3u8/subtitle/{quote(str(Path(file).with_name(item.name)))}.m3u8?duration={duration}" + ) name = f"{item.suffix[1:].upper()} ({index}) - {lang}" playlist.append( - f'#EXT-X-MEDIA:TYPE=SUBTITLES,GROUP-ID="subs",NAME="{name}",DEFAULT=NO,AUTOSELECT=NO,FORCED=NO,LANGUAGE="{lang}",URI="{url}"') + f'#EXT-X-MEDIA:TYPE=SUBTITLES,GROUP-ID="subs",NAME="{name}",DEFAULT=NO,AUTOSELECT=NO,FORCED=NO,LANGUAGE="{lang}",URI="{url}"' + ) - playlist.append(f'#EXT-X-STREAM-INF:PROGRAM-ID=1{subs}') + playlist.append(f"#EXT-X-STREAM-INF:PROGRAM-ID=1{subs}") playlist.append(f"{self.url}player/m3u8/video/{quote(file)}.m3u8") - return '\n'.join(playlist) + return "\n".join(playlist) - def getSideCarFiles(self, file: pathlib.Path) -> list[pathlib.Path]: + def getSideCarFiles(self, file: Path) -> list[Path]: """ Get sidecar files for the given file. @@ -75,7 +83,7 @@ class Playlist: files = [] for sub_file in file.parent.glob(f"{glob.escape(file.stem)}.*"): - if sub_file == file or sub_file.is_file() is False or sub_file.stem.startswith('.'): + if sub_file == file or sub_file.is_file() is False or sub_file.stem.startswith("."): continue files.append(sub_file) diff --git a/app/library/Segments.py b/app/library/Segments.py index 0c2bc781..e4e7b3d4 100644 --- a/app/library/Segments.py +++ b/app/library/Segments.py @@ -3,20 +3,14 @@ import hashlib import logging import os import tempfile -from .Utils import calcDownloadPath -from .config import Config -LOG = logging.getLogger('player.segments') +from .config import Config +from .Utils import calcDownloadPath, StreamingError + +LOG = logging.getLogger("player.segments") class Segments: - duration: int - index: int - vconvert: bool - aconvert: bool - vcodec: str - acodec: str - def __init__(self, index: int, duration: float, vconvert: bool, aconvert: bool): config = Config.get_instance() self.index = int(index) @@ -25,13 +19,14 @@ class Segments: self.aconvert = bool(aconvert) self.vcodec = config.streamer_vcodec self.acodec = config.streamer_acodec + # sadly due to unforeseen circumstances, we have to convert the video for now. self.vconvert = True async def stream(self, path: str, file: str) -> bytes: realFile: str = calcDownloadPath(basePath=path, folder=file, createPath=False) if not os.path.exists(realFile): - raise Exception(f"File {realFile} does not exist.") + raise StreamingError(f"File {realFile} does not exist.") tmpDir: str = tempfile.gettempdir() tmpFile = os.path.join(tmpDir, f'ytptube_stream.{hashlib.md5(realFile.encode("utf-8")).hexdigest()}') @@ -40,68 +35,69 @@ class Segments: os.symlink(realFile, tmpFile) if self.index == 0: - startTime: float = '{:.6f}'.format(0) + startTime: float = "{:.6f}".format(0) else: - startTime: float = '{:.6f}'.format((self.duration * self.index)) + startTime: float = "{:.6f}".format((self.duration * self.index)) fargs = [] - fargs.append('-xerror') - fargs.append('-hide_banner') - fargs.append('-loglevel') - fargs.append('error') + fargs.append("-xerror") + fargs.append("-hide_banner") + fargs.append("-loglevel") + fargs.append("error") - fargs.append('-ss') - fargs.append(str(startTime if startTime else '0.00000')) - fargs.append('-t') - fargs.append(str('{:.6f}'.format(self.duration))) + fargs.append("-ss") + fargs.append(str(startTime if startTime else "0.00000")) + fargs.append("-t") + fargs.append(str("{:.6f}".format(self.duration))) - fargs.append('-copyts') + fargs.append("-copyts") - fargs.append('-i') - fargs.append(f'file:{tmpFile}') - fargs.append('-map_metadata') - fargs.append('-1') + fargs.append("-i") + fargs.append(f"file:{tmpFile}") + fargs.append("-map_metadata") + fargs.append("-1") - fargs.append('-pix_fmt') - fargs.append('yuv420p') - fargs.append('-g') - fargs.append('52') + fargs.append("-pix_fmt") + fargs.append("yuv420p") + fargs.append("-g") + fargs.append("52") - fargs.append('-map') - fargs.append('0:v:0') - fargs.append('-strict') - fargs.append('-2') + fargs.append("-map") + fargs.append("0:v:0") + fargs.append("-strict") + fargs.append("-2") - fargs.append('-codec:v') - fargs.append(self.vcodec if self.vconvert else 'copy') + fargs.append("-codec:v") + fargs.append(self.vcodec if self.vconvert else "copy") # audio section. - fargs.append('-map') - fargs.append('0:a:0') - fargs.append('-codec:a') - fargs.append(self.acodec if self.aconvert else 'copy') + fargs.append("-map") + fargs.append("0:a:0") + fargs.append("-codec:a") + fargs.append(self.acodec if self.aconvert else "copy") - fargs.append('-sn') + fargs.append("-sn") - fargs.append('-muxdelay') - fargs.append('0') - fargs.append('-f') - fargs.append('mpegts') - fargs.append('pipe:1') + fargs.append("-muxdelay") + fargs.append("0") + fargs.append("-f") + fargs.append("mpegts") + fargs.append("pipe:1") - LOG.debug(f"Streaming '{realFile}' segment '{self.index}'. " + " ".join(fargs)) + LOG.debug(f"Streaming '{realFile}' segment '{self.index}'. ffmpeg: {' '.join(fargs)}") proc = await asyncio.subprocess.create_subprocess_exec( - 'ffmpeg', *fargs, + "ffmpeg", + *fargs, stdin=asyncio.subprocess.DEVNULL, stdout=asyncio.subprocess.PIPE, - stderr=asyncio.subprocess.PIPE + stderr=asyncio.subprocess.PIPE, ) data, err = await proc.communicate() if 0 != proc.returncode: LOG.error(f'Failed to stream {realFile} segment {self.index}. {err.decode("utf-8")}.') - raise Exception(f'Failed to stream {realFile} segment {self.index}.') + raise StreamingError(f"Failed to stream {realFile} segment {self.index}.") return data diff --git a/app/library/Subtitle.py b/app/library/Subtitle.py index 2b010eb8..589fd9c6 100644 --- a/app/library/Subtitle.py +++ b/app/library/Subtitle.py @@ -5,7 +5,7 @@ import pysubs2 from pysubs2.time import ms_to_times from pysubs2.formats.substation import SubstationFormat -LOG = logging.getLogger('player.subtitle') +LOG = logging.getLogger("player.subtitle") def ms_to_timestamp(ms: int) -> str: @@ -19,9 +19,13 @@ SubstationFormat.ms_to_timestamp = ms_to_timestamp class Subtitle: - allowedExtensions: tuple[str] = (".srt", ".vtt", ".ass",) + allowedExtensions: tuple[str] = ( + ".srt", + ".vtt", + ".ass", + ) - async def make(self, path: str, file: str) -> bytes: + async def make(self, path: str, file: str) -> str: realFile: str = calcDownloadPath(basePath=path, folder=file, createPath=False) rFile = pathlib.Path(realFile) @@ -29,12 +33,12 @@ class Subtitle: if not rFile.exists(): raise Exception(f"File '{file}' does not exist.") - if not rFile.suffix in self.allowedExtensions: + if rFile.suffix not in self.allowedExtensions: raise Exception(f"File '{file}' subtitle type is not supported.") if rFile.suffix == ".vtt": - subData = '' - with open(realFile, 'r') as f: + subData = "" + with open(realFile, "r") as f: subData = f.read() return subData @@ -45,9 +49,9 @@ class Subtitle: raise Exception(f"No subtitle events were found in '{rFile}'.") if len(subs.events) < 2: - return subs.to_string('vtt') + return subs.to_string("vtt") - if subs.events[0].end == subs.events[len(subs.events)-1].end: + if subs.events[0].end == subs.events[len(subs.events) - 1].end: subs.events.pop(0) - return subs.to_string('vtt') + return subs.to_string("vtt") diff --git a/app/library/Utils.py b/app/library/Utils.py index c95cafe3..cf5cfc77 100644 --- a/app/library/Utils.py +++ b/app/library/Utils.py @@ -1,6 +1,4 @@ import copy -from datetime import datetime, timezone -from functools import lru_cache import ipaddress import json import logging @@ -8,16 +6,31 @@ import os import pathlib import re import socket -from typing import Any import uuid +from datetime import datetime, timezone +from functools import lru_cache +from typing import Any + import yt_dlp -LOG = logging.getLogger('Utils') +LOG = logging.getLogger("Utils") -IGNORED_KEYS: tuple[str] = ('cookiefile', 'paths', 'outtmpl', 'progress_hooks', 'postprocessor_hooks',) +IGNORED_KEYS: tuple[str] = ( + "cookiefile", + "paths", + "outtmpl", + "progress_hooks", + "postprocessor_hooks", +) YTDLP_INFO_CLS: yt_dlp.YoutubeDL = None +class StreamingError(Exception): + """Raised when an error occurs during streaming.""" + + pass + + def get_opts(preset: str, ytdl_opts: dict) -> dict: """ Returns ytdlp options download options @@ -31,16 +44,17 @@ def get_opts(preset: str, ytdl_opts: dict) -> dict: """ opts = copy.deepcopy(ytdl_opts) - if 'default' == preset: + if "default" == preset: LOG.debug("Using default preset.") return opts from .config import Config + presets = Config.get_instance().presets found = False for _preset in presets: - if _preset['name'] == preset: + if _preset["name"] == preset: found = True preset_opts = _preset break @@ -49,24 +63,24 @@ def get_opts(preset: str, ytdl_opts: dict) -> dict: LOG.error(f"Preset '{preset}' is not defined in the presets.") return opts - opts['format'] = preset_opts.get('format') + opts["format"] = preset_opts.get("format") - if 'postprocessors' in preset_opts: - opts['postprocessors'] = preset_opts['postprocessors'] + if "postprocessors" in preset_opts: + opts["postprocessors"] = preset_opts["postprocessors"] - if 'args' in preset_opts: - for key, value in preset_opts['args'].items(): + if "args" in preset_opts: + for key, value in preset_opts["args"].items(): opts[key] = value LOG.debug(f"Using preset '{preset}', altered options: {opts}") return opts -def getVideoInfo(url: str, ytdlp_opts: dict = None) -> (Any | dict[str, Any] | None): +def getVideoInfo(url: str, ytdlp_opts: dict = None) -> Any | dict[str, Any] | None: params: dict = { - 'quiet': True, - 'color': 'no_color', - 'extract_flat': True, + "quiet": True, + "color": "no_color", + "extract_flat": True, } if ytdlp_opts: @@ -98,30 +112,35 @@ def calcDownloadPath(basePath: str, folder: str = None, createPath: bool = True) def ExtractInfo(config: dict, url: str, debug: bool = False) -> dict: params: dict = { - 'color': 'no_color', - 'extract_flat': True, - 'skip_download': True, - 'ignoreerrors': True, - 'ignore_no_formats_error': True, + "color": "no_color", + "extract_flat": True, + "skip_download": True, + "ignoreerrors": True, + "ignore_no_formats_error": True, **config, } # Remove keys that are not needed for info extraction as those keys generate files when used with extract_info. - for key in ('writeinfojson', 'writethumbnail', 'writedescription', 'writeautomaticsub',): + for key in ( + "writeinfojson", + "writethumbnail", + "writedescription", + "writeautomaticsub", + ): if key in params: params.pop(key) if debug: - params['verbose'] = True - params['logger'] = logging.getLogger('YTPTube-ytdl') + params["verbose"] = True + params["logger"] = logging.getLogger("YTPTube-ytdl") else: - params['quiet'] = True + params["quiet"] = True return yt_dlp.YoutubeDL(params=params).extract_info(url, download=False) def mergeDict(source: dict, destination: dict) -> dict: - """ Merge data from source into destination """ + """Merge data from source into destination""" destination_copy = copy.deepcopy(destination) for key, value in source.items(): @@ -137,15 +156,14 @@ def mergeDict(source: dict, destination: dict) -> dict: def mergeConfig(config: dict, new_config: dict) -> dict: - """ Merge user provided config into default config """ + """Merge user provided config into default config""" ignored_keys: tuple = ( - 'cookiefile', - 'download_archive' - 'paths', - 'outtmpl', - 'progress_hooks', - 'postprocessor_hooks', + "cookiefile", + "download_archive" "paths", + "outtmpl", + "progress_hooks", + "postprocessor_hooks", ) for key in ignored_keys: @@ -159,23 +177,28 @@ def isDownloaded(archive_file: str, url: str) -> tuple[bool, dict[str | None, st global YTDLP_INFO_CLS idDict = { - 'id': None, - 'ie_key': None, - 'archive_id': None, + "id": None, + "ie_key": None, + "archive_id": None, } if not url or not archive_file or not os.path.exists(archive_file): - return False, idDict, + return ( + False, + idDict, + ) if not YTDLP_INFO_CLS: - YTDLP_INFO_CLS = yt_dlp.YoutubeDL(params={ - 'color': 'no_color', - 'extract_flat': True, - 'skip_download': True, - 'ignoreerrors': True, - 'ignore_no_formats_error': True, - 'quiet': True, - }) + YTDLP_INFO_CLS = yt_dlp.YoutubeDL( + params={ + "color": "no_color", + "extract_flat": True, + "skip_download": True, + "ignoreerrors": True, + "ignore_no_formats_error": True, + "quiet": True, + } + ) for key, ie in YTDLP_INFO_CLS._ies.items(): if not ie.suitable(url): @@ -188,26 +211,35 @@ def isDownloaded(archive_file: str, url: str) -> tuple[bool, dict[str | None, st if not temp_id: break - idDict['id'] = temp_id - idDict['ie_key'] = key - idDict['archive_id'] = YTDLP_INFO_CLS._make_archive_id(idDict) + idDict["id"] = temp_id + idDict["ie_key"] = key + idDict["archive_id"] = YTDLP_INFO_CLS._make_archive_id(idDict) break - if not idDict['archive_id']: - return False, idDict, + if not idDict["archive_id"]: + return ( + False, + idDict, + ) - with open(archive_file, 'r') as f: + with open(archive_file, "r") as f: for line in f.readlines(): - if idDict['archive_id'] in line: - return True, idDict, + if idDict["archive_id"] in line: + return ( + True, + idDict, + ) - return False, idDict, + return ( + False, + idDict, + ) def jsonCookie(cookies: dict[dict[str, any]]) -> str | None: """Converts JSON cookies to Netscape cookies - Returns None if no cookies are found, otherwise returns a string of cookies in Netscape format. + Returns None if no cookies are found, otherwise returns a string of cookies in Netscape format. """ netscapeCookies = "# Netscape HTTP Cookie File\n# https://curl.haxx.se/docs/http-cookies.html\n# This file was generated by libcurl! Edit at your own risk.\n\n" @@ -227,19 +259,26 @@ def jsonCookie(cookies: dict[dict[str, any]]) -> str | None: cookieDict = cookies[domain][subDomain][cookie] - if 0 == int(cookieDict['expirationDate']): - cookieDict['expirationDate'] = datetime.now(timezone.utc).timestamp() + (86400 * 1000) + if 0 == int(cookieDict["expirationDate"]): + cookieDict["expirationDate"] = datetime.now(timezone.utc).timestamp() + (86400 * 1000) hasCookies = True - netscapeCookies += "\t".join([ - cookieDict['domain'] if str(cookieDict['domain']).startswith('.') else '.' + cookieDict['domain'], - 'TRUE', - cookieDict['path'], - 'TRUE' if cookieDict['secure'] else 'FALSE', - str(int(cookieDict['expirationDate'])), - cookieDict['name'], - cookieDict['value'] - ])+"\n" + netscapeCookies += ( + "\t".join( + [ + cookieDict["domain"] + if str(cookieDict["domain"]).startswith(".") + else "." + cookieDict["domain"], + "TRUE", + cookieDict["path"], + "TRUE" if cookieDict["secure"] else "FALSE", + str(int(cookieDict["expirationDate"])), + cookieDict["name"], + cookieDict["value"], + ] + ) + + "\n" + ) return netscapeCookies if hasCookies else None @@ -264,21 +303,38 @@ def load_file(file: str, check_type=None) -> tuple[dict | list, bool, str]: if check_type: assert isinstance(opts, check_type) - return (opts, True, '',) + return ( + opts, + True, + "", + ) except Exception: with open(file) as json_data: from pyjson5 import load as json5_load + try: opts = json5_load(json_data) if check_type: assert isinstance(opts, check_type) - return (opts, True, '',) + return ( + opts, + True, + "", + ) except AssertionError: - return ({}, False, f"Failed to assert that the contents '{type(opts)}' are of type '{check_type}'.",) + return ( + {}, + False, + f"Failed to assert that the contents '{type(opts)}' are of type '{check_type}'.", + ) except Exception as e: - return ({}, False, f'{e}',) + return ( + {}, + False, + f"{e}", + ) def checkId(basePath: str, file: pathlib.Path) -> bool | str: @@ -292,17 +348,26 @@ def checkId(basePath: str, file: pathlib.Path) -> bool | str: :return: False if no id found, otherwise the id. """ - match = re.search(r'(?<=\[)(?:youtube-)?(?P[a-zA-Z0-9\-_]{11})(?=\])', file.stem, re.IGNORECASE) + match = re.search(r"(?<=\[)(?:youtube-)?(?P[a-zA-Z0-9\-_]{11})(?=\])", file.stem, re.IGNORECASE) if not match: return False - id = match.groupdict().get('id') + id = match.groupdict().get("id") for f in file.parent.iterdir(): if id not in f.stem: continue - if f.suffix not in ('.mp4', '.mkv', '.webm', '.m4v', '.m4a', '.mp3', '.aac', '.ogg',): + if f.suffix not in ( + ".mp4", + ".mkv", + ".webm", + ".m4v", + ".m4a", + ".mp3", + ".aac", + ".ogg", + ): continue return f.absolute() @@ -314,7 +379,7 @@ def get_value(value): return value() if callable(value) else value -def ag(array: dict | list, path: list[str | int] | str | int, default: Any = None, separator: str = '.') -> Any: +def ag(array: dict | list, path: list[str | int] | str | int, default: Any = None, separator: str = ".") -> Any: """ dict/array getter: Retrieve a value from a nested dict or object using a path. @@ -328,7 +393,7 @@ def ag(array: dict | list, path: list[str | int] | str | int, default: Any = Non :return: The found value or the default if not found. """ - if path is None or path == '': + if path is None or path == "": return array if not isinstance(array, dict): @@ -369,7 +434,7 @@ def is_private_address(hostname: str) -> bool: try: ip = socket.gethostbyname(hostname) ip_obj = ipaddress.ip_address(ip) - return (ip_obj.is_private or ip_obj.is_loopback or ip_obj.is_reserved or ip_obj.is_link_local) + return ip_obj.is_private or ip_obj.is_loopback or ip_obj.is_reserved or ip_obj.is_link_local except socket.gaierror: # Could not resolve - treat as invalid or restricted return True @@ -393,6 +458,7 @@ def validate_url(url: str) -> bool: try: from yarl import URL + parsed_url = URL(url) except ValueError: raise ValueError("Invalid URL.") diff --git a/app/library/Webhooks.py b/app/library/Webhooks.py index 6b586f50..a3cb4152 100644 --- a/app/library/Webhooks.py +++ b/app/library/Webhooks.py @@ -5,12 +5,17 @@ from .ItemDTO import ItemDTO import httpx from .version import APP_VERSION -LOG = logging.getLogger('Webhooks') +LOG = logging.getLogger("Webhooks") class Webhooks: targets: list[dict] = [] - allowed_events: tuple = ('added', 'completed', 'error', 'not_live',) + allowed_events: tuple = ( + "added", + "completed", + "error", + "not_live", + ) def __init__(self, file: str): if os.path.exists(file): @@ -19,13 +24,14 @@ class Webhooks: def load(self, file: str): try: if os.path.getsize(file) < 3: - raise Exception(f'file is empty.') + raise Exception("file is empty.") LOG.info(f"Loading webhooks from '{file}'.") from .Utils import load_file + (target, status, error) = load_file(file, list) if not status: - raise Exception(f'{error}') + raise Exception(f"{error}") self.targets = target except Exception as e: @@ -42,7 +48,7 @@ class Webhooks: tasks = [] for target in self.targets: - allowed_events = target.get('on') if 'on' in target else [] + allowed_events = target.get("on") if "on" in target else [] if len(allowed_events) > 0 and event not in allowed_events: continue @@ -52,41 +58,36 @@ class Webhooks: async def __send(self, event: str, target: dict, item: ItemDTO) -> dict: from .config import Config - req: dict = target.get('request') + + req: dict = target.get("request") try: LOG.info(f"Sending event '{event}' id '{item.id}' to '{target.get('name')}'.") async with httpx.AsyncClient() as client: - request_type = req.get('type', 'json') + request_type = req.get("type", "json") reqBody = { - 'method': req.get('method', 'POST'), - 'url': req.get('url'), - 'headers': { - 'User-Agent': f"YTPTube/{APP_VERSION}" - }, + "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')) + if req.get("headers", None): + reqBody["headers"].update(req.get("headers")) - match(request_type): - case 'json': - reqBody['json'] = item.__dict__ - reqBody['headers']['Content-Type'] = 'application/json' + match request_type: + case "json": + reqBody["json"] = item.__dict__ + reqBody["headers"]["Content-Type"] = "application/json" case _: - reqBody['data'] = item.__dict__ - reqBody['headers']['Content-Type'] = 'application/x-www-form-urlencoded' + reqBody["data"] = item.__dict__ + 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 - } + respData = {"url": req.get("url"), "status": response.status_code, "text": response.text} msg = f"Webhook target '{target.get('name')}' Responded to event '{event}' id '{item.id}' with status '{response.status_code}'." - if Config.get_instance().debug and respData.get('text'): + if Config.get_instance().debug and respData.get("text"): msg += f" body '{respData.get('text','??')}'." LOG.info(msg) @@ -94,11 +95,7 @@ class Webhooks: return respData except Exception as e: LOG.error(f"Error sending event '{event}' id '{item.id}' to '{target.get('name')}'. '{e}'") - return { - 'url': req.get('url'), - 'status': 500, - 'text': str(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: diff --git a/app/library/common.py b/app/library/common.py index 5715ab0e..956ece68 100644 --- a/app/library/common.py +++ b/app/library/common.py @@ -1,14 +1,16 @@ import logging + from .DownloadQueue import DownloadQueue from .encoder import Encoder -LOG = logging.getLogger('common') +LOG = logging.getLogger("common") -class common(): +class common: """ This class is used to share common methods between the socket and the API gateways. """ + queue: DownloadQueue = None encoder: Encoder = None @@ -17,8 +19,9 @@ class common(): self.queue = queue self.encoder = encoder - async def add(self, url: str, preset: str, folder: str, ytdlp_cookies: str, ytdlp_config: dict, - output_template: str) -> dict[str, str]: + async def add( + self, url: str, preset: str, folder: str, ytdlp_cookies: str, ytdlp_config: dict, output_template: str + ) -> dict[str, str]: if ytdlp_config is None: ytdlp_config = {} @@ -27,11 +30,11 @@ class common(): status = await self.queue.add( url=url, - preset=preset if preset else 'default', + preset=preset if preset else "default", folder=folder, ytdlp_cookies=ytdlp_cookies, ytdlp_config=ytdlp_config, - output_template=output_template + output_template=output_template, ) return status diff --git a/app/library/config.py b/app/library/config.py index eade2eb7..05966b19 100644 --- a/app/library/config.py +++ b/app/library/config.py @@ -3,12 +3,14 @@ import os import re import sys import time +from pathlib import Path + import coloredlogs -from .version import APP_VERSION from dotenv import load_dotenv from yt_dlp.version import __version__ as YTDLP_VERSION + from .Utils import load_file -from pathlib import Path +from .version import APP_VERSION class Config: @@ -39,6 +41,8 @@ class Config: log_level: str = "info" + access_log: bool = True + allow_manifestless: bool = False max_workers: int = 1 @@ -136,6 +140,7 @@ class Config: "debug", "temp_keep", "allow_manifestless", + "access_log", ) _frontend_vars: tuple = ( diff --git a/app/library/encoder.py b/app/library/encoder.py index b17f6a91..d78668b1 100644 --- a/app/library/encoder.py +++ b/app/library/encoder.py @@ -9,7 +9,7 @@ class Encoder(json.JSONEncoder): """ def default(self, o): - if isinstance(o, object) and hasattr(o, '__dict__'): + if isinstance(o, object) and hasattr(o, "__dict__"): return o.__dict__ return json.JSONEncoder.default(self, o) diff --git a/app/library/ffprobe.py b/app/library/ffprobe.py index 66e3da01..72b6e548 100644 --- a/app/library/ffprobe.py +++ b/app/library/ffprobe.py @@ -1,6 +1,7 @@ """ Python wrapper for ffprobe command line tool. ffprobe must exist in the path. """ + import asyncio import functools import json @@ -22,25 +23,26 @@ class FFStream: setattr(self, key, val) try: - self.__dict__['framerate'] = round( - functools.reduce( - operator.truediv, map(int, self.__dict__.get('avg_frame_rate', '').split('/')) - ) + self.__dict__["framerate"] = round( + functools.reduce(operator.truediv, map(int, self.__dict__.get("avg_frame_rate", "").split("/"))) ) except ValueError: - self.__dict__['framerate'] = None + self.__dict__["framerate"] = None except ZeroDivisionError: - self.__dict__['framerate'] = 0 + self.__dict__["framerate"] = 0 def __repr__(self): - if not 'codec_long_name' in self.__dict__: - self.codec_long_name = self.__dict__.get('codec_name', '') + if "codec_long_name" not in self.__dict__: + self.codec_long_name = self.__dict__.get("codec_name", "") if self.is_video(): return f"" if self.is_audio(): - return f" " + return ( + f" " + ) if self.is_subtitle() or self.is_attachment(): return f"" @@ -51,25 +53,25 @@ class FFStream: """ Is this stream labelled as an audio stream? """ - return self.__dict__.get('codec_type', None) == 'audio' + return self.__dict__.get("codec_type", None) == "audio" def is_video(self): """ Is the stream labelled as a video stream. """ - return self.__dict__.get('codec_type', None) == 'video' + return self.__dict__.get("codec_type", None) == "video" def is_subtitle(self): """ Is the stream labelled as a subtitle stream. """ - return self.__dict__.get('codec_type', None) == 'subtitle' + return self.__dict__.get("codec_type", None) == "subtitle" def is_attachment(self): """ Is the stream labelled as a attachment stream. """ - return self.__dict__.get('codec_type', None) == 'attachment' + return self.__dict__.get("codec_type", None) == "attachment" def frame_size(self): """ @@ -78,8 +80,8 @@ class FFStream: """ size = None if self.is_video(): - width = self.__dict__['width'] - height = self.__dict__['height'] + width = self.__dict__["width"] + height = self.__dict__["height"] if width and height: try: @@ -96,7 +98,7 @@ class FFStream: Returns a string representing the pixel format of the video stream. e.g. yuv420p. Returns none is it is not a video stream. """ - return self.__dict__.get('pix_fmt', None) + return self.__dict__.get("pix_fmt", None) def frames(self): """ @@ -104,9 +106,9 @@ class FFStream: """ if self.is_video() or self.is_audio(): try: - frame_count = int(self.__dict__.get('nb_frames', '')) + frame_count = int(self.__dict__.get("nb_frames", "")) except ValueError: - raise FFProbeError('None integer frame count') + raise FFProbeError("None integer frame count") else: frame_count = 0 @@ -119,9 +121,9 @@ class FFStream: """ if self.is_video() or self.is_audio(): try: - duration = float(self.__dict__.get('duration', '')) + duration = float(self.__dict__.get("duration", "")) except ValueError: - raise FFProbeError('None numeric duration') + raise FFProbeError("None numeric duration") else: duration = 0.0 @@ -131,34 +133,34 @@ class FFStream: """ Returns language tag of stream. e.g. eng """ - return self.__dict__.get('TAG:language', None) + return self.__dict__.get("TAG:language", None) def codec(self): """ Returns a string representation of the stream codec. """ - return self.__dict__.get('codec_name', None) + return self.__dict__.get("codec_name", None) def codec_description(self): """ Returns a long representation of the stream codec. """ - return self.__dict__.get('codec_long_name', None) + return self.__dict__.get("codec_long_name", None) def codec_tag(self): """ Returns a short representative tag of the stream codec. """ - return self.__dict__.get('codec_tag_string', None) + return self.__dict__.get("codec_tag_string", None) def bit_rate(self): """ Returns bit_rate as an integer in bps """ try: - return int(self.__dict__.get('bit_rate', '')) + return int(self.__dict__.get("bit_rate", "")) except ValueError: - raise FFProbeError('None integer bit_rate') + raise FFProbeError("None integer bit_rate") class FFProbe: @@ -166,38 +168,40 @@ class FFProbe: FFProbe wraps the ffprobe command and pulls the data into an object form:: metadata=FFProbe('multimedia-file.mov') """ + audio: list[FFStream] = [] attachment: list[FFStream] = [] streams: list[FFStream] = [] subtitle: list[FFStream] = [] video: list[FFStream] = [] metadata: dict = {} - path_to_video: str = '' + path_to_video: str = "" def __init__(self, path_to_video): self.path_to_video = path_to_video async def run(self): try: - with open(os.devnull, 'w') as tempf: - await asyncio.create_subprocess_exec( - "ffprobe", "-h", stdout=tempf, stderr=tempf - ) + with open(os.devnull, "w") as tempf: + await asyncio.create_subprocess_exec("ffprobe", "-h", stdout=tempf, stderr=tempf) except FileNotFoundError: - raise IOError('ffprobe not found.') + raise IOError("ffprobe not found.") if not os.path.isfile(self.path_to_video): raise IOError(f"No such media file '{self.path_to_video}'.") args = [ - '-v', 'quiet', - '-of', 'json', - '-show_streams', - '-show_format', + "-v", + "quiet", + "-of", + "json", + "-show_streams", + "-show_format", self.path_to_video, ] p = await asyncio.create_subprocess_exec( - 'ffprobe', *args, + "ffprobe", + *args, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, ) @@ -206,7 +210,7 @@ class FFProbe: data, err = await p.communicate() if 0 == exitCode: - parsed: dict = json.loads(data.decode('utf-8')) + parsed: dict = json.loads(data.decode("utf-8")) else: raise FFProbeError(f"FFProbe return with non-0 exit code. '{err.decode('utf-8')}'") @@ -217,10 +221,10 @@ class FFProbe: self.subtitle = [] self.attachment = [] - for stream in parsed.get('streams', []): + for stream in parsed.get("streams", []): self.streams.append(FFStream(stream)) - self.metadata = parsed.get('format', {}) + self.metadata = parsed.get("format", {}) for stream in self.streams: if stream.is_audio(): diff --git a/app/library/version.py b/app/library/version.py index 8a7718b5..7efd8aef 100644 --- a/app/library/version.py +++ b/app/library/version.py @@ -2,5 +2,6 @@ Version of the application, This file is updated by the CI/CD pipeline, do not edit it manually. """ + APP_VERSION = "dev-master" "Application version" diff --git a/app/main.py b/app/main.py index a38cd3cc..9f82e14d 100644 --- a/app/main.py +++ b/app/main.py @@ -16,7 +16,7 @@ from library.config import Config from library.DownloadQueue import DownloadQueue from library.Emitter import Emitter from library.encoder import Encoder -from library.HttpAPI import HttpAPI +from library.HttpAPI import HttpAPI, LOG as http_logger from library.HttpSocket import HttpSocket from library.Webhooks import Webhooks @@ -136,15 +136,19 @@ class Main: self.load_tasks() - start: str = f"YTPTube v{self.config.version} - started on http://{self.config.host}:{self.config.port}" + def started(_): + LOG.info("=" * 40) + LOG.info(f"YTPTube v{self.config.version} - started on http://{self.config.host}:{self.config.port}") + LOG.info("=" * 40) + web.run_app( self.app, host=self.config.host, port=self.config.port, reuse_port=True, loop=asyncio.get_event_loop(), - access_log=None, - print=lambda _: LOG.info(start), + access_log=http_logger if self.config.access_log else None, + print=started, ) diff --git a/entrypoint.sh b/entrypoint.sh index 5c2b77d9..94a38918 100644 --- a/entrypoint.sh +++ b/entrypoint.sh @@ -1,15 +1,15 @@ #!/usr/bin/env bash set -e -echo "Setting umask to ${UMASK}, to change it set the UMASK environment variable" +echo "Setting umask to '${UMASK}', to change it set the UMASK environment variable" umask ${UMASK} -echo_err() { cat <<< "$@" 1>&2; } +echo_err() { cat <<<"$@" 1>&2; } if [ ! -w "${YTP_CONFIG_PATH}" ]; then CH_USER=$(stat -c "%u" "${YTP_CONFIG_PATH}") CH_GRP=$(stat -c "%g" "${YTP_CONFIG_PATH}") - echo_err "ERROR: Unable to write to [${YTP_CONFIG_PATH}] data directory. Current user id [${UID}] while directory owner is [${CH_USER}]" + echo_err "ERROR: Unable to write to '${YTP_CONFIG_PATH}' data directory. Current user id '${UID}' while directory owner is '${CH_USER}'." echo_err "[Running under docker]" echo_err "change docker-compose.yaml user: to user:\"${CH_USER}:${CH_GRP}\"" echo_err "Run the following command to change the directory ownership" diff --git a/healthcheck.sh b/healthcheck.sh index 4d0e93b8..059c0643 100644 --- a/healthcheck.sh +++ b/healthcheck.sh @@ -2,4 +2,6 @@ set -eu pipefail -/usr/bin/curl -f http://localhost:${YTP_PORT:-8081}/ping +LOCAL_PORT="${YTP_PORT:-8081}" + +/usr/bin/curl -f "http://localhost:${LOCAL_PORT}/ping"