Refactor HttpSocket to be more manageable and more modular.

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 .router import RouteType, get_routes
-from .Utils import decrypt_data, encrypt_data, get_file, get_mime_type
+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}")

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
-
+    di_context: dict[str, object] = {}
-    subscribers: dict[str, list[str]] = {}
-    """Event subscriber list."""
-
-    log_task = None
-    """Task to tail the log file."""
 
     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.
+        load_modules(self.rootPath, self.rootPath / "routes" / "socket")
-        app.on_shutdown.append(self.on_shutdown)
 
-    @ws_event
+        for route in get_routes(RouteType.SOCKET).values():
-    async def cli_post(self, sid: str, data):
+            LOG.debug(
-        if not self.config.console_enabled:
+                f"Add ({route.name}) {route.method.value if isinstance(route.method,RouteType) else route.method}: {route.path}."
-            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",
-                }
             )
+            self.sio.on(route.path)(HttpSocket._injector(route.handler, route.path, self.di_context))
 
-            try:
+    @staticmethod
-                import pty
+    def _injector(func, event: str, container: dict):
+        sig: inspect.Signature = inspect.signature(func)
 
-                master_fd, slave_fd = pty.openpty()
+        async def wrapper(sid, data, **kwargs):
-                stdin_arg = asyncio.subprocess.DEVNULL
+            args = {}
-                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 = await asyncio.create_subprocess_exec(
+            merged = {**container, "sid": sid, "data": data, "event_name": event}
-                *args,
+            if isinstance(kwargs, dict):
-                cwd=self.config.download_path,
+                merged.update(kwargs)
-                stdin=stdin_arg,
-                stdout=stdout_arg,
-                stderr=stderr_arg,
-                env=_env,
-            )
 
-            if use_pty:
+            for name in sig.parameters:
-                try:
+                if name in merged:
-                    os.close(slave_fd)
+                    args[name] = merged[name]
-                except Exception as e:
-                    LOG.error(f"Error closing PTY. '{e!s}'.")
 
-            async def reader(sid: str):
+            return await func(**args)
-                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
 
-                loop = asyncio.get_running_loop()
+        return wrapper
-                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)

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.method: str = method.upper()
-        self.path = path
+        self.path: str = path
-        self.name = name
+        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)
+        route_type: str = RouteType.SOCKET if RouteType.SOCKET == method else RouteType.HTTP
-        if path.endswith("/") and "/" != path and not kwargs.get("no_slash", False):
+        if route_type not in ROUTES:
-            ROUTES[f"{name}_no_slash"] = Route(
+            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)
+    route_type: str = RouteType.SOCKET if RouteType.SOCKET == method else RouteType.HTTP
-    if path.endswith("/") and "/" != path and not kwargs.get("no_slash", False):
+
-        ROUTES[f"{name}_no_slash"] = Route(
+    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, {})

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)
 

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)

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

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)

ui/pages/console.vue

@@ -1,3 +1,9 @@
+<style>
+.terminal {
+  padding-left: 10px;
+}
+</style>
+
 <template>
   <div class="mt-1 columns is-multiline">
     <div class="column is-12 is-clearfix">