"""Terminal UI components using Textual framework.""" import asyncio from datetime import datetime from typing import List from textual.app import App, ComposeResult from textual.containers import Container, Vertical, Horizontal, ScrollableContainer from textual.widgets import Header, Footer, Input, Static, RichLog from textual.binding import Binding from .constants import COLORS class ChatHistory(RichLog): """Custom RichLog widget for chat history display.""" def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs, highlight=True, markup=True) self.can_focus = False class TerminalChatApp(App): """Main terminal chat application UI.""" CSS = """ Screen { background: #1a1a1a; } Header { background: #2d2d2d; color: #00ff00; text-style: bold; } Footer { background: #2d2d2d; } #main-container { height: 1fr; layout: horizontal; background: #1a1a1a; } #chat-panel { width: 3fr; height: 1fr; layout: vertical; } #chat-container { height: 1fr; border: heavy #00ff00; background: #0a0a0a; padding: 0; } ChatHistory { background: #0a0a0a; color: #ffffff; padding: 1 2; border: none; height: 1fr; } #input-container { height: auto; padding: 1 2; background: #1a1a1a; border-top: heavy #00ff00; } #user-input { width: 1fr; background: #2d2d2d; border: solid #00ff00; } #user-input:focus { border: heavy #00ff00; } #sidebar { width: 22; height: 1fr; layout: vertical; margin-left: 1; } #channels-panel { height: 1fr; border: heavy #00ff00; background: #0a0a0a; padding: 1; margin-bottom: 1; } #users-panel { height: 1fr; border: heavy #00ff00; background: #0a0a0a; padding: 1; } .panel-title { color: #00ff00; text-style: bold; text-align: center; margin-bottom: 1; background: #1a1a1a; padding: 1; border-bottom: solid #00ff00; } .panel-list { height: 1fr; padding: 1 0; } .list-item { padding: 0 1; margin: 0 0 1 0; } #status-bar { dock: bottom; height: 1; background: #2d2d2d; color: #00ff00; padding: 0 2; text-style: bold; } #help-bar { height: auto; background: #1a1a1a; color: #888888; padding: 1 2; border-top: solid #444444; text-align: center; } """ BINDINGS = [ Binding("ctrl+c", "quit", "Quit", show=False), Binding("ctrl+d", "quit", "Quit", show=False), Binding("ctrl+h", "show_help", "Help", show=True), ] TITLE = "Terminal Chat" def __init__(self, username: str, message_queue: asyncio.Queue, chat_server): super().__init__() self.username = username self.message_queue = message_queue self.chat_server = chat_server self.current_view = 'global' # Can be channel name or 'dm:username' self.users_list = [] def compose(self) -> ComposeResult: yield Header() with Container(id="main-container"): with Vertical(id="chat-panel"): with Container(id="chat-container"): yield ChatHistory(id="chat-history") with Horizontal(id="input-container"): yield Input( placeholder="Type a message or /help for commands...", id="user-input" ) with Vertical(id="sidebar"): with Vertical(id="channels-panel"): yield Static("CHANNELS", classes="panel-title") yield ScrollableContainer(id="channels-list", classes="panel-list") with Vertical(id="users-panel"): yield Static("ONLINE USERS", classes="panel-title") yield ScrollableContainer(id="users-list", classes="panel-list") yield Static( "/dm - DM | /create - create | /join - join | /help - help", id="help-bar" ) yield Static(f"Connected as: {self.username} | View: #global", id="status-bar") yield Footer() def on_mount(self) -> None: self.query_one("#user-input").focus() self.display_info_message(f"Welcome to Terminal Chat, {self.username}!") self.display_info_message("Type /help to see available commands") self.set_interval(0.1, self.check_messages) self.update_channels_list() def update_status_bar(self): view_text = f"#{self.current_view}" if not self.current_view.startswith('dm:') else f"DM: {self.current_view[3:]}" self.query_one("#status-bar", Static).update(f"Connected as: {self.username} | View: {view_text}") def update_channels_list(self): channels_container = self.query_one("#channels-list", ScrollableContainer) channels_container.remove_children() user_channels = self.chat_server.get_user_channels(self.username) for channel_name in sorted(user_channels): prefix = "> " if channel_name == self.current_view else " " color = COLORS['channel'] if channel_name == self.current_view else "#888888" channel_widget = Static(f"{prefix}[{color}]#{channel_name}[/{color}]", classes="list-item", markup=True) channels_container.mount(channel_widget) def display_user_message(self, username: str, message: str, timestamp: str, color: str, channel: str = None): chat_history = self.query_one(ChatHistory) chat_history.write(f"[dim]{timestamp}[/dim] [{color}]{username}[/{color}]: {message}") def display_system_message(self, message: str, timestamp: str): chat_history = self.query_one(ChatHistory) chat_history.write(f"[dim]{timestamp}[/dim] [{COLORS['system']}][SYSTEM][/{COLORS['system']}] {message}") def display_error_message(self, message: str, timestamp: str = None): if not timestamp: timestamp = datetime.now().strftime("%H:%M:%S") chat_history = self.query_one(ChatHistory) chat_history.write(f"[dim]{timestamp}[/dim] [{COLORS['error']}][ERROR][/{COLORS['error']}] {message}") def display_info_message(self, message: str): timestamp = datetime.now().strftime("%H:%M:%S") chat_history = self.query_one(ChatHistory) chat_history.write(f"[dim]{timestamp}[/dim] [{COLORS['info']}][INFO][/{COLORS['info']}] {message}") def display_success_message(self, message: str): timestamp = datetime.now().strftime("%H:%M:%S") chat_history = self.query_one(ChatHistory) chat_history.write(f"[dim]{timestamp}[/dim] [{COLORS['success']}][SUCCESS][/{COLORS['success']}] {message}") def display_dm(self, from_user: str, message: str, timestamp: str, color: str): chat_history = self.query_one(ChatHistory) chat_history.write(f"[dim]{timestamp}[/dim] [{color}]{from_user}[/{color}]: {message}") def switch_view(self, view: str): self.current_view = view self.update_status_bar() self.update_channels_list() chat_history = self.query_one(ChatHistory) chat_history.clear() if view.startswith('dm:'): other_user = view[3:] dm = self.chat_server.get_or_create_dm(self.username, other_user) for msg in dm.messages: self.display_dm(msg['from'], msg['message'], msg['timestamp'], msg['color']) else: if view in self.chat_server.channels: channel = self.chat_server.channels[view] for msg in channel.messages: if msg['type'] == 'system': self.display_system_message(msg['message'], msg['timestamp']) elif msg['type'] == 'message': self.display_user_message(msg['username'], msg['message'], msg['timestamp'], msg['color']) def check_messages(self): try: while not self.message_queue.empty(): msg_data = self.message_queue.get_nowait() # Only display if in the right view if msg_data['type'] == 'message': if msg_data.get('channel') == self.current_view: self.display_user_message( msg_data['username'], msg_data['message'], msg_data['timestamp'], msg_data.get('color', '#FFFFFF') ) elif msg_data['type'] == 'dm': other_user = msg_data['from'] if msg_data['from'] != self.username else msg_data['to'] if self.current_view == f"dm:{other_user}": self.display_dm( msg_data['from'], msg_data['message'], msg_data['timestamp'], msg_data.get('color', '#FFFFFF') ) elif msg_data['type'] == 'system': if self.current_view == 'global': self.display_system_message(msg_data['message'], msg_data['timestamp']) elif msg_data['type'] == 'error': self.display_error_message(msg_data['message'], msg_data['timestamp']) elif msg_data['type'] == 'info': self.display_info_message(msg_data['message']) elif msg_data['type'] == 'user_list': self.update_user_list(msg_data['users']) self.update_channels_list() except: pass def update_user_list(self, users: list): self.users_list = users users_container = self.query_one("#users-list", ScrollableContainer) users_container.remove_children() for user_data in sorted(users, key=lambda x: x['username']): username = user_data['username'] color = user_data['color'] prefix = "> " if username == self.username else " " user_widget = Static(f"{prefix}[{color}]{username}[/{color}]", classes="list-item", markup=True) users_container.mount(user_widget) def process_command(self, message: str) -> bool: if not message.startswith('/'): return False parts = message.split(' ', 2) command = parts[0].lower() if command == '/help' or command == '/h': self.show_help_message() return True elif command == '/dm' or command == '/d': if len(parts) < 2: self.display_error_message("Usage: /dm [message]") return True target_user = parts[1] if target_user == self.username: self.display_error_message("You cannot DM yourself") return True if target_user not in self.chat_server.users: self.display_error_message(f"User '{target_user}' is not online") return True if len(parts) == 3: self.chat_server.send_dm(self.username, target_user, parts[2]) else: self.switch_view(f"dm:{target_user}") return True elif command == '/create' or command == '/new': if len(parts) < 2: self.display_error_message("Usage: /create [private]") return True channel_name = parts[1].lower() is_private = len(parts) > 2 and parts[2].lower() == 'private' if self.chat_server.create_channel(channel_name, self.username, is_private): self.display_success_message(f"Created channel #{channel_name}") self.switch_view(channel_name) else: self.display_error_message(f"Channel #{channel_name} already exists") return True elif command == '/join' or command == '/j': if len(parts) < 2: self.display_error_message("Usage: /join ") return True channel_name = parts[1].lower() success, msg = self.chat_server.join_channel(channel_name, self.username) if success: self.display_success_message(msg) self.switch_view(channel_name) else: self.display_error_message(msg) return True elif command == '/leave' or command == '/l': if len(parts) < 2: self.display_error_message("Usage: /leave ") return True channel_name = parts[1].lower() success, msg = self.chat_server.leave_channel(channel_name, self.username) if success: self.display_success_message(msg) self.switch_view('global') else: self.display_error_message(msg) return True elif command == '/invite' or command == '/inv': if len(parts) < 3: self.display_error_message("Usage: /invite ") return True channel_name = parts[1].lower() invitee = parts[2] success, msg = self.chat_server.invite_to_channel(channel_name, self.username, invitee) if success: self.display_success_message(msg) else: self.display_error_message(msg) return True elif command == '/delete' or command == '/del': if len(parts) < 2: self.display_error_message("Usage: /delete ") return True channel_name = parts[1].lower() if self.chat_server.delete_channel(channel_name, self.username): self.display_success_message(f"Deleted channel #{channel_name}") self.switch_view('global') else: self.display_error_message(f"Cannot delete #{channel_name}. You must be the creator") return True elif command == '/switch' or command == '/s': if len(parts) < 2: self.display_error_message("Usage: /switch ") return True channel_name = parts[1].lower() if channel_name in self.chat_server.channels and self.chat_server.channels[channel_name].has_member(self.username): self.switch_view(channel_name) else: self.display_error_message(f"You are not a member of #{channel_name}") return True elif command == '/channels' or command == '/ch': channels = self.chat_server.get_user_channels(self.username) self.display_info_message(f"Your channels: {', '.join(['#' + c for c in channels])}") return True elif command == '/users' or command == '/u': usernames = [u['username'] for u in self.users_list] self.display_info_message(f"Online users ({len(usernames)}): {', '.join(sorted(usernames))}") return True elif command == '/clear' or command == '/c': chat_history = self.query_one(ChatHistory) chat_history.clear() self.display_info_message("Chat history cleared locally") return True elif command == '/quit' or command == '/q': self.exit() return True else: self.display_error_message(f"Unknown command: {command}. Type /help for available commands") return True def show_help_message(self): self.display_info_message("Available commands:") chat_history = self.query_one(ChatHistory) chat_history.write(f" [{COLORS['help']}]/dm [msg][/{COLORS['help']}] or [{COLORS['command']}]/d[/{COLORS['command']}] - DM a user") chat_history.write(f" [{COLORS['help']}]/create [private][/{COLORS['help']}] or [{COLORS['command']}]/new[/{COLORS['command']}] - Create a channel") chat_history.write(f" [{COLORS['help']}]/join [/{COLORS['help']}] or [{COLORS['command']}]/j[/{COLORS['command']}] - Join a channel") chat_history.write(f" [{COLORS['help']}]/leave [/{COLORS['help']}] or [{COLORS['command']}/]/l[/{COLORS['command']}] - Leave a channel") chat_history.write(f" [{COLORS['help']}]/invite [/{COLORS['help']}] or [{COLORS['command']}]/inv[/{COLORS['command']}] - Invite user to channel") chat_history.write(f" [{COLORS['help']}]/delete [/{COLORS['help']}] or [{COLORS['command']}]/del[/{COLORS['command']}] - Delete a channel you created") chat_history.write(f" [{COLORS['help']}]/switch [/{COLORS['help']}] or [{COLORS['command']}]/s[/{COLORS['command']}] - Switch to a channel") chat_history.write(f" [{COLORS['help']}]/channels[/{COLORS['help']}] or [{COLORS['command']}]/ch[/{COLORS['command']}] - List your channels") chat_history.write(f" [{COLORS['help']}]/users[/{COLORS['help']}] or [{COLORS['command']}]/u[/{COLORS['command']}] - List online users") chat_history.write(f" [{COLORS['help']}]/clear[/{COLORS['help']}] or [{COLORS['command']}]/c[/{COLORS['command']}] - Clear chat history locally") chat_history.write(f" [{COLORS['help']}]/quit[/{COLORS['help']}] or [{COLORS['command']}]/q[/{COLORS['command']}] - Quit the chat application") def on_input_submitted(self, event: Input.Submitted) -> None: message = event.value.strip() event.input.value = "" if not message: return if self.process_command(message): return # Regular message if self.current_view.startswith('dm:'): target_user = self.current_view[3:] self.chat_server.send_dm(self.username, target_user, message) else: self.chat_server.send_channel_message(self.current_view, self.username, message)