diff --git a/api_service/api_server.py b/api_service/api_server.py index b3e2470..575fc9e 100644 --- a/api_service/api_server.py +++ b/api_service/api_server.py @@ -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 diff --git a/api_service/terminal_images_endpoint.py b/api_service/terminal_images_endpoint.py new file mode 100644 index 0000000..c580af7 --- /dev/null +++ b/api_service/terminal_images_endpoint.py @@ -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.") diff --git a/cogs/terminal_cog.py b/cogs/terminal_cog.py index 16469c6..30433e3 100644 --- a/cogs/terminal_cog.py +++ b/cogs/terminal_cog.py @@ -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