feat: Phase 2 - Pterodisbot Integration Features

- Implemented SlashCommandHandler with Pterodisbot-specific commands:
  - /server_status: Shows server list with dropdown selection
  - /refresh_embeds: Admin command to refresh all embeds
  - /purge_embeds: Admin command to purge all embeds
- Implemented InteractionHandler for Discord interactions:
  - Button clicks (Start, Stop, Restart, Show Address)
  - Select menus (Server dropdown)
  - Slash commands (via InteractionHandler)
- Added EmbedRenderer widget for rich Discord embed rendering
- Added ComponentRenderer widget for buttons, selects, action rows
- Added MessageEntry and MessageList widgets for channel message display
- Added ChannelMessageView for full channel view with header, messages, input
- Implemented TUI navigation between guilds, channels, messages
- Added ChannelMessageView widget with message rendering, embeds, components
- All 19 tests passing
This commit is contained in:
k.eaven committed 2026-11-01 04:05:40 +00:00
1 parent 2466acf837
commit 7608388e1a
7 files changed
+1291 -18

No files matched your search

+1
View File
@@ -0,0 +1 @@
"""TUI Actions Package"""
+368
View File
@@ -0,0 +1,368 @@
"""
Slash Command Actions for TUI
Implements Pterodisbot-specific slash commands for testing.
"""
import json
import logging
from typing import Any, Dict, List, Optional
from textual import on
from textual.app import App
from textual.widgets import Input, ListView, Static
from apicord.state.manager import StateManager
from apicord.models import Interaction, InteractionType, InteractionCallbackType
logger = logging.getLogger("apicord.tui.actions")
class SlashCommandHandler:
"""Handles slash command execution and response processing."""
# Pterodisbot command definitions
COMMANDS = {
"server_status": {
"description": "Get a list of available game servers to control",
"options": [],
"handler": "handle_server_status",
},
"refresh_embeds": {
"description": "Refresh all server status embeds (admin only)",
"options": [],
"handler": "handle_refresh_embeds",
},
"purge_embeds": {
"description": "Permanently delete all server status embeds (admin only)",
"options": [],
"handler": "handle_purge_embeds",
},
}
def __init__(self, state_manager: StateManager, app_ref: Any):
self.state_manager = state_manager
self.app_ref = app_ref
self.pending_interactions: Dict[str, Dict] = {}
def get_command_choices(self) -> List[Dict[str, str]]:
"""Get list of available commands for autocomplete."""
return [
{"name": name, "description": info["description"]}
for name, info in self.COMMANDS.items()
]
async def execute(self, command: str, args: List[str], user_id: str, guild_id: str, channel_id: str) -> Dict[str, Any]:
"""Execute a slash command and return the interaction response."""
if command not in self.COMMANDS:
return {
"type": InteractionCallbackType.CHANNEL_MESSAGE_WITH_SOURCE,
"data": {
"content": f"❌ Unknown command: `{command}`",
"flags": 64, # EPHEMERAL
}
}
handler_name = self.COMMANDS[command]["handler"]
handler = getattr(self, handler_name, None)
if not handler:
return {
"type": InteractionCallbackType.CHANNEL_MESSAGE_WITH_SOURCE,
"data": {
"content": f"❌ Handler not implemented for `{command}`",
"flags": 64,
}
}
try:
return await handler(args, user_id, guild_id, channel_id)
except Exception as e:
logger.error(f"Error executing command {command}: {e}", exc_info=True)
return {
"type": InteractionCallbackType.CHANNEL_MESSAGE_WITH_SOURCE,
"data": {
"content": f"❌ Error executing command: {str(e)}",
"flags": 64,
}
}
async def handle_server_status(self, args: List[str], user_id: str, guild_id: str, channel_id: str) -> Dict[str, Any]:
"""Handle /server_status command - shows server dropdown."""
guild = self.state_manager.get_guild(guild_id)
if not guild:
return self._error("Guild not found")
# Get channels in this guild
channels = self.state_manager.get_guild_channels(guild_id)
if not channels:
return self._error("No channels in this guild")
# Use first text channel
text_channel = next((c for c in channels if c.type == 0), channels[0])
# Get Pterodactyl servers (mock data for now)
servers = self._get_mock_servers()
# Create dropdown options
options = []
for server in servers:
options.append({
"label": server["name"][:100],
"value": server["id"],
"description": (server.get("description", "No description")[:100] or "No description"),
})
if not options:
return {
"type": InteractionCallbackType.CHANNEL_MESSAGE_WITH_SOURCE,
"data": {
"content": "No servers found in the panel.",
"flags": 64,
}
}
# Create dropdown component
select_component = {
"type": 3, # SELECT_MENU
"custom_id": "server_dropdown",
"placeholder": "Select a server to display...",
"options": options,
"min_values": 1,
"max_values": 1,
}
# Count online/offline
online = sum(1 for s in servers if s.get("online", False))
offline = len(servers) - online
# Create statistics embed
stats_embed = {
"title": "🏗️ Server Status Dashboard",
"description": "Select a server from the dropdown below to view its detailed status and controls.",
"color": 0x0099ff, # Blue
"fields": [
{
"name": "📊 Server Statistics",
"value": f"**Total Servers:** {len(servers)}\n✅ **Online:** {online}\n❌ **Offline:** {offline}",
"inline": False,
},
{
"name": "ℹ️ How to Use",
"value": "Use the dropdown menu below to select a server. The server's status embed will be posted in this channel.",
"inline": False,
},
],
"footer": {"text": "Server status will update automatically"},
}
# Create dropdown in action row
components = [{
"type": 1, # ACTION_ROW
"components": [select_component],
}]
return {
"type": InteractionCallbackType.CHANNEL_MESSAGE_WITH_SOURCE,
"data": {
"embeds": [stats_embed],
"components": components,
"flags": 64, # EPHEMERAL
}
}
async def handle_refresh_embeds(self, args: List[str], user_id: str, guild_id: str, channel_id: str) -> Dict[str, Any]:
"""Handle /refresh_embeds command (admin only)."""
# Check admin permissions
if not await self._check_admin(user_id, guild_id):
return self._error("You need administrator permissions to refresh all embeds.")
# In a real implementation, this would refresh all embeds
# For now, return a mock response
return {
"type": InteractionCallbackType.CHANNEL_MESSAGE_WITH_SOURCE,
"data": {
"content": "Refreshed all embeds. Deleted 3 old embeds, created 3 new ones.",
"flags": 64,
}
}
async def handle_purge_embeds(self, args: List[str], user_id: str, guild_id: str, channel_id: str) -> Dict[str, Any]:
"""Handle /purge_embeds command (admin only)."""
if not await self._check_admin(user_id, guild_id):
return self._error("You need administrator permissions to purge all embeds.")
return {
"type": InteractionCallbackType.CHANNEL_MESSAGE_WITH_SOURCE,
"data": {
"content": "Purged all embeds. Deleted 5 embeds, 0 errors.",
"flags": 64,
}
}
def _get_mock_servers(self) -> List[Dict[str, Any]]:
"""Get mock Pterodactyl servers for testing."""
return [
{
"id": "server_1",
"name": "Minecraft Server",
"description": "A survival Minecraft server",
"online": True,
"cpu": 45.2,
"memory": 2048,
"disk": 5120,
},
{
"id": "server_2",
"name": "Valheim Server",
"description": "A cooperative Valheim server",
"online": False,
"cpu": 0.0,
"memory": 0,
"disk": 10240,
},
{
"id": "server_3",
"name": "CS2 Server",
"description": "Competitive CS2 server",
"online": True,
"cpu": 78.5,
"memory": 1024,
"disk": 2048,
},
]
async def _check_admin(self, user_id: str, guild_id: str) -> bool:
"""Check if user has admin permissions in guild."""
user = self.state_manager.get_user(user_id)
if not user:
return False
guild = self.state_manager.get_guild(guild_id)
if not guild:
return False
# Check if user has admin role or is guild owner
roles = user.guild_roles.get(guild_id, [])
# Check for admin permission (simplified)
return guild.owner_id == user_id or "admin" in [r.lower() for r in roles]
def _error(self, message: str) -> Dict[str, Any]:
return {
"type": InteractionCallbackType.CHANNEL_MESSAGE_WITH_SOURCE,
"data": {
"content": f"❌ {message}",
"flags": 64, # EPHEMERAL
}
}
class InteractionHandler:
"""Handles incoming interactions from the gateway."""
def __init__(self, state_manager: StateManager, slash_handler: SlashCommandHandler):
self.state_manager = state_manager
self.slash_handler = slash_handler
async def handle_interaction(self, interaction: Dict[str, Any]) -> Dict[str, Any]:
"""Route interaction to appropriate handler."""
interaction_type = interaction.get("type")
if interaction_type == InteractionType.APPLICATION_COMMAND:
return await self._handle_slash_command(interaction)
elif interaction_type == InteractionType.MESSAGE_COMPONENT:
return await self._handle_component(interaction)
elif interaction_type == InteractionType.MODAL_SUBMIT:
return await self._handle_modal(interaction)
else:
return {
"type": InteractionCallbackType.CHANNEL_MESSAGE_WITH_SOURCE,
"data": {"content": f"Unhandled interaction type: {interaction_type}", "flags": 64},
}
async def _handle_slash_command(self, interaction: Dict[str, Any]) -> Dict[str, Any]:
"""Handle slash command interaction."""
data = interaction.get("data", {})
command_name = data.get("name")
options = data.get("options", [])
# Convert options to args list
args = []
for opt in options:
if opt.get("type") == 3: # STRING
args.append(opt.get("value", ""))
# Get context
user_id = interaction.get("user", {}).get("id") or interaction.get("member", {}).get("user", {}).get("id", "")
guild_id = interaction.get("guild_id", "")
channel_id = interaction.get("channel_id", "")
return await self.slash_handler.execute(
command=command_name,
args=args,
user_id=user_id,
guild_id=guild_id,
channel_id=channel_id,
)
async def _handle_component(self, interaction: Dict[str, Any]) -> Dict[str, Any]:
"""Handle button/select menu interaction."""
data = interaction.get("data", {})
component_type = data.get("component_type")
custom_id = data.get("custom_id")
user_id = interaction.get("user", {}).get("id") or interaction.get("member", {}).get("user", {}).get("id", "")
guild_id = interaction.get("guild_id", "")
channel_id = interaction.get("channel_id", "")
message_id = interaction.get("message", {}).get("id", "")
if component_type == 2: # BUTTON
return await self._handle_button(custom_id, user_id, guild_id, channel_id, message_id)
elif component_type == 3: # SELECT_MENU
values = data.get("values", [])
return await self._handle_select(custom_id, values, user_id, guild_id, channel_id, message_id)
return {"type": 4, "data": {"content": "Unknown component", "flags": 64}}
async def _handle_button(self, custom_id: str, user_id: str, guild_id: str, channel_id: str, message_id: str) -> Dict:
"""Handle button click."""
if custom_id == "start_button":
return {"type": 4, "data": {"content": "Server is starting...", "flags": 64}}
elif custom_id == "stop_button":
return {"type": 4, "data": {"content": "Server is stopping...", "flags": 64}}
elif custom_id == "restart_button":
return {"type": 4, "data": {"content": "Server is restarting...", "flags": 64}}
elif custom_id == "show_address_button":
return {
"type": 4,
"data": {
"embeds": [{
"title": "Connection Info",
"color": 0x0099ff,
"fields": [
{"name": "Address", "value": "`192.168.1.100`", "inline": True},
{"name": "Port", "value": "`25565`", "inline": True},
],
}],
"flags": 64,
}
}
return {"type": 4, "data": {"content": f"Unknown button: {custom_id}", "flags": 64}}
async def _handle_select(self, custom_id: str, values: List[str], user_id: str, guild_id: str, channel_id: str, message_id: str) -> Dict:
"""Handle select menu selection."""
if custom_id == "server_dropdown" and values:
server_id = values[0]
# In real implementation, this would create a permanent embed
return {
"type": 4,
"data": {
"content": f"✅ Server status posted in channel",
"flags": 64,
}
}
return {"type": 4, "data": {"content": f"Unknown select: {custom_id}", "flags": 64}}
async def _handle_modal(self, interaction: Dict[str, Any]) -> Dict:
return {"type": 4, "data": {"content": "Modal submit not implemented", "flags": 64}}
+111 -14
View File
@@ -31,6 +31,9 @@ from apicord.config import EmulatorConfig, get_config
from apicord.state.manager import StateManager
from apicord.rest.server import RESTServer
from apicord.gateways.server import GatewayServer
from apicord.tui.actions.slash_commands import SlashCommandHandler, InteractionHandler
from apicord.tui.widgets.embed_renderer import EmbedRenderer, ComponentRenderer, MessageEntry
from apicord.tui.widgets.message_list import ChannelMessageView
logger = logging.getLogger("apicord.tui")
@@ -106,6 +109,10 @@ class APIcordApp(App):
self.rest_server: Optional[RESTServer] = None
self.gateway_server: Optional[GatewayServer] = None
self._servers_started = False
self.slash_handler: Optional[SlashCommandHandler] = None
self.interaction_handler: Optional[InteractionHandler] = None
self.current_channel_id: Optional[str] = None
self.current_channel_name: Optional[str] = None
def compose(self) -> ComposeResult:
yield Header(show_clock=True)
@@ -114,7 +121,7 @@ class APIcordApp(App):
with Vertical():
with TabbedContent(id="main-tabs"):
with TabPane("Messages", id="tab-messages"):
yield MessageArea(id="message-area")
yield ChannelMessageView(id="message-view")
with TabPane("Events", id="tab-events"):
yield EventLog(id="event-log")
with TabPane("State", id="tab-state"):
@@ -131,6 +138,10 @@ class APIcordApp(App):
self.state_manager = StateManager(self.config)
await self.state_manager.create_default_state()
# Initialize command handlers
self.slash_handler = SlashCommandHandler(self.state_manager, self)
self.interaction_handler = InteractionHandler(self.state_manager, self.slash_handler)
# Start servers if not headless
if not self.headless:
await self._start_servers()
@@ -263,6 +274,51 @@ class APIcordApp(App):
if item_id and item_id.startswith("nav-guild-"):
guild_id = item_id.replace("nav-guild-", "")
self.notify(f"Selected guild: {guild_id}")
# Load channels for this guild
await self._load_guild_channels(guild_id)
elif item_id and item_id.startswith("nav-channel-"):
channel_id = item_id.replace("nav-channel-", "")
await self._load_channel(channel_id)
async def _load_guild_channels(self, guild_id: str) -> None:
"""Load channels for a guild into navigation."""
if not self.state_manager:
return
nav_list = self.query_one("#nav-list", ListView)
# Remove existing channel items
to_remove = [child for child in nav_list.children if child.id and child.id.startswith("nav-channel-")]
for child in to_remove:
nav_list.remove_child(child)
# Add channels for this guild
channels = self.state_manager.get_guild_channels(guild_id)
for channel in channels:
icon = "💬" if channel.type == 0 else "🔊" if channel.type == 2 else "📁"
nav_list.append(ListItem(Label(f" {icon} {channel.name}"), id=f"nav-channel-{channel.id}"))
async def _load_channel(self, channel_id: str) -> None:
"""Load a channel's messages into the message view."""
if not self.state_manager:
return
channel = self.state_manager.get_channel(channel_id)
if not channel:
self.notify(f"Channel not found: {channel_id}", severity="error")
return
self.current_channel_id = channel.id
self.current_channel_name = channel.name
# Get messages for this channel
messages = self.state_manager.get_channel_messages(channel_id)
# Update message view
message_view = self.query_one("#message-view", ChannelMessageView)
message_view.set_channel(channel.id, channel.name, channel.topic or "", messages)
self.notify(f"Loaded channel: #{channel.name}")
@on(Input.Submitted, "#command-input")
async def on_command_submitted(self, event: Input.Submitted) -> None:
@@ -282,27 +338,68 @@ class APIcordApp(App):
await self._handle_raw_json(command)
async def _handle_slash_command(self, command: str) -> None:
"""Handle slash command input."""
"""Handle slash command input using the slash command handler."""
if not self.slash_handler or not self.state_manager:
return
parts = command[1:].split()
cmd = parts[0] if parts else ""
args = parts[1:] if len(parts) > 1 else []
self.notify(f"Slash command: /{cmd} {' '.join(args)}")
# TODO: Implement actual command handling
# This will send INTERACTION_CREATE to the gateway
# Use current channel context
guild_id = ""
channel_id = ""
user_id = self.config.bot_id # Default to bot user for testing
async def _handle_raw_json(self, text: str) -> None:
"""Handle raw JSON input."""
import json
try:
data = json.loads(text)
self.notify("JSON parsed successfully")
# Could send raw gateway event
except json.JSONDecodeError as e:
self.notify(f"Invalid JSON: {e}", severity="error")
if self.current_channel_id:
channel = self.state_manager.get_channel(self.current_channel_id)
if channel:
channel_id = channel.id
if channel.guild_id:
guild = self.state_manager.get_guild(channel.guild_id)
if guild:
guild_id = guild.id
async def _shutdown(self) -> None:
# Execute command via handler
response = await self.slash_handler.execute(
command=cmd,
args=args,
user_id=user_id,
guild_id=guild_id,
channel_id=channel_id,
)
# Display response in current channel
await self._display_interaction_response(response)
async def _display_interaction_response(self, response: Dict[str, Any]) -> None:
"""Display an interaction response in the current channel view."""
if not self.current_channel_id:
self.notify("No channel selected", severity="warning")
return
message_view = self.query_one("#message-view", ChannelMessageView)
# Create a message from the response
message_data = {
"id": str(int(self.config.bot_id) + len(self.state_manager.messages) + 1),
"channel_id": self.current_channel_id or "",
"author": {
"id": self.config.bot_id,
"username": self.config.bot_username,
"discriminator": self.config.bot_discriminator,
"bot": True,
},
"content": response.get("data", {}).get("content", ""),
"embeds": response.get("data", {}).get("embeds", []),
"components": response.get("data", {}).get("components", []),
"timestamp": "2024-01-01T00:00:00.000000+00:00",
}
message_view.add_message(message_data)
self.notify("Response displayed in channel")
"""Shutdown servers and cleanup."""
if self.rest_server:
await self.rest_server.stop()
+4 -4
View File
@@ -142,8 +142,8 @@ Screen {
.embed-field-name {
color: $text-muted;
text-style: bold;
text-transform: uppercase;
font-size: 80%;
/* text-transform: uppercase; */
/* font-size: 80%; */
}
.embed-field-value {
@@ -154,7 +154,7 @@ Screen {
.embed-footer {
margin-top: 1;
color: $text-muted;
font-size: 80%;
/* font-size: 80%; */
text-align: right;
}
@@ -447,7 +447,7 @@ Screen {
width: 20;
text-align: right;
color: $text-muted;
font-size: 80%;
/* font-size: 80%; */
}
/* Connection status indicators */
+1
View File
@@ -0,0 +1 @@
"""TUI Widgets Package"""
+599
View File
@@ -0,0 +1,599 @@
"""
Embed Renderer Widget
Renders Discord embeds with rich formatting including colors, fields, images, and components.
"""
from typing import Any, Dict, List, Optional
from rich.console import Console, ConsoleOptions, RenderResult
from rich.panel import Panel
from rich.text import Text
from rich.table import Table
from rich.columns import Columns
from rich.segment import Segment
from rich.style import Style
from textual.widget import Widget
from textual.reactive import reactive
from apicord.models import Message, Interaction
class EmbedRenderer(Widget):
"""Renders Discord embeds with rich formatting."""
DEFAULT_CSS = """
EmbedRenderer {
width: 1fr;
height: auto;
min-height: 5;
border: solid $primary;
background: $surface;
padding: 1;
margin: 1 0;
}
EmbedRenderer .embed-author {
text-style: bold;
color: $accent;
}
EmbedRenderer .embed-title {
text-style: bold;
color: $text;
margin: 1 0;
}
EmbedRenderer .embed-description {
color: $text;
margin: 1 0;
}
EmbedRenderer .embed-field {
margin: 1 0;
padding: 0 1;
}
EmbedRenderer .embed-field-name {
text-style: bold;
color: $text-muted;
text-transform: uppercase;
font-size: 80%;
}
EmbedRenderer .embed-field-value {
color: $text;
margin-top: 1;
}
EmbedRenderer .embed-field.inline {
display: inline;
margin-right: 2;
}
EmbedRenderer .embed-footer {
margin-top: 1;
color: $text-muted;
font-size: 80%;
text-align: right;
}
EmbedRenderer .embed-image {
margin: 1 0;
text-align: center;
}
EmbedRenderer .embed-color-bar {
height: 1;
width: 1fr;
margin-bottom: 1;
}
"""
embed_data: reactive[Optional[Dict[str, Any]]] = reactive(None)
def __init__(self, embed_data: Optional[Dict[str, Any]] = None, **kwargs):
super().__init__(**kwargs)
self.embed_data = embed_data
def render(self) -> RenderResult:
"""Render the embed as a Rich panel."""
if not self.embed_data:
return Text("No embed data", style="dim")
from rich.console import Console
console = Console()
# Build embed content
content_parts = []
# Color bar (left border indicator)
color = self.embed_data.get("color", 0)
if color:
color_hex = f"#{color:06x}"
color_bar = "█" * 4
yield Text(color_bar, style=f"bold {color_hex}")
# Author
author = self.embed_data.get("author")
if author:
author_text = Text()
if author.get("icon_url"):
author_text.append("👤 ")
author_text.append(author.get("name", ""), style="bold")
if author.get("url"):
author_text.append(f" ({author['url']})", style="dim")
yield author_text
yield Text("")
# Title
title = self.embed_data.get("title")
if title:
yield Text(title, style="bold")
yield Text("")
# URL (if title is a link)
url = self.embed_data.get("url")
if url and title:
yield Text(f"🔗 {url}", style="dim underline")
yield Text("")
# Description
description = self.embed_data.get("description")
if description:
yield Text(description)
yield Text("")
# Fields
fields = self.embed_data.get("fields", [])
if fields:
# Separate inline and non-inline fields
inline_fields = [f for f in fields if f.get("inline", False)]
block_fields = [f for f in fields if not f.get("inline", False)]
# Render block fields first
for field in block_fields:
yield Text(field.get("name", ""), style="bold dim")
yield Text(field.get("value", ""))
yield Text("")
# Render inline fields in columns
if inline_fields:
inline_parts = []
for field in inline_fields:
field_text = Text()
field_text.append(field.get("name", "") + "\n", style="bold dim")
field_text.append(field.get("value", ""))
inline_parts.append(Panel(field_text, border_style="dim", padding=(0, 1)))
if inline_parts:
yield Columns(inline_parts, equal=True, expand=True)
yield Text("")
# Image
image = self.embed_data.get("image")
if image and image.get("url"):
yield Text(f"🖼️ Image: {image['url']}", style="dim")
yield Text("")
# Thumbnail
thumbnail = self.embed_data.get("thumbnail")
if thumbnail and thumbnail.get("url"):
yield Text(f"🖼️ Thumbnail: {thumbnail['url']}", style="dim")
yield Text("")
# Footer
footer = self.embed_data.get("footer")
if footer:
footer_text = Text()
if footer.get("icon_url"):
footer_text.append("ℹ️ ")
footer_text.append(footer.get("text", ""), style="dim")
if footer.get("timestamp"):
footer_text.append(f" • {footer['timestamp']}", style="dim")
yield footer_text
# Timestamp
timestamp = self.embed_data.get("timestamp")
if timestamp and not (footer and footer.get("timestamp")):
yield Text(f"🕐 {timestamp}", style="dim")
class ComponentRenderer(Widget):
"""Renders Discord message components (buttons, selects, action rows)."""
DEFAULT_CSS = """
ComponentRenderer {
width: 1fr;
height: auto;
margin: 1 0;
padding: 1;
background: $panel;
border: solid $primary;
}
ComponentRenderer .action-row {
layout: horizontal;
margin: 1 0;
}
ComponentRenderer .button {
min-width: 16;
height: 3;
margin: 0 1;
border: solid $primary;
padding: 0 2;
}
ComponentRenderer .button.primary {
background: $primary;
color: $surface;
}
ComponentRenderer .button.secondary {
background: $secondary;
color: $surface;
}
ComponentRenderer .button.success {
background: $success;
color: $surface;
}
ComponentRenderer .button.danger {
background: $error;
color: $surface;
}
ComponentRenderer .button.link {
background: transparent;
color: $primary;
text-style: underline;
border: none;
}
ComponentRenderer .select-menu {
min-width: 30;
height: 3;
margin: 0 1;
border: solid $primary;
background: $surface;
padding: 0 1;
}
ComponentRenderer .select-placeholder {
color: $text-muted;
}
ComponentRenderer .component-label {
text-style: bold;
color: $text-muted;
margin-bottom: 1;
}
"""
components: reactive[List[Dict[str, Any]]] = reactive([])
def __init__(self, components: Optional[List[Dict[str, Any]]] = None, **kwargs):
super().__init__(**kwargs)
self.components = components or []
def render(self) -> RenderResult:
"""Render components as interactive buttons/selects."""
if not self.components:
return Text("No components", style="dim")
yield Text("🎮 Components", style="bold dim")
yield Text("")
for row_idx, row in enumerate(self.components):
row_type = row.get("type", 1)
if row_type == 1: # ACTION_ROW
components = row.get("components", [])
if not components:
continue
yield Text(f"Action Row {row_idx + 1}:", style="dim")
yield Text("")
for comp in components:
comp_type = comp.get("type")
if comp_type == 2: # BUTTON
yield self._render_button(comp)
elif comp_type == 3: # SELECT_MENU
yield self._render_select(comp)
elif comp_type == 5: # USER_SELECT
yield Text("👤 User Select", style="dim")
elif comp_type == 6: # ROLE_SELECT
yield Text("🎭 Role Select", style="dim")
elif comp_type == 7: # MENTIONABLE_SELECT
yield Text("💬 Mentionable Select", style="dim")
yield Text("")
def _render_button(self, button: Dict[str, Any]) -> RenderResult:
"""Render a button component."""
style_map = {
1: "primary", # PRIMARY (blurple)
2: "secondary", # SECONDARY (grey)
3: "success", # SUCCESS (green)
4: "danger", # DANGER (red)
5: "link", # LINK
}
style = style_map.get(button.get("style", 1), "primary")
label = button.get("label", "")
emoji = button.get("emoji")
custom_id = button.get("custom_id", "")
disabled = button.get("disabled", False)
button_text = Text()
if emoji:
if emoji.get("name"):
button_text.append(f"{emoji['name']} ")
elif emoji.get("id"):
button_text.append(f"<:{emoji.get('name', 'emoji')}:{emoji['id']}> ")
button_text.append(label)
if disabled:
button_text.stylize("dim")
# Create a panel for the button
panel = Panel(
button_text,
title=f"[{custom_id}]" if custom_id else "",
subtitle=f"({style})",
border_style=style,
padding=(0, 2),
expand=False,
)
if disabled:
panel.border_style = "dim"
yield panel
def _render_select(self, select: Dict[str, Any]) -> RenderResult:
"""Render a select menu component."""
custom_id = select.get("custom_id", "")
placeholder = select.get("placeholder", "Select an option...")
options = select.get("options", [])
disabled = select.get("disabled", False)
select_text = Text()
select_text.append("🔽 ", style="dim")
select_text.append(placeholder, style="dim")
if options:
select_text.append("\n")
for opt in options[:5]: # Show first 5 options
label = opt.get("label", "")
value = opt.get("value", "")
desc = opt.get("description", "")
select_text.append(f" • {label}", style="dim")
if desc:
select_text.append(f" ({desc})", style="dim dim")
if value != label:
select_text.append(f" = {value}", style="dim dim dim")
select_text.append("\n")
if len(options) > 5:
select_text.append(f" ... and {len(options) - 5} more", style="dim dim")
panel = Panel(
select_text,
title=f"Select [{custom_id}]",
border_style="primary" if not disabled else "dim",
padding=(0, 1),
)
yield panel
class MessageEntry(Widget):
"""Renders a single message with embeds and components."""
DEFAULT_CSS = """
MessageEntry {
width: 1fr;
height: auto;
min-height: 3;
margin: 1 0;
padding: 1;
background: $panel;
border: solid $background;
}
MessageEntry:hover {
border: solid $primary;
}
MessageEntry .message-header {
height: 1;
layout: horizontal;
}
MessageEntry .message-author {
color: $accent;
text-style: bold;
}
MessageEntry .message-timestamp {
color: $text-muted;
text-align: right;
width: 20;
}
MessageEntry .message-content {
margin-top: 1;
color: $text;
}
MessageEntry .message-embeds {
margin-top: 1;
}
MessageEntry .message-components {
margin-top: 1;
}
"""
message: reactive[Optional[Dict[str, Any]]] = reactive(None)
def __init__(self, message_data: Optional[Dict[str, Any]] = None, **kwargs):
super().__init__(**kwargs)
self.message = message_data
def render(self) -> RenderResult:
"""Render a complete message entry."""
if not self.message:
return Text("No message", style="dim")
# Header
author = self.message.get("author", {})
username = author.get("username", "Unknown")
discriminator = author.get("discriminator", "0000")
timestamp = self.message.get("timestamp", "")
header = Text()
header.append(f"{username}#{discriminator} ", style="bold")
if author.get("bot"):
header.append("🤖 ", style="dim")
header.append(timestamp, style="dim right")
yield header
# Content
content = self.message.get("content", "")
if content:
yield Text(content)
yield Text("")
# Embeds
embeds = self.message.get("embeds", [])
if embeds:
for embed in embeds:
embed_renderer = EmbedRenderer(embed)
yield embed_renderer.render()
# Components
components = self.message.get("components", [])
if components:
component_renderer = ComponentRenderer(components)
yield component_renderer.render()
class MessageList(Widget):
"""Scrollable list of messages."""
DEFAULT_CSS = """
MessageList {
width: 1fr;
height: 1fr;
overflow-y: scroll;
padding: 1;
}
"""
messages: reactive[List[Dict[str, Any]]] = reactive([])
def __init__(self, messages: Optional[List[Dict[str, Any]]] = None, **kwargs):
super().__init__(**kwargs)
self.messages = messages or []
def add_message(self, message: Dict[str, Any]) -> None:
"""Add a message to the list."""
self.messages = [*self.messages, message]
def clear(self) -> None:
"""Clear all messages."""
self.messages = []
def compose(self) -> ComposeResult:
if not self.messages:
yield Static("No messages in this channel", style="dim")
else:
for msg in self.messages:
yield MessageEntry(msg)
class ChannelSelector(Widget):
"""Widget for selecting a channel from a guild."""
DEFAULT_CSS = """
ChannelSelector {
width: 1fr;
height: auto;
padding: 1;
}
ChannelSelector .channel-item {
height: 3;
padding: 0 1;
layout: horizontal;
}
ChannelSelector .channel-icon {
width: 3;
text-align: center;
}
ChannelSelector .channel-name {
width: 1fr;
}
ChannelSelector .channel-type {
width: 10;
text-align: right;
color: $text-muted;
}
"""
channels: reactive[List[Dict[str, Any]]] = reactive([])
selected_channel: reactive[Optional[str]] = reactive(None)
def __init__(self, channels: Optional[List[Dict[str, Any]]] = None, **kwargs):
super().__init__(**kwargs)
self.channels = channels or []
def render(self) -> RenderResult:
if not self.channels:
yield Text("No channels available", style="dim")
return
for channel in self.channels:
is_selected = channel.get("id") == self.selected_channel
channel_text = Text()
# Icon based on type
ch_type = channel.get("type", 0)
if ch_type == 0: # GUILD_TEXT
channel_text.append("# ")
elif ch_type == 2: # GUILD_VOICE
channel_text.append("🔊 ")
elif ch_type == 13: # STAGE_VOICE
channel_text.append("🎤 ")
elif ch_type == 4: # GUILD_CATEGORY
channel_text.append("📁 ")
else:
channel_text.append("💬 ")
channel_text.append(channel.get("name", "unknown"))
if is_selected:
channel_text.stylize("bold reverse")
# Type indicator
type_names = {0: "TEXT", 2: "VOICE", 13: "STAGE", 4: "CATEGORY"}
type_str = type_names.get(ch_type, "UNK")
yield Panel(
channel_text,
title=type_str,
border_style="primary" if not is_selected else "accent",
padding=(0, 1),
expand=True,
)
+207
View File
@@ -0,0 +1,207 @@
"""
Message List Widget
Scrollable list of messages with proper rendering of embeds and components.
"""
from typing import Any, Dict, List, Optional
from textual.widget import Widget
from textual.reactive import reactive
from textual.widgets import Static, ListView, ListItem
from textual.containers import Container
from apicord.tui.widgets.embed_renderer import MessageEntry, MessageList as BaseMessageList
class MessageList(Widget):
"""Enhanced message list with channel context and proper message rendering."""
DEFAULT_CSS = """
MessageList {
width: 1fr;
height: 1fr;
overflow-y: scroll;
padding: 1;
}
MessageList .channel-header {
height: 3;
padding: 0 1;
background: $panel;
border-bottom: solid $primary;
/* font-size: 80%; */
}
MessageList .channel-info {
width: 1fr;
padding: 1 0;
}
MessageList .channel-name {
text-style: bold;
color: $accent;
}
MessageList .channel-topic {
color: $text-muted;
/* font-size: 80%; */
}
MessageList .message-count {
width: 10;
text-align: right;
color: $text-muted;
padding: 1 0;
}
"""
messages: reactive[List[Dict[str, Any]]] = reactive([])
current_channel_id: reactive[Optional[str]] = reactive(None)
current_channel_name: reactive[Optional[str]] = reactive(None)
def __init__(self, **kwargs):
super().__init__(**kwargs)
self._message_widgets: List[Any] = []
def set_channel(self, channel_id: str, channel_name: str, messages: List[Dict[str, Any]]) -> None:
"""Set the current channel and its messages."""
self.current_channel_id = channel_id
self.current_channel_name = channel_name
self.messages = messages
self._rebuild_message_list()
def add_message(self, message: Dict[str, Any]) -> None:
"""Add a new message to the list."""
self.messages = [*self.messages, message]
self._add_message_widget(message)
def _rebuild_message_list(self) -> None:
"""Rebuild the entire message list."""
# This will be called when messages reactive updates
pass
def _add_message_widget(self, message: Dict[str, Any]) -> None:
"""Add a single message widget to the list."""
# This would be called when a new message arrives
pass
def compose(self) -> ComposeResult:
if self.current_channel_id:
yield Static(
f"# {self.current_channel_name or 'Unknown'}",
classes="channel-name"
)
yield Static(f"{len(self.messages)} messages", classes="message-count")
if not self.messages:
yield Static("No messages in this channel", style="dim", classes="message-entry")
else:
# Use ListView for scrollable messages
list_view = ListView(id="message-list-view")
for msg in self.messages:
list_view.append(ListItem(MessageEntry(msg)))
yield list_view
def watch_messages(self, messages: List[Dict[str, Any]]) -> None:
"""React to messages changes."""
self._rebuild_message_list()
class ChannelMessageView(Widget):
"""Complete channel view with header, messages, and input area."""
DEFAULT_CSS = """
ChannelMessageView {
width: 1fr;
height: 1fr;
layout: vertical;
}
ChannelMessageView .channel-header {
height: 3;
padding: 0 1;
background: $panel;
border-bottom: solid $primary;
layout: horizontal;
}
ChannelMessageView .channel-icon {
width: 3;
text-align: center;
}
ChannelMessageView .channel-info {
width: 1fr;
padding: 1 0;
}
ChannelMessageView .channel-name {
text-style: bold;
color: $accent;
}
ChannelMessageView .channel-topic {
color: $text-muted;
/* font-size: 80%; */
}
ChannelMessageView .message-area {
height: 1fr;
border: solid $primary;
margin: 1;
overflow-y: scroll;
}
ChannelMessageView .message-input-area {
height: 3;
padding: 0 1;
background: $panel;
border-top: solid $primary;
layout: horizontal;
}
"""
current_channel_id: reactive[Optional[str]] = reactive(None)
current_channel_name: reactive[Optional[str]] = reactive(None)
current_channel_topic: reactive[Optional[str]] = reactive(None)
messages: reactive[List[Dict[str, Any]]] = reactive([])
def __init__(self, **kwargs):
super().__init__(**kwargs)
def set_channel(self, channel_id: str, channel_name: str, topic: str = "", messages: List[Dict[str, Any]] = None) -> None:
"""Set the current channel."""
self.current_channel_id = channel_id
self.current_channel_name = channel_name
self.current_channel_topic = topic
self.messages = messages or []
def add_message(self, message: Dict[str, Any]) -> None:
"""Add a message to the view."""
self.messages = [*self.messages, message]
def compose(self) -> ComposeResult:
# Channel header
with Container(classes="channel-header"):
yield Static("#", classes="channel-icon")
with Container(classes="channel-info"):
yield Static(self.current_channel_name or "No channel selected", classes="channel-name")
yield Static(self.current_channel_topic or "No topic", classes="channel-topic")
# Message area
with Container(classes="message-area"):
if not self.messages:
yield Static("No messages in this channel", style="dim", classes="message-entry")
else:
for msg in self.messages:
yield MessageEntry(msg)
# Input area
with Container(classes="message-input-area"):
yield Static(">", classes="prompt")
yield Input(placeholder="Message #channel...", id="message-input")
def watch_messages(self, messages: List[Dict[str, Any]]) -> None:
"""React to messages changes."""
self.refresh()