diff --git a/app/library/HttpAPI.py b/app/library/HttpAPI.py
index 496a806c..d90c384b 100644
--- a/app/library/HttpAPI.py
+++ b/app/library/HttpAPI.py
@@ -1,9 +1,7 @@
import base64
import hmac
-import importlib
import inspect
import logging
-import pkgutil
from collections.abc import Awaitable
from datetime import UTC, datetime, timedelta
from pathlib import Path
@@ -19,20 +17,20 @@ from .DownloadQueue import DownloadQueue
from .encoder import Encoder
from .Events import EventBus
from .ffprobe import ffprobe
-from .router import get_routes
-from .Utils import decrypt_data, encrypt_data, get_file, get_mime_type
+from .router import RouteType, get_routes
+from .Utils import decrypt_data, encrypt_data, get_file, get_mime_type, load_modules
LOG: logging.Logger = logging.getLogger("http_api")
class HttpAPI:
- def __init__(self, root_path: str, queue: DownloadQueue):
+ def __init__(self, root_path: Path, 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()
- self.rootPath: str = root_path
+ self.rootPath: Path = root_path
self.cache = Cache()
self.app: web.Application | None = None
@@ -97,21 +95,20 @@ class HttpAPI:
registered_options: list = []
base_path: str = self.config.base_path.rstrip("/")
- from app.routes._static import preload_static
+ from app.routes.api._static import preload_static
- self._load_modules(Path(self.rootPath) / "routes")
+ load_modules(self.rootPath, self.rootPath / "routes" / "api")
preload_static(self.rootPath, self.config)
async def options_handler(_: Request) -> Response:
return web.Response(status=204)
- for route in get_routes().values():
+ for route in get_routes(RouteType.HTTP).values():
routePath: str = f"/{route.path.lstrip('/')}"
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('/')}"
LOG.debug(f"Add ({route.name}) {route.method}: {route.path}.")
@@ -297,16 +294,3 @@ class HttpAPI:
return response
return middleware_handler
-
- 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
- try:
- 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 b0b35dc6..67cc1935 100644
--- a/app/library/HttpSocket.py
+++ b/app/library/HttpSocket.py
@@ -1,26 +1,21 @@
-import asyncio
-import errno
import functools
+import inspect
import logging
-import os
-import shlex
-import time
-from datetime import UTC, datetime
from pathlib import Path
-import anyio
import socketio
from aiohttp import web
+from app.library.router import RouteType, get_routes
+from app.library.Utils import load_modules
+
from .config import Config
from .DownloadQueue import DownloadQueue
from .encoder import Encoder
-from .Events import Event, EventBus, Events, error
+from .Events import Event, EventBus, Events
from .ItemDTO import Item
-from .Presets import Presets
-from .Utils import is_downloaded, tail_log
-LOG = logging.getLogger("socket_api")
+LOG: logging.Logger = logging.getLogger("socket_api")
class HttpSocket:
@@ -31,15 +26,11 @@ class HttpSocket:
config: Config
sio: socketio.AsyncServer
queue: DownloadQueue
-
- subscribers: dict[str, list[str]] = {}
- """Event subscriber list."""
-
- log_task = None
- """Task to tail the log file."""
+ di_context: dict[str, object] = {}
def __init__(
self,
+ root_path: Path,
queue: DownloadQueue | None = None,
encoder: Encoder | None = None,
config: Config | None = None,
@@ -61,10 +52,19 @@ class HttpSocket:
ping_timeout=5,
)
encoder = encoder or Encoder()
+ self.rootPath = root_path
def emit(e: Event, _, **kwargs):
return self.sio.emit(event=e.event, data=encoder.encode(e.data), **kwargs)
+ self.di_context = {
+ "config": self.config,
+ "queue": self.queue,
+ "sio": self.sio,
+ "encoder": encoder,
+ "notify": self._notify,
+ }
+
self._notify.subscribe("frontend", emit, f"{__class__.__name__}.emit")
@staticmethod
@@ -90,347 +90,38 @@ class HttpSocket:
LOG.debug("Socket server shutdown complete.")
def attach(self, app: web.Application):
+ app.on_shutdown.append(self.on_shutdown)
+
self.sio.attach(app, socketio_path=f"{self.config.base_path.rstrip('/')}/socket.io")
-
- for attr_name in dir(self):
- method = getattr(self, attr_name)
- if hasattr(method, "_ws_event") and self.sio:
- self.sio.on(method._ws_event)(method) # type: ignore
-
self._notify.subscribe(
Events.ADD_URL,
lambda data, _, **kwargs: self.queue.add(item=Item.format(data.data)), # noqa: ARG005
f"{__class__.__name__}.add",
)
- # register the shutdown event.
- app.on_shutdown.append(self.on_shutdown)
+ load_modules(self.rootPath, self.rootPath / "routes" / "socket")
- @ws_event
- async def cli_post(self, sid: str, data):
- if not self.config.console_enabled:
- await self._notify.emit(Events.ERROR, data=error("Console is disabled."), to=sid)
- return
-
- if not data:
- await self._notify.emit(Events.CLI_CLOSE, data={"exitcode": 0}, to=sid)
- return
-
- try:
- LOG.info(f"Cli command from client '{sid}'. '{data}'")
-
- args = ["yt-dlp", *shlex.split(data)]
- _env = os.environ.copy()
- _env.update(
- {
- "TERM": "xterm-256color",
- "LANG": "en_US.UTF-8",
- "SHELL": "/bin/bash",
- "LC_ALL": "en_US.UTF-8",
- "PWD": self.config.download_path,
- "FORCE_COLOR": "1",
- "PYTHONUNBUFFERED": "1",
- }
+ for route in get_routes(RouteType.SOCKET).values():
+ LOG.debug(
+ f"Add ({route.name}) {route.method.value if isinstance(route.method,RouteType) else route.method}: {route.path}."
)
+ self.sio.on(route.path)(HttpSocket._injector(route.handler, route.path, self.di_context))
- try:
- import pty
+ @staticmethod
+ def _injector(func, event: str, container: dict):
+ sig: inspect.Signature = inspect.signature(func)
- master_fd, slave_fd = pty.openpty()
- stdin_arg = asyncio.subprocess.DEVNULL
- stdout_arg = stderr_arg = slave_fd
- use_pty = True
- except ImportError:
- use_pty = False
- master_fd = slave_fd = None
- stdin_arg = asyncio.subprocess.DEVNULL
- stdout_arg = asyncio.subprocess.PIPE
- stderr_arg = asyncio.subprocess.STDOUT
+ async def wrapper(sid, data, **kwargs):
+ args = {}
- proc = await asyncio.create_subprocess_exec(
- *args,
- cwd=self.config.download_path,
- stdin=stdin_arg,
- stdout=stdout_arg,
- stderr=stderr_arg,
- env=_env,
- )
+ merged = {**container, "sid": sid, "data": data, "event_name": event}
+ if isinstance(kwargs, dict):
+ merged.update(kwargs)
- if use_pty:
- try:
- os.close(slave_fd)
- except Exception as e:
- LOG.error(f"Error closing PTY. '{e!s}'.")
+ for name in sig.parameters:
+ if name in merged:
+ args[name] = merged[name]
- async def reader(sid: str):
- if use_pty is False:
- assert proc.stdout is not None
- async for raw_line in proc.stdout:
- line = raw_line.rstrip(b"\n")
- await self._notify.emit(
- Events.CLI_OUTPUT,
- data={"type": "stdout", "line": line.decode("utf-8", errors="replace")},
- to=sid,
- )
- return
+ return await func(**args)
- loop = asyncio.get_running_loop()
- buffer = b""
- while True:
- try:
- chunk = await loop.run_in_executor(None, lambda: os.read(master_fd, 1024))
- except OSError as e:
- if e.errno == errno.EIO:
- break
- raise
-
- if not chunk:
- # No more output
- if buffer:
- # Emit any remaining partial line
- await self._notify.emit(
- Events.CLI_OUTPUT,
- data={"type": "stdout", "line": buffer.decode("utf-8", errors="replace")},
- to=sid,
- )
- break
- buffer += chunk
- *lines, buffer = buffer.split(b"\n")
- for line in lines:
- await self._notify.emit(
- Events.CLI_OUTPUT,
- data={"type": "stdout", "line": line.decode("utf-8", errors="replace")},
- to=sid,
- )
- try:
- os.close(master_fd)
- except Exception as e:
- LOG.error(f"Error closing PTY. '{e!s}'.")
-
- # Start reading output from PTY
- read_task = asyncio.create_task(reader(sid=sid))
-
- # Wait until process finishes
- returncode = await proc.wait()
-
- # Ensure reading is done
- await read_task
-
- await self._notify.emit(Events.CLI_CLOSE, data={"exitcode": returncode}, to=sid)
- except Exception as e:
- LOG.error(f"CLI execute exception was thrown for client '{sid}'.")
- LOG.exception(e)
- await self._notify.emit(Events.CLI_OUTPUT, data={"type": "stderr", "line": str(e)}, to=sid)
- await self._notify.emit(Events.CLI_CLOSE, data={"exitcode": -1}, to=sid)
-
- @ws_event
- async def add_url(self, sid: str, data: dict):
- url: str | None = data.get("url")
-
- if not url:
- await self._notify.emit(Events.ERROR, data=error("No URL provided.", data={"unlock": True}), to=sid)
- return
-
- try:
- await self._notify.emit(
- event=Events.STATUS,
- data=await self.queue.add(item=Item.format(data)),
- to=sid,
- )
- except ValueError as e:
- LOG.exception(e)
- await self._notify.emit(Events.ERROR, data=error(str(e)), to=sid)
- return
-
- @ws_event
- async def item_cancel(self, sid: str, id: str):
- if not id:
- await self._notify.emit(Events.ERROR, data=error("Invalid request."), to=sid)
- return
-
- status: dict[str, str] = {}
- status = await self.queue.cancel([id])
- status.update({"identifier": id})
-
- await self._notify.emit(Events.ITEM_CANCEL, data=status)
-
- @ws_event
- async def item_delete(self, sid: str, data: dict):
- if not data:
- await self._notify.emit(Events.ERROR, data=error("Invalid request."), to=sid)
- return
-
- id: str | None = data.get("id")
- if not id:
- await self._notify.emit(Events.ERROR, data=error("Invalid request."), to=sid)
- return
-
- status: dict[str, str] = {}
- status = await self.queue.clear([id], remove_file=bool(data.get("remove_file", False)))
- status.update({"identifier": id})
-
- await self._notify.emit(Events.ITEM_DELETE, data=status)
-
- @ws_event
- async def archive_item(self, _: str, data: dict):
- if not isinstance(data, dict) or "url" not in data:
- return
-
- from .YTDLPOpts import YTDLPOpts
-
- params = YTDLPOpts.get_instance()
-
- if "preset" in data and isinstance(data["preset"], str):
- params.preset(name=data["preset"])
-
- if "cli" in data and isinstance(data["cli"], str) and len(data["cli"]) > 1:
- params.add_cli(data["cli"], from_user=True)
-
- params = params.get_all()
-
- file: str = params.get("download_archive", None)
-
- if not file:
- return
-
- exists, idDict = is_downloaded(file, data["url"])
- if exists or "archive_id" not in idDict or idDict["archive_id"] is None:
- return
-
- async with await anyio.open_file(file, "a") as f:
- await f.write(f"{idDict['archive_id']}\n")
-
- manual_archive = self.config.manual_archive
- if manual_archive:
- manual_archive = Path(manual_archive)
-
- if not manual_archive.exists():
- manual_archive.touch(exist_ok=True)
-
- previouslyArchived = False
- async with await anyio.open_file(manual_archive) as f:
- async for line in f:
- if idDict["archive_id"] in line:
- previouslyArchived = True
- break
-
- if not previouslyArchived:
- async with await anyio.open_file(manual_archive, "a") as f:
- await f.write(f"{idDict['archive_id']} - at: {datetime.now(UTC).isoformat()}\n")
- LOG.info(f"Archiving url '{data['url']}' with id '{idDict['archive_id']}'.")
- else:
- LOG.info(f"URL '{data['url']}' with id '{idDict['archive_id']}' already archived.")
-
- @ws_event
- async def connect(self, sid: str, _=None):
- data = {
- **self.queue.get(),
- "config": self.config.frontend(),
- "presets": Presets.get_instance().get_all(),
- "paused": self.queue.is_paused(),
- }
-
- data["folders"] = [folder.name for folder in Path(self.config.download_path).iterdir() if folder.is_dir()]
-
- await self._notify.emit(Events.INITIAL_DATA, data=data, to=sid)
-
- @ws_event
- async def pause(self, *_, **__):
- self.queue.pause()
- await self._notify.emit(Events.PAUSED, data={"paused": True, "at": time.time()})
-
- @ws_event
- async def resume(self, *_, **__):
- self.queue.resume()
- await self._notify.emit(Events.PAUSED, data={"paused": False, "at": time.time()})
-
- @ws_event
- async def subscribe(self, sid: str, event: str):
- """
- Subscribe to a specific event.
-
- Args:
- sid (str): The session ID of the client.
- event (str): The event to subscribe to.
-
- """
- if not isinstance(event, str):
- await self._notify.emit(Events.ERROR, data=error("Invalid event."), to=sid)
- return
-
- if event not in self.subscribers:
- self.subscribers[event] = []
-
- if sid not in self.subscribers[event]:
- self.subscribers[event].append(sid)
- LOG.debug(f"Client '{sid}' subscribed to event '{event}'.")
- await self.sio.emit(Events.SUBSCRIBED, data={"event": event}, to=sid)
-
- async def emit_logs(data: dict):
- await self.subscribe_emit(event=event, data=data)
-
- if "log_lines" == event and self.log_task is None:
- log_file = Path(self.config.config_path) / "logs" / "app.log"
- LOG.debug(f"Starting tailing '{log_file!s}'.")
- self.log_task = asyncio.create_task(
- tail_log(
- file=log_file,
- emitter=emit_logs,
- ),
- name="tail_log",
- )
-
- @ws_event
- async def unsubscribe(self, sid: str, event: str):
- """
- Unsubscribe from a specific event.
-
- Args:
- sid (str): The session ID of the client.
- event (str): The event to unsubscribe from.
-
- """
- if event not in self.subscribers:
- return
-
- if sid not in self.subscribers[event]:
- return
-
- self.subscribers[event].remove(sid)
- await self.sio.emit(Events.UNSUBSCRIBED, data={"event": event}, to=sid)
-
- LOG.debug(f"Client '{sid}' unsubscribed from event '{event}'.")
-
- if "log_lines" != event or not self.log_task or "log_lines" not in self.subscribers:
- return
-
- if len(self.subscribers["log_lines"]) < 1:
- try:
- LOG.debug("Stopping log tailing task.")
- self.log_task.cancel()
- self.log_task = None
- except asyncio.CancelledError:
- pass
-
- @ws_event
- async def disconnect(self, sid: str, reason: str):
- """
- Handle client disconnection.
-
- Args:
- sid (str): The session ID of the client.
- reason (str): The reason for disconnection.
-
- """
- LOG.debug(f"Client '{sid}' disconnected. {reason}")
-
- for event in self.subscribers:
- if sid in self.subscribers[event]:
- await self.unsubscribe(sid=sid, event=event)
-
- async def subscribe_emit(self, event: str, data: dict):
- if event not in self.subscribers or len(self.subscribers[event]) < 1:
- return
-
- for sid in self.subscribers[event]:
- await self.sio.emit(event=event, data=data, to=sid)
+ return wrapper
diff --git a/app/library/router.py b/app/library/router.py
index ce466545..8572b27a 100644
--- a/app/library/router.py
+++ b/app/library/router.py
@@ -1,14 +1,25 @@
import logging
import re
from collections.abc import Awaitable
+from enum import Enum
from functools import wraps
LOG: logging.Logger = logging.getLogger(__name__)
+# make a enum for route types
+class RouteType(str, Enum):
+ HTTP = "http"
+ SOCKET = "socket"
+
+ @classmethod
+ def all(cls) -> list[str]:
+ return [member.value for member in cls]
+
+
class Route:
"""
- A class to represent an HTTP route.
+ A class to represent an route.
Attributes:
method (str): The HTTP method (GET, POST, etc.).
@@ -19,20 +30,20 @@ class Route:
"""
def __init__(self, method: str, path: str, name: str, handler: Awaitable):
- self.method = method.upper()
- self.path = path
- self.name = name
+ self.method: str = method.upper()
+ self.path: str = path
+ self.name: str = name
self.handler: Awaitable = handler
-ROUTES: dict[str, Route] = {}
+ROUTES: dict[str, dict[str, Route]] = {}
def make_route_name(method: str, path: str) -> str:
method = method.lower()
path = path.strip("/")
- segments = []
+ segments: list = []
for part in path.split("/"):
part = re.sub(r"[^\w]", "_", part) # remove invalid chars
if not part:
@@ -44,12 +55,12 @@ def make_route_name(method: str, path: str) -> str:
return f"{method}:" + ".".join(segments or ["root"])
-def route(method: str, path: str, name: str | None = None, **kwargs) -> Awaitable:
+def route(method: RouteType | 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.
+ method (RouteType|str): The HTTP method.
path (str): The path to the route.
name (str): The name of the route.
kwargs: Additional keyword arguments.
@@ -66,9 +77,13 @@ def route(method: str, path: str, name: str | None = None, **kwargs) -> Awaitabl
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(
+ route_type: str = RouteType.SOCKET if RouteType.SOCKET == method else RouteType.HTTP
+ if route_type not in ROUTES:
+ ROUTES[route_type] = {}
+
+ ROUTES[route_type][name] = Route(method=method.upper(), path=path, name=name, handler=wrapper)
+ if "http" == route_type and path.endswith("/") and "/" != path and not kwargs.get("no_slash", False):
+ ROUTES[route_type][f"{name}_no_slash"] = Route(
method=method.upper(), path=path[:-1], name=f"{name}_no_slash", handler=wrapper
)
@@ -77,12 +92,12 @@ def route(method: str, path: str, name: str | None = None, **kwargs) -> Awaitabl
return decorator
-def add_route(method: str, path: str, handler: Awaitable, name: str | None = None, **kwargs):
+def add_route(method: RouteType | 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.
+ method (RouteType|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.
@@ -92,33 +107,43 @@ def add_route(method: str, path: str, handler: Awaitable, name: str | None = Non
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(
+ route_type: str = RouteType.SOCKET if RouteType.SOCKET == method else RouteType.HTTP
+
+ if route_type not in ROUTES:
+ ROUTES[route_type] = {}
+
+ ROUTES[route_type][name] = Route(method=method.upper(), path=path, name=name, handler=handler)
+
+ if "http" == route_type and path.endswith("/") and "/" != path and not kwargs.get("no_slash", False):
+ ROUTES[route_type][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:
+def get_route(route_type: RouteType, name: str) -> dict[str, Route] | None:
"""
Get the route information by name.
Args:
+ route_type (RouteType): The type of the route (e.g., RouteType.HTTP, RouteType.SOCKET).
name (str): The name of the route.
Returns:
dict: The route information, or None if not found.
"""
- return ROUTES.get(name)
+ return ROUTES.get(route_type, {}).get(name, None)
-def get_routes() -> dict[str, Route]:
+def get_routes(route_type: RouteType) -> dict[str, Route]:
"""
Get all registered routes.
+ Args:
+ route_type (RouteType): The type of the route (e.g., RouteType.HTTP, RouteType.SOCKET).
+
Returns:
dict[str, dict]: A dictionary of all registered routes.
"""
- return ROUTES
+ return ROUTES.get(route_type, {})
diff --git a/app/main.py b/app/main.py
index 00c9d9cd..42f91235 100644
--- a/app/main.py
+++ b/app/main.py
@@ -65,7 +65,7 @@ class Main:
self._queue = DownloadQueue(connection=connection)
self._http = HttpAPI(root_path=ROOT_PATH, queue=self._queue)
- self._socket = HttpSocket(queue=self._queue)
+ self._socket = HttpSocket(root_path=ROOT_PATH, queue=self._queue)
self._app.on_cleanup.append(_close_connection)
diff --git a/app/routes/socket/connection.py b/app/routes/socket/connection.py
new file mode 100644
index 00000000..d1d49f9e
--- /dev/null
+++ b/app/routes/socket/connection.py
@@ -0,0 +1,126 @@
+import asyncio
+import logging
+from pathlib import Path
+
+import socketio
+
+from app.library.config import Config
+from app.library.DownloadQueue import DownloadQueue
+from app.library.Events import EventBus, Events, error
+from app.library.Presets import Presets
+from app.library.router import RouteType, route
+from app.library.Utils import tail_log
+
+LOG: logging.Logger = logging.getLogger(__name__)
+
+
+class _Data:
+ subscribers: dict[str, list[str]] = {}
+ log_task: asyncio.Task | None = None
+
+
+@route(RouteType.SOCKET, "connect", "socket_connect")
+async def connect(config: Config, queue: DownloadQueue, notify: EventBus, sid: str):
+ data = {
+ **queue.get(),
+ "config": config.frontend(),
+ "presets": Presets.get_instance().get_all(),
+ "paused": queue.is_paused(),
+ }
+
+ data["folders"] = [folder.name for folder in Path(config.download_path).iterdir() if folder.is_dir()]
+
+ await notify.emit(Events.INITIAL_DATA, data=data, to=sid)
+
+
+@route(RouteType.SOCKET, "disconnect", "socket_disconnect")
+async def disconnect(sid: str, data: str = None):
+ """
+ Handle client disconnection.
+
+ Args:
+ sid (str): The session ID of the client.
+ data (str): The reason for disconnection.
+
+ """
+ LOG.debug(f"Client '{sid}' disconnected. {data}")
+
+ for event in _Data.subscribers:
+ if sid in _Data.subscribers[event]:
+ await unsubscribe(sid=sid, data=event)
+
+
+@route(RouteType.SOCKET, "subscribe", "socket_subscribe")
+async def subscribe(config: Config, notify: EventBus, sio: socketio.AsyncServer, sid: str, data: str):
+ """
+ Subscribe to a specific event.
+
+ Args:
+ config (Config): The configuration instance.
+ notify (EventBus): The event bus to use for notifications.
+ sio (socketio.AsyncServer): The Socket.IO server instance.
+ sid (str): The session ID of the client.
+ data (str): The event to subscribe to.
+
+ """
+ if not isinstance(data, str):
+ await notify.emit(Events.ERROR, data=error("Invalid event."), to=sid)
+ return
+
+ if data not in _Data.subscribers:
+ _Data.subscribers[data] = []
+
+ if sid not in _Data.subscribers[data]:
+ _Data.subscribers[data].append(sid)
+ LOG.debug(f"Client '{sid}' subscribed to event '{data}'.")
+ await sio.emit(Events.SUBSCRIBED, data={"event": data}, to=sid)
+
+ async def emit_logs(data: dict):
+ await subscribe_emit(sio=sio, event=data, data=data)
+
+ if "log_lines" == data and _Data.log_task is None:
+ log_file = Path(config.config_path) / "logs" / "app.log"
+ LOG.debug(f"Starting tailing '{log_file!s}'.")
+ _Data.log_task = asyncio.create_task(tail_log(file=log_file, emitter=emit_logs), name="tail_log")
+
+
+@route(RouteType.SOCKET, "unsubscribe", "socket_unsubscribe")
+async def unsubscribe(sio: socketio.AsyncServer, sid: str, data: str):
+ """
+ Unsubscribe from a specific event.
+
+ Args:
+ sio (socketio.AsyncServer): The Socket.IO server instance.
+ sid (str): The session ID of the client.
+ data (str): The event to unsubscribe from.
+
+ """
+ if data not in _Data.subscribers:
+ return
+
+ if sid not in _Data.subscribers[data]:
+ return
+
+ _Data.subscribers[data].remove(sid)
+ await sio.emit(Events.UNSUBSCRIBED, data={"event": data}, to=sid)
+
+ LOG.debug(f"Client '{sid}' unsubscribed from event '{data}'.")
+
+ if "log_lines" != data or not _Data.log_task or "log_lines" not in _Data.subscribers:
+ return
+
+ if len(_Data.subscribers["log_lines"]) < 1:
+ try:
+ LOG.debug("Stopping log tailing task.")
+ _Data.log_task.cancel()
+ _Data.log_task = None
+ except asyncio.CancelledError:
+ pass
+
+
+async def subscribe_emit(sio: socketio.AsyncServer, event: str, data: dict):
+ if event not in _Data.subscribers or len(_Data.subscribers[event]) < 1:
+ return
+
+ for sid in _Data.subscribers[event]:
+ await sio.emit(event=event, data=data, to=sid)
diff --git a/app/routes/socket/history.py b/app/routes/socket/history.py
new file mode 100644
index 00000000..2a16be2f
--- /dev/null
+++ b/app/routes/socket/history.py
@@ -0,0 +1,130 @@
+import logging
+import time
+from datetime import UTC, datetime
+from pathlib import Path
+
+import anyio
+
+from app.library.config import Config
+from app.library.DownloadQueue import DownloadQueue
+from app.library.Events import EventBus, Events, error
+from app.library.ItemDTO import Item
+from app.library.router import RouteType, route
+from app.library.Utils import is_downloaded
+from app.library.YTDLPOpts import YTDLPOpts
+
+LOG: logging.Logger = logging.getLogger(__name__)
+
+
+@route(RouteType.SOCKET, "pause", "pause_downloads")
+async def pause(notify: EventBus, queue: DownloadQueue):
+ queue.pause()
+ await notify.emit(Events.PAUSED, data={"paused": True, "at": time.time()})
+
+
+@route(RouteType.SOCKET, "resume", "resume_downloads")
+async def resume(notify: EventBus, queue: DownloadQueue):
+ queue.resume()
+ await notify.emit(Events.PAUSED, data={"paused": False, "at": time.time()})
+
+
+@route(RouteType.SOCKET, "add_url", "add_url")
+async def add_url(queue: DownloadQueue, notify: EventBus, sid: str, data: dict):
+ url: str | None = data.get("url")
+
+ if not url:
+ await notify.emit(Events.ERROR, data=error("No URL provided.", data={"unlock": True}), to=sid)
+ return
+
+ try:
+ await notify.emit(
+ event=Events.STATUS,
+ data=await queue.add(item=Item.format(data)),
+ to=sid,
+ )
+ except ValueError as e:
+ LOG.exception(e)
+ await notify.emit(Events.ERROR, data=error(str(e)), to=sid)
+ return
+
+
+@route(RouteType.SOCKET, "item_cancel", "item_cancel")
+async def item_cancel(queue: DownloadQueue, notify: EventBus, sid: str, data: str):
+ if not data:
+ await notify.emit(Events.ERROR, data=error("Invalid request."), to=sid)
+ return
+
+ status: dict[str, str] = {}
+ status = await queue.cancel([data])
+ status.update({"identifier": data})
+
+ await notify.emit(Events.ITEM_CANCEL, data=status)
+
+
+@route(RouteType.SOCKET, "item_delete", "item_delete")
+async def item_delete(queue: DownloadQueue, notify: EventBus, sid: str, data: dict):
+ if not data:
+ await notify.emit(Events.ERROR, data=error("Invalid request."), to=sid)
+ return
+
+ id: str | None = data.get("id")
+ if not id:
+ await notify.emit(Events.ERROR, data=error("Invalid request."), to=sid)
+ return
+
+ status: dict[str, str] = {}
+ status = await queue.clear([id], remove_file=bool(data.get("remove_file", False)))
+ status.update({"identifier": id})
+
+ await notify.emit(Events.ITEM_DELETE, data=status)
+
+
+@route(RouteType.SOCKET, "archive_item", "archive_item")
+async def archive_item(config: Config, data: dict):
+ if not isinstance(data, dict) or "url" not in data:
+ return
+
+ params: YTDLPOpts = YTDLPOpts.get_instance()
+
+ if "preset" in data and isinstance(data["preset"], str):
+ params.preset(name=data["preset"])
+
+ if "cli" in data and isinstance(data["cli"], str) and len(data["cli"]) > 1:
+ params.add_cli(data["cli"], from_user=True)
+
+ params = params.get_all()
+
+ file: str = params.get("download_archive", None)
+
+ if not file:
+ return
+
+ exists, idDict = is_downloaded(file, data["url"])
+ if exists or "archive_id" not in idDict or idDict["archive_id"] is None:
+ return
+
+ async with await anyio.open_file(file, "a") as f:
+ await f.write(f"{idDict['archive_id']}\n")
+
+ manual_archive: str = config.manual_archive
+ if not manual_archive:
+ return
+
+ manual_archive = Path(manual_archive)
+
+ if not manual_archive.exists():
+ manual_archive.touch(exist_ok=True)
+
+ previouslyArchived = False
+ async with await anyio.open_file(manual_archive) as f:
+ async for line in f:
+ if idDict["archive_id"] in line:
+ previouslyArchived = True
+ break
+
+ if not previouslyArchived:
+ async with await anyio.open_file(manual_archive, "a") as f:
+ await f.write(f"{idDict['archive_id']} - at: {datetime.now(UTC).isoformat()}\n")
+ LOG.info(f"Archiving url '{data['url']}' with id '{idDict['archive_id']}'.")
+ else:
+ LOG.info(f"URL '{data['url']}' with id '{idDict['archive_id']}' already archived.")
diff --git a/app/routes/socket/terminal.py b/app/routes/socket/terminal.py
new file mode 100644
index 00000000..911223a0
--- /dev/null
+++ b/app/routes/socket/terminal.py
@@ -0,0 +1,133 @@
+import logging
+from typing import TYPE_CHECKING
+
+from app.library.config import Config
+from app.library.Events import EventBus, Events, error
+from app.library.router import RouteType, route
+
+if TYPE_CHECKING:
+ from asyncio import Task
+ from asyncio.events import AbstractEventLoop
+ from asyncio.subprocess import Process
+
+LOG: logging.Logger = logging.getLogger(__name__)
+
+
+@route(RouteType.SOCKET, "cli_post", "socket_cli_post")
+async def cli_post(config: Config, notify: EventBus, sid: str, data: str):
+ if not config.console_enabled:
+ await notify.emit(Events.ERROR, data=error("Console is disabled."), to=sid)
+ return
+
+ if not data:
+ await notify.emit(Events.CLI_CLOSE, data={"exitcode": 0}, to=sid)
+ return
+
+ import asyncio
+ import errno
+ import os
+ import shlex
+
+ try:
+ LOG.info(f"Cli command from client '{sid}'. '{data}'")
+
+ args: list[str] = ["yt-dlp", *shlex.split(data)]
+ _env: dict[str, str] = os.environ.copy()
+ _env.update(
+ {
+ "TERM": "xterm-256color",
+ "LANG": "en_US.UTF-8",
+ "SHELL": "/bin/bash",
+ "LC_ALL": "en_US.UTF-8",
+ "PWD": config.download_path,
+ "FORCE_COLOR": "1",
+ "PYTHONUNBUFFERED": "1",
+ }
+ )
+
+ try:
+ import pty
+
+ master_fd, slave_fd = pty.openpty()
+ stdin_arg = asyncio.subprocess.DEVNULL
+ stdout_arg = stderr_arg = slave_fd
+ use_pty = True
+ except ImportError:
+ use_pty = False
+ master_fd = slave_fd = None
+ stdin_arg = asyncio.subprocess.DEVNULL
+ stdout_arg = asyncio.subprocess.PIPE
+ stderr_arg = asyncio.subprocess.STDOUT
+
+ proc: Process = await asyncio.create_subprocess_exec(
+ *args,
+ cwd=config.download_path,
+ stdin=stdin_arg,
+ stdout=stdout_arg,
+ stderr=stderr_arg,
+ env=_env,
+ )
+
+ if use_pty:
+ try:
+ os.close(slave_fd)
+ except Exception as e:
+ LOG.error(f"Error closing PTY. '{e!s}'.")
+
+ async def reader(sid: str):
+ if use_pty is False:
+ assert proc.stdout is not None
+ async for raw_line in proc.stdout:
+ line = raw_line.rstrip(b"\n")
+ await notify.emit(
+ Events.CLI_OUTPUT,
+ data={"type": "stdout", "line": line.decode("utf-8", errors="replace")},
+ to=sid,
+ )
+ return
+
+ loop: AbstractEventLoop = asyncio.get_running_loop()
+ buffer: bytes = b""
+ while True:
+ try:
+ chunk: bytes = await loop.run_in_executor(None, lambda: os.read(master_fd, 1024))
+ except OSError as e:
+ if e.errno == errno.EIO:
+ break
+ raise
+
+ if not chunk:
+ if buffer:
+ await notify.emit(
+ Events.CLI_OUTPUT,
+ data={"type": "stdout", "line": buffer.decode("utf-8", errors="replace")},
+ to=sid,
+ )
+ break
+
+ buffer += chunk
+ *lines, buffer = buffer.split(b"\n")
+
+ for line in lines:
+ await notify.emit(
+ Events.CLI_OUTPUT,
+ data={"type": "stdout", "line": line.decode("utf-8", errors="replace")},
+ to=sid,
+ )
+ try:
+ os.close(master_fd)
+ except Exception as e:
+ LOG.error(f"Error closing PTY. '{e!s}'.")
+
+ read_task: Task = asyncio.create_task(reader(sid=sid))
+
+ returncode: int = await proc.wait()
+
+ await read_task
+
+ await notify.emit(Events.CLI_CLOSE, data={"exitcode": returncode}, to=sid)
+ except Exception as e:
+ LOG.error(f"CLI execute exception was thrown for client '{sid}'.")
+ LOG.exception(e)
+ await notify.emit(Events.CLI_OUTPUT, data={"type": "stderr", "line": str(e)}, to=sid)
+ await notify.emit(Events.CLI_CLOSE, data={"exitcode": -1}, to=sid)
diff --git a/ui/pages/console.vue b/ui/pages/console.vue
index 69a5a57a..ff99e06c 100644
--- a/ui/pages/console.vue
+++ b/ui/pages/console.vue
@@ -1,3 +1,9 @@
+
+