From 501455029c27633657282862781ba4fbf3b151a8 Mon Sep 17 00:00:00 2001 From: arabcoders Date: Wed, 21 Jan 2026 21:06:07 +0300 Subject: [PATCH] refactor: remove socket.io dependency --- .vscode/settings.json | 1 + API.md | 129 +++++---- app/features/core/utils.py | 6 + app/library/Events.py | 18 +- app/library/HttpSocket.py | 147 +++++++--- app/routes/api/logs.py | 59 +++- app/routes/api/system.py | 175 ++++++++++++ app/routes/socket/connection.py | 97 +------ app/routes/socket/terminal.py | 146 ---------- app/tests/test_events.py | 5 - pyproject.toml | 1 - ui/app/pages/console.vue | 106 +++++-- ui/app/pages/logs.vue | 91 +++++- ui/app/pages/tasks.vue | 8 +- ui/app/stores/SocketStore.ts | 490 +++++++++++++++++++------------- ui/app/types/globals.d.ts | 2 +- ui/app/types/sockets.d.ts | 51 ++++ ui/package.json | 2 +- ui/pnpm-lock.yaml | 83 +----- uv.lock | 48 ---- 20 files changed, 963 insertions(+), 702 deletions(-) delete mode 100644 app/routes/socket/terminal.py diff --git a/.vscode/settings.json b/.vscode/settings.json index 8c6c6098..ed4d949e 100644 --- a/.vscode/settings.json +++ b/.vscode/settings.json @@ -202,6 +202,7 @@ "vconvert", "Vitest", "writedescription", + "WSEP", "xerror", "youtu", "youtubepot", diff --git a/API.md b/API.md index 6085f82f..b5c1755f 100644 --- a/API.md +++ b/API.md @@ -72,6 +72,7 @@ This document describes the available endpoints and their usage. All endpoints r - [DELETE /api/conditions/{id}](#delete-apiconditionsid) - [POST /api/conditions/test](#post-apiconditionstest) - [GET /api/logs](#get-apilogs) + - [GET /api/logs/stream](#get-apilogsstream) - [GET /api/notifications/](#get-apinotifications) - [GET /api/notifications/events/](#get-apinotificationsevents) - [POST /api/notifications/](#post-apinotifications) @@ -82,6 +83,7 @@ This document describes the available endpoints and their usage. All endpoints r - [POST /api/yt-dlp/archive\_id/](#post-apiyt-dlparchive_id) - [POST /api/notifications/test](#post-apinotificationstest) - [GET /api/yt-dlp/options](#get-apiyt-dlpoptions) + - [POST /api/system/terminal](#post-apisystemterminal) - [POST /api/system/pause](#post-apisystempause) - [POST /api/system/resume](#post-apisystemresume) - [POST /api/system/shutdown](#post-apisystemshutdown) @@ -95,8 +97,8 @@ This document describes the available endpoints and their usage. All endpoints r - [Message Format](#message-format) - [Core Events](#core-events) - [Connection Events](#connection-events) - - [`connect` (Built-in)](#connect-built-in) - - [`disconnect` (Built-in)](#disconnect-built-in) + - [`connect` (Server → Client)](#connect-server--client) + - [`disconnect` (Server → Client)](#disconnect-server--client) - [`configuration` (Server → Client)](#configuration-server--client) - [`config_update` (Server → Client)](#config_update-server--client) - [`connected` (Server → Client)](#connected-server--client) @@ -106,7 +108,6 @@ This document describes the available endpoints and their usage. All endpoints r - [`log_success` (Server → Client)](#log_success-server--client) - [`log_warning` (Server → Client)](#log_warning-server--client) - [`log_error` (Server → Client)](#log_error-server--client) - - [`log_lines` (Server → Client)](#log_lines-server--client) - [Download Queue Events](#download-queue-events) - [`item_added` (Server → Client)](#item_added-server--client) - [`item_updated` (Server → Client)](#item_updated-server--client) @@ -117,10 +118,6 @@ This document describes the available endpoints and their usage. All endpoints r - [`item_status` (Server → Client)](#item_status-server--client) - [`paused` (Server → Client)](#paused-server--client) - [`resumed` (Server → Client)](#resumed-server--client) - - [Terminal/CLI Events](#terminalcli-events) - - [`cli_post` (Client → Server)](#cli_post-client--server) - - [`cli_output` (Server → Client)](#cli_output-server--client) - - [`cli_close` (Server → Client)](#cli_close-server--client) - [Error Responses](#error-responses) --- @@ -1726,6 +1723,26 @@ Binary image data with appropriate headers --- +### GET /api/logs/stream +**Purpose**: Stream live log lines via Server-Sent Events (SSE). + +**Response**: +- `Content-Type: text/event-stream` +- Emits `log_lines` events with a log line payload. + +**Event Payload**: +```json +{ + "id": "", + "line": "", + "datetime": "2024-01-01T12:00:00.000000+00:00" +} +``` + +- Returns `404 Not Found` if file logging is not enabled or the log file is missing. + +--- + ### GET /api/notifications/ **Purpose**: Retrieve notification targets with pagination. @@ -1904,6 +1921,33 @@ or an error: --- +### POST /api/system/terminal +**Purpose**: Stream yt-dlp CLI output via Server-Sent Events (SSE). Requires `YTP_CONSOLE_ENABLED=true`. + +**Body**: +```json +{ + "command": "--help" +} +``` + +**Response**: +- `Content-Type: text/event-stream` +- Emits `output` events for stdout/stderr and a final `close` event when the process exits. + +**Event Payloads**: +```json +{ "type": "stdout", "line": "" } +``` +```json +{ "exitcode": 0 } +``` + +- `403 Forbidden` if console is disabled. +- `400 Bad Request` if the request body is invalid. + +--- + ### POST /api/system/pause **Purpose**: Pause all non-active downloads in the queue. @@ -2013,7 +2057,7 @@ or an error: ## WebSocket API -The WebSocket API provides real-time bidirectional communication between the client and server using Socket.IO protocol. It enables live updates for downloads, queue status, notifications, and terminal access. +The WebSocket API provides real-time bidirectional communication between the client and server WebSockets. It enables live updates for downloads, queue status, and notifications. > ![IMPORTANT] > The WebSocket API is unstable and many events will be moved to REST endpoints in future releases. @@ -2021,20 +2065,9 @@ The WebSocket API provides real-time bidirectional communication between the cli ### Connection -**URL**: `ws://localhost:8081/socket.io/` (development) or `https://yourdomain.com/socket.io/` (production) +**URL**: `ws://localhost:8081/ws` (development) or `wss://yourdomain.com/ws` (production) -The client automatically connects to the WebSocket server and receives a `connected` event with initial state. The connection uses automatic reconnection with exponential backoff (default: up to 30 attempts, 5s delay). - -**Connection Options**: -```javascript -{ - transports: ['websocket', 'polling'], // Fallback to long-polling if WebSocket unavailable - withCredentials: true, // Include cookies for authentication - reconnection: true, // Enable automatic reconnection - reconnectionAttempts: 30, // Max reconnection attempts - reconnectionDelay: 5000 // Initial reconnection delay in ms -} -``` +The client automatically connects to the WebSocket server and receives a `connected` event with initial state. The frontend wrapper handles reconnection (default: up to 50 attempts, 5s delay). ### Authentication @@ -2047,37 +2080,42 @@ If Basic Authentication is configured, include credentials when establishing the 2. **Via Query Parameter**: ``` - ws://localhost:8081/socket.io/?apikey=:")> + ws://localhost:8081/ws?apikey=:")> ``` ### Message Format -All WebSocket messages are JSON-encoded and follow a consistent structure: +All WebSocket messages are JSON-encoded and follow a consistent envelope: -**Server-to-Client (Event)** - Emitted by server, received by client: +**Server-to-Client (Event Envelope)** ```json { - "id": "unique-event-id", - "created_at": "2024-01-15T10:30:00.000000+00:00", "event": "item_added", - "title": "Item Queued", - "message": "Video added to download queue", - "data": {...} + "data": { + "id": "unique-event-id", + "created_at": "2024-01-15T10:30:00.000000+00:00", + "event": "item_added", + "title": "Item Queued", + "message": "Video added to download queue", + "data": {"_id": "abc123", "title": "..."} + } } ``` +> **Note**: Every message follows the same envelope and `data` is always a JSON object. + ### Core Events #### Connection Events -##### `connect` (Built-in) +##### `connect` (Server → Client) Fired when WebSocket connection is established. No data payload. ```typescript socket.on('connect', () => console.log('WebSocket connected')); ``` -##### `disconnect` (Built-in) +##### `disconnect` (Server → Client) Fired when WebSocket connection is closed. No data payload. ```typescript @@ -2140,12 +2178,7 @@ Warning notification message. ##### `log_error` (Server → Client) Error notification message. -##### `log_lines` (Server → Client) -Continuous application log lines (requires subscription event). - -**Data Fields**: -- `line`: Log line content -- `timestamp`: Log timestamp +For continuous log streaming, use `GET /api/logs/stream` via SSE. ### Download Queue Events @@ -2239,28 +2272,6 @@ Emitted when the download queue is paused. #### `resumed` (Server → Client) Emitted when the download queue is resumed. -### Terminal/CLI Events - -The terminal feature requires `YTP_CONSOLE_ENABLED=true`. - -#### `cli_post` (Client → Server) -Execute a command via yt-dlp CLI. - -**Data**: Command arguments as string (without `yt-dlp` prefix) - -#### `cli_output` (Server → Client) -Output line from the CLI command execution. - -**Data Fields**: -- `type`: Output type (`stdout` or `stderr`) -- `line`: Output line content - -#### `cli_close` (Server → Client) -Emitted when CLI command execution completes. - -**Data Fields**: -- `exitcode`: Command exit code (0 = success, non-zero = error) - --- ## Error Responses diff --git a/app/features/core/utils.py b/app/features/core/utils.py index f71570ca..6e2faa58 100644 --- a/app/features/core/utils.py +++ b/app/features/core/utils.py @@ -75,3 +75,9 @@ def format_validation_errors(exc: ValidationError) -> list[dict[str, Any]]: } for error in exc.errors() ] + + +def gen_random(length: int = 16) -> str: + import secrets + + return "".join(secrets.token_urlsafe(length)[:length]) diff --git a/app/library/Events.py b/app/library/Events.py index 72f9c691..2cef42a2 100644 --- a/app/library/Events.py +++ b/app/library/Events.py @@ -1,11 +1,12 @@ import asyncio import datetime import logging -import uuid from collections.abc import Callable from dataclasses import dataclass, field from typing import Any +from app.features.core.utils import gen_random + from .BackgroundWorker import BackgroundWorker from .Singleton import Singleton @@ -50,10 +51,6 @@ class Events: PAUSED: str = "paused" RESUMED: str = "resumed" - CLI_POST: str = "cli_post" - CLI_CLOSE: str = "cli_close" - CLI_OUTPUT: str = "cli_output" - TASKS_ADD: str = "task_add" TASK_DISPATCHED: str = "task_dispatched" TASK_FINISHED: str = "task_finished" @@ -61,9 +58,6 @@ class Events: SCHEDULE_ADD: str = "schedule_add" - SUBSCRIBED: str = "subscribed" - UNSUBSCRIBED: str = "unsubscribed" - def get_all() -> list: """ Get all the events. @@ -101,8 +95,6 @@ class Events: Events.ITEM_STATUS, Events.PAUSED, Events.RESUMED, - Events.CLI_CLOSE, - Events.CLI_OUTPUT, ] def only_debug() -> list: @@ -113,7 +105,7 @@ class Events: list: The list of debug events. """ - return [Events.ITEM_UPDATED, Events.CLI_OUTPUT] + return [Events.ITEM_UPDATED] @dataclass(kw_only=True) @@ -122,7 +114,7 @@ class Event: Event is a data transfer object that represents an event that was emitted. """ - id: str = field(default_factory=lambda: str(uuid.uuid4()), init=False) + id: str = field(default_factory=lambda: str(gen_random(16)), init=False) """The id of the event.""" created_at: str = field(default_factory=lambda: str(datetime.datetime.now(tz=datetime.UTC).isoformat())) @@ -258,7 +250,7 @@ class EventBus(metaclass=Singleton): event = [event] if not name: - name = str(uuid.uuid4()) + name = gen_random(12) for e in event: if e not in all_events: diff --git a/app/library/HttpSocket.py b/app/library/HttpSocket.py index b73ac236..33163277 100644 --- a/app/library/HttpSocket.py +++ b/app/library/HttpSocket.py @@ -1,12 +1,13 @@ import functools +import json import logging from pathlib import Path from typing import Any -import socketio from aiohttp import web -from app.library.router import RouteType, get_routes +from app.features.core.utils import gen_random +from app.library.router import Route, RouteType, get_routes from app.library.Services import Services from app.library.Utils import load_modules @@ -18,13 +19,55 @@ from .ItemDTO import Item LOG: logging.Logger = logging.getLogger("socket_api") +class WebSocketHub: + def __init__(self, encoder: Encoder): + self._encoder: Encoder = encoder + self._clients: dict[str, web.WebSocketResponse] = {} + + def add(self, sid: str, ws: web.WebSocketResponse) -> None: + self._clients[sid] = ws + + def remove(self, sid: str) -> None: + self._clients.pop(sid, None) + + async def emit(self, event: str, data: Any, to: str | None = None) -> None: + payload: str = self._encoder.encode({"event": event, "data": data}) + + if to: + await self._send(to, payload) + return + + for sid in list(self._clients.keys()): + await self._send(sid, payload) + + async def disconnect(self, sid: str) -> None: + ws: web.WebSocketResponse | None = self._clients.pop(sid, None) + if ws and not ws.closed: + await ws.close() + + async def disconnect_all(self) -> None: + for sid in list(self._clients.keys()): + await self.disconnect(sid) + + async def _send(self, sid: str, payload: str) -> None: + ws: web.WebSocketResponse | None = self._clients.get(sid) + if not ws or ws.closed: + self._clients.pop(sid, None) + return + + try: + await ws.send_str(payload) + except ConnectionResetError: + self._clients.pop(sid, None) + + class HttpSocket: """ This class is used to handle WebSocket events. """ config: Config - sio: socketio.AsyncServer + sio: WebSocketHub di_context: dict[str, Any] = {} def __init__( @@ -32,26 +75,18 @@ class HttpSocket: root_path: Path, encoder: Encoder | None = None, config: Config | None = None, - sio: socketio.AsyncServer | None = None, + sio: WebSocketHub | None = None, ): self.config = config or Config.get_instance() self._notify = EventBus.get_instance() - self.sio = sio or socketio.AsyncServer( - async_handlers=True, - async_mode="aiohttp", - cors_allowed_origins="*", - transports=["websocket", "polling"], - logger=self.config.debug, - engineio_logger=self.config.debug, - ping_interval=10, - ping_timeout=5, - ) encoder = encoder or Encoder() + self.sio = sio or WebSocketHub(encoder=encoder) self.rootPath: Path = root_path async def event_handler(e: Event, _, **kwargs): - await self.sio.emit(event=e.event, data=encoder.encode(e), **kwargs) + payload = json.loads(encoder.encode(e)) + await self.sio.emit(event=e.event, data=payload, **kwargs) services: Services = Services.get_instance() services.add_all( @@ -85,38 +120,86 @@ class HttpSocket: async def on_shutdown(self, _: web.Application): LOG.debug("Shutting down socket server.") - - for sid in self.sio.manager.get_participants("/", None): - LOG.debug(f"Disconnecting client '{sid}'.") - await self.sio.disconnect(sid[0], namespace="/") - + await self.sio.disconnect_all() 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") + base_path: str = self.config.base_path.rstrip("/") + ws_path: str = f"{base_path}/ws" if base_path else "/ws" async def event_handler(data: Event, _): - if data and data.data: - await Services.get_instance().get("queue").add(item=Item.format(data.data)) + if not (data and data.data): + return + + await Services.get_instance().get("queue").add(item=Item.format(data.data)) self._notify.subscribe(Events.ADD_URL, event_handler, f"{__class__.__name__}.add") load_modules(self.rootPath, self.rootPath / "routes" / "socket") - for route in get_routes(RouteType.SOCKET).values(): + socket_routes: dict[str, Route] = {route.path: route for route in get_routes(RouteType.SOCKET).values()} + + for route in socket_routes.values(): if self.config.debug: 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)) - @staticmethod - def _injector(func, event: str): - async def wrapper(sid, data=None, **kwargs): - if not data: - data = {} - return await Services.get_instance().handle_async(func, sid=sid, data=data, event=event, **kwargs) + async def handle_message(sid: str, message: str) -> None: + try: + payload = json.loads(message) + except json.JSONDecodeError: + LOG.debug("Invalid websocket payload received.") + return - return wrapper + event = payload.get("event") if isinstance(payload, dict) else None + if not event or not isinstance(event, str): + LOG.debug("Missing websocket event name.") + return + + if not (route := socket_routes.get(event)): + LOG.debug(f"Unknown websocket event '{event}'.") + return + + data = payload.get("data") if isinstance(payload, dict) else None + await Services.get_instance().handle_async(route.handler, sid=sid, data=data, event=event) + + async def handle_connect(sid: str) -> None: + if not (route := socket_routes.get("connect")): + return + await Services.get_instance().handle_async(route.handler, sid=sid, event="connect") + + async def handle_disconnect(sid: str, reason: str | None) -> None: + if not (route := socket_routes.get("disconnect")): + return + await Services.get_instance().handle_async(route.handler, sid=sid, data=reason, event="disconnect") + + async def ws_handler(request: web.Request) -> web.WebSocketResponse: + ws = web.WebSocketResponse(heartbeat=10) + await ws.prepare(request) + + sid: str = gen_random(14) + self.sio.add(sid, ws) + LOG.debug(f"WebSocket client '{sid}' connected.") + + await handle_connect(sid) + + try: + async for msg in ws: + if msg.type == web.WSMsgType.TEXT: + await handle_message(sid, msg.data) + elif msg.type == web.WSMsgType.ERROR: + LOG.error(f"WebSocket connection closed with exception {ws.exception()}") + finally: + await handle_disconnect(sid, getattr(ws, "close_reason", None)) + self.sio.remove(sid) + LOG.debug(f"WebSocket client '{sid}' disconnected.") + + return ws + + if self.config.debug: + LOG.debug(f"Add (ws) GET: {ws_path}.") + + app.router.add_get(ws_path, ws_handler, name="ws") diff --git a/app/routes/api/logs.py b/app/routes/api/logs.py index 6e00a23b..073d4fc4 100644 --- a/app/routes/api/logs.py +++ b/app/routes/api/logs.py @@ -1,3 +1,5 @@ +import asyncio +import logging from pathlib import Path from aiohttp import web @@ -6,7 +8,9 @@ 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 +from app.library.Utils import read_logfile, tail_log + +LOG: logging.Logger = logging.getLogger(__name__) @route("GET", "api/logs/", "logs") @@ -48,3 +52,56 @@ async def logs(request: Request, config: Config, encoder: Encoder) -> Response: status=web.HTTPOk.status_code, dumps=encoder.encode, ) + + +@route("GET", "api/logs/stream", "logs.stream") +async def stream_logs(request: Request, config: Config, encoder: Encoder) -> Response | web.StreamResponse: + if not config.file_logging: + return web.json_response( + data={"error": "File logging is not enabled."}, + status=web.HTTPNotFound.status_code, + ) + + log_file = Path(config.config_path) / "logs" / "app.log" + if not log_file.exists(): + return web.json_response( + data={"error": "Log file is not available."}, + status=web.HTTPNotFound.status_code, + ) + + response = web.StreamResponse( + status=web.HTTPOk.status_code, + headers={ + "Content-Type": "text/event-stream", + "Cache-Control": "no-cache", + "Connection": "keep-alive", + }, + ) + await response.prepare(request) + + async def emit_log(data: dict) -> None: + if request.transport is None or request.transport.is_closing(): + raise asyncio.CancelledError + payload = f"event: log_lines\ndata: {encoder.encode(data)}\n\n" + await response.write(payload.encode("utf-8")) + + log_task: asyncio.Task[None] = asyncio.create_task(tail_log(file=log_file, emitter=emit_log), name="log_stream") + + try: + LOG.debug("Log streaming connected.") + while not log_task.done(): + await asyncio.sleep(0.5) + if request.transport is None or request.transport.is_closing(): + log_task.cancel() + break + await log_task + except asyncio.CancelledError: + pass + finally: + LOG.debug("Log streaming disconnected.") + try: + await response.write_eof() + except ConnectionResetError: + pass + + return response diff --git a/app/routes/api/system.py b/app/routes/api/system.py index 490b98b8..84530365 100644 --- a/app/routes/api/system.py +++ b/app/routes/api/system.py @@ -1,6 +1,10 @@ import asyncio +import errno import logging +import os +import shlex import time +from typing import TYPE_CHECKING from aiohttp import web from aiohttp.web import Request, Response @@ -13,6 +17,11 @@ from app.library.Events import EventBus, Events from app.library.router import route from app.library.UpdateChecker import UpdateChecker +if TYPE_CHECKING: + from asyncio import Task + from asyncio.events import AbstractEventLoop + from asyncio.subprocess import Process + LOG: logging.Logger = logging.getLogger(__name__) @@ -178,3 +187,169 @@ async def check_updates(config: Config, encoder: Encoder, update_checker: Update status=web.HTTPOk.status_code, dumps=encoder.encode, ) + + +LOG: logging.Logger = logging.getLogger(__name__) + + +@route("POST", "api/system/terminal", "system.terminal") +async def stream_terminal(request: Request, config: Config, encoder: Encoder) -> Response | web.StreamResponse: + if not config.console_enabled: + return web.json_response( + {"error": "Console feature is disabled."}, + status=web.HTTPForbidden.status_code, + ) + + if not request.can_read_body: + return web.json_response( + {"error": "Request body is required."}, + status=web.HTTPBadRequest.status_code, + ) + + payload = await request.json() + if not isinstance(payload, dict): + return web.json_response( + {"error": "Invalid request payload."}, + status=web.HTTPBadRequest.status_code, + ) + + raw_command = payload.get("command") + if not isinstance(raw_command, str) or not raw_command.strip(): + return web.json_response( + {"error": "Command is required."}, + status=web.HTTPBadRequest.status_code, + ) + + response = web.StreamResponse( + status=web.HTTPOk.status_code, + headers={ + "Content-Type": "text/event-stream", + "Cache-Control": "no-cache", + "Connection": "keep-alive", + }, + ) + await response.prepare(request) + + async def emit_event(event: str, data: dict) -> None: + if request.transport is None or request.transport.is_closing(): + return + payload: str = f"event: {event}\ndata: {encoder.encode(data)}\n\n" + await response.write(payload.encode("utf-8")) + + returncode: int = -1 + try: + LOG.info("Cli command from client. '%s'", raw_command) + + args: list[str] = ["yt-dlp", *shlex.split(raw_command, posix=os.name != "nt")] + env_vars: dict[str, str] = os.environ.copy() + env_vars.update( + { + "PWD": config.download_path, + "FORCE_COLOR": "1", + "PYTHONUNBUFFERED": "1", + } + ) + + if "nt" != os.name: + env_vars.update( + { + "TERM": "xterm-256color", + "LANG": "en_US.UTF-8", + "LC_ALL": "en_US.UTF-8", + "SHELL": "/bin/bash", + } + ) + + 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 + + creationflags = 0 + if os.name == "nt": + import subprocess + + creationflags = subprocess.CREATE_NO_WINDOW + + proc: Process = await asyncio.create_subprocess_exec( + *args, + cwd=config.download_path, + stdin=stdin_arg, + stdout=stdout_arg, + stderr=stderr_arg, + env=env_vars, + creationflags=creationflags, + ) + + if use_pty: + assert slave_fd is not None + try: + os.close(slave_fd) + except Exception as e: + LOG.error("Error closing PTY. '%s'.", str(e)) + + async def reader() -> None: + 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 emit_event("output", {"type": "stdout", "line": line.decode("utf-8", errors="replace")}) + return + + assert master_fd is not None + read_fd = master_fd + loop: AbstractEventLoop = asyncio.get_running_loop() + buffer: bytes = b"" + while True: + try: + chunk: bytes = await loop.run_in_executor(None, lambda: os.read(read_fd, 1024)) + except OSError as e: + if e.errno == errno.EIO: + break + raise + + if not chunk: + if buffer: + await emit_event( + "output", + {"type": "stdout", "line": buffer.decode("utf-8", errors="replace")}, + ) + break + + buffer += chunk + *lines, buffer = buffer.split(b"\n") + + for line in lines: + await emit_event( + "output", + {"type": "stdout", "line": line.decode("utf-8", errors="replace")}, + ) + if master_fd is None: + return + try: + os.close(master_fd) + except Exception as e: + LOG.error("Error closing PTY. '%s'.", str(e)) + + read_task: Task = asyncio.create_task(reader(), name="cli_reader") + + returncode = await proc.wait() + await read_task + except Exception as e: + LOG.error("CLI execute exception was thrown.") + LOG.exception(e) + await emit_event("output", {"type": "stderr", "line": str(e)}) + finally: + await emit_event("close", {"exitcode": returncode}) + await response.write_eof() + + return response diff --git a/app/routes/socket/connection.py b/app/routes/socket/connection.py index b5f80a09..57e8a1de 100644 --- a/app/routes/socket/connection.py +++ b/app/routes/socket/connection.py @@ -1,24 +1,20 @@ -import asyncio import logging from pathlib import Path -from typing import Any - -import socketio from app.features.dl_fields.service import DLFields from app.library.config import Config from app.library.downloads import DownloadQueue from app.library.Events import EventBus, Events +from app.library.HttpSocket import WebSocketHub from app.library.Presets import Presets from app.library.router import RouteType, route -from app.library.Utils import list_folders, tail_log +from app.library.Utils import list_folders 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") @@ -62,99 +58,14 @@ async def connect(config: Config, queue: DownloadQueue, notify: EventBus, sid: s @route(RouteType.SOCKET, "disconnect", "socket_disconnect") -async def disconnect(sio: socketio.AsyncServer, sid: str, data: str = None): +async def disconnect(sio: WebSocketHub, sid: str, data: str | None = None): # noqa: ARG001 """ Handle client disconnection. Args: - sio (socketio.AsyncServer): The Socket.IO server instance. + sio (WebSocketHub): The WebSocket hub instance. 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(sio=sio, sid=sid, data=event) - - -@route(RouteType.SOCKET, "subscribe", "socket_subscribe") -async def subscribe(config: Config, notify: EventBus, sio: socketio.AsyncServer, sid: str, data: str | Any): - """ - 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) or not data: - notify.emit( - Events.LOG_ERROR, - title="Subscription Error", - message="Invalid event type was expecting a string.", - 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="log_lines", 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/terminal.py b/app/routes/socket/terminal.py deleted file mode 100644 index 146c39ff..00000000 --- a/app/routes/socket/terminal.py +++ /dev/null @@ -1,146 +0,0 @@ -import logging -import os -from typing import TYPE_CHECKING - -from app.library.config import Config -from app.library.Events import EventBus, Events -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: - notify.emit( - Events.LOG_ERROR, - title="Feature disabled", - message="Console feature is disabled.", - to=sid, - ) - return - - if not data: - notify.emit(Events.CLI_CLOSE, data={"exitcode": 0}, to=sid) - return - - import asyncio - import errno - import shlex - import subprocess # ignore - - returncode: int = -1 - try: - LOG.info(f"Cli command from client '{sid}'. '{data}'") - - args: list[str] = ["yt-dlp", *shlex.split(data, posix=os.name != "nt")] - _env: dict[str, str] = os.environ.copy() - _env.update( - { - "PWD": config.download_path, - "FORCE_COLOR": "1", - "PYTHONUNBUFFERED": "1", - } - ) - - if "nt" != os.name: - _env.update( - { - "TERM": "xterm-256color", - "LANG": "en_US.UTF-8", - "LC_ALL": "en_US.UTF-8", - "SHELL": "/bin/bash", - } - ) - - 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, - creationflags=subprocess.CREATE_NO_WINDOW if os.name == "nt" else 0, - ) - - 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") - 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: - 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: - 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), name=f"cli_reader_{sid}") - - returncode: int = await proc.wait() - - await read_task - except Exception as e: - LOG.error(f"CLI execute exception was thrown for client '{sid}'.") - LOG.exception(e) - notify.emit(Events.CLI_OUTPUT, data={"type": "stderr", "line": str(e)}, to=sid) - finally: - notify.emit(Events.CLI_CLOSE, data={"exitcode": returncode}, to=sid) diff --git a/app/tests/test_events.py b/app/tests/test_events.py index 563dd6f6..a6d3f4a4 100644 --- a/app/tests/test_events.py +++ b/app/tests/test_events.py @@ -105,7 +105,6 @@ class TestEvents: assert isinstance(debug_events, list) assert Events.ITEM_UPDATED in debug_events - assert Events.CLI_OUTPUT in debug_events class TestEvent: @@ -783,10 +782,6 @@ class TestEventBus: assert Events.TEST in bus._listeners assert len(bus._listeners[Events.TEST]) == 1 - # Name should be a UUID - subscriber_name = bus._listeners[Events.TEST][0][0] - assert len(subscriber_name) == 36 # UUID string length - @patch("app.library.config.Config") @patch("app.library.BackgroundWorker.BackgroundWorker") def test_event_bus_unsubscribe_single_event(self, mock_bg_worker, mock_config): diff --git a/pyproject.toml b/pyproject.toml index f08acec9..a618c6d3 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -13,7 +13,6 @@ dependencies = [ # Non‐Windows "python-magic>=0.4.27; sys_platform != 'win32'", # Cross‐platform - "python-socketio>=5.11.1", "aiohttp>=3.9.3", "coloredlogs>=15.0.1", "aiocron>=1.8", diff --git a/ui/app/pages/console.vue b/ui/app/pages/console.vue index 04410f3b..0a6c3f8b 100644 --- a/ui/app/pages/console.vue +++ b/ui/app/pages/console.vue @@ -112,18 +112,19 @@