diff --git a/app/__init__.py b/app/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/app/library/DownloadQueue.py b/app/library/DownloadQueue.py index 5c3ec100..0716d527 100644 --- a/app/library/DownloadQueue.py +++ b/app/library/DownloadQueue.py @@ -63,6 +63,12 @@ class DownloadQueue(metaclass=Singleton): _instance = None """Instance of the DownloadQueue.""" + queue: DataStore + """DataStore for the download queue.""" + + done: DataStore + """DataStore for the completed downloads.""" + def __init__(self, connection: Connection, config: Config | None = None): DownloadQueue._instance = self @@ -566,7 +572,7 @@ class DownloadQueue(metaclass=Singleton): removed_files = 0 filename: str = "" - LOG.info( + LOG.debug( f"{remove_file=} {itemRef} - Removing local files: {self.config.remove_files}, {item.info.status=}" ) diff --git a/app/library/HttpAPI.py b/app/library/HttpAPI.py index d11b693a..496a806c 100644 --- a/app/library/HttpAPI.py +++ b/app/library/HttpAPI.py @@ -1,140 +1,45 @@ -import asyncio import base64 -import functools import hmac -import json +import importlib +import inspect import logging -import random -import time -import uuid +import pkgutil from collections.abc import Awaitable from datetime import UTC, datetime, timedelta from pathlib import Path -from urllib.parse import unquote_plus, urlparse +from typing import Any import anyio -import httpx -import magic from aiohttp import web from aiohttp.web import Request, RequestHandler, Response -from library.ag_utils import ag -from yt_dlp.cookies import LenientSimpleCookie from .cache import Cache -from .common import Common -from .conditions import Condition, Conditions from .config import Config from .DownloadQueue import DownloadQueue from .encoder import Encoder -from .Events import EventBus, Events, message +from .Events import EventBus from .ffprobe import ffprobe -from .ItemDTO import Item -from .M3u8 import M3u8 -from .Notifications import Notification, NotificationEvents -from .Playlist import Playlist -from .Presets import Preset, Presets -from .Segments import Segments -from .Subtitle import Subtitle -from .Tasks import Task, Tasks -from .Utils import ( - REMOVE_KEYS, - StreamingError, - arg_converter, - decrypt_data, - encrypt_data, - extract_info, - get_file, - get_file_sidecar, - get_files, - get_mime_type, - init_class, - read_logfile, - validate_url, - validate_uuid, -) -from .YTDLPOpts import YTDLPOpts +from .router import get_routes +from .Utils import decrypt_data, encrypt_data, get_file, get_mime_type -LOG = logging.getLogger("http_api") -MIME = magic.Magic(mime=True) +LOG: logging.Logger = logging.getLogger("http_api") -class HttpAPI(Common): - _static_holder: dict = {} - """Holds loaded static assets.""" +class HttpAPI: + def __init__(self, root_path: str, queue: DownloadQueue): + self.queue: DownloadQueue = queue or DownloadQueue.get_instance() + self.encoder: Encoder = Encoder() + self.config: Config = Config.get_instance() + self._notify: EventBus = EventBus.get_instance() - _ext_to_mime: dict = { - ".html": "text/html", - ".css": "text/css", - ".js": "application/javascript", - ".json": "application/json", - ".ico": "image/x-icon", - } - """Map ext to mimetype""" - - _frontend_routes: list[str] = [ - "/console", - "/presets", - "/tasks", - "/notifications", - "/changelog", - "/logs", - "/conditions", - "/browser", - "/browser/{path:.*}", - ] - """Frontend routes to be preloaded""" - - def __init__( - self, - root_path: str, - queue: DownloadQueue | None = None, - encoder: Encoder | None = None, - config: Config | None = None, - ): - self.queue = queue or DownloadQueue.get_instance() - self.encoder = encoder or Encoder() - self.config = config or Config.get_instance() - self._notify = EventBus.get_instance() - - self.rootPath = root_path - self.routes = web.RouteTableDef() + self.rootPath: str = root_path self.cache = Cache() self.app: web.Application | None = None - self.isRequestingBackground = False - - super().__init__(queue=self.queue, encoder=self.encoder, config=self.config) - - @staticmethod - def route(method: str, path: str, name: str | None = None) -> Awaitable: - """ - Decorator to mark a method as an HTTP route handler. - - Args: - method (str): The HTTP method. - path (str): The path to the route. - name (str | None): The name of the route (optional). - - Returns: - Awaitable: The decorated function. - - """ - - 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 - wrapper._http_name = name - return wrapper - - return decorator async def on_shutdown(self, _: web.Application): pass - def attach(self, app: web.Application) -> "HttpAPI": + def attach(self, app: web.Application): """ Attach the routes to the application. @@ -152,6 +57,7 @@ class HttpAPI(Common): app=app, base_path=self.config.base_path.rstrip("/"), download_path=self.config.download_path, + this=self, ) ) @@ -176,114 +82,8 @@ class HttpAPI(Common): LOG.exception(e) app.on_shutdown.append(self.on_shutdown) - return self - async def _static_file(self, req: Request) -> Response: - """ - Preload static files from the ui/exported folder. - - Args: - req (Request): The request object. - - Returns: - Response: The response object. - - """ - path = req.path - - if "/" != self.config.base_path and path.rstrip("/") == self.config.base_path.rstrip("/"): - path: str = f"{self.config.base_path.rstrip('/')}/index.html" - - if path not in self._static_holder: - for k in self._static_holder: - if path.startswith(k): - path = k - break - else: - return web.json_response( - {"error": "File not found.", "file": path}, status=web.HTTPNotFound.status_code - ) - - item: dict = self._static_holder[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") else "disk", - }, - status=web.HTTPOk.status_code, - ) - - def _preload_static(self, app: web.Application) -> "HttpAPI": - """ - Preload static files from the ui/exported folder. - - Args: - app (web.Application): The application to attach the routes to. - - Returns: - HttpAPI: The instance of the HttpAPI. - - """ - staticDir = (Path(self.rootPath) / "ui" / "exported").absolute() - if not staticDir.exists(): - staticDir = (Path(self.rootPath).parent / "ui" / "exported").absolute() - if not staticDir.exists(): - msg = f"Could not find the frontend UI static assets. '{staticDir}'." - raise ValueError(msg) - - preloaded = 0 - - base_path: str = str(Path(self.config.base_path).as_posix()).rstrip("/") - - for file in staticDir.rglob("*.*"): - if ".map" == file.suffix: - continue - - urlPath: str = f"{base_path}/{str(file.as_posix()).replace(f'{staticDir.as_posix()!s}/', '')}" - - with open(file, "rb") as f: - content = f.read() - - contentType = self._ext_to_mime.get(file.suffix, MIME.from_file(file)) - - self._static_holder[urlPath] = {"content": content, "content_type": contentType} - LOG.debug(f"Preloading static '{urlPath}'.") - app.router.add_get(urlPath, self._static_file) - preloaded += 1 - - if "/index.html" in self._static_holder: - for path in self._frontend_routes: - path: str = f"{base_path}/{path.lstrip('/')}" - self._static_holder[path] = self._static_holder["/index.html"] - app.router.add_get(path, self._static_file) - if "{" not in path: - self._static_holder[path + "/"] = self._static_holder["/index.html"] - app.router.add_get(path + "/", self._static_file) - LOG.debug(f"Preloading static route '{path}'.") - preloaded += 1 - - if "/" != self.config.base_path: - LOG.debug(f"adding base_path folder '{base_path}' to routes.") - app.router.add_get(base_path, self._static_file, name="_base_path") - app.router.add_get(f"{base_path}/", self._static_file, name="_base_path_slash") - - if preloaded < 1: - message = f"Failed to find any static files in '{staticDir}'." - if self.config.ignore_ui: - LOG.warning(message) - return self - - raise ValueError(message) - - LOG.info(f"Preloaded '{preloaded}' static files.") - - return self - - def add_routes(self, app: web.Application) -> "HttpAPI": + def add_routes(self, app: web.Application): """ Add the routes to the application. @@ -297,43 +97,34 @@ class HttpAPI(Common): registered_options: list = [] base_path: str = self.config.base_path.rstrip("/") + from app.routes._static import preload_static - for attr_name in dir(self): - method = getattr(self, attr_name) - if hasattr(method, "_http_method") and hasattr(method, "_http_path"): - http_path = method._http_path - if http_path.startswith("/"): - http_path = method._http_path[1:] + self._load_modules(Path(self.rootPath) / "routes") + preload_static(self.rootPath, self.config) - opts = {} - if hasattr(method, "_http_name") and method._http_name: - opts["name"] = method._http_name + async def options_handler(_: Request) -> Response: + return web.Response(status=204) - LOG.debug(f"Adding API route {method._http_method} {base_path}/{http_path}' {opts}.") - self.routes.route(method._http_method, f"{base_path}/{http_path}", **opts)(method) + for route in get_routes().values(): + routePath: str = f"/{route.path.lstrip('/')}" - if http_path in registered_options: - continue + if self.config.base_path == route.path: + pass + elif "" == base_path or not routePath.rstrip("/").startswith(base_path.rstrip("/")): + LOG.debug(f"Route path '{routePath}' does not start with base path '{base_path}'. Skipping.") + route.path = f"{base_path}/{route.path.lstrip('/')}" - async def options_handler(_: Request) -> Response: - return web.Response(status=204) + LOG.debug(f"Add ({route.name}) {route.method}: {route.path}.") - if "name" in opts: - opts["name"] = f"{opts['name']}_options" + app.router.add_route(route.method, route.path, handler=route.handler, name=route.name) - self.routes.route("OPTIONS", f"{base_path}/{http_path}", **opts)(options_handler) - registered_options.append(http_path) + if route.path in registered_options: + continue - self.routes.static(f"{base_path}/api/download/", self.config.download_path, name="download_static") - self._preload_static(app) + app.router.add_route("OPTIONS", route.path, handler=options_handler, name=f"{route.name}_opts") + registered_options.append(route.path) - try: - app.add_routes(self.routes) - except ValueError as e: - if "ui/exported" in str(e): - msg = f"Could not find the frontend UI static assets. '{e}'." - raise RuntimeError(msg) from e - raise + app.router.add_static(f"{base_path}/api/download/", self.config.download_path, name="download_static") @staticmethod def basic_auth(username: str, password: str) -> Awaitable: @@ -422,7 +213,7 @@ class HttpAPI(Common): return middleware_handler @staticmethod - def middle_wares(app: web.Application, base_path: str, download_path: str) -> Awaitable: + def middle_wares(app: web.Application, base_path: str, download_path: str, this: "HttpAPI") -> Awaitable: @web.middleware async def middleware_handler(request: Request, handler: RequestHandler) -> Response: static_path = str(app.router["download_static"].url_for(filename="")) @@ -443,8 +234,32 @@ class HttpAPI(Common): }, ) + kwargs: dict[str, Any] = { + "request": request, + "queue": this.queue, + "encoder": this.encoder, + "config": this.config, + "notify": this._notify, + "cache": this.cache, + "app": app, + "http_api": this, + } + try: - response = await handler(request) + sig = inspect.signature(handler) + expected_args = sig.parameters.keys() + + try: + if 1 == len(expected_args) and "request" in expected_args: + response = await handler(request) + else: + filtered = {k: v for k, v in kwargs.items() if k in expected_args} + response = await handler(**filtered) + except TypeError as te: + if "missing 1 required positional argument" in str(te) and "request" in str(te): + response = await handler(request) + else: + raise except web.HTTPException as e: return web.json_response(data={"error": str(e)}, status=e.status_code) except Exception as e: @@ -457,15 +272,21 @@ class HttpAPI(Common): contentType: str | None = response.headers.get("content-type", None) if contentType and "/" != base_path and contentType.startswith("text/html"): rewrite_path: str = base_path.rstrip("/") - content: str = ( - response.body.decode("utf-8") - .replace('', f'') - .replace('baseURL:""', f'baseURL:"{rewrite_path}/"') + async with await anyio.open_file(response._path, "rb") as f: + content = await f.read() + content: str = ( + content.decode("utf-8") + .replace('', f'') + .replace('baseURL:""', f'baseURL:"{rewrite_path}/"') + ) + + return web.Response( + body=content.encode("utf-8"), + status=response.status, + headers=response.headers, ) - response.body = content.encode("utf-8") - - if isinstance(response, web.FileResponse): + if request.path.startswith(static_path) and isinstance(response, web.FileResponse): try: ff_info = await ffprobe(response._path) mime_type = get_mime_type(ff_info.get("metadata", {}), response._path) @@ -477,1607 +298,15 @@ class HttpAPI(Common): return middleware_handler - @route("GET", "api/ping", "ping") - async def ping(self, _: Request) -> Response: - """ - Ping the server. - - Args: - _: The request object. - - Returns: - Response: The response object. - - """ - await self.queue.test() - return web.json_response(data={"status": "pong"}, status=web.HTTPOk.status_code) - - @route("POST", "api/yt-dlp/convert", "yt_dlp_convert") - async def yt_dlp_convert(self, request: Request) -> Response: - """ - Convert the yt-dlp args to a dict. - - Args: - request (Request): The request object. - - Returns: - Response: The response object. - - """ - post = await request.json() - args: str | None = post.get("args") - - if not args: - return web.json_response(data={"error": "args param is required."}, status=web.HTTPBadRequest.status_code) - - try: - response = {"opts": {}, "output_template": None, "download_path": None} - - data = arg_converter(args, dumps=True) - - if "outtmpl" in data and "default" in data["outtmpl"]: - response["output_template"] = data["outtmpl"]["default"] - - if "paths" in data and "home" in data["paths"]: - response["download_path"] = data["paths"]["home"] - - if "format" in data: - response["format"] = data["format"] - - bad_options = {k: v for d in REMOVE_KEYS for k, v in d.items()} - removed_options = [] - - for key in data: - if key in bad_options.items(): - removed_options.append(bad_options[key]) - continue - if not key.startswith("_"): - response["opts"][key] = data[key] - - if len(removed_options) > 0: - response["removed_options"] = removed_options - - return web.json_response(data=response, status=web.HTTPOk.status_code) - except Exception as e: - err = str(e).strip() - err = err.split("\n")[-1] if "\n" in err else err - err = err.replace("main.py: error: ", "").strip().capitalize() - return web.json_response( - data={"error": f"Failed to parse command options for yt-dlp. '{err}'."}, - status=web.HTTPBadRequest.status_code, - ) - - @route("GET", "api/yt-dlp/url/info", "get_info") - async def get_info(self, request: Request) -> Response: - """ - Get the video info. - - Args: - request (Request): The request object. - - Returns: - Response: The response object - - """ - url = request.query.get("url") - if not url: - return web.json_response(data={"error": "URL is required."}, status=web.HTTPBadRequest.status_code) - - try: - validate_url(url) - except ValueError as e: - return web.json_response(data={"error": str(e)}, status=web.HTTPBadRequest.status_code) - - preset = request.query.get("preset") - if preset: - exists = Presets.get_instance().get(preset) - if not exists: - return web.json_response( - data={"status": False, "message": f"Preset '{preset}' does not exist."}, - status=web.HTTPBadRequest.status_code, - ) - else: - preset = self.config.default_preset - - try: - key = self.cache.hash(url + str(preset)) - - if self.cache.has(key) and not request.query.get("force", False): - data = self.cache.get(key) - data["_cached"] = { - "status": "hit", - "key": key, - "ttl": data.get("_cached", {}).get("ttl", 300), - "ttl_left": data.get("_cached", {}).get("expires", time.time() + 300) - time.time(), - "expires": data.get("_cached", {}).get("expires", time.time() + 300), - } - return web.Response(body=json.dumps(data, indent=4), status=web.HTTPOk.status_code) - - opts = {} - - if ytdlp_proxy := self.config.get_ytdlp_args().get("proxy", None): - opts["proxy"] = ytdlp_proxy - - ytdlp_opts = YTDLPOpts.get_instance().preset(name=preset).add(opts).get_all() - - data = extract_info( - config=ytdlp_opts, - url=url, - debug=False, - no_archive=True, - follow_redirect=True, - sanitize_info=True, - ) - - if "formats" in data: - for index, item in enumerate(data["formats"]): - if "cookies" in item and len(item["cookies"]) > 0: - cookies = [f"{c.key}={c.value}" for c in LenientSimpleCookie(item["cookies"]).values()] - if len(cookies) > 0: - data["formats"][index]["h_cookies"] = "; ".join(cookies) - data["formats"][index]["h_cookies"] = data["formats"][index]["h_cookies"].strip() - - data["_cached"] = { - "status": "miss", - "key": key, - "ttl": 300, - "ttl_left": 300, - "expires": time.time() + 300, - } - - self.cache.set(key=key, value=data, ttl=300) - - return web.Response(body=json.dumps(data, indent=4), status=web.HTTPOk.status_code) - except Exception as e: - LOG.exception(e) - LOG.error(f"Error encountered while getting video info for '{url}'. '{e!s}'.") - return web.json_response( - data={ - "error": "failed to get video info.", - "message": str(e), - "formats": [], - }, - status=web.HTTPInternalServerError.status_code, - ) - - @route("GET", "api/yt-dlp/archive/recheck", "archive_recheck") - async def archive_recheck(self, _) -> Response: - """ - Recheck the manual archive entries. - - Args: - _ (Request): The request object. - - Returns: - Response: The response object - - """ - manual_archive = self.config.manual_archive - if not manual_archive: - return web.json_response( - data={"error": "Manual archive is not enabled."}, status=web.HTTPNotFound.status_code - ) - - manual_archive = Path(manual_archive) - if not manual_archive.exists(): - return web.json_response( - data={"error": "Manual archive file not found.", "file": manual_archive}, - status=web.HTTPNotFound.status_code, - ) - - tasks = [] - response = [] - - def info_wrapper(id: str, url: str) -> tuple[str, dict]: - try: - return ( - id, - extract_info( - config={ - "proxy": self.config.get_ytdlp_args().get("proxy", None), - "simulate": True, - "dump_single_json": True, - }, - url=url, - no_archive=True, - ), - ) - except Exception as e: - return (id, {"error": str(e)}) - - async with await anyio.open_file(manual_archive) as f: - # line format is "youtube ID - at: ISO8601" - async for line in f: - line = line.strip() - - if not line or not line.startswith("youtube"): - continue - - id = line.split(" ")[1].strip() - - if not id: - continue - - url = f"https://www.youtube.com/watch?v={id}" - key = self.cache.hash(id) - - if self.cache.has(key): - data = self.cache.get(key) - response.append({id: bool(data.get("id", None)) if isinstance(data, dict) else False}) - continue - - tasks.append( - asyncio.get_event_loop().run_in_executor(None, lambda i=id, url=url: info_wrapper(id=i, url=url)) - ) - - if len(tasks) > 0: - results = await asyncio.gather(*tasks) - for data in results: - if not data: - continue - - id, info = data - self.cache.set(key=self.cache.hash(id), value=info, ttl=3600 * 6) - response.append({id: bool(data.get("id", None)) if isinstance(data, dict) else False}) - - return web.json_response(data=response, status=web.HTTPOk.status_code, dumps=self.encoder.encode) - - @route("GET", "api/history/add", "quick_add") - async def quick_add(self, request: Request) -> Response: - """ - Add a URL to the download queue. - - Args: - request (Request): The request object. - - Returns: - Response: The response object - - """ - url: str | None = request.query.get("url") - if not url: - return web.json_response(data={"error": "url param is required."}, status=web.HTTPBadRequest.status_code) - - data = { - "url": url, - } - - preset = request.query.get("preset") - if preset: - exists = Presets.get_instance().get(preset) - if not exists: - return web.json_response( - data={"status": False, "message": f"Preset '{preset}' does not exist."}, - status=web.HTTPBadRequest.status_code, - ) - data["preset"] = preset - - try: - status = await self.add(item=Item.format(data)) - except ValueError as e: - return web.json_response(data={"status": False, "message": str(e)}, status=web.HTTPBadRequest.status_code) - - return web.json_response( - data={"status": status.get("status") == "ok", "message": status.get("msg", "URL added")}, - status=web.HTTPOk.status_code, - dumps=self.encoder.encode, - ) - - @route("POST", "api/history", "history_item_add") - async def history_item_add(self, request: Request) -> Response: - """ - Add a URL to the download queue. - - Args: - request (Request): The request object. - - Returns: - Response: The response object. - - """ - data = await request.json() - - if isinstance(data, dict): - data = [data] - - items = [] - for item in data: - try: - items.append(Item.format(item)) - except ValueError as e: - return web.json_response(data={"error": str(e), "data": item}, status=web.HTTPBadRequest.status_code) - - status = await asyncio.wait_for( - fut=asyncio.gather(*[self.add(item=item) for item in items]), - timeout=None, - ) - response = [] - for i, item in enumerate(items): - response.append({"item": item, "status": status[i].get("status") == "ok", "msg": status[i].get("msg")}) - - return web.json_response(data=response, status=web.HTTPOk.status_code, dumps=self.encoder.encode) - - @route("GET", "api/logs", "logs") - async def logs(self, request: Request) -> Response: - """ - Get recent logs - - Args: - request (Request): The request object. - - Returns: - Response: The response object. - - """ - if not self.config.file_logging: - return web.json_response( - data={"error": "File logging is not enabled."}, status=web.HTTPNotFound.status_code - ) - - offset = int(request.query.get("offset", 0)) - limit = int(request.query.get("limit", 100)) - if limit < 1 or limit > 150: - limit = 50 - - logs_data = await read_logfile( - file=Path(self.config.config_path) / "logs" / "app.log", - offset=offset, - limit=limit, - ) - return web.json_response( - data={ - "logs": logs_data["logs"], - "offset": offset, - "limit": limit, - "next_offset": logs_data["next_offset"], - "end_is_reached": logs_data["end_is_reached"], - }, - status=web.HTTPOk.status_code, - dumps=self.encoder.encode, - ) - - @route("GET", "api/conditions", "conditions") - async def conditions(self, _: Request) -> Response: - """ - Get the conditions - - Args: - _ (Request): The request object. - - Returns: - Response: The response object. - - """ - return web.json_response( - data=Conditions.get_instance().get_all(), - status=web.HTTPOk.status_code, - dumps=self.encoder.encode, - ) - - @route("PUT", "api/conditions", "conditions_add") - async def conditions_add(self, request: Request) -> Response: - """ - Save Conditions. - - Args: - request (Request): The request object. - - Returns: - Response: The response object - - """ - data = await request.json() - - if not isinstance(data, list): - return web.json_response( - {"error": "Invalid request body expecting list with dicts."}, - status=web.HTTPBadRequest.status_code, - ) - - items: list = [] - - cls = Conditions.get_instance() - - for item in data: - if not isinstance(item, dict): - return web.json_response( - {"error": "Invalid request body expecting list with dicts."}, - status=web.HTTPBadRequest.status_code, - ) - - if not item.get("name"): - return web.json_response( - {"error": "name is required.", "data": item}, status=web.HTTPBadRequest.status_code - ) - - if not item.get("filter"): - return web.json_response( - {"error": "filter is required.", "data": item}, status=web.HTTPBadRequest.status_code - ) - - if not item.get("cli"): - return web.json_response( - {"error": "command options for yt-dlp is required.", "data": item}, - status=web.HTTPBadRequest.status_code, - ) - - if not item.get("id", None) or not validate_uuid(item.get("id"), version=4): - item["id"] = str(uuid.uuid4()) - - try: - cls.validate(item) - except ValueError as e: - return web.json_response( - {"error": f"Failed to validate condition '{item.get('name')}'. '{e!s}'"}, - status=web.HTTPBadRequest.status_code, - ) - - items.append(init_class(Condition, item)) - try: - items = cls.save(items=items).load().get_all() - except Exception as e: - LOG.exception(e) - return web.json_response( - {"error": "Failed to save conditions.", "message": str(e)}, - status=web.HTTPInternalServerError.status_code, - ) - - await self._notify.emit(Events.CONDITIONS_UPDATE, data=items) - return web.json_response(data=items, status=web.HTTPOk.status_code, dumps=self.encoder.encode) - - @route("POST", "api/conditions/test", "conditions_test") - async def conditions_test(self, request: Request) -> Response: - """ - Test condition against URL. - - Args: - request (Request): The request object. - - Returns: - Response: The response object - - """ - params = await request.json() - - if not isinstance(params, dict): - return web.json_response( - {"error": "Invalid request body expecting dict."}, - status=web.HTTPBadRequest.status_code, - ) - - url = params.get("url") - if not url: - return web.json_response({"error": "url is required."}, status=web.HTTPBadRequest.status_code) - - cond = params.get("condition") - if not cond: - return web.json_response({"error": "condition is required."}, status=web.HTTPBadRequest.status_code) - - try: - preset = params.get("preset", self.config.default_preset) - key = self.cache.hash(url + str(preset)) - if not self.cache.has(key): - opts = {} - if ytdlp_proxy := self.config.get_ytdlp_args().get("proxy", None): - opts["proxy"] = ytdlp_proxy - ytdlp_opts = YTDLPOpts.get_instance().preset(name=preset).add(opts).get_all() - - data = extract_info( - config=ytdlp_opts, - url=url, - debug=False, - no_archive=True, - follow_redirect=True, - sanitize_info=True, - ) - if not data: - return web.json_response( - data={"error": f"Failed to extract info from '{url!s}'."}, - status=web.HTTPBadRequest.status_code, - ) - self.cache.set(key=key, value=data, ttl=600) - else: - data = self.cache.get(key) - except Exception as e: - LOG.exception(e) - return web.json_response( - data={"error": f"Failed to extract video info. '{e!s}'"}, - status=web.HTTPInternalServerError.status_code, - ) - - try: - from yt_dlp.utils import match_str - - status = match_str(cond, data) - except Exception as e: - LOG.exception(e) - return web.json_response( - data={"error": str(e)}, - status=web.HTTPBadRequest.status_code, - ) - - return web.json_response( - data={"status": status, "condition": cond, "data": data}, - status=web.HTTPOk.status_code, - dumps=self.encoder.encode, - ) - - @route("GET", "api/presets", "presets") - async def presets(self, request: Request) -> Response: - """ - Get the presets. - - Args: - request (Request): The request object. - - Returns: - Response: The response object. - - """ - data = Presets.get_instance().get_all() - filter_fields = request.query.get("filter", None) - - if filter_fields: - fields = [field.strip() for field in filter_fields.split(",")] - data = [{key: value for key, value in preset.__dict__.items() if key in fields} for preset in data] - - return web.json_response(data=data, status=web.HTTPOk.status_code, dumps=self.encoder.encode) - - @route("PUT", "api/presets", "presets_add") - async def presets_add(self, request: Request) -> Response: - """ - Add presets. - - Args: - request (Request): The request object. - - Returns: - Response: The response object - - """ - data = await request.json() - - if not isinstance(data, list): - return web.json_response( - {"error": "Invalid request body expecting list with dicts."}, - status=web.HTTPBadRequest.status_code, - ) - - presets: list = [] - - cls = Presets.get_instance() - - for item in data: - if not isinstance(item, dict): - return web.json_response( - {"error": "Invalid request body expecting list with dicts."}, - status=web.HTTPBadRequest.status_code, - ) - - if not item.get("name"): - return web.json_response( - {"error": "name is required.", "data": item}, status=web.HTTPBadRequest.status_code - ) - - if not item.get("id", None) or not validate_uuid(item.get("id"), version=4): - item["id"] = str(uuid.uuid4()) - - try: - cls.validate(item) - except ValueError as e: - return web.json_response( - {"error": f"Failed to validate preset '{item.get('name')}'. '{e!s}'"}, - status=web.HTTPBadRequest.status_code, - ) - - presets.append(Preset(**item)) - try: - presets = cls.save(items=presets).load().get_all() - except Exception as e: - LOG.exception(e) - return web.json_response( - {"error": "Failed to save presets.", "message": str(e)}, - status=web.HTTPInternalServerError.status_code, - ) - - await self._notify.emit(Events.PRESETS_UPDATE, data=presets) - return web.json_response(data=presets, status=web.HTTPOk.status_code, dumps=self.encoder.encode) - - @route("GET", "api/tasks", "tasks") - async def tasks(self, _: Request) -> Response: - """ - Get the tasks. - - Args: - _: The request object. - - Returns: - Response: The response object. - - """ - return web.json_response( - data=Tasks.get_instance().get_all(), status=web.HTTPOk.status_code, dumps=self.encoder.encode - ) - - @route("PUT", "api/tasks", "tasks_add") - async def tasks_add(self, request: Request) -> Response: - """ - Add tasks to the queue. - - Args: - request (Request): The request object. - - Returns: - Response: The response object - - """ - data = await request.json() - - if not isinstance(data, list): - return web.json_response( - {"error": "Invalid request body expecting list with dicts."}, - status=web.HTTPBadRequest.status_code, - ) - - tasks: list = [] - - ins = Tasks.get_instance() - - for item in data: - if not isinstance(item, dict): - return web.json_response( - {"error": "Invalid request body expecting list with dicts."}, - status=web.HTTPBadRequest.status_code, - ) - - if not item.get("url"): - return web.json_response( - {"error": "url is required.", "data": item}, status=web.HTTPBadRequest.status_code - ) - - if not item.get("id", None) or not validate_uuid(item.get("id"), version=4): - item["id"] = str(uuid.uuid4()) - - if not item.get("template", None): - item["template"] = "" - - if not item.get("cli", None): - item["cli"] = "" - - try: - Tasks.validate(item) - except ValueError as e: - return web.json_response( - {"error": f"Failed to validate task '{item.get('name')}'. '{e!s}'"}, - status=web.HTTPBadRequest.status_code, - ) - - tasks.append(Task(**item)) - - try: - tasks = ins.save(tasks=tasks).load().get_all() - except Exception as e: - LOG.exception(e) - return web.json_response( - {"error": "Failed to save tasks.", "message": str(e)}, - status=web.HTTPInternalServerError.status_code, - ) - - return web.json_response(data=tasks, status=web.HTTPOk.status_code, dumps=self.encoder.encode) - - @route("DELETE", "api/history", "history_delete") - async def history_delete(self, request: Request) -> Response: - """ - Delete an item from the queue. - - Args: - request (Request): The request object. - - Returns: - Response: The response object. - - """ - data = await request.json() - ids = data.get("ids") - where = data.get("where") - 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 - ) - - remove_file: bool = bool(data.get("remove_file", True)) - - return web.json_response( - data=await (self.queue.cancel(ids) if where == "queue" else self.queue.clear(ids, remove_file=remove_file)), - status=web.HTTPOk.status_code, - dumps=self.encoder.encode, - ) - - @route("POST", "api/history/{id}", "history_item_update") - async def history_item_update(self, request: Request) -> Response: - """ - Update an item in the history. - - Args: - request (Request): The request object. - - Returns: - Response: The response object. - - """ - id: str = request.match_info.get("id") - if not id: - return web.json_response(data={"error": "id is required."}, status=web.HTTPBadRequest.status_code) - - item = self.queue.done.get_by_id(id) - if not item: - return web.json_response(data={"error": "item not found."}, status=web.HTTPNotFound.status_code) - - post = await request.json() - if not post: - return web.json_response(data={"error": "no data provided."}, status=web.HTTPBadRequest.status_code) - - updated = False - - for k, v in post.items(): - if not hasattr(item.info, k): + def _load_modules(self, directory: Path): + package_name: str = str(directory.relative_to(Path(self.rootPath))).replace("/", ".") + LOG.debug(f"Loading modules from directory '{directory}' with package name '{package_name}'.") + for _, name, _ in pkgutil.iter_modules([directory]): + full_name: str = f"{package_name}.{name}" + if name.startswith("_"): continue - - if getattr(item.info, k) == v: - continue - - updated = True - setattr(item.info, k, v) - LOG.debug(f"Updated '{k}' to '{v}' for '{item.info.id}'") - - if updated: - self.queue.done.put(item) - await self._notify.emit(Events.UPDATE, data=item.info) - - 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", "api/history", "history") - async def history(self, _: Request) -> Response: - """ - Get the history. - - Args: - _: The request object. - - Returns: - Response: The response object. - - """ - data: dict = {"queue": [], "history": []} - q = self.queue.get() - - data["queue"].extend([q.get("queue", {}).get(k) for k in q.get("queue", {})]) - data["history"].extend([q.get("done", {}).get(k) for k in q.get("done", {})]) - - return web.json_response(data=data, status=web.HTTPOk.status_code, dumps=self.encoder.encode) - - @route("GET", "api/player/playlist/{file:.*}.m3u8", "playlist") - async def playlist(self, request: Request) -> Response: - """ - Get the playlist. - - Args: - request (Request): The request object. - - Returns: - Response: The response object. - - """ - file: str = request.match_info.get("file") - - if not file: - return web.json_response(data={"error": "file is required"}, status=web.HTTPBadRequest.status_code) - - base_path: str = self.config.base_path.rstrip("/") - - try: - realFile, status = get_file(download_path=self.config.download_path, file=file) - if web.HTTPFound.status_code == status: - return Response( - status=web.HTTPFound.status_code, - headers={ - "Location": str( - self.app.router["playlist"].url_for( - file=str(realFile).replace(self.config.download_path, "").strip("/") - ) - ), - }, - ) - - if web.HTTPNotFound.status_code == status: - return web.json_response(data={"error": f"File '{file}' does not exist."}, status=status) - - return web.Response( - text=await Playlist(download_path=Path(self.config.download_path), url=f"{base_path}/").make( - file=realFile - ), - headers={ - "Content-Type": "application/x-mpegURL", - "Cache-Control": "no-cache", - "Access-Control-Max-Age": "300", - }, - status=web.HTTPOk.status_code, - ) - except StreamingError as e: - return web.json_response(data={"error": str(e)}, status=web.HTTPNotFound.status_code) - - @route("GET", "api/player/m3u8/{mode}/{file:.*}.m3u8", "m3u8") - async def m3u8(self, request: Request) -> Response: - """ - Get the m3u8 file. - - Args: - request (Request): The request object. - - Returns: - Response: The response object. - - """ - file: str = request.match_info.get("file") - mode: str = request.match_info.get("mode") - - 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: - return web.json_response(data={"error": "file is required"}, status=web.HTTPBadRequest.status_code) - - duration = request.query.get("duration", None) - - if "subtitle" in mode: - if not duration: - return web.json_response(data={"error": "duration is required."}, status=web.HTTPBadRequest.status_code) - - duration = float(duration) - - base_path: str = self.config.base_path.rstrip("/") - - try: - cls = M3u8(download_path=Path(self.config.download_path), url=f"{base_path}/") - - realFile, status = get_file(download_path=self.config.download_path, file=file) - if web.HTTPFound.status_code == status: - return Response( - status=status, - headers={ - "Location": str( - self.app.router["m3u8"].url_for( - mode=mode, - file=str(realFile).replace(self.config.download_path, "").strip("/"), - ) - ), - }, - ) - - if web.HTTPNotFound.status_code == status: - return web.json_response(data={"error": f"File '{file}' does not exist."}, status=status) - - if "subtitle" in mode: - text = await cls.make_subtitle(file=realFile, duration=duration) - else: - text = await cls.make_stream(file=realFile) - 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", - }, - status=web.HTTPOk.status_code, - ) - - @route("GET", r"api/player/segments/{segment:\d+}/{file:.*}.ts", "segments") - async def segments(self, request: Request) -> Response: - """ - Get the segments. - - Args: - request (Request): The request object. - - Returns: - Response: The response object. - - """ - 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)) - - if not file: - return web.json_response(data={"error": "file is required"}, status=web.HTTPBadRequest.status_code) - - if not segment: - return web.json_response(data={"error": "segment id is required."}, status=web.HTTPBadRequest.status_code) - - realFile, status = get_file(download_path=self.config.download_path, file=file) - if web.HTTPFound.status_code == status: - return Response( - status=status, - headers={ - "Location": str( - self.app.router["segments"].url_for( - segment=segment, - file=str(realFile).replace(self.config.download_path, "").strip("/"), - ) - ), - }, - ) - - if web.HTTPNotFound.status_code == status: - return web.json_response(data={"error": f"File '{file}' does not exist."}, status=status) - - mtime = realFile.stat().st_mtime - - if request.if_modified_since and request.if_modified_since.timestamp() == mtime: - lastMod = time.strftime("%a, %d %b %Y %H:%M:%S GMT", datetime.fromtimestamp(mtime, tz=UTC).timetuple()) - return web.Response(status=web.HTTPNotModified.status_code, headers={"Last-Modified": lastMod}) - - resp = web.StreamResponse( - status=web.HTTPOk.status_code, - 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(mtime, tz=UTC).timetuple() - ), - "Expires": time.strftime( - "%a, %d %b %Y %H:%M:%S GMT", datetime.fromtimestamp(time.time() + 31536000, tz=UTC).timetuple() - ), - }, - ) - - await resp.prepare(request) - - await Segments( - download_path=self.config.download_path, - index=int(segment), - duration=float(f"{float(sd if sd else M3u8.duration):.6f}"), - vconvert=vc == 1, - aconvert=ac == 1, - ).stream(realFile, resp) - - return resp - - @route("GET", "api/player/subtitle/{file:.*}.vtt", "subtitles") - async def subtitles(self, request: Request) -> Response: - """ - Get the subtitles. - - Args: - request (Request): The request object. - - Returns: - Response: The response object. - - """ - file: str = request.match_info.get("file") - - if not file: - return web.json_response(data={"error": "file is required"}, status=web.HTTPBadRequest.status_code) - - realFile, status = get_file(download_path=self.config.download_path, file=file) - if web.HTTPFound.status_code == status: - return Response( - status=status, - headers={ - "Location": str( - self.app.router["subtitles"].url_for( - file=str(realFile).replace(self.config.download_path, "").strip("/") - ) - ), - }, - ) - - if web.HTTPNotFound.status_code == status: - return web.json_response(data={"error": f"File '{file}' does not exist."}, status=status) - - mtime = realFile.stat().st_mtime - - if request.if_modified_since and request.if_modified_since.timestamp() == mtime: - lastMod = time.strftime("%a, %d %b %Y %H:%M:%S GMT", datetime.fromtimestamp(mtime, tz=UTC).timetuple()) - return web.Response(status=web.HTTPNotModified.status_code, headers={"Last-Modified": lastMod}) - - return web.Response( - body=await Subtitle().make(file=realFile), - 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(mtime, tz=UTC).timetuple() - ), - "Expires": time.strftime( - "%a, %d %b %Y %H:%M:%S GMT", datetime.fromtimestamp(time.time() + 31536000, tz=UTC).timetuple() - ), - }, - status=web.HTTPOk.status_code, - ) - - @route("GET", "/", "index") - async def index(self, _: Request) -> Response: - """ - Get the index file. - - Args: - _: The request object. - - Returns: - Response: The response object. - - """ - if "/index.html" not in self._static_holder: - 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._static_holder["/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", "api/thumbnail", "get_thumbnail") - async def get_thumbnail(self, request: Request) -> Response: - """ - Get the thumbnail. - - Args: - request (Request): The request object. - - Returns: - Response: The response object. - - """ - url = request.query.get("url") - if not url: - return web.json_response(data={"error": "URL is required."}, status=web.HTTPForbidden.status_code) - - try: - validate_url(url) - except ValueError as e: - return web.json_response(data={"error": str(e)}, status=web.HTTPForbidden.status_code) - - try: - ytdlp_args = self.config.get_ytdlp_args() - opts = { - "proxy": ytdlp_args.get("proxy", None), - "headers": { - "User-Agent": ytdlp_args.get( - "user_agent", request.headers.get("User-Agent", f"YTPTube/{self.config.version}") - ), - }, - } - - async with httpx.AsyncClient(**opts) as client: - 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, tz=UTC).timetuple(), - ), - }, - ) - except Exception as e: - LOG.error(f"Error fetching thumbnail from '{url}'. '{e}'.") - return web.json_response( - data={"error": "failed to retrieve the thumbnail."}, status=web.HTTPInternalServerError.status_code - ) - - @route("GET", "api/random/background", "get_background") - async def get_background(self, request: Request) -> Response: - """ - Get random background. - - Args: - request (Request): The request object. - - Returns: - Response: The response object. - - """ - backend = None - - if self.isRequestingBackground: - return web.Response(status=web.HTTPTooManyRequests.status_code) - - try: - self.isRequestingBackground = True - backend = random.choice(self.config.pictures_backends) # noqa: S311 - CACHE_KEY_BING = "random_background_bing" - CACHE_KEY = "random_background" - - if self.cache.has(CACHE_KEY) and not request.query.get("force", False): - data = await self.cache.aget(CACHE_KEY) - return web.Response( - body=data.get("content"), - headers={ - "X-Cache": "HIT", - "X-Cache-TTL": str(await self.cache.attl(CACHE_KEY)), - "X-Image-Via": data.get("backend"), - **data.get("headers"), - }, - ) - - ytdlp_args = self.config.get_ytdlp_args() - opts = { - "proxy": ytdlp_args.get("proxy", None), - "headers": { - "User-Agent": ytdlp_args.get("user_agent", f"YTPTube/{self.config.version}"), - }, - } - - async with httpx.AsyncClient(**opts) as client: - if backend.startswith("https://www.bing.com/HPImageArchive.aspx"): - if not self.cache.has(CACHE_KEY_BING): - response = await client.request(method="GET", url=backend) - if response.status_code != web.HTTPOk.status_code: - return web.json_response( - data={"error": "failed to retrieve the random background image."}, - status=web.HTTPInternalServerError.status_code, - ) - - img_url: str | None = ag(response.json(), "images.0.url") - if not img_url: - return web.json_response( - data={"error": "failed to retrieve the random background image."}, - status=web.HTTPInternalServerError.status_code, - ) - - backend = f"https://www.bing.com{img_url}" - await self.cache.aset(key=CACHE_KEY_BING, value=backend, ttl=3600 * 24) - else: - backend: str = await self.cache.aget(CACHE_KEY_BING) - - LOG.debug(f"Requesting random picture from '{backend!s}'.") - - response = await client.request(method="GET", url=backend, follow_redirects=True) - - if response.status_code != web.HTTPOk.status_code: - return web.json_response( - data={"error": "failed to retrieve the random background image."}, - status=web.HTTPInternalServerError.status_code, - ) - - data = { - "content": response.content, - "backend": urlparse(backend).netloc, - "headers": { - "Content-Type": response.headers.get("Content-Type", "image/jpeg"), - "Content-Length": str(len(response.content)), - }, - } - - await self.cache.aset(key=CACHE_KEY, value=data, ttl=3600) - - LOG.debug(f"Random background image from '{backend!s}' cached.") - - return web.Response( - body=data.get("content"), - headers={ - "X-Cache": "MISS", - "X-Cache-TTL": "3600", - "X-Image-Via": data.get("backend"), - **data.get("headers"), - }, - ) - except Exception as e: - LOG.error(f"Failed to request random background image from '{backend!s}'.'. '{e!s}'.") - return web.json_response( - data={"error": "failed to retrieve the random background image."}, - status=web.HTTPInternalServerError.status_code, - ) - finally: - self.isRequestingBackground = False - - @route("GET", "api/file/ffprobe/{file:.*}", "ffprobe") - async def get_ffprobe(self, request: Request) -> Response: - """ - Get the ffprobe data. - - Args: - request (Request): The request object. - - Returns: - Response: The response object. - - """ - file: str = request.match_info.get("file") - if not file: - return web.json_response(data={"error": "file is required."}, status=web.HTTPBadRequest.status_code) - - try: - realFile, status = get_file(download_path=self.config.download_path, file=file) - if web.HTTPFound.status_code == status: - return Response( - status=web.HTTPFound.status_code, - headers={ - "Location": str( - self.app.router["ffprobe"].url_for( - file=str(realFile.relative_to(self.config.download_path).as_posix()).strip("/") - ) - ), - }, - ) - - if web.HTTPNotFound.status_code == status: - return web.json_response(data={"error": f"File '{file}' does not exist."}, status=status) - - return web.json_response( - data=await ffprobe(realFile), status=web.HTTPOk.status_code, dumps=self.encoder.encode - ) - except Exception as e: - return web.json_response(data={"error": str(e)}, status=web.HTTPInternalServerError.status_code) - - @route("GET", "api/file/info/{file:.*}", "file_info") - async def get_file_info(self, request: Request) -> Response: - """ - Get file info - - Args: - request (Request): The request object. - - Returns: - Response: The response object. - - """ - file: str = request.match_info.get("file") - if not file: - return web.json_response(data={"error": "file is required."}, status=web.HTTPBadRequest.status_code) - - try: - realFile, status = get_file(download_path=self.config.download_path, file=file) - if web.HTTPFound.status_code == status: - return Response( - status=web.HTTPFound.status_code, - headers={ - "Location": str( - self.app.router["file_info"].url_for( - file=str(realFile.relative_to(self.config.download_path).as_posix()).strip("/") - ) - ), - }, - ) - - if web.HTTPNotFound.status_code == status: - return web.json_response(data={"error": f"File '{file}' does not exist."}, status=status) - - ff_info = await ffprobe(realFile) - - response = { - "title": realFile.stem, - "ffprobe": ff_info, - "mimetype": get_mime_type(ff_info.get("metadata", {}), realFile), - "sidecar": get_file_sidecar(realFile), - } - - for key in response["sidecar"]: - for i, f in enumerate(response["sidecar"][key]): - response["sidecar"][key][i]["file"] = str( - realFile.with_name(f["file"].name).relative_to(self.config.download_path) - ).strip("/") - - return web.json_response(data=response, status=web.HTTPOk.status_code, dumps=self.encoder.encode) - except Exception as e: - LOG.exception(e) - return web.json_response(data={"error": str(e)}, status=web.HTTPInternalServerError.status_code) - - @route("GET", "api/notifications", "notifications") - async def notifications(self, _: Request) -> Response: - """ - Get the notification targets. - - Args: - _: The request object. - - Returns: - Response: The response object. - - """ - return web.json_response( - data={ - "notifications": Notification.get_instance().get_targets(), - "allowedTypes": list(NotificationEvents.get_events().values()), - }, - status=web.HTTPOk.status_code, - dumps=self.encoder.encode, - ) - - @route("PUT", "api/notifications", "notification_add") - async def notification_add(self, request: Request) -> Response: - """ - Add notification targets. - - Args: - request (Request): The request object. - - Returns: - Response: The response object. - - """ - post = await request.json() - if not isinstance(post, list): - return web.json_response( - {"error": "Invalid request body expecting list with dicts."}, - status=web.HTTPBadRequest.status_code, - ) - - targets: list = [] - - ins = Notification.get_instance() - for item in post: - if not isinstance(item, dict): - return web.json_response( - {"error": "Invalid request body expecting list with dicts."}, - status=web.HTTPBadRequest.status_code, - ) - - if not item.get("id", None) or validate_uuid(item.get("id"), version=4): - item["id"] = str(uuid.uuid4()) - try: - Notification.validate(item) - except ValueError as e: - return web.json_response( - {"error": f"Invalid notification target settings. {e!s}", "data": item}, - status=web.HTTPBadRequest.status_code, - ) - - targets.append(ins.make_target(item)) - - try: - ins.save(targets=targets) - ins.load() - except Exception as e: - LOG.exception(e) - return web.json_response({"error": "Failed to save tasks."}, status=web.HTTPInternalServerError.status_code) - - data = {"notifications": targets, "allowedTypes": list(NotificationEvents.get_events().values())} - - return web.json_response(data=data, status=web.HTTPOk.status_code, dumps=self.encoder.encode) - - @route("POST", "api/notifications/test", "notification_test") - async def notification_test(self, _: Request) -> Response: - """ - Test the notification. - - Args: - _: The request object. - - Returns: - Response: The response object. - - """ - data = message("test", "This is a test notification.") - - await self._notify.emit(Events.TEST, data=data) - - return web.json_response(data=data, status=web.HTTPOk.status_code, dumps=self.encoder.encode) - - @route("GET", "api/file/browser/{path:.*}", "file_browser") - async def file_browser(self, request: Request) -> Response: - """ - Get the file browser. - - Args: - request (Request): The request object. - - Returns: - Response: The response object. - - """ - if not self.config.browser_enabled: - return web.json_response(data={"error": "File browser is disabled."}, status=web.HTTPForbidden.status_code) - - req_path: str = request.match_info.get("path") - req_path: str = "/" if not req_path else unquote_plus(req_path) - - test: Path = Path(self.config.download_path).joinpath(req_path) - if not test.exists(): - return web.json_response( - data={"error": f"path '{req_path}' does not exist."}, status=web.HTTPNotFound.status_code - ) - - if not test.is_dir() and not test.is_symlink(): - return web.json_response( - data={"error": f"path '{req_path}' is not a directory."}, status=web.HTTPBadRequest.status_code - ) - - try: - return web.json_response( - data={ - "path": req_path, - "contents": get_files(base_path=Path(self.config.download_path), dir=req_path), - }, - status=web.HTTPOk.status_code, - dumps=self.encoder.encode, - ) - except OSError as e: - LOG.exception(e) - return web.json_response(data={"error": str(e)}, status=web.HTTPInternalServerError.status_code) - - @route("GET", "api/dev/loop") - async def debug_asyncio(self, _: Request) -> Response: - if not self.config.is_dev(): - return web.json_response( - data={"error": "This endpoint is only available in development mode."}, - status=web.HTTPForbidden.status_code, - ) - - import traceback - - tasks = [] - for task in asyncio.all_tasks(): - task_info = {"task": str(task), "stack": []} - for frame in task.get_stack(): - formatted = traceback.format_stack(f=frame) - task_info["stack"].extend(formatted) - - tasks.append(task_info) - - return web.json_response( - data={ - "total_tasks": len(tasks), - "loop": str(asyncio.get_event_loop()), - "tasks": tasks, - }, - status=web.HTTPOk.status_code, - dumps=self.encoder.encode, - ) - - @route("GET", "api/dev/workers", "pool_list") - async def pool_list(self, _) -> Response: - """ - Get the workers status. - - Args: - _: The request object. - - Returns: - Response: The response object. - - """ - if not self.config.is_dev(): - return web.json_response( - {"error": "This endpoint is only available in development mode."}, - status=web.HTTPNotFound.status_code, - ) - - if self.queue.pool is None: - return web.json_response({"error": "Worker pool not initialized."}, status=web.HTTPNotFound.status_code) - - status = self.queue.pool.get_workers_status() - - data = [] - - 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, - } - ) - - 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"<>"), - ) - - @route("POST", "api/dev/workers", "pool_start") - async def pool_restart(self, _) -> Response: - """ - Restart the workers pool. - - Args: - _: The request object. - - Returns: - Response: The response object. - - """ - if not self.config.is_dev(): - return web.json_response( - {"error": "This endpoint is only available in development mode."}, - status=web.HTTPNotFound.status_code, - ) - - if self.queue.pool is None: - return web.json_response({"error": "Worker pool not initialized."}, status=web.HTTPNotFound.status_code) - - self.queue.pool.start() - - return web.json_response({"message": "Workers pool being restarted."}, status=web.HTTPOk.status_code) - - @route("PATCH", "api/dev/workers/{id}", "worker_restart") - async def worker_restart(self, request: Request) -> Response: - """ - Restart a worker. - - Args: - request (Request): The request object. - - Returns: - Response: The response object - - """ - if not self.config.is_dev(): - return web.json_response( - {"error": "This endpoint is only available in development mode."}, - status=web.HTTPNotFound.status_code, - ) - - worker_id: str = request.match_info.get("id") - if not worker_id: - 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=web.HTTPNotFound.status_code) - - status = await self.queue.pool.restart(worker_id, "requested by user.") - - return web.json_response({"status": "restarted" if status else "in_error_state"}, status=web.HTTPOk.status_code) - - @route("DELETE", "api/dev/workers/{id}", "worker_stop") - async def worker_stop(self, request: Request) -> Response: - """ - Stop a worker. - - Args: - request (Request): The request object. - - Returns: - Response: The response object. - - """ - if not self.config.is_dev(): - return web.json_response( - {"error": "This endpoint is only available in development mode."}, - status=web.HTTPNotFound.status_code, - ) - - worker_id: str = request.match_info.get("id") - if not worker_id: - raise web.HTTPBadRequest(text="worker id is required.") - - if self.queue.pool is None: - return web.json_response({"error": "Worker pool not initialized."}, status=web.HTTPNotFound.status_code) - - status = await self.queue.pool.stop(worker_id, "requested by user.") - - return web.json_response({"status": "stopped" if status else "in_error_state"}, status=web.HTTPOk.status_code) + LOG.debug(f"Loading module '{full_name}'.") + importlib.import_module(full_name) + except ImportError as e: + LOG.error(f"Failed to import module '{full_name}': {e}") diff --git a/app/library/HttpSocket.py b/app/library/HttpSocket.py index d1612b3c..b0b35dc6 100644 --- a/app/library/HttpSocket.py +++ b/app/library/HttpSocket.py @@ -12,7 +12,6 @@ import anyio import socketio from aiohttp import web -from .common import Common from .config import Config from .DownloadQueue import DownloadQueue from .encoder import Encoder @@ -24,7 +23,7 @@ from .Utils import is_downloaded, tail_log LOG = logging.getLogger("socket_api") -class HttpSocket(Common): +class HttpSocket: """ This class is used to handle WebSocket events. """ @@ -68,8 +67,6 @@ class HttpSocket(Common): self._notify.subscribe("frontend", emit, f"{__class__.__name__}.emit") - super().__init__(queue=queue, encoder=encoder, config=config) - @staticmethod def ws_event(func): # type: ignore """ @@ -102,7 +99,7 @@ class HttpSocket(Common): self._notify.subscribe( Events.ADD_URL, - lambda data, _, **kwargs: self.add(item=Item.format(data.data)), # noqa: ARG005 + lambda data, _, **kwargs: self.queue.add(item=Item.format(data.data)), # noqa: ARG005 f"{__class__.__name__}.add", ) @@ -237,7 +234,7 @@ class HttpSocket(Common): try: await self._notify.emit( event=Events.STATUS, - data=await self.add(item=Item.format(data)), + data=await self.queue.add(item=Item.format(data)), to=sid, ) except ValueError as e: diff --git a/app/library/Presets.py b/app/library/Presets.py index dbb044ff..6ef5e2c6 100644 --- a/app/library/Presets.py +++ b/app/library/Presets.py @@ -175,13 +175,14 @@ class Presets(metaclass=Singleton): Presets: The current instance. """ + has: int = len(self._items) self.clear() if not self._file.exists() or self._file.stat().st_size < 10: return self try: - LOG.info(f"Loading '{self._file}'.") + LOG.info(f"{'Reloading' if has else 'Loading'} '{self._file}'.") presets: dict = json.loads(self._file.read_text()) except Exception as e: LOG.error(f"Failed to parse '{self._file}'. '{e}'.") diff --git a/app/library/Scheduler.py b/app/library/Scheduler.py index 2baac1eb..e26d49fe 100644 --- a/app/library/Scheduler.py +++ b/app/library/Scheduler.py @@ -103,7 +103,7 @@ class Scheduler(metaclass=Singleton): self._jobs[job_id] = job - LOG.debug(f"Added job '{job_id}' to the schedule.") + LOG.debug(f"Added '{job_id}' to the scheduler.") return job_id diff --git a/app/library/Tasks.py b/app/library/Tasks.py index 71666845..8981fc0b 100644 --- a/app/library/Tasks.py +++ b/app/library/Tasks.py @@ -17,7 +17,7 @@ from .Scheduler import Scheduler from .Singleton import Singleton from .Utils import init_class -LOG = logging.getLogger("tasks") +LOG: logging.Logger = logging.getLogger("tasks") @dataclass(kw_only=True) @@ -124,13 +124,14 @@ class Tasks(metaclass=Singleton): Tasks: The current instance. """ + has_tasks: bool = len(self._tasks) > 0 self.clear() if not self._file.exists() or self._file.stat().st_size < 1: return self try: - LOG.info(f"Loading '{self._file}'.") + LOG.info(f"{'Reloading' if has_tasks else 'Loading'} '{self._file}'.") tasks = json.loads(self._file.read_text()) except Exception as e: LOG.error(f"Error loading '{self._file}'. '{e!s}'.") diff --git a/app/library/Utils.py b/app/library/Utils.py index c65cf851..44a2cbc0 100644 --- a/app/library/Utils.py +++ b/app/library/Utils.py @@ -20,7 +20,7 @@ from Crypto.Cipher import AES from .LogWrapper import LogWrapper -LOG = logging.getLogger("Utils") +LOG: logging.Logger = logging.getLogger("Utils") REMOVE_KEYS: list = [ { @@ -803,7 +803,8 @@ def get_files(base_path: Path | str, dir: str | None = None): if file.name.startswith(".") or file.name.startswith("_"): continue - if file.is_symlink(): + is_symlink: bool = file.is_symlink() + if is_symlink: try: test: Path = file.resolve() test.relative_to(base_path) @@ -831,9 +832,10 @@ def get_files(base_path: Path | str, dir: str | None = None): content_type = "download" stat = file.stat() + contents.append( { - "type": "file" if file.is_file() else "dir", + "type": "file" if file.is_file() else "link" if is_symlink else "dir", "content_type": content_type, "name": file.name, "path": str(file.relative_to(base_path)).strip("/"), @@ -843,6 +845,7 @@ def get_files(base_path: Path | str, dir: str | None = None): "ctime": datetime.fromtimestamp(stat.st_ctime, tz=UTC).isoformat(), "is_dir": file.is_dir(), "is_file": file.is_file(), + "is_symlink": is_symlink, } ) diff --git a/app/library/YTDLPOpts.py b/app/library/YTDLPOpts.py index 2954165a..a7a8a43f 100644 --- a/app/library/YTDLPOpts.py +++ b/app/library/YTDLPOpts.py @@ -2,7 +2,7 @@ import logging from pathlib import Path from .config import Config -from .Presets import Presets, Preset +from .Presets import Preset, Presets from .Singleton import Singleton from .Utils import REMOVE_KEYS, arg_converter, calc_download_path, load_cookies, merge_dict diff --git a/app/library/common.py b/app/library/common.py index f8ff5fda..a8736f04 100644 --- a/app/library/common.py +++ b/app/library/common.py @@ -20,11 +20,10 @@ class Common: config: Config | None = None, ): super().__init__() - self.queue = queue or DownloadQueue.get_instance() - self.encoder = encoder or Encoder() - - config = config or Config.get_instance() - self.default_preset = config.default_preset + self.queue: DownloadQueue = queue or DownloadQueue.get_instance() + self.encoder: Encoder = encoder or Encoder() + config: Config = config or Config.get_instance() + self.default_preset: str = config.default_preset async def add(self, item: Item) -> dict[str, str]: """ diff --git a/app/library/router.py b/app/library/router.py new file mode 100644 index 00000000..ce466545 --- /dev/null +++ b/app/library/router.py @@ -0,0 +1,124 @@ +import logging +import re +from collections.abc import Awaitable +from functools import wraps + +LOG: logging.Logger = logging.getLogger(__name__) + + +class Route: + """ + A class to represent an HTTP route. + + Attributes: + method (str): The HTTP method (GET, POST, etc.). + path (str): The path for the route. + name (str): The name of the route. + handler (Awaitable): The function that handles the route. + + """ + + def __init__(self, method: str, path: str, name: str, handler: Awaitable): + self.method = method.upper() + self.path = path + self.name = name + self.handler: Awaitable = handler + + +ROUTES: dict[str, Route] = {} + + +def make_route_name(method: str, path: str) -> str: + method = method.lower() + path = path.strip("/") + + segments = [] + for part in path.split("/"): + part = re.sub(r"[^\w]", "_", part) # remove invalid chars + if not part: + part = "part" + elif part[0].isdigit(): + part = f"p_{part}" + segments.append(part) + + return f"{method}:" + ".".join(segments or ["root"]) + + +def route(method: str, path: str, name: str | None = None, **kwargs) -> Awaitable: + """ + Decorator to mark a method as an HTTP route handler. + + Args: + method (str): The HTTP method. + path (str): The path to the route. + name (str): The name of the route. + kwargs: Additional keyword arguments. + + Returns: + Awaitable: The decorated function. + + """ + if not name: + name = make_route_name(method, path) + + def decorator(func): + @wraps(func) + async def wrapper(*args, **kwargs): + return await func(*args, **kwargs) + + ROUTES[name] = Route(method=method.upper(), path=path, name=name, handler=wrapper) + if path.endswith("/") and "/" != path and not kwargs.get("no_slash", False): + ROUTES[f"{name}_no_slash"] = Route( + method=method.upper(), path=path[:-1], name=f"{name}_no_slash", handler=wrapper + ) + + return wrapper + + return decorator + + +def add_route(method: str, path: str, handler: Awaitable, name: str | None = None, **kwargs): + """ + Decorator to mark a method as an HTTP route handler. + + Args: + method (str): The HTTP method. + path (str): The path to the route. + name (str): The name of the route. + handler (Awaitable): The function that handles the route. + kwargs: Additional keyword arguments. + + """ + if not name: + name = make_route_name(method, path) + + ROUTES[name] = Route(method=method.upper(), path=path, name=name, handler=handler) + if path.endswith("/") and "/" != path and not kwargs.get("no_slash", False): + ROUTES[f"{name}_no_slash"] = Route( + method=method.upper(), path=path[:-1], name=f"{name}_no_slash", handler=handler + ) + + +def get_route(name: str) -> dict[str, Route] | None: + """ + Get the route information by name. + + Args: + name (str): The name of the route. + + Returns: + dict: The route information, or None if not found. + + """ + return ROUTES.get(name) + + +def get_routes() -> dict[str, Route]: + """ + Get all registered routes. + + Returns: + dict[str, dict]: A dictionary of all registered routes. + + """ + return ROUTES diff --git a/app/main.py b/app/main.py index 1b646740..00c9d9cd 100644 --- a/app/main.py +++ b/app/main.py @@ -1,24 +1,31 @@ #!/usr/bin/env python3 +import sys +from pathlib import Path + +APP_ROOT = str((Path(__file__).parent / "..").resolve()) +if APP_ROOT not in sys.path: + sys.path.insert(0, APP_ROOT) + import asyncio import logging import sqlite3 -import sys from pathlib import Path import caribou import magic from aiohttp import web -from library.conditions import Conditions -from library.config import Config -from library.DownloadQueue import DownloadQueue -from library.Events import EventBus, Events -from library.HttpAPI import HttpAPI -from library.HttpSocket import HttpSocket -from library.Notifications import Notification -from library.Presets import Presets -from library.Scheduler import Scheduler -from library.Tasks import Tasks + +from app.library.conditions import Conditions +from app.library.config import Config +from app.library.DownloadQueue import DownloadQueue +from app.library.Events import EventBus, Events +from app.library.HttpAPI import HttpAPI +from app.library.HttpSocket import HttpSocket +from app.library.Notifications import Notification +from app.library.Presets import Presets +from app.library.Scheduler import Scheduler +from app.library.Tasks import Tasks LOG = logging.getLogger("app") MIME = magic.Magic(mime=True) @@ -117,7 +124,7 @@ class Main: HTTP_LOGGER = None if self._config.access_log: - from library.HttpAPI import LOG as HTTP_LOGGER + from app.library.HttpAPI import LOG as HTTP_LOGGER HTTP_LOGGER.addFilter(lambda record: f"GET {self._app.router['ping'].url_for()}" not in record.getMessage()) diff --git a/app/native.py b/app/native.py index 120b2b0a..9aa302af 100644 --- a/app/native.py +++ b/app/native.py @@ -1,9 +1,17 @@ +#!/usr/bin/env python3 +import sys +from pathlib import Path + +APP_ROOT = str((Path(__file__).parent / "..").resolve()) +if APP_ROOT not in sys.path: + sys.path.insert(0, APP_ROOT) + + import json import os import queue import socket import threading -from pathlib import Path ready = threading.Event() exception_holder = queue.Queue() diff --git a/app/routes/_static.py b/app/routes/_static.py new file mode 100644 index 00000000..816a2505 --- /dev/null +++ b/app/routes/_static.py @@ -0,0 +1,135 @@ +import logging +from pathlib import Path + +import magic +from aiohttp import web +from aiohttp.web import Request, Response + +from app.library.config import Config +from app.library.router import add_route + +MIME = magic.Magic(mime=True) +LOG: logging.Logger = logging.getLogger(__name__) + +STATIC_FILES: dict[str, dict] = {} + +EXT_TO_MIME: dict = { + ".html": "text/html", + ".css": "text/css", + ".js": "application/javascript", + ".json": "application/json", + ".ico": "image/x-icon", +} + +FRONTEND_ROUTES: list[str] = [ + "/console/", + "/presets/", + "/tasks/", + "/notifications/", + "/changelog/", + "/logs/", + "/conditions/", + "/browser/", + "/browser/{path:.*}", +] + + +async def serve_static_file(request: Request, config: Config) -> Response: + """ + Preload static files from the ui/exported folder. + + Args: + request (Request): The request object. + config (Config): The configuration instance. + + Returns: + Response: The response object. + + """ + path: str = request.path + + if "/" != config.base_path and path.startswith(config.base_path): + path = path.replace(config.base_path[:-1], "", 1) + + if path not in STATIC_FILES: + for k in STATIC_FILES: + if path.startswith(k): + path = k + break + else: + return web.json_response({"error": "File not found.", "file": path}, status=web.HTTPNotFound.status_code) + + item: dict = STATIC_FILES[path] + + return web.FileResponse( + path=item["file"], + headers={ + "Pragma": "public", + "Cache-Control": "public, max-age=31536000", + "Content-Type": item.get("content_type"), + }, + status=web.HTTPOk.status_code, + ) + + +def preload_static(root_path: Path, config: Config) -> None: + """ + Preload static files from the ui/exported folder. + + Args: + root_path (Path): The root path of the application. + config (Config): The configuration instance. + + """ + global STATIC_FILES # noqa: PLW0602 + + static_dir: Path = (root_path / "ui" / "exported").absolute() + if not static_dir.exists(): + static_dir = (root_path.parent / "ui" / "exported").absolute() + if not static_dir.exists(): + msg: str = f"Could not find the frontend UI static assets. '{static_dir}'." + raise ValueError(msg) + + preloaded = 0 + + for file in static_dir.rglob("*.*"): + if ".map" == file.suffix: + continue + + uri_path: str = f"/{file.relative_to(static_dir).as_posix()!s}" + # uri_path: str = f"/{str(file.as_posix()).replace(f'{static_dir.as_posix()!s}/', '')}" + contentType = EXT_TO_MIME.get(file.suffix) + if not contentType: + contentType = MIME.from_file(file) + + STATIC_FILES[uri_path] = { + "uri": uri_path, + "content_type": contentType, + "file": file, + } + + add_route(method="GET", path=uri_path, handler=serve_static_file) + preloaded += 1 + + if "/index.html" in STATIC_FILES: + for path in FRONTEND_ROUTES: + STATIC_FILES[path] = STATIC_FILES["/index.html"] + STATIC_FILES[path.lstrip("/")] = STATIC_FILES["/index.html"] + add_route(method="GET", path=path, handler=serve_static_file) + LOG.debug(f"Preloading frontend static route '{path}'.") + preloaded += 1 + + # Add main app route + STATIC_FILES["/"] = STATIC_FILES["/index.html"] + STATIC_FILES[config.base_path] = STATIC_FILES["/index.html"] + add_route(method="GET", path=config.base_path, handler=serve_static_file, name="index") + + if preloaded < 1: + message = f"Failed to find any static files in '{static_dir}'." + if config.ignore_ui: + LOG.warning(message) + return + + raise ValueError(message) + + LOG.info(f"Loaded '{preloaded}' static files.") diff --git a/app/routes/browser.py b/app/routes/browser.py new file mode 100644 index 00000000..4a4d0ef0 --- /dev/null +++ b/app/routes/browser.py @@ -0,0 +1,157 @@ +import logging +from pathlib import Path +from urllib.parse import unquote_plus + +from aiohttp import web +from aiohttp.web import Request, Response + +from app.library.config import Config +from app.library.encoder import Encoder +from app.library.ffprobe import ffprobe +from app.library.router import route +from app.library.Utils import get_file, get_file_sidecar, get_files, get_mime_type + +LOG: logging.Logger = logging.getLogger(__name__) + + +@route("GET", "api/file/ffprobe/{file:.*}", "ffprobe") +async def get_ffprobe(request: Request, config: Config, encoder: Encoder, app: web.Application) -> Response: + """ + Get the ffprobe data. + + Args: + request (Request): The request object. + config (Config): The configuration object. + encoder (Encoder): The encoder object. + app (web.Application): The aiohttp application object. + + Returns: + Response: The response object. + + """ + file: str = request.match_info.get("file") + if not file: + return web.json_response(data={"error": "file is required."}, status=web.HTTPBadRequest.status_code) + + try: + realFile, status = get_file(download_path=config.download_path, file=file) + if web.HTTPFound.status_code == status: + return Response( + status=web.HTTPFound.status_code, + headers={ + "Location": str( + app.router["ffprobe"].url_for( + file=str(realFile.relative_to(config.download_path).as_posix()).strip("/") + ) + ), + }, + ) + + if web.HTTPNotFound.status_code == status: + return web.json_response(data={"error": f"File '{file}' does not exist."}, status=status) + + return web.json_response(data=await ffprobe(realFile), status=web.HTTPOk.status_code, dumps=encoder.encode) + except Exception as e: + return web.json_response(data={"error": str(e)}, status=web.HTTPInternalServerError.status_code) + + +@route("GET", "api/file/info/{file:.*}", "file_info") +async def get_file_info(request: Request, config: Config, encoder: Encoder, app: web.Application) -> Response: + """ + Get file info + + Args: + request (Request): The request object. + config (Config): The configuration object. + encoder (Encoder): The encoder object. + app (web.Application): The aiohttp application object. + + Returns: + Response: The response object. + + """ + file: str = request.match_info.get("file") + if not file: + return web.json_response(data={"error": "file is required."}, status=web.HTTPBadRequest.status_code) + + try: + realFile, status = get_file(download_path=config.download_path, file=file) + if web.HTTPFound.status_code == status: + return Response( + status=web.HTTPFound.status_code, + headers={ + "Location": str( + app.router["file_info"].url_for( + file=str(realFile.relative_to(config.download_path).as_posix()).strip("/") + ) + ), + }, + ) + + if web.HTTPNotFound.status_code == status: + return web.json_response(data={"error": f"File '{file}' does not exist."}, status=status) + + ff_info = await ffprobe(realFile) + + response = { + "title": realFile.stem, + "ffprobe": ff_info, + "mimetype": get_mime_type(ff_info.get("metadata", {}), realFile), + "sidecar": get_file_sidecar(realFile), + } + + for key in response["sidecar"]: + for i, f in enumerate(response["sidecar"][key]): + response["sidecar"][key][i]["file"] = str( + realFile.with_name(f["file"].name).relative_to(config.download_path) + ).strip("/") + + return web.json_response(data=response, status=web.HTTPOk.status_code, dumps=encoder.encode) + except Exception as e: + LOG.exception(e) + return web.json_response(data={"error": str(e)}, status=web.HTTPInternalServerError.status_code) + + +@route("GET", "api/file/browser/{path:.*}", "file_browser") +async def file_browser(request: Request, config: Config, encoder: Encoder) -> Response: + """ + Get the file browser. + + Args: + request (Request): The request object. + config (Config): The configuration object. + encoder (Encoder): The encoder object. + + Returns: + Response: The response object. + + """ + if not config.browser_enabled: + return web.json_response(data={"error": "File browser is disabled."}, status=web.HTTPForbidden.status_code) + + req_path: str = request.match_info.get("path") + req_path: str = "/" if not req_path else unquote_plus(req_path) + + test: Path = Path(config.download_path).joinpath(req_path) + if not test.exists(): + return web.json_response( + data={"error": f"path '{req_path}' does not exist."}, status=web.HTTPNotFound.status_code + ) + + if not test.is_dir() and not test.is_symlink(): + return web.json_response( + data={"error": f"path '{req_path}' is not a directory."}, status=web.HTTPBadRequest.status_code + ) + + try: + return web.json_response( + data={ + "path": req_path, + "contents": get_files(base_path=Path(config.download_path), dir=req_path), + }, + status=web.HTTPOk.status_code, + dumps=encoder.encode, + ) + except OSError as e: + LOG.exception(e) + return web.json_response(data={"error": str(e)}, status=web.HTTPInternalServerError.status_code) diff --git a/app/routes/conditions.py b/app/routes/conditions.py new file mode 100644 index 00000000..9895cd45 --- /dev/null +++ b/app/routes/conditions.py @@ -0,0 +1,191 @@ +import logging +import uuid + +from aiohttp import web +from aiohttp.web import Request, Response + +from app.library.cache import Cache +from app.library.conditions import Condition, Conditions +from app.library.config import Config +from app.library.encoder import Encoder +from app.library.Events import EventBus, Events +from app.library.router import route +from app.library.Utils import extract_info, init_class, validate_uuid +from app.library.YTDLPOpts import YTDLPOpts + +LOG: logging.Logger = logging.getLogger(__name__) + + +@route("GET", "api/conditions/", "conditions_list") +async def conditions_list(encoder: Encoder) -> Response: + """ + Get the conditions + + Args: + encoder (Encoder): The encoder instance. + + Returns: + Response: The response object. + + """ + return web.json_response( + data=Conditions.get_instance().get_all(), + status=web.HTTPOk.status_code, + dumps=encoder.encode, + ) + + +@route("PUT", "api/conditions/", "conditions_add") +async def conditions_add(request: Request, encoder: Encoder, notify: EventBus) -> Response: + """ + Save Conditions. + + Args: + request (Request): The request object. + encoder (Encoder): The encoder instance. + notify (EventBus): The event bus instance. + + Returns: + Response: The response object + + """ + data = await request.json() + + if not isinstance(data, list): + return web.json_response( + {"error": "Invalid request body expecting list with dicts."}, + status=web.HTTPBadRequest.status_code, + ) + + items: list = [] + + cls = Conditions.get_instance() + + for item in data: + if not isinstance(item, dict): + return web.json_response( + {"error": "Invalid request body expecting list with dicts."}, + status=web.HTTPBadRequest.status_code, + ) + + if not item.get("name"): + return web.json_response( + {"error": "name is required.", "data": item}, status=web.HTTPBadRequest.status_code + ) + + if not item.get("filter"): + return web.json_response( + {"error": "filter is required.", "data": item}, status=web.HTTPBadRequest.status_code + ) + + if not item.get("cli"): + return web.json_response( + {"error": "command options for yt-dlp is required.", "data": item}, + status=web.HTTPBadRequest.status_code, + ) + + if not item.get("id", None) or not validate_uuid(item.get("id"), version=4): + item["id"] = str(uuid.uuid4()) + + try: + cls.validate(item) + except ValueError as e: + return web.json_response( + {"error": f"Failed to validate condition '{item.get('name')}'. '{e!s}'"}, + status=web.HTTPBadRequest.status_code, + ) + + items.append(init_class(Condition, item)) + + try: + items = cls.save(items=items).load().get_all() + except Exception as e: + LOG.exception(e) + return web.json_response( + {"error": "Failed to save conditions.", "message": str(e)}, + status=web.HTTPInternalServerError.status_code, + ) + + await notify.emit(Events.CONDITIONS_UPDATE, data=items) + return web.json_response(data=items, status=web.HTTPOk.status_code, dumps=encoder.encode) + + +@route("POST", "api/conditions/test/", "conditions_test") +async def conditions_test(request: Request, encoder: Encoder, cache: Cache, config: Config) -> Response: + """ + Test condition against URL. + + Args: + request (Request): The request object. + encoder (Encoder): The encoder instance. + cache (Cache): The cache instance. + config (Config): The configuration instance. + + Returns: + Response: The response object + + """ + params = await request.json() + + if not isinstance(params, dict): + return web.json_response( + {"error": "Invalid request body expecting dict."}, + status=web.HTTPBadRequest.status_code, + ) + + url: str | None = params.get("url") + if not url: + return web.json_response({"error": "url is required."}, status=web.HTTPBadRequest.status_code) + + cond: str | None = params.get("condition") + if not cond: + return web.json_response({"error": "condition is required."}, status=web.HTTPBadRequest.status_code) + + try: + preset: str = params.get("preset", config.default_preset) + key: str = cache.hash(url + str(preset)) + if not cache.has(key): + opts = {} + if ytdlp_proxy := config.get_ytdlp_args().get("proxy", None): + opts["proxy"] = ytdlp_proxy + ytdlp_opts = YTDLPOpts.get_instance().preset(name=preset).add(opts).get_all() + + data = extract_info( + config=ytdlp_opts, + url=url, + debug=False, + no_archive=True, + follow_redirect=True, + sanitize_info=True, + ) + if not data: + return web.json_response( + data={"error": f"Failed to extract info from '{url!s}'."}, + status=web.HTTPBadRequest.status_code, + ) + cache.set(key=key, value=data, ttl=600) + else: + data = cache.get(key) + except Exception as e: + LOG.exception(e) + return web.json_response( + data={"error": f"Failed to extract video info. '{e!s}'"}, + status=web.HTTPInternalServerError.status_code, + ) + + try: + from yt_dlp.utils import match_str + + status = match_str(cond, data) + except Exception as e: + LOG.exception(e) + return web.json_response( + data={"error": str(e)}, + status=web.HTTPBadRequest.status_code, + ) + + return web.json_response( + data={"status": status, "condition": cond, "data": data}, + status=web.HTTPOk.status_code, + dumps=encoder.encode, + ) diff --git a/app/routes/dev.py b/app/routes/dev.py new file mode 100644 index 00000000..06dc4e81 --- /dev/null +++ b/app/routes/dev.py @@ -0,0 +1,180 @@ +import asyncio +import json +import logging + +from aiohttp import web +from aiohttp.web import Request, Response + +from app.library import DownloadQueue +from app.library.config import Config +from app.library.encoder import Encoder +from app.library.router import route + +LOG: logging.Logger = logging.getLogger(__name__) + + +@route("GET", "api/dev/loop/", "debug_loop") +async def debug_asyncio(config: Config, encoder: Encoder) -> Response: + if not config.is_dev(): + return web.json_response( + data={"error": "This endpoint is only available in development mode."}, + status=web.HTTPForbidden.status_code, + ) + + import traceback + + tasks = [] + for task in asyncio.all_tasks(): + task_info = {"task": str(task), "stack": []} + for frame in task.get_stack(): + formatted = traceback.format_stack(f=frame) + task_info["stack"].extend(formatted) + + tasks.append(task_info) + + return web.json_response( + data={ + "total_tasks": len(tasks), + "loop": str(asyncio.get_event_loop()), + "tasks": tasks, + }, + status=web.HTTPOk.status_code, + dumps=encoder.encode, + ) + + +@route("GET", "api/dev/workers/", "pool_list") +async def pool_list(config: Config, queue: DownloadQueue) -> Response: + """ + Get the workers status. + + Args: + config (Config): The configuration object. + queue (DownloadQueue): The download queue object. + + Returns: + Response: The response object. + + """ + if not config.is_dev(): + return web.json_response( + {"error": "This endpoint is only available in development mode."}, + status=web.HTTPNotFound.status_code, + ) + + if queue.pool is None: + return web.json_response({"error": "Worker pool not initialized."}, status=web.HTTPNotFound.status_code) + + status = queue.pool.get_workers_status() + + data = [] + + 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, + } + ) + + return web.json_response( + data={ + "open": queue.pool.has_open_workers(), + "count": queue.pool.get_available_workers(), + "workers": data, + }, + status=web.HTTPOk.status_code, + dumps=lambda obj: json.dumps(obj, default=lambda o: f"<>"), + ) + + +@route("POST", "api/dev/workers/", "pool_start") +async def pool_restart(config: Config, queue: DownloadQueue) -> Response: + """ + Restart the workers pool. + + Args: + config (Config): The configuration object. + queue (DownloadQueue): The download queue object. + + Returns: + Response: The response object. + + """ + if not config.is_dev(): + return web.json_response( + {"error": "This endpoint is only available in development mode."}, + status=web.HTTPNotFound.status_code, + ) + + if queue.pool is None: + return web.json_response({"error": "Worker pool not initialized."}, status=web.HTTPNotFound.status_code) + + queue.pool.start() + + return web.json_response({"message": "Workers pool being restarted."}, status=web.HTTPOk.status_code) + + +@route("PATCH", "api/dev/workers/{id}", "worker_restart") +async def worker_restart(request: Request, config: Config, queue: DownloadQueue) -> Response: + """ + Restart a worker. + + Args: + request (Request): The request object. + config (Config): The configuration object. + queue (DownloadQueue): The download queue object. + + Returns: + Response: The response object + + """ + if not config.is_dev(): + return web.json_response( + {"error": "This endpoint is only available in development mode."}, + status=web.HTTPNotFound.status_code, + ) + + worker_id: str = request.match_info.get("id") + if not worker_id: + return web.json_response({"error": "worker id is required."}, status=web.HTTPBadRequest.status_code) + + if queue.pool is None: + return web.json_response({"error": "Worker pool not initialized."}, status=web.HTTPNotFound.status_code) + + status = await queue.pool.restart(worker_id, "requested by user.") + + return web.json_response({"status": "restarted" if status else "in_error_state"}, status=web.HTTPOk.status_code) + + +@route("DELETE", "api/dev/workers/{id}", "worker_stop") +async def worker_stop(request: Request, config: Config, queue: DownloadQueue) -> Response: + """ + Stop a worker. + + Args: + request (Request): The request object. + config (Config): The configuration object. + queue (DownloadQueue): The download queue object. + + Returns: + Response: The response object. + + """ + if not config.is_dev(): + return web.json_response( + {"error": "This endpoint is only available in development mode."}, + status=web.HTTPNotFound.status_code, + ) + + worker_id: str = request.match_info.get("id") + if not worker_id: + raise web.HTTPBadRequest(text="worker id is required.") + + if queue.pool is None: + return web.json_response({"error": "Worker pool not initialized."}, status=web.HTTPNotFound.status_code) + + status = await queue.pool.stop(worker_id, "requested by user.") + + return web.json_response({"status": "stopped" if status else "in_error_state"}, status=web.HTTPOk.status_code) diff --git a/app/routes/history.py b/app/routes/history.py new file mode 100644 index 00000000..6765c24b --- /dev/null +++ b/app/routes/history.py @@ -0,0 +1,201 @@ +import asyncio +import logging +from typing import TYPE_CHECKING + +from aiohttp import web +from aiohttp.web import Request, Response + +from app.library.DownloadQueue import DownloadQueue +from app.library.encoder import Encoder +from app.library.Events import EventBus, Events +from app.library.ItemDTO import Item +from app.library.Presets import Preset, Presets +from app.library.router import route + +if TYPE_CHECKING: + from library.Download import Download + +LOG: logging.Logger = logging.getLogger(__name__) + + +@route("GET", r"api/history/", "items_list") +async def items_list(queue: DownloadQueue, encoder: Encoder) -> Response: + """ + Get the history. + + Args: + queue (DownloadQueue): The download queue instance. + encoder (Encoder): The encoder instance. + + Returns: + Response: The response object. + + """ + data: dict = {"queue": [], "history": []} + q = queue.get() + + data["queue"].extend([q.get("queue", {}).get(k) for k in q.get("queue", {})]) + data["history"].extend([q.get("done", {}).get(k) for k in q.get("done", {})]) + + return web.json_response(data=data, status=web.HTTPOk.status_code, dumps=encoder.encode) + + +@route("DELETE", "api/history/", "item_delete") +async def item_delete(request: Request, queue: DownloadQueue, encoder: Encoder) -> Response: + """ + Delete an item from the queue. + + Args: + request (Request): The request object. + queue (DownloadQueue): The download queue instance. + encoder (Encoder): The encoder instance. + + Returns: + Response: The response object. + + """ + data = await request.json() + ids = data.get("ids") + where = data.get("where") + 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) + + remove_file: bool = bool(data.get("remove_file", True)) + + return web.json_response( + data=await (queue.cancel(ids) if where == "queue" else queue.clear(ids, remove_file=remove_file)), + status=web.HTTPOk.status_code, + dumps=encoder.encode, + ) + + +@route("POST", "api/history/{id}", "item_update") +async def item_update(request: Request, queue: DownloadQueue, encoder: Encoder, notify: EventBus) -> Response: + """ + Update an item in the history. + + Args: + request (Request): The request object. + queue (DownloadQueue): The download queue instance. + encoder (Encoder): The encoder instance. + notify (EventBus): The event bus instance. + + Returns: + Response: The response object. + + """ + id: str = request.match_info.get("id") + if not id: + return web.json_response(data={"error": "id is required."}, status=web.HTTPBadRequest.status_code) + + item: Download | None = queue.done.get_by_id(id) + if not item: + return web.json_response(data={"error": "item not found."}, status=web.HTTPNotFound.status_code) + + post = await request.json() + if not post: + return web.json_response(data={"error": "no data provided."}, status=web.HTTPBadRequest.status_code) + + updated = False + + for k, v in post.items(): + if not hasattr(item.info, k): + continue + + if getattr(item.info, k) == v: + continue + + updated = True + setattr(item.info, k, v) + LOG.debug(f"Updated '{k}' to '{v}' for '{item.info.id}'") + + if updated: + queue.done.put(item) + await notify.emit(Events.UPDATE, data=item.info) + + return web.json_response( + data=item.info, + status=web.HTTPOk.status_code if updated else web.HTTPNotModified.status_code, + dumps=encoder.encode, + ) + + +@route("GET", "api/history/add/", "item_add") +async def item_add(request: Request, queue: DownloadQueue, encoder: Encoder) -> Response: + """ + Add a URL to the download queue. + + Args: + request (Request): The request object. + queue (DownloadQueue): The download queue instance. + encoder (Encoder): The encoder instance. + + Returns: + Response: The response object + + """ + url: str | None = request.query.get("url") + if not url: + return web.json_response(data={"error": "url param is required."}, status=web.HTTPBadRequest.status_code) + + data: dict[str, str] = {"url": url} + + preset: str | None = request.query.get("preset") + if preset: + exists: Preset | None = Presets.get_instance().get(preset) + if not exists: + return web.json_response( + data={"status": False, "message": f"Preset '{preset}' does not exist."}, + status=web.HTTPBadRequest.status_code, + ) + data["preset"] = preset + + try: + status: dict[str, str] = await queue.add(item=Item.format(data)) + except ValueError as e: + return web.json_response(data={"status": False, "message": str(e)}, status=web.HTTPBadRequest.status_code) + + return web.json_response( + data={"status": status.get("status") == "ok", "message": status.get("msg", "URL added")}, + status=web.HTTPOk.status_code, + dumps=encoder.encode, + ) + + +@route("POST", "api/history/", "items_add") +async def items_add(request: Request, queue: DownloadQueue, encoder: Encoder) -> Response: + """ + Add a URL to the download queue. + + Args: + request (Request): The request object. + queue (DownloadQueue): The download queue instance. + encoder (Encoder): The encoder instance. + + Returns: + Response: The response object. + + """ + data = await request.json() + + if isinstance(data, dict): + data = [data] + + items = [] + for item in data: + try: + items.append(Item.format(item)) + except ValueError as e: + return web.json_response(data={"error": str(e), "data": item}, status=web.HTTPBadRequest.status_code) + + status: list = await asyncio.wait_for( + fut=asyncio.gather(*[queue.add(item=item) for item in items]), + timeout=None, + ) + + response: list = [] + + for i, item in enumerate(items): + response.append({"item": item, "status": "ok" == status[i].get("status"), "msg": status[i].get("msg")}) + + return web.json_response(data=response, status=web.HTTPOk.status_code, dumps=encoder.encode) diff --git a/app/routes/images.py b/app/routes/images.py new file mode 100644 index 00000000..ac794814 --- /dev/null +++ b/app/routes/images.py @@ -0,0 +1,186 @@ +import logging +import random +import time +from datetime import UTC, datetime +from urllib.parse import urlparse + +import httpx +from aiohttp import web +from aiohttp.web import Request, Response + +from app.library.ag_utils import ag +from app.library.cache import Cache +from app.library.config import Config +from app.library.router import route +from app.library.Utils import validate_url + +LOG: logging.Logger = logging.getLogger(__name__) + +IS_REQUESTING_BACKGROUND: bool = False + + +@route("GET", "api/thumbnail/", "get_thumbnail") +async def get_thumbnail(request: Request, config: Config) -> Response: + """ + Get the thumbnail. + + Args: + request (Request): The request object. + config (Config): The configuration object. + + Returns: + Response: The response object. + + """ + url = request.query.get("url") + if not url: + return web.json_response(data={"error": "URL is required."}, status=web.HTTPForbidden.status_code) + + try: + validate_url(url) + except ValueError as e: + return web.json_response(data={"error": str(e)}, status=web.HTTPForbidden.status_code) + + try: + ytdlp_args = config.get_ytdlp_args() + opts = { + "proxy": ytdlp_args.get("proxy", None), + "headers": { + "User-Agent": ytdlp_args.get( + "user_agent", request.headers.get("User-Agent", f"YTPTube/{config.version}") + ), + }, + } + + async with httpx.AsyncClient(**opts) as client: + 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, tz=UTC).timetuple(), + ), + }, + ) + except Exception as e: + LOG.error(f"Error fetching thumbnail from '{url}'. '{e}'.") + return web.json_response( + data={"error": "failed to retrieve the thumbnail."}, status=web.HTTPInternalServerError.status_code + ) + + +@route("GET", "api/random/background/", "get_background") +async def get_background(request: Request, config: Config, cache: Cache) -> Response: + """ + Get random background. + + Args: + request (Request): The request object. + config (Config): The configuration object. + cache (Cache): The cache object. + + Returns: + Response: The response object. + + """ + global IS_REQUESTING_BACKGROUND # noqa: PLW0603 + + backend = None + + if IS_REQUESTING_BACKGROUND: + return web.Response(status=web.HTTPTooManyRequests.status_code) + + try: + IS_REQUESTING_BACKGROUND = True + backend = random.choice(config.pictures_backends) # noqa: S311 + CACHE_KEY_BING = "random_background_bing" + CACHE_KEY = "random_background" + + if cache.has(CACHE_KEY) and not request.query.get("force", False): + data = await cache.aget(CACHE_KEY) + return web.Response( + body=data.get("content"), + headers={ + "X-Cache": "HIT", + "X-Cache-TTL": str(await cache.attl(CACHE_KEY)), + "X-Image-Via": data.get("backend"), + **data.get("headers"), + }, + ) + + ytdlp_args = config.get_ytdlp_args() + opts = { + "proxy": ytdlp_args.get("proxy", None), + "headers": { + "User-Agent": ytdlp_args.get("user_agent", f"YTPTube/{config.version}"), + }, + } + + async with httpx.AsyncClient(**opts) as client: + if backend.startswith("https://www.bing.com/HPImageArchive.aspx"): + if not cache.has(CACHE_KEY_BING): + response = await client.request(method="GET", url=backend) + if response.status_code != web.HTTPOk.status_code: + return web.json_response( + data={"error": "failed to retrieve the random background image."}, + status=web.HTTPInternalServerError.status_code, + ) + + img_url: str | None = ag(response.json(), "images.0.url") + if not img_url: + return web.json_response( + data={"error": "failed to retrieve the random background image."}, + status=web.HTTPInternalServerError.status_code, + ) + + backend = f"https://www.bing.com{img_url}" + await cache.aset(key=CACHE_KEY_BING, value=backend, ttl=3600 * 24) + else: + backend: str = await cache.aget(CACHE_KEY_BING) + + LOG.debug(f"Requesting random picture from '{backend!s}'.") + + response = await client.request(method="GET", url=backend, follow_redirects=True) + + if response.status_code != web.HTTPOk.status_code: + return web.json_response( + data={"error": "failed to retrieve the random background image."}, + status=web.HTTPInternalServerError.status_code, + ) + + data = { + "content": response.content, + "backend": urlparse(backend).netloc, + "headers": { + "Content-Type": response.headers.get("Content-Type", "image/jpeg"), + "Content-Length": str(len(response.content)), + }, + } + + await cache.aset(key=CACHE_KEY, value=data, ttl=3600) + + LOG.debug(f"Random background image from '{backend!s}' cached.") + + return web.Response( + body=data.get("content"), + headers={ + "X-Cache": "MISS", + "X-Cache-TTL": "3600", + "X-Image-Via": data.get("backend"), + **data.get("headers"), + }, + ) + except Exception as e: + LOG.error(f"Failed to request random background image from '{backend!s}'.'. '{e!s}'.") + return web.json_response( + data={"error": "failed to retrieve the random background image."}, + status=web.HTTPInternalServerError.status_code, + ) + finally: + IS_REQUESTING_BACKGROUND = False diff --git a/app/routes/logs.py b/app/routes/logs.py new file mode 100644 index 00000000..6e00a23b --- /dev/null +++ b/app/routes/logs.py @@ -0,0 +1,50 @@ +from pathlib import Path + +from aiohttp import web +from aiohttp.web import Request, Response + +from app.library.config import Config +from app.library.encoder import Encoder +from app.library.router import route +from app.library.Utils import read_logfile + + +@route("GET", "api/logs/", "logs") +async def logs(request: Request, config: Config, encoder: Encoder) -> Response: + """ + Get recent logs + + Args: + request (Request): The request object. + config (Config): The configuration instance. + encoder (Encoder): The encoder instance. + + Returns: + Response: The response object. + + """ + if not config.file_logging: + return web.json_response(data={"error": "File logging is not enabled."}, status=web.HTTPNotFound.status_code) + + offset = int(request.query.get("offset", 0)) + limit = int(request.query.get("limit", 100)) + if limit < 1 or limit > 150: + limit = 50 + + logs_data = await read_logfile( + file=Path(config.config_path) / "logs" / "app.log", + offset=offset, + limit=limit, + ) + + return web.json_response( + data={ + "logs": logs_data["logs"], + "offset": offset, + "limit": limit, + "next_offset": logs_data["next_offset"], + "end_is_reached": logs_data["end_is_reached"], + }, + status=web.HTTPOk.status_code, + dumps=encoder.encode, + ) diff --git a/app/routes/notifications.py b/app/routes/notifications.py new file mode 100644 index 00000000..9d912002 --- /dev/null +++ b/app/routes/notifications.py @@ -0,0 +1,111 @@ +import logging +import uuid + +from aiohttp import web +from aiohttp.web import Request, Response + +from app.library.encoder import Encoder +from app.library.Events import EventBus, Events, message +from app.library.Notifications import Notification, NotificationEvents +from app.library.router import route +from app.library.Utils import validate_uuid + +LOG: logging.Logger = logging.getLogger(__name__) + + +@route("GET", "api/notifications/", "notifications_list") +async def notifications_list(encoder: Encoder) -> Response: + """ + Get the notification targets. + + Args: + encoder (Encoder): The encoder instance. + + Returns: + Response: The response object. + + """ + return web.json_response( + data={ + "notifications": Notification.get_instance().get_targets(), + "allowedTypes": list(NotificationEvents.get_events().values()), + }, + status=web.HTTPOk.status_code, + dumps=encoder.encode, + ) + + +@route("PUT", "api/notifications/", "notification_add") +async def notification_add(request: Request, encoder: Encoder) -> Response: + """ + Add notification targets. + + Args: + request (Request): The request object. + encoder (Encoder): The encoder instance. + notify (EventBus): The event bus instance. + + Returns: + Response: The response object. + + """ + post = await request.json() + if not isinstance(post, list): + return web.json_response( + {"error": "Invalid request body expecting list with dicts."}, + status=web.HTTPBadRequest.status_code, + ) + + targets: list = [] + + ins: Notification = Notification.get_instance() + for item in post: + if not isinstance(item, dict): + return web.json_response( + {"error": "Invalid request body expecting list with dicts."}, + status=web.HTTPBadRequest.status_code, + ) + + if not item.get("id", None) or validate_uuid(item.get("id"), version=4): + item["id"] = str(uuid.uuid4()) + + try: + Notification.validate(item) + except ValueError as e: + return web.json_response( + {"error": f"Invalid notification target settings. {e!s}", "data": item}, + status=web.HTTPBadRequest.status_code, + ) + + targets.append(ins.make_target(item)) + + try: + ins.save(targets=targets) + ins.load() + except Exception as e: + LOG.exception(e) + return web.json_response({"error": "Failed to save tasks."}, status=web.HTTPInternalServerError.status_code) + + data = {"notifications": targets, "allowedTypes": list(NotificationEvents.get_events().values())} + + return web.json_response(data=data, status=web.HTTPOk.status_code, dumps=encoder.encode) + + +@route("POST", "api/notifications/test/", "notification_test") +async def notification_test(encoder: Encoder, notify: EventBus) -> Response: + """ + Test the notification. + + Args: + encoder (Encoder): The encoder instance. + notify (EventBus): The event bus instance. + + Returns: + Response: The response object. + + """ + data = message("test", "This is a test notification.") + + await notify.emit(Events.TEST, data=data) + + return web.json_response(data=data, status=web.HTTPOk.status_code, dumps=encoder.encode) diff --git a/app/routes/ping.py b/app/routes/ping.py new file mode 100644 index 00000000..a0da11bc --- /dev/null +++ b/app/routes/ping.py @@ -0,0 +1,19 @@ +from aiohttp import web +from aiohttp.web import Response + +from app.library.DownloadQueue import DownloadQueue +from app.library.router import route + + +@route("GET", "api/ping/", "ping") +async def ping(queue: DownloadQueue) -> Response: + """ + Ping the server. + + Returns: + Response: The response object. + + """ + await queue.queue.test() + + return web.json_response(data={"status": "pong"}, status=web.HTTPOk.status_code) diff --git a/app/routes/player.py b/app/routes/player.py new file mode 100644 index 00000000..ee6355b6 --- /dev/null +++ b/app/routes/player.py @@ -0,0 +1,278 @@ +import logging +import time +from datetime import UTC, datetime +from pathlib import Path + +from aiohttp import web +from aiohttp.web import Request, Response + +from app.library.config import Config +from app.library.M3u8 import M3u8 +from app.library.Playlist import Playlist +from app.library.router import route +from app.library.Segments import Segments +from app.library.Subtitle import Subtitle +from app.library.Utils import StreamingError, get_file + +LOG: logging.Logger = logging.getLogger(__name__) + + +@route("GET", "api/player/playlist/{file:.*}.m3u8", "playlist_create") +async def playlist_create(request: Request, config: Config, app: web.Application) -> Response: + """ + Get the playlist. + + Args: + request (Request): The request object. + config (Config): The configuration instance. + app (web.Application): The aiohttp application instance. + + Returns: + Response: The response object. + + """ + file: str = request.match_info.get("file") + + if not file: + return web.json_response(data={"error": "file is required"}, status=web.HTTPBadRequest.status_code) + + base_path: str = config.base_path.rstrip("/") + + try: + realFile, status = get_file(download_path=config.download_path, file=file) + if web.HTTPFound.status_code == status: + return Response( + status=web.HTTPFound.status_code, + headers={ + "Location": str( + app.router["playlist_create"].url_for( + file=str(realFile).replace(config.download_path, "").strip("/") + ) + ), + }, + ) + + if web.HTTPNotFound.status_code == status: + return web.json_response(data={"error": f"File '{file}' does not exist."}, status=status) + + return web.Response( + text=await Playlist(download_path=Path(config.download_path), url=f"{base_path}/").make(file=realFile), + headers={ + "Content-Type": "application/x-mpegURL", + "Cache-Control": "no-cache", + "Access-Control-Max-Age": "300", + }, + status=web.HTTPOk.status_code, + ) + except StreamingError as e: + return web.json_response(data={"error": str(e)}, status=web.HTTPNotFound.status_code) + + +@route("GET", "api/player/m3u8/{mode}/{file:.*}.m3u8", "m3u8_create") +async def m3u8_create(request: Request, config: Config, app: web.Application) -> Response: + """ + Get the m3u8 file. + + Args: + request (Request): The request object. + config (Config): The configuration instance. + app (web.Application): The aiohttp application instance. + + Returns: + Response: The response object. + + """ + file: str = request.match_info.get("file") + mode: str = request.match_info.get("mode") + + 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: + return web.json_response(data={"error": "file is required"}, status=web.HTTPBadRequest.status_code) + + duration = request.query.get("duration", None) + + if "subtitle" in mode: + if not duration: + return web.json_response(data={"error": "duration is required."}, status=web.HTTPBadRequest.status_code) + + duration = float(duration) + + base_path: str = config.base_path.rstrip("/") + + try: + cls = M3u8(download_path=Path(config.download_path), url=f"{base_path}/") + + realFile, status = get_file(download_path=config.download_path, file=file) + if web.HTTPFound.status_code == status: + return Response( + status=status, + headers={ + "Location": str( + app.router["m3u8_create"].url_for( + mode=mode, file=str(realFile).replace(config.download_path, "").strip("/") + ) + ), + }, + ) + + if web.HTTPNotFound.status_code == status: + return web.json_response(data={"error": f"File '{file}' does not exist."}, status=status) + + if "subtitle" in mode: + text = await cls.make_subtitle(file=realFile, duration=duration) + else: + text = await cls.make_stream(file=realFile) + 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", + }, + status=web.HTTPOk.status_code, + ) + + +@route("GET", r"api/player/segments/{segment:\d+}/{file:.*}.ts", "segments_stream") +async def segments_stream(request: Request, config: Config, app: web.Application) -> Response: + """ + Get the segments. + + Args: + request (Request): The request object. + config (Config): The configuration instance. + app (web.Application): The aiohttp application instance. + + Returns: + Response: The response object. + + """ + 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)) + + if not file: + return web.json_response(data={"error": "file is required"}, status=web.HTTPBadRequest.status_code) + + if not segment: + return web.json_response(data={"error": "segment id is required."}, status=web.HTTPBadRequest.status_code) + + realFile, status = get_file(download_path=config.download_path, file=file) + if web.HTTPFound.status_code == status: + return Response( + status=status, + headers={ + "Location": str( + app.router["segments"].url_for( + segment=segment, + file=str(realFile).replace(config.download_path, "").strip("/"), + ) + ), + }, + ) + + if web.HTTPNotFound.status_code == status: + return web.json_response(data={"error": f"File '{file}' does not exist."}, status=status) + + mtime = realFile.stat().st_mtime + + if request.if_modified_since and request.if_modified_since.timestamp() == mtime: + lastMod = time.strftime("%a, %d %b %Y %H:%M:%S GMT", datetime.fromtimestamp(mtime, tz=UTC).timetuple()) + return web.Response(status=web.HTTPNotModified.status_code, headers={"Last-Modified": lastMod}) + + resp = web.StreamResponse( + status=web.HTTPOk.status_code, + 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(mtime, tz=UTC).timetuple() + ), + "Expires": time.strftime( + "%a, %d %b %Y %H:%M:%S GMT", datetime.fromtimestamp(time.time() + 31536000, tz=UTC).timetuple() + ), + }, + ) + + await resp.prepare(request) + + await Segments( + download_path=config.download_path, + index=int(segment), + duration=float(f"{float(sd if sd else M3u8.duration):.6f}"), + vconvert=vc == 1, + aconvert=ac == 1, + ).stream(realFile, resp) + + return resp + + +@route("GET", "api/player/subtitle/{file:.*}.vtt", "subtitles_get") +async def subtitles_get(request: Request, config: Config, app: web.Application) -> Response: + """ + Get the subtitles. + + Args: + request (Request): The request object. + config (Config): The configuration instance. + app (web.Application): The aiohttp application instance. + + Returns: + Response: The response object. + + """ + file: str = request.match_info.get("file") + + if not file: + return web.json_response(data={"error": "file is required"}, status=web.HTTPBadRequest.status_code) + + realFile, status = get_file(download_path=config.download_path, file=file) + if web.HTTPFound.status_code == status: + return Response( + status=status, + headers={ + "Location": str( + app.router["subtitles_get"].url_for(file=str(realFile).replace(config.download_path, "").strip("/")) + ), + }, + ) + + if web.HTTPNotFound.status_code == status: + return web.json_response(data={"error": f"File '{file}' does not exist."}, status=status) + + mtime = realFile.stat().st_mtime + + if request.if_modified_since and request.if_modified_since.timestamp() == mtime: + lastMod = time.strftime("%a, %d %b %Y %H:%M:%S GMT", datetime.fromtimestamp(mtime, tz=UTC).timetuple()) + return web.Response(status=web.HTTPNotModified.status_code, headers={"Last-Modified": lastMod}) + + return web.Response( + body=await Subtitle().make(file=realFile), + 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(mtime, tz=UTC).timetuple() + ), + "Expires": time.strftime( + "%a, %d %b %Y %H:%M:%S GMT", datetime.fromtimestamp(time.time() + 31536000, tz=UTC).timetuple() + ), + }, + status=web.HTTPOk.status_code, + ) diff --git a/app/routes/presets.py b/app/routes/presets.py new file mode 100644 index 00000000..cd4bcdb8 --- /dev/null +++ b/app/routes/presets.py @@ -0,0 +1,100 @@ +import logging +import uuid + +from aiohttp import web +from aiohttp.web import Request, Response + +from app.library.encoder import Encoder +from app.library.Events import EventBus, Events +from app.library.Presets import Preset, Presets +from app.library.router import route +from app.library.Utils import init_class, validate_uuid + +LOG: logging.Logger = logging.getLogger(__name__) + + +@route("GET", "api/presets/", "presets") +async def presets(request: Request, encoder: Encoder) -> Response: + """ + Get the presets. + + Args: + request (Request): The request object. + encoder (Encoder): The encoder instance. + + Returns: + Response: The response object. + + """ + data: list[Preset] = Presets.get_instance().get_all() + filter_fields: str | None = request.query.get("filter", None) + + if filter_fields: + fields: list[str] = [field.strip() for field in filter_fields.split(",")] + data = [{key: value for key, value in preset.serialize().items() if key in fields} for preset in data] + + return web.json_response(data=data, status=web.HTTPOk.status_code, dumps=encoder.encode) + + +@route("PUT", "api/presets/", "presets_add") +async def presets_add(request: Request, encoder: Encoder, notify: EventBus) -> Response: + """ + Add presets. + + Args: + request (Request): The request object. + encoder (Encoder): The encoder instance. + notify (EventBus): The event bus instance. + + Returns: + Response: The response object + + """ + data = await request.json() + + if not isinstance(data, list): + return web.json_response( + {"error": "Invalid request body expecting list with dicts."}, + status=web.HTTPBadRequest.status_code, + ) + + presets: list = [] + + cls = Presets.get_instance() + + for item in data: + if not isinstance(item, dict): + return web.json_response( + {"error": "Invalid request body expecting list with dicts."}, + status=web.HTTPBadRequest.status_code, + ) + + if not item.get("name"): + return web.json_response( + {"error": "name is required.", "data": item}, status=web.HTTPBadRequest.status_code + ) + + if not item.get("id", None) or not validate_uuid(item.get("id"), version=4): + item["id"] = str(uuid.uuid4()) + + try: + cls.validate(item) + except ValueError as e: + return web.json_response( + {"error": f"Failed to validate preset '{item.get('name')}'. '{e!s}'"}, + status=web.HTTPBadRequest.status_code, + ) + + presets.append(init_class(Preset, item)) + + try: + presets = cls.save(items=presets).load().get_all() + except Exception as e: + LOG.exception(e) + return web.json_response( + {"error": "Failed to save presets.", "message": str(e)}, + status=web.HTTPInternalServerError.status_code, + ) + + await notify.emit(Events.PRESETS_UPDATE, data=presets) + return web.json_response(data=presets, status=web.HTTPOk.status_code, dumps=encoder.encode) diff --git a/app/routes/tasks.py b/app/routes/tasks.py new file mode 100644 index 00000000..c326433b --- /dev/null +++ b/app/routes/tasks.py @@ -0,0 +1,93 @@ +import logging +import uuid + +from aiohttp import web +from aiohttp.web import Request, Response + +from app.library.encoder import Encoder +from app.library.router import route +from app.library.Tasks import Task, Tasks +from app.library.Utils import init_class, validate_uuid + +LOG: logging.Logger = logging.getLogger(__name__) + + +@route("GET", "api/tasks/", "tasks") +async def tasks(encoder: Encoder) -> Response: + """ + Get the tasks. + + Args: + encoder (Encoder): The encoder instance. + + Returns: + Response: The response object. + + """ + return web.json_response(data=Tasks.get_instance().get_all(), status=web.HTTPOk.status_code, dumps=encoder.encode) + + +@route("PUT", "api/tasks/", "tasks_add") +async def tasks_add(request: Request, encoder: Encoder) -> Response: + """ + Add tasks to the queue. + + Args: + request (Request): The request object. + encoder (Encoder): The encoder instance. + + Returns: + Response: The response object + + """ + data = await request.json() + + if not isinstance(data, list): + return web.json_response( + {"error": "Invalid request body expecting list with dicts."}, + status=web.HTTPBadRequest.status_code, + ) + + tasks: list = [] + + ins = Tasks.get_instance() + + for item in data: + if not isinstance(item, dict): + return web.json_response( + {"error": "Invalid request body expecting list with dicts."}, + status=web.HTTPBadRequest.status_code, + ) + + if not item.get("url"): + return web.json_response({"error": "url is required.", "data": item}, status=web.HTTPBadRequest.status_code) + + if not item.get("id", None) or not validate_uuid(item.get("id"), version=4): + item["id"] = str(uuid.uuid4()) + + if not item.get("template", None): + item["template"] = "" + + if not item.get("cli", None): + item["cli"] = "" + + try: + Tasks.validate(item) + except ValueError as e: + return web.json_response( + {"error": f"Failed to validate task '{item.get('name')}'. '{e!s}'"}, + status=web.HTTPBadRequest.status_code, + ) + + tasks.append(init_class(Task, item)) + + try: + tasks = ins.save(tasks=tasks).load().get_all() + except Exception as e: + LOG.exception(e) + return web.json_response( + {"error": "Failed to save tasks.", "message": str(e)}, + status=web.HTTPInternalServerError.status_code, + ) + + return web.json_response(data=tasks, status=web.HTTPOk.status_code, dumps=encoder.encode) diff --git a/app/routes/yt_dlp.py b/app/routes/yt_dlp.py new file mode 100644 index 00000000..a735b3a4 --- /dev/null +++ b/app/routes/yt_dlp.py @@ -0,0 +1,258 @@ +import asyncio +import json +import logging +import time +from collections import OrderedDict +from pathlib import Path +from typing import Any + +import anyio +from aiohttp import web +from aiohttp.web import Request, Response + +from app.library.cache import Cache +from app.library.config import Config +from app.library.Presets import Preset, Presets +from app.library.router import route +from app.library.Utils import REMOVE_KEYS, arg_converter, extract_info, validate_url +from app.library.YTDLPOpts import YTDLPOpts + +LOG: logging.Logger = logging.getLogger(__name__) + + +@route("POST", "api/yt-dlp/convert/", "convert") +async def convert(request: Request) -> Response: + """ + Convert the yt-dlp args to a dict. + + Args: + request (Request): The request object. + + Returns: + Response: The response object. + + """ + post = await request.json() + args: str | None = post.get("args") + + if not args: + return web.json_response(data={"error": "args param is required."}, status=web.HTTPBadRequest.status_code) + + try: + response = {"opts": {}, "output_template": None, "download_path": None} + + data = arg_converter(args, dumps=True) + + if "outtmpl" in data and "default" in data["outtmpl"]: + response["output_template"] = data["outtmpl"]["default"] + + if "paths" in data and "home" in data["paths"]: + response["download_path"] = data["paths"]["home"] + + if "format" in data: + response["format"] = data["format"] + + bad_options = {k: v for d in REMOVE_KEYS for k, v in d.items()} + removed_options = [] + + for key in data: + if key in bad_options.items(): + removed_options.append(bad_options[key]) + continue + if not key.startswith("_"): + response["opts"][key] = data[key] + + if len(removed_options) > 0: + response["removed_options"] = removed_options + + return web.json_response(data=response, status=web.HTTPOk.status_code) + except Exception as e: + err = str(e).strip() + err = err.split("\n")[-1] if "\n" in err else err + err = err.replace("main.py: error: ", "").strip().capitalize() + return web.json_response( + data={"error": f"Failed to parse command options for yt-dlp. '{err}'."}, + status=web.HTTPBadRequest.status_code, + ) + + +@route("GET", "api/yt-dlp/url/info/", "get_info") +async def get_info(request: Request, cache: Cache) -> Response: + """ + Get the video info. + + Args: + request (Request): The request object. + cache (Cache): The cache instance. + + Returns: + Response: The response object + + """ + url: str | None = request.query.get("url") + if not url: + return web.json_response(data={"error": "URL is required."}, status=web.HTTPBadRequest.status_code) + + try: + validate_url(url) + except ValueError as e: + return web.json_response(data={"error": str(e)}, status=web.HTTPBadRequest.status_code) + + config = Config.get_instance() + + preset = request.query.get("preset") + if preset: + exists: Preset | None = Presets.get_instance().get(preset) + if not exists: + return web.json_response( + data={"status": False, "message": f"Preset '{preset}' does not exist."}, + status=web.HTTPBadRequest.status_code, + ) + else: + preset: str = config.default_preset + + try: + key: str = cache.hash(f"{preset}:{url}") + + if cache.has(key) and not request.query.get("force", False): + data: Any | None = cache.get(key) + data["_cached"] = { + "status": "hit", + "key": key, + "ttl": data.get("_cached", {}).get("ttl", 300), + "ttl_left": data.get("_cached", {}).get("expires", time.time() + 300) - time.time(), + "expires": data.get("_cached", {}).get("expires", time.time() + 300), + } + return web.Response(body=json.dumps(data, indent=4, default=str), status=web.HTTPOk.status_code) + + opts: dict = {} + + if ytdlp_proxy := config.get_ytdlp_args().get("proxy", None): + opts["proxy"] = ytdlp_proxy + + ytdlp_opts = YTDLPOpts.get_instance().preset(name=preset).add(opts).get_all() + + data = extract_info( + config=ytdlp_opts, + url=url, + debug=False, + no_archive=True, + follow_redirect=True, + sanitize_info=True, + ) + + if "formats" in data: + from yt_dlp.cookies import LenientSimpleCookie + + for index, item in enumerate(data["formats"]): + if "cookies" in item and len(item["cookies"]) > 0: + cookies: list[str] = [f"{c.key}={c.value}" for c in LenientSimpleCookie(item["cookies"]).values()] + if len(cookies) > 0: + data["formats"][index]["h_cookies"] = "; ".join(cookies) + data["formats"][index]["h_cookies"] = data["formats"][index]["h_cookies"].strip() + + data["_cached"] = { + "status": "miss", + "key": key, + "ttl": 300, + "ttl_left": 300, + "expires": time.time() + 300, + } + + data = OrderedDict(sorted(data.items(), key=lambda item: len(str(item[1])))) + + cache.set(key=key, value=data, ttl=300) + + return web.Response(body=json.dumps(data, indent=4, default=str), status=web.HTTPOk.status_code) + except Exception as e: + LOG.exception(e) + LOG.error(f"Error encountered while getting video info for '{url}'. '{e!s}'.") + return web.json_response( + data={ + "error": "failed to get video info.", + "message": str(e), + "formats": [], + }, + status=web.HTTPInternalServerError.status_code, + ) + + +@route("GET", "api/yt-dlp/archive/recheck/", "archive_recheck") +async def archive_recheck(cache: Cache) -> Response: + """ + Recheck the manual archive entries. + + Args: + cache (Cache): The cache instance. + + Returns: + Response: The response object + + """ + config: Config = Config.get_instance() + if not config.manual_archive: + return web.json_response(data={"error": "Manual archive is not enabled."}, status=web.HTTPNotFound.status_code) + + manual_archive = Path(config.manual_archive) + if not manual_archive.exists(): + return web.json_response( + data={"error": "Manual archive file not found.", "file": manual_archive}, + status=web.HTTPNotFound.status_code, + ) + + tasks: list = [] + response: list = [] + + def info_wrapper(id: str, url: str) -> tuple[str, dict]: + try: + return ( + id, + extract_info( + config={ + "proxy": config.get_ytdlp_args().get("proxy", None), + "simulate": True, + "dump_single_json": True, + }, + url=url, + no_archive=True, + ), + ) + except Exception as e: + return (id, {"error": str(e)}) + + async with await anyio.open_file(manual_archive) as f: + # line format is "youtube ID - at: ISO8601" + async for line in f: + line = line.strip() + + if not line or not line.startswith("youtube"): + continue + + id = line.split(" ")[1].strip() + + if not id: + continue + + url = f"https://www.youtube.com/watch?v={id}" + key = cache.hash(id) + + if cache.has(key): + data = cache.get(key) + response.append({id: bool(data.get("id", None)) if isinstance(data, dict) else False}) + continue + + tasks.append( + asyncio.get_event_loop().run_in_executor(None, lambda i=id, url=url: info_wrapper(id=i, url=url)) + ) + + if len(tasks) > 0: + results = await asyncio.gather(*tasks) + for data in results: + if not data: + continue + + id, info = data + cache.set(key=cache.hash(id), value=info, ttl=3600 * 6) + response.append({id: bool(data.get("id", None)) if isinstance(data, dict) else False}) + + return web.json_response(data=response, status=web.HTTPOk.status_code) diff --git a/app/upgrader.py b/app/upgrader.py index 0eade685..cb353122 100644 --- a/app/upgrader.py +++ b/app/upgrader.py @@ -1,4 +1,10 @@ #!/usr/bin/env python3 +import sys +from pathlib import Path + +APP_ROOT = str((Path(__file__).parent / "..").resolve()) +if APP_ROOT not in sys.path: + sys.path.insert(0, APP_ROOT) import logging import os diff --git a/ui/components/Dropdown.vue b/ui/components/Dropdown.vue new file mode 100644 index 00000000..01218b1b --- /dev/null +++ b/ui/components/Dropdown.vue @@ -0,0 +1,70 @@ + + + + + diff --git a/ui/components/GetInfo.vue b/ui/components/GetInfo.vue index 5135dd2c..b6696825 100644 --- a/ui/components/GetInfo.vue +++ b/ui/components/GetInfo.vue @@ -9,7 +9,7 @@
- +