danvics/ytptube
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1

Cleaning up the event dispatcher system.

Browse source
This commit is contained in:
arabcoders 2025-07-18 21:06:09 +03:00
parent 84f623f2e4
commit 09c5d7794d
13 changed files with 275 additions and 188 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
app/library/Download.py
View file

@ -492,10 +492,10 @@ class Download:
if "error" == self.info.status and "error" in status:
self.info.error = status.get("error")
await self._notify.emit(
Events.ERROR,
data={"message": self.info.error, "data": self.info},
Events.LOG_ERROR,
data=self.info,
title="Download Error",
message=f"Error in download task '{self.info.title}': {self.info.error}",
message=f"'{self.info.title}' failed to download: {self.info.error}",
)
if "downloaded_bytes" in status and status.get("downloaded_bytes") > 0:

128
app/library/DownloadQueue.py
View file

@ -19,8 +19,6 @@ from .config import Config
from .DataStore import DataStore, StoreType
from .Download import Download
from .Events import EventBus, Events
from .Events import info as event_info
from .Events import warning as event_warning
from .ItemDTO import Item, ItemDTO
from .Presets import Presets
from .Scheduler import Scheduler
@ -452,39 +450,48 @@ class DownloadQueue(metaclass=Singleton):
try:
dlInfo: Download = Download(info=dl, info_dict=entry if item.auto_start else None, logs=logs)
notifyTitle: str | None = None
notifyMessage: str | None = None
text_logs: str = ""
if filtered_logs := extract_ytdlp_logs(logs):
text_logs = f" Logs: {', '.join(filtered_logs)}"
if "is_upcoming" == entry.get("live_status"):
NotifyEvent = Events.COMPLETED
notifyEvent = Events.COMPLETED
notifyTitle = "Upcoming Premiere" if is_premiere else "Upcoming Live Stream"
notifyMessage = f"{'Premiere video' if is_premiere else 'Stream' } '{dlInfo.info.title}' is not available yet. {text_logs}"
dlInfo.info.status = "not_live"
dlInfo.info.msg = f"{'Premiere video' if is_premiere else 'Stream' } is not available yet." + text_logs
dlInfo.info.msg = notifyMessage.replace(f" '{dlInfo.info.title}'", "")
await self._notify.emit(
Events.LOG_INFO,
data=event_info(dlInfo.info.msg, {"lowPriority": True}),
title="Premiere video" if is_premiere else "Live Stream",
message=f"Item '{dlInfo.info.title}' is not available yet. {dlInfo.info.msg}",
data={"lowPriority": True},
title=notifyTitle,
message=notifyMessage,
)
itemDownload: Download = self.done.put(dlInfo)
elif len(entry.get("formats", [])) < 1:
availability: str = entry.get("availability", "public")
msg: str = "No formats found."
if availability and availability not in ("public",):
msg += f" Availability is set for '{availability}'."
ava: str = entry.get("availability", "public")
notifyTitle = "Download Error"
notifyMessage: str = f"No formats for '{dl.title}'."
if ava and ava not in ("public",):
notifyMessage += f" Availability is set for '{ava}'."
dlInfo.info.error = msg + text_logs
dlInfo.info.error = notifyMessage.replace(f" for '{dl.title}'.", ".") + text_logs
dlInfo.info.status = "error"
itemDownload = self.done.put(dlInfo)
NotifyEvent = Events.COMPLETED
notifyEvent = Events.COMPLETED
await self._notify.emit(
Events.LOG_WARNING,
data=event_warning(f"No formats found for '{dl.title}'."),
title="No Formats Found",
message=f"No formats found for '{dl.title}'. {dlInfo.info.error}",
data={"logs": text_logs},
title=notifyTitle,
message=notifyMessage,
)
elif is_premiere and self.config.prevent_live_premiere:
notifyTitle = "Premiere Video"
dlInfo.info.error = "Premiering right now."
_requeue = True
@ -496,6 +503,7 @@ class DownloadQueue(metaclass=Singleton):
)
starts_in = starts_in + timedelta(minutes=5, seconds=dl.extras.get("duration", 0))
dlInfo.info.error += f" Download will start at {starts_in.astimezone().isoformat()}."
notifyMessage = dlInfo.info.error.strip()
_requeue = False
except Exception as e:
LOG.error(f"Failed to parse live_in date '{release_in}'. {e!s}")
@ -503,25 +511,24 @@ class DownloadQueue(metaclass=Singleton):
else:
dlInfo.info.error += f" Delaying download by '{300+dl.extras.get('duration',0)}' seconds."
notifyMessage = dlInfo.info.error.strip()
if _requeue:
NotifyEvent = Events.ADDED
notifyEvent = Events.ADDED
itemDownload = self.queue.put(dlInfo)
if item.auto_start:
self.event.set()
else:
LOG.debug(f"Item {itemDownload.info.name()} is not set to auto-start.")
else:
dlInfo.info.status = "not_live"
itemDownload = self.done.put(dlInfo)
NotifyEvent = Events.COMPLETED
await self._notify.emit(
Events.LOG_INFO,
data=event_info(f"'{dl.title}' is {dlInfo.info.error}.", {"lowPriority": True}),
title="Item Not Live",
message=f"Item '{dl.title}' is not live. {dlInfo.info.error}",
)
notifyEvent = Events.COMPLETED
notifyTitle = "Item Not Live"
notifyMessage = f"Item '{dlInfo.info.title}' is not live."
await self._notify.emit(Events.LOG_INFO, title=notifyTitle, message=notifyMessage)
else:
NotifyEvent = Events.ADDED
notifyEvent = Events.ADDED
notifyTitle = "Item Added"
notifyMessage = f"Item '{dlInfo.info.title}' has been added to the download queue."
itemDownload = self.queue.put(dlInfo)
if item.auto_start:
self.event.set()
@ -529,10 +536,10 @@ class DownloadQueue(metaclass=Singleton):
LOG.debug(f"Item {itemDownload.info.name()} is not set to auto-start.")
await self._notify.emit(
NotifyEvent,
notifyEvent,
data=itemDownload.info.serialize(),
title="Item Added" if Events.ADDED == NotifyEvent else None,
message=f"Added '{itemDownload.info.title}'." if NotifyEvent == Events.ADDED else None,
title=notifyTitle,
message=notifyMessage,
)
return {"status": "ok"}
@ -740,14 +747,18 @@ class DownloadQueue(metaclass=Singleton):
self.queue.delete(id)
await self._notify.emit(
Events.CANCELLED,
data=item.info.serialize(),
data=item.info,
title="Download Cancelled",
message=f"Download '{item.info.title}' has been cancelled.",
)
item.info.status = "cancelled"
# item.info.error = "Cancelled by user."
self.done.put(item)
await self._notify.emit(Events.COMPLETED, data=item.info.serialize())
await self._notify.emit(
Events.COMPLETED,
data=item.info,
title="Download Cancelled",
message=f"Download '{item.info.title}' has been cancelled.",
)
LOG.info(f"Deleted from queue {item_ref}")
status[id] = "ok"
@ -815,11 +826,13 @@ class DownloadQueue(metaclass=Singleton):
LOG.error(f"Unable to remove '{itemRef}' local file '{filename}'. {e!s}")
self.done.delete(id)
await self._notify.emit(
_status: str = "Removed" if removed_files > 0 else "Cleared"
self._notify.offload(
Events.CLEARED,
data=item.info.serialize(),
title="Download Cleared",
message=f"Cleared download '{item.info.title}' from history.",
data=item.info,
title=f"Download {_status}",
message=f"{_status} '{item.info.title}' from history.",
)
msg = f"Deleted completed download '{itemRef}'."
@ -857,6 +870,27 @@ class DownloadQueue(metaclass=Singleton):
return items
def get_item(self, id: str) -> Download | None:
"""
Get a specific item from the download queue or history.
Args:
id (str): The ID of the item to retrieve.
Returns:
Download | None: The requested item if found, otherwise None.
"""
try:
return self.queue.get(key=id)
except KeyError:
pass
try:
return self.done.get(key=id)
except KeyError:
return None
async def _download_pool(self) -> None:
"""
Create a pool of workers to download the files.
@ -931,25 +965,25 @@ class DownloadQueue(metaclass=Singleton):
LOG.debug(f"Download Task '{id}' is completed. Removing from queue.")
self.queue.delete(key=id)
notifyTitle: str | None = None
notifyMessage: str | None = None
if entry.is_cancelled() is True:
await self._notify.emit(
Events.CANCELLED,
data=entry.info.serialize(),
data=entry.info,
title="Download Cancelled",
message=f"Download '{entry.info.title}' has been cancelled.",
)
entry.info.status = "cancelled"
# entry.info.error = "Cancelled by user."
notifyTitle = "Download Cancelled"
notifyMessage = f"Download '{entry.info.title}' has been cancelled."
elif entry.info.status == "finished":
notifyTitle = "Download Completed"
notifyMessage = f"Download '{entry.info.title}' has been completed."
self.done.put(value=entry)
await self._notify.emit(
Events.COMPLETED,
data=entry.info.serialize(),
title="Download Completed" if entry.info.status == "finished" else None,
message=f"Download '{entry.info.title}' has been completed."
if entry.info.status == "finished"
else None,
)
await self._notify.emit(Events.COMPLETED, data=entry.info, title=notifyTitle, message=notifyMessage)
else:
LOG.warning(f"Download '{id}' not found in queue.")

149
app/library/Events.py
View file

@ -6,86 +6,10 @@ from collections.abc import Awaitable
from dataclasses import dataclass, field
from typing import Any
from .BackgroundWorker import BackgroundWorker
from .Singleton import Singleton
LOG = logging.getLogger("Events")
def error(msg: str, data: dict | None = None) -> dict:
"""
Create an error message.
Args:
msg (str): The message.
data (dict|None): The data to include in the message.
Returns:
dict : The message wrapped in a dictionary.
"""
return message("error", msg, data)
def warning(msg: str, data: dict | None = None) -> dict:
"""
Create an error message.
Args:
msg (str): The message.
data (dict|None): The data to include in the message.
Returns:
dict : The message wrapped in a dictionary.
"""
return message("warning", msg, data)
def info(msg: str, data: dict | None = None) -> dict:
"""
Create an info message.
Args:
msg (str): The message.
data (dict|None): The data to include in the message.
Returns:
dict : The message wrapped in a dictionary.
"""
return message("info", msg, data)
def success(msg: str, data: dict | None = None) -> dict:
"""
Create a success message.
Args:
msg (str): The message.
data (dict|None): The data to include in the message.
Returns:
dict : The message wrapped in a dictionary.
"""
return message("success", msg, data)
def message(type: str, message: str, data: dict | None = None) -> dict:
"""
Create a message.
Args:
type (str): The type of the message.
message (str): The message.
data (dict|None): The data to include in the message.
Returns:
dict : The message wrapped in a dictionary.
"""
return {"type": type, "message": message, "data": data if data else {}}
LOG: logging.Logger = logging.getLogger("Events")
class Events:
"""
@ -185,7 +109,7 @@ class Events:
list: The list of debug events.
"""
return [Events.UPDATED]
return [Events.UPDATED, Events.CLI_OUTPUT]
@dataclass(kw_only=True)
@ -209,7 +133,7 @@ class Event:
message: str | None = None
"""The message of the event, if any."""
data: any
data: Any
"""The data that was passed to the event."""
def serialize(self) -> dict:
@ -277,12 +201,16 @@ class EventBus(metaclass=Singleton):
debug: bool = False
"""Whether to log debug messages or not."""
_offload: BackgroundWorker
"""The background worker to offload tasks to."""
def __init__(self):
EventBus._instance = self
from .config import Config
self.debug = Config.get_instance().debug
self._offload = BackgroundWorker.get_instance()
@staticmethod
def get_instance() -> "EventBus":
@ -368,13 +296,24 @@ class EventBus(metaclass=Singleton):
return self
def sync_emit(self, event: str, data: Any, loop=None, wait: bool = True, **kwargs):
def sync_emit(
self,
event: str,
data: Any | None = None,
title: str | None = None,
message: str | None = None,
loop=None,
wait: bool = True,
**kwargs,
):
"""
Emit event and (optionally) wait for results.
Args:
event (str): The event to emit.
data (Any): The data to pass to the event.
data (Any|None): The data to pass to the event.
title (str | None): The title of the event, if any.
message (str | None): The message of the event, if any.
loop (asyncio.AbstractEventLoop | None): The event loop to use. If None, the current running loop is used.
wait (bool): Whether to wait for the results of the event handlers. Defaults to True.
**kwargs: Additional keyword arguments to pass to the event
@ -389,8 +328,11 @@ class EventBus(metaclass=Singleton):
if event not in self._listeners:
return []
if not data:
data = {}
async def emit_all():
ev = Event(event=event, data=data)
ev = Event(event=event, title=title, message=message, data=data)
LOG.debug(f"Emitting event '{ev.id}: {ev.event}'.", extra={"data": data})
res: list = []
@ -428,14 +370,14 @@ class EventBus(metaclass=Singleton):
return fut.result() if wait else fut
async def emit(
self, event: str, data: Any, title: str | None = None, message: str | None = None, **kwargs
self, event: str, data: Any | None = None, title: str | None = None, message: str | None = None, **kwargs
) -> Awaitable:
"""
Emit an event.
Args:
event (str): The event to emit.
data (Any): The data to pass to the event.
data (Any|None): The data to pass to the event.
title (str | None): The title of the event, if any.
message (str | None): The message of the event, if any.
**kwargs: The keyword arguments to pass to the event.
@ -447,7 +389,10 @@ class EventBus(metaclass=Singleton):
if event not in self._listeners:
return []
ev = Event(event=event, data=data, title=title, message=message)
if not data:
data = {}
ev = Event(event=event, title=title, message=message, data=data)
if self.debug or event not in Events.only_debug():
LOG.debug(f"Emitting event '{ev.id}: {ev.event}'.", extra={"data": data})
@ -461,3 +406,35 @@ class EventBus(metaclass=Singleton):
LOG.error(f"Failed to emit event '{ev.event}' to '{handler.name}'. Error message '{e!s}'.")
return asyncio.gather(*tasks)
def offload(
self, event: str, title: str | None = None, message: str | None = None, data: Any | None = None, **kwargs
) -> None:
"""
Offload an dispatching event to a background worker.
Args:
event (str): The event to offload.
data (Any|None): The data to pass to the event.
title (str | None): The title of the event, if any.
message (str | None): The message of the event, if any.
**kwargs: Additional keyword arguments to pass to the event.
"""
if event not in self._listeners:
return
if not data:
data = {}
ev = Event(event=event, title=title, message=message, data=data)
if self.debug or event not in Events.only_debug():
LOG.debug(f"Offloading event '{ev.id}: {ev.event}'.", extra={"data": data})
for handler in self._listeners[event].values():
try:
self._offload.submit(handler.handle, ev, **kwargs)
except Exception as e:
LOG.exception(e)
LOG.error(f"Failed to offload event '{ev.event}' to '{handler.name}'. Error message '{e!s}'.")

2
app/library/HttpSocket.py
View file

@ -55,7 +55,7 @@ class HttpSocket:
self.rootPath = root_path
def emit(e: Event, _, **kwargs):
return self.sio.emit(event=e.event, data=encoder.encode(e.data), **kwargs)
return self.sio.emit(event=e.event, data=encoder.encode(e), **kwargs)
services = Services.get_instance()
services.add_all(

19
app/library/Tasks.py
View file

@ -15,7 +15,7 @@ from app.library.Services import Services
from .config import Config
from .encoder import Encoder
from .Events import EventBus, Events, error, success
from .Events import EventBus, Events
from .Scheduler import Scheduler
from .Singleton import Singleton
from .Utils import init_class, validate_url
@ -325,7 +325,7 @@ class Tasks(metaclass=Singleton):
"template": template,
"cli": cli,
},
title=f"Task '{task.name}' started",
title="Tasks",
message=f"Task '{task.name}' started at '{timeNow}'",
id=task.id,
)
@ -337,19 +337,16 @@ class Tasks(metaclass=Singleton):
await self._notify.emit(
Events.LOG_SUCCESS,
data=success(
f"Task '{task.name}' completed in '{ended - started:.2f}' seconds.", data={"lowPriority": True}
),
title=f"Task '{task.name}' completed",
message=f"Task '{task.name}' completed at '{timeNow}'",
data={"lowPriority": True},
title="Task completed",
message=f"Task '{task.name}' completed in '{ended - started:.2f}'.",
)
except Exception as e:
LOG.error(f"Failed to execute '{task.name}' at '{timeNow}'. '{e!s}'.")
await self._notify.emit(
Events.ERROR,
data=error(f"Failed to execute '{task.name}' at '{timeNow}'. '{e!s}'."),
title=f"Task '{task.name}' failed",
message=str(e),
Events.LOG_ERROR,
title="Task failed",
message=f"Failed to execute '{task.name}'. '{e!s}'",
)

23
app/main.py
View file

@ -111,7 +111,12 @@ class Main:
host = host or self._config.host
port = port or self._config.port
EventBus.get_instance().sync_emit(Events.STARTUP, data={"app": self._app})
EventBus.get_instance().sync_emit(
Events.STARTUP,
data={"app": self._app},
title="Application Startup",
message="The application is starting up.",
)
Scheduler.get_instance().attach(self._app)
self._socket.attach(self._app)
@ -123,7 +128,12 @@ class Main:
Notification.get_instance().attach(self._app)
Conditions.get_instance().attach(self._app)
EventBus.get_instance().sync_emit(Events.LOADED, data={"app": self._app})
EventBus.get_instance().sync_emit(
Events.LOADED,
data={"app": self._app},
title="Application Loaded",
message="The application has loaded all components.",
)
def started(_):
LOG.info("=" * 40)
@ -132,7 +142,14 @@ class Main:
loop = asyncio.get_event_loop()
EventBus.get_instance().sync_emit(Events.STARTED, data={"app": self._app}, loop=loop, wait=False)
EventBus.get_instance().sync_emit(
Events.STARTED,
data={"app": self._app},
title="Application Started",
message="The application has started successfully.",
loop=loop,
wait=False,
)
if loop and self._config.debug:
loop.set_debug(True)

8
app/routes/api/notifications.py
View file

@ -5,7 +5,7 @@ from aiohttp import web
from aiohttp.web import Request, Response
from app.library.encoder import Encoder
from app.library.Events import EventBus, Events, message
from app.library.Events import EventBus, Events
from app.library.Notifications import Notification, NotificationEvents
from app.library.router import route
from app.library.Utils import validate_uuid
@ -104,8 +104,6 @@ async def notification_test(encoder: Encoder, notify: EventBus) -> Response:
Response: The response object.
"""
data = message("test", "This is a test notification.")
await notify.emit(Events.TEST, title="Test Notification", message="This is a test notification.")
await notify.emit(Events.TEST, data=data, title="Test Notification", message="This is a test notification.")
return web.json_response(data=data, status=web.HTTPOk.status_code, dumps=encoder.encode)
return web.json_response(data={}, status=web.HTTPOk.status_code, dumps=encoder.encode)

14
app/routes/socket/connection.py
View file

@ -1,12 +1,13 @@
import asyncio
import logging
from pathlib import Path
from typing import Any
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.Events import EventBus, Events
from app.library.Presets import Presets
from app.library.router import RouteType, route
from app.library.Utils import tail_log
@ -52,7 +53,7 @@ async def disconnect(sio: socketio.AsyncServer, sid: str, data: str = None):
@route(RouteType.SOCKET, "subscribe", "socket_subscribe")
async def subscribe(config: Config, notify: EventBus, sio: socketio.AsyncServer, sid: str, data: str):
async def subscribe(config: Config, notify: EventBus, sio: socketio.AsyncServer, sid: str, data: str | Any):
"""
Subscribe to a specific event.
@ -64,8 +65,13 @@ async def subscribe(config: Config, notify: EventBus, sio: socketio.AsyncServer,
data (str): The event to subscribe to.
"""
if not isinstance(data, str):
await notify.emit(Events.ERROR, data=error("Invalid event."), to=sid)
if not isinstance(data, str) or not data:
notify.offload(
Events.ERROR,
title="Subscription Error",
message="Invalid event type was expecting a string.",
to=sid,
)
return
if data not in _Data.subscribers:

64
app/routes/socket/history.py
View file

@ -7,7 +7,7 @@ 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.Events import EventBus, Events
from app.library.ItemDTO import Item
from app.library.router import RouteType, route
from app.library.Utils import is_downloaded
@ -43,25 +43,37 @@ 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)
await notify.emit(
Events.LOG_ERROR,
title="Invalid URL",
message="Please provide a valid URL to add to the download queue.",
to=sid,
)
return
try:
status = await queue.add(item=Item.format(data))
await notify.emit(
event=Events.STATUS,
data=await queue.add(item=Item.format(data)),
title="Adding URL",
message=f"Adding URL '{url}' to the download queue.",
data=status,
to=sid,
)
except ValueError as e:
LOG.exception(e)
await notify.emit(Events.ERROR, data=error(str(e)), to=sid)
return
await notify.emit(Events.LOG_ERROR, title="Error Adding URL", message=str(e), to=sid)
@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)
await notify.emit(
Events.LOG_ERROR,
title="Invalid Request",
message="No item ID provided to cancel.",
to=sid,
)
return
status: dict[str, str] = {}
@ -74,12 +86,32 @@ async def item_cancel(queue: DownloadQueue, notify: EventBus, sid: str, data: st
@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)
await notify.emit(
Events.LOG_ERROR,
title="Invalid Request",
message="No item ID provided to delete.",
to=sid,
)
return
id: str | None = data.get("id")
if not id:
await notify.emit(Events.ERROR, data=error("Invalid request."), to=sid)
await notify.emit(
Events.LOG_ERROR,
title="Invalid Request",
message="No item ID provided to delete.",
to=sid,
)
return
item = queue.get_item(id=id)
if not item:
await notify.emit(
Events.LOG_ERROR,
title="Item Not Found",
message=f"Item with ID '{id}' not found.",
to=sid,
)
return
status: dict[str, str] = {}
@ -87,7 +119,7 @@ async def item_delete(queue: DownloadQueue, notify: EventBus, sid: str, data: di
status.update({"identifier": id})
await notify.emit(
Events.ITEM_DELETE, data=status, title="Item Deleted", message=f"Item with ID '{id}' has been deleted."
Events.ITEM_DELETE, data=status, title="Item Deleted", message=f"Item '{item.info.title}': has been deleted."
)
@ -145,7 +177,12 @@ async def archive_item(config: Config, data: dict):
@route(RouteType.SOCKET, "item_start", "item_start")
async def item_start(queue: DownloadQueue, notify: EventBus, sid: str, data: list | str) -> None:
if not data:
await notify.emit(Events.ERROR, data=error("Invalid request."), to=sid)
await notify.emit(
Events.LOG_ERROR,
title="Invalid Request",
message="No items provided to start.",
to=sid,
)
return
if isinstance(data, str):
@ -157,7 +194,12 @@ async def item_start(queue: DownloadQueue, notify: EventBus, sid: str, data: lis
@route(RouteType.SOCKET, "item_pause", "item_pause")
async def item_pause(queue: DownloadQueue, notify: EventBus, sid: str, data: list | str) -> None:
if not data:
await notify.emit(Events.ERROR, data=error("Invalid request."), to=sid)
await notify.emit(
Events.LOG_ERROR,
title="Invalid Request",
message="No items provided to pause.",
to=sid,
)
return
if isinstance(data, str):

39
app/routes/socket/terminal.py
View file

@ -3,7 +3,7 @@ import os
from typing import TYPE_CHECKING
from app.library.config import Config
from app.library.Events import EventBus, Events, error
from app.library.Events import EventBus, Events
from app.library.router import RouteType, route
if TYPE_CHECKING:
@ -17,11 +17,16 @@ 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)
notify.offload(
Events.LOG_ERROR,
title="Feature disabled",
message="Console feature is disabled.",
to=sid,
)
return
if not data:
await notify.emit(Events.CLI_CLOSE, data={"exitcode": 0}, to=sid)
notify.offload(Events.CLI_CLOSE, data={"exitcode": 0}, to=sid)
return
import asyncio
@ -29,6 +34,7 @@ async def cli_post(config: Config, notify: EventBus, sid: str, data: str):
import shlex
import subprocess # ignore
returncode: int = -1
try:
LOG.info(f"Cli command from client '{sid}'. '{data}'")
@ -36,16 +42,22 @@ async def cli_post(config: Config, notify: EventBus, sid: str, data: str):
_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",
}
)
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
@ -81,7 +93,7 @@ async def cli_post(config: Config, notify: EventBus, sid: str, data: str):
assert proc.stdout is not None
async for raw_line in proc.stdout:
line = raw_line.rstrip(b"\n")
await notify.emit(
notify.offload(
Events.CLI_OUTPUT,
data={"type": "stdout", "line": line.decode("utf-8", errors="replace")},
to=sid,
@ -100,7 +112,7 @@ async def cli_post(config: Config, notify: EventBus, sid: str, data: str):
if not chunk:
if buffer:
await notify.emit(
notify.offload(
Events.CLI_OUTPUT,
data={"type": "stdout", "line": buffer.decode("utf-8", errors="replace")},
to=sid,
@ -111,7 +123,7 @@ async def cli_post(config: Config, notify: EventBus, sid: str, data: str):
*lines, buffer = buffer.split(b"\n")
for line in lines:
await notify.emit(
notify.offload(
Events.CLI_OUTPUT,
data={"type": "stdout", "line": line.decode("utf-8", errors="replace")},
to=sid,
@ -126,10 +138,9 @@ async def cli_post(config: Config, notify: EventBus, sid: str, data: str):
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)
notify.offload(Events.CLI_OUTPUT, data={"type": "stderr", "line": str(e)}, to=sid)
finally:
notify.offload(Events.CLI_CLOSE, data={"exitcode": returncode}, to=sid)

4
ui/assets/css/style.css
View file

@ -336,3 +336,7 @@ hr {
.fa-spin-10 {
--fa-animation-iteration-count: 10;
}
.Vue-Toastification__toast-body {
user-select: none;
}

4
ui/pages/console.vue
View file

@ -173,9 +173,9 @@ const writer = (s: string) => {
return
}
const data = JSON.parse(s)
const json = JSON.parse(s)
terminal.value.writeln(data.line)
terminal.value.writeln(json.data.line)
}
const loader = () => isLoading.value = false

3
ui/pages/tasks.vue
View file

@ -686,7 +686,8 @@ const runNow = async (item: task_item, mass: boolean = false) => {
onBeforeUnmount(() => socket.off('status', statusHandler))
const statusHandler = async (stream: string) => {
const { status, msg } = JSON.parse(stream)
const json = JSON.parse(stream)
const { status, msg } = json.data
if ('error' === status) {
toast.error(msg)