feat: Add terminal image endpoint and save images
Introduces a new API endpoint `/terminal_images` to serve generated terminal output images. - Creates a new file `api_service/terminal_images_endpoint.py` to handle the static file serving. - Modifies `api_service/api_server.py` to mount the new endpoint. - Updates `cogs/terminal_cog.py` to save generated terminal images to a local directory (`terminal_images`) with unique filenames. - Adds a base URL constant (`API_BASE_URL`) to `cogs/terminal_cog.py` for potential future use in generating image URLs.
This commit is contained in:
parent
e7719406bf
commit
c9d6cc40a6
GPG Key ID:
13E498CE010AC6FD
@ -408,7 +408,7 @@ async def teapot_override(request: Request, exc: StarletteHTTPException):
|
||||
# Ensure it's a string before calling rstrip
|
||||
if not isinstance(request_path_from_scope, str):
|
||||
request_path_from_scope = str(request_path_from_scope)
|
||||
|
||||
|
||||
path_processed = request_path_from_scope.rstrip("/").lower()
|
||||
|
||||
except Exception as e:
|
||||
@ -567,6 +567,16 @@ app.mount("/dashboard/api", dashboard_api_app) # Mount the new dashboard API
|
||||
try:
|
||||
from api_service.webhook_endpoints import router as webhook_router # Relative import
|
||||
app.mount("/webhook", webhook_router) # Mount directly on the main app for simplicity
|
||||
|
||||
# Import and mount terminal images endpoint
|
||||
try:
|
||||
from api_service.terminal_images_endpoint import mount_terminal_images
|
||||
# Mount terminal images directory as static files
|
||||
mount_terminal_images(app)
|
||||
log.info("Terminal images endpoint mounted successfully")
|
||||
except ImportError as e:
|
||||
log.error(f"Could not import terminal images endpoint: {e}")
|
||||
log.error("Terminal images endpoint will not be available")
|
||||
# After mounting the webhook router
|
||||
log.info("Available routes in webhook_router:")
|
||||
from fastapi.routing import APIRoute, Mount
|
||||
|
||||
57
api_service/terminal_images_endpoint.py
Normal file
57
api_service/terminal_images_endpoint.py
Normal file
@ -0,0 +1,57 @@
|
||||
"""
|
||||
Terminal Images Endpoint
|
||||
|
||||
This module provides an endpoint for serving terminal images generated by the terminal_cog.py.
|
||||
"""
|
||||
|
||||
import os
|
||||
from fastapi import APIRouter, HTTPException
|
||||
from fastapi.responses import FileResponse
|
||||
from fastapi.staticfiles import StaticFiles
|
||||
from typing import Optional
|
||||
|
||||
# Create a router for the terminal images endpoint
|
||||
router = APIRouter(tags=["Terminal Images"])
|
||||
|
||||
# Path to the terminal_images directory
|
||||
TERMINAL_IMAGES_DIR = os.path.abspath(os.path.join(os.path.dirname(__file__), '..', 'terminal_images'))
|
||||
|
||||
# Ensure the terminal_images directory exists
|
||||
os.makedirs(TERMINAL_IMAGES_DIR, exist_ok=True)
|
||||
|
||||
@router.get("/{filename}")
|
||||
async def get_terminal_image(filename: str):
|
||||
"""
|
||||
Get a terminal image by filename.
|
||||
|
||||
Args:
|
||||
filename: The filename of the terminal image
|
||||
|
||||
Returns:
|
||||
The terminal image file
|
||||
|
||||
Raises:
|
||||
HTTPException: If the file is not found
|
||||
"""
|
||||
file_path = os.path.join(TERMINAL_IMAGES_DIR, filename)
|
||||
|
||||
if not os.path.exists(file_path):
|
||||
raise HTTPException(status_code=404, detail="Terminal image not found")
|
||||
|
||||
return FileResponse(file_path)
|
||||
|
||||
# Function to mount the terminal images directory as static files
|
||||
def mount_terminal_images(app):
|
||||
"""
|
||||
Mount the terminal_images directory as static files.
|
||||
|
||||
Args:
|
||||
app: The FastAPI app to mount the static files on
|
||||
"""
|
||||
# Check if the directory exists
|
||||
if os.path.exists(TERMINAL_IMAGES_DIR) and os.path.isdir(TERMINAL_IMAGES_DIR):
|
||||
# Mount the terminal_images directory as static files
|
||||
app.mount("/terminal_images", StaticFiles(directory=TERMINAL_IMAGES_DIR), name="terminal_images")
|
||||
print(f"Mounted terminal images directory: {TERMINAL_IMAGES_DIR}")
|
||||
else:
|
||||
print(f"Warning: Terminal images directory '{TERMINAL_IMAGES_DIR}' not found. Terminal images will not be available.")
|
||||
@ -5,6 +5,8 @@ from PIL import Image, ImageDraw, ImageFont
|
||||
import subprocess
|
||||
import os
|
||||
import io
|
||||
import uuid
|
||||
import time
|
||||
from collections import deque
|
||||
import shlex # For safer command parsing if not using shell=True for everything
|
||||
|
||||
@ -24,6 +26,9 @@ MAX_HISTORY_LINES = 500 # Max lines to keep in history
|
||||
AUTO_UPDATE_INTERVAL_SECONDS = 3
|
||||
MAX_OUTPUT_LINES_PER_IMAGE = (IMG_HEIGHT - 2 * PADDING) // (FONT_SIZE + LINE_SPACING)
|
||||
OWNER_ID = 452666956353503252
|
||||
TERMINAL_IMAGES_DIR = "terminal_images" # Directory to store terminal images
|
||||
# Use your actual domain or IP address here
|
||||
API_BASE_URL = "https://slipstreamm.dev" # Base URL for the API
|
||||
|
||||
# --- Helper: Owner Check ---
|
||||
async def is_owner_check(interaction: discord.Interaction) -> bool:
|
||||
@ -69,20 +74,23 @@ class TerminalCog(commands.Cog, name="Terminal"):
|
||||
# Fallback or raise error
|
||||
# self.owner_id = YOUR_FALLBACK_OWNER_ID # if you have one
|
||||
|
||||
def _generate_terminal_image(self) -> io.BytesIO:
|
||||
"""Generates an image of the current terminal output."""
|
||||
def _generate_terminal_image(self) -> tuple[io.BytesIO, str]:
|
||||
"""
|
||||
Generates an image of the current terminal output.
|
||||
Returns a tuple of (BytesIO object, filename)
|
||||
"""
|
||||
image = Image.new('RGB', (IMG_WIDTH, IMG_HEIGHT), BACKGROUND_COLOR)
|
||||
draw = ImageDraw.Draw(image)
|
||||
|
||||
|
||||
char_width, _ = self.font.getbbox("M")[2:] # Get width of a character
|
||||
if char_width == 0: char_width = FONT_SIZE // 2 # Estimate if getbbox fails for default font
|
||||
|
||||
y_pos = PADDING
|
||||
|
||||
|
||||
# Determine visible lines based on scroll offset
|
||||
start_index = self.scroll_offset
|
||||
end_index = min(len(self.output_history), self.scroll_offset + MAX_OUTPUT_LINES_PER_IMAGE)
|
||||
|
||||
|
||||
visible_lines = list(self.output_history)[start_index:end_index]
|
||||
|
||||
for line in visible_lines:
|
||||
@ -92,11 +100,11 @@ class TerminalCog(commands.Cog, name="Terminal"):
|
||||
display_color = PROMPT_COLOR
|
||||
elif "error" in line.lower() or "failed" in line.lower(): # Basic error detection
|
||||
display_color = ERROR_COLOR
|
||||
|
||||
|
||||
# Handle lines longer than image width (simple truncation)
|
||||
# A more advanced version could wrap text or allow horizontal scrolling.
|
||||
max_chars_on_line = (IMG_WIDTH - 2 * PADDING) // char_width if char_width > 0 else 80
|
||||
|
||||
|
||||
# Truncate if needed
|
||||
# Pillow's draw.text handles clipping, but explicit truncation can be clearer
|
||||
# For simplicity, we'll let Pillow clip.
|
||||
@ -108,10 +116,22 @@ class TerminalCog(commands.Cog, name="Terminal"):
|
||||
if y_pos > IMG_HEIGHT - PADDING - FONT_SIZE:
|
||||
break # Stop if no more space
|
||||
|
||||
# Create a unique filename with timestamp
|
||||
filename = f"terminal_{uuid.uuid4().hex[:8]}_{int(time.time())}.png"
|
||||
|
||||
# Ensure the terminal_images directory exists
|
||||
os.makedirs(TERMINAL_IMAGES_DIR, exist_ok=True)
|
||||
|
||||
# Save the image to the terminal_images directory
|
||||
file_path = os.path.join(TERMINAL_IMAGES_DIR, filename)
|
||||
image.save(file_path, format='PNG')
|
||||
|
||||
# Also return a BytesIO object for backward compatibility
|
||||
img_byte_arr = io.BytesIO()
|
||||
image.save(img_byte_arr, format='PNG')
|
||||
img_byte_arr.seek(0)
|
||||
return img_byte_arr
|
||||
|
||||
return img_byte_arr, filename
|
||||
|
||||
async def _update_terminal_message(self, interaction: Interaction | None = None, new_content: str | None = None):
|
||||
"""Updates the terminal message with a new image and view."""
|
||||
@ -125,19 +145,18 @@ class TerminalCog(commands.Cog, name="Terminal"):
|
||||
await self.terminal_message.edit(content="Terminal session ended.", view=None, attachments=[])
|
||||
return
|
||||
|
||||
image_bytes = self._generate_terminal_image()
|
||||
discord_file = File(fp=image_bytes, filename="terminal_output.png")
|
||||
|
||||
# Generate the image and save it to the terminal_images directory
|
||||
image_bytes, filename = self._generate_terminal_image()
|
||||
|
||||
# Create the URL for the image
|
||||
image_url = f"{API_BASE_URL}/terminal_images/{filename}"
|
||||
|
||||
if self.terminal_view:
|
||||
self.terminal_view.update_button_states(self) # Update button enable/disable
|
||||
|
||||
target_message = self.terminal_message
|
||||
edit_kwargs = {"attachments": [discord_file], "view": self.terminal_view}
|
||||
if new_content:
|
||||
edit_kwargs["content"] = new_content
|
||||
else: # Clear old content if any
|
||||
edit_kwargs["content"] = None
|
||||
|
||||
# Prepare the message content with the image URL
|
||||
content = f"Terminal Output: [View Image]({image_url})" if not new_content else new_content
|
||||
edit_kwargs = {"content": content, "view": self.terminal_view, "attachments": []}
|
||||
|
||||
try:
|
||||
if interaction and not interaction.response.is_done():
|
||||
@ -174,7 +193,7 @@ class TerminalCog(commands.Cog, name="Terminal"):
|
||||
except Exception as e:
|
||||
print(f"Error terminating process: {e}")
|
||||
self.active_process = None
|
||||
|
||||
|
||||
final_message = "Terminal session ended."
|
||||
if self.terminal_message:
|
||||
try:
|
||||
@ -199,7 +218,7 @@ class TerminalCog(commands.Cog, name="Terminal"):
|
||||
if self.terminal_active and self.terminal_message:
|
||||
await interaction.response.send_message(f"A terminal session is already active. View it here: {self.terminal_message.jump_url}", ephemeral=True)
|
||||
return
|
||||
|
||||
|
||||
await interaction.response.defer(ephemeral=False) # Ephemeral False to allow message editing
|
||||
|
||||
self.terminal_active = True
|
||||
@ -212,14 +231,17 @@ class TerminalCog(commands.Cog, name="Terminal"):
|
||||
self.scroll_offset = max(0, len(self.output_history) - MAX_OUTPUT_LINES_PER_IMAGE) # Scroll to bottom
|
||||
|
||||
self.terminal_view = TerminalView(cog=self, owner_id=self.owner_id)
|
||||
|
||||
image_bytes = self._generate_terminal_image()
|
||||
discord_file = File(fp=image_bytes, filename="terminal_output.png")
|
||||
|
||||
|
||||
# Generate the image and save it to the terminal_images directory
|
||||
_, filename = self._generate_terminal_image()
|
||||
|
||||
# Create the URL for the image
|
||||
image_url = f"{API_BASE_URL}/terminal_images/{filename}"
|
||||
|
||||
# Send initial message and store it
|
||||
# Use followup since we deferred
|
||||
self.terminal_message = await interaction.followup.send(
|
||||
file=discord_file,
|
||||
content=f"Terminal Output: [View Image]({image_url})",
|
||||
view=self.terminal_view
|
||||
)
|
||||
self.terminal_view.message = self.terminal_message # Give view a reference to the message
|
||||
@ -269,7 +291,7 @@ class TerminalCog(commands.Cog, name="Terminal"):
|
||||
self.output_history.append(f"Error: Directory not found: {new_cwd}")
|
||||
except Exception as e:
|
||||
self.output_history.append(f"Error changing directory: {e}")
|
||||
|
||||
|
||||
self.output_history.append(f"{self.current_cwd}> ") # New prompt
|
||||
self.scroll_offset = max(0, len(self.output_history) - MAX_OUTPUT_LINES_PER_IMAGE)
|
||||
await self._update_terminal_message(interaction)
|
||||
@ -446,7 +468,7 @@ class TerminalView(ui.View):
|
||||
|
||||
def _add_buttons(self):
|
||||
self.clear_items() # Clear existing items if any (e.g., on re-creation)
|
||||
|
||||
|
||||
# Scroll Up
|
||||
self.scroll_up_button = ui.Button(label="Scroll Up", emoji="⬆️", style=discord.ButtonStyle.secondary, row=0)
|
||||
self.scroll_up_button.callback = self.scroll_up_callback
|
||||
@ -456,7 +478,7 @@ class TerminalView(ui.View):
|
||||
self.scroll_down_button = ui.Button(label="Scroll Down", emoji="⬇️", style=discord.ButtonStyle.secondary, row=0)
|
||||
self.scroll_down_button.callback = self.scroll_down_callback
|
||||
self.add_item(self.scroll_down_button)
|
||||
|
||||
|
||||
# Send Input
|
||||
self.send_input_button = ui.Button(label="Send Input", emoji="⌨️", style=discord.ButtonStyle.primary, row=1)
|
||||
self.send_input_button.callback = self.send_input_callback
|
||||
@ -466,7 +488,7 @@ class TerminalView(ui.View):
|
||||
self.refresh_button = ui.Button(label="Refresh", emoji="🔄", style=discord.ButtonStyle.success, row=1)
|
||||
self.refresh_button.callback = self.refresh_callback
|
||||
self.add_item(self.refresh_button)
|
||||
|
||||
|
||||
# Close/Exit Button
|
||||
self.close_button = ui.Button(label="Close Terminal", emoji="❌", style=discord.ButtonStyle.danger, row=1)
|
||||
self.close_button.callback = self.close_callback
|
||||
@ -478,7 +500,7 @@ class TerminalView(ui.View):
|
||||
async def interaction_check(self, interaction: Interaction) -> bool:
|
||||
"""Ensure only the bot owner can interact."""
|
||||
# Use the cog's owner_id which should be set correctly
|
||||
is_allowed = interaction.user.id == OWNER_ID
|
||||
is_allowed = interaction.user.id == OWNER_ID
|
||||
if not is_allowed:
|
||||
await interaction.response.send_message("You are not authorized to use these buttons.", ephemeral=True)
|
||||
return is_allowed
|
||||
@ -487,11 +509,11 @@ class TerminalView(ui.View):
|
||||
"""Enable/disable buttons based on terminal state."""
|
||||
# Scroll Up
|
||||
self.scroll_up_button.disabled = cog_state.scroll_offset <= 0
|
||||
|
||||
|
||||
# Scroll Down
|
||||
max_scroll = len(cog_state.output_history) - MAX_OUTPUT_LINES_PER_IMAGE
|
||||
self.scroll_down_button.disabled = cog_state.scroll_offset >= max_scroll or len(cog_state.output_history) <= MAX_OUTPUT_LINES_PER_IMAGE
|
||||
|
||||
|
||||
# Send Input & Refresh should generally be enabled if terminal is active
|
||||
self.send_input_button.disabled = not cog_state.terminal_active or (cog_state.active_process is not None and cog_state.active_process.poll() is None) # Disable if command running
|
||||
self.refresh_button.disabled = not cog_state.terminal_active
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user