diff --git a/src/q3rcon_tui/tui.py b/src/q3rcon_tui/tui.py index 506357c..1dc150a 100644 --- a/src/q3rcon_tui/tui.py +++ b/src/q3rcon_tui/tui.py @@ -1,139 +1,19 @@ -import re -from pathlib import Path -from typing import Annotated, Type - from aioq3rcon import Client, RCONError -from loguru import logger -from pydantic import AfterValidator, BeforeValidator -from pydantic_settings import BaseSettings, CliSettingsSource, SettingsConfigDict from textual.app import App, ComposeResult -from textual.containers import Grid, Horizontal, Vertical -from textual.screen import ModalScreen -from textual.widgets import Button, Input, Label, RichLog, Static +from textual.containers import Grid +from textual.widgets import Button, Input, RichLog -from .__about__ import __version__ as version - - -def is_valid_password(password: str) -> str | None: - if len(password) < 8: - raise ValueError('Password must be at least 8 characters long') - return password - - -def version_callback(value: bool) -> bool | None: - if value: - print(f'q3rcon-tui version: {version}') - raise SystemExit(0) - return False - - -class Settings(BaseSettings): - host: str = 'localhost' - port: int = 28960 - password: Annotated[str, AfterValidator(is_valid_password)] = '' - refresh_output: bool = False - version: Annotated[bool, BeforeValidator(version_callback)] = False - - model_config = SettingsConfigDict( - env_file=( - '.env', - Path.home() / '.config' / 'q3rcon-tui' / 'config.env', - ), - env_file_encoding='utf-8', - env_prefix='Q3RCON_TUI_', - cli_prefix='', - cli_parse_args=True, - cli_implicit_flags=True, - # Allow field assignment for runtime updates - validate_assignment=False, - frozen=False, - ) - - @classmethod - def settings_customise_sources( - cls, - settings_cls: Type[BaseSettings], - init_settings: ..., - env_settings: ..., - dotenv_settings: ..., - file_secret_settings: ..., - ) -> tuple: - return ( - CliSettingsSource(settings_cls), - env_settings, - dotenv_settings, - init_settings, - ) - - -try: - settings = Settings() -except ValueError as e: - logger.error(e) - raise SystemExit(1) - - -class ConfigScreen(ModalScreen[bool]): - """Modal dialog for configuring connection settings.""" - - def __init__(self, current_host: str, current_port: int, current_password: str): - super().__init__() - self.current_host = current_host - self.current_port = current_port - self.current_password = current_password - - def compose(self) -> ComposeResult: - with Vertical(id='config-dialog'): - yield Static('Connection Configuration', id='config-title') - with Vertical(id='config-form'): - yield Label('Host:') - yield Input( - value=self.current_host, placeholder='localhost', id='host-input' - ) - yield Label('Port:') - yield Input( - value=str(self.current_port), placeholder='28960', id='port-input' - ) - yield Label('Password:') - yield Input( - value=self.current_password, - placeholder='Enter password', - password=True, - id='password-input', - ) - with Horizontal(id='config-buttons'): - yield Button('Save', variant='success', id='config-save') - yield Button('Cancel', variant='error', id='config-cancel') - - def on_button_pressed(self, event: Button.Pressed) -> None: - if event.button.id == 'config-save': - try: - new_host = ( - self.query_one('#host-input', Input).value.strip() or 'localhost' - ) - new_port = int(self.query_one('#port-input', Input).value or '28960') - new_password = self.query_one('#password-input', Input).value - - if new_port < 1 or new_port > 65535: - raise ValueError('Port must be between 1 and 65535') - - if len(new_password) < 8: - raise ValueError('Password must be at least 8 characters long') - - settings.host = new_host - settings.port = new_port - settings.password = new_password - - self.dismiss(True) - except ValueError: - self.app.bell() - elif event.button.id == 'config-cancel': - self.dismiss(False) +from .configscreen import ConfigScreen +from .settings import settings +from .writable import Writable class RconApp(App): - RE_COLOR_CODES = re.compile(r'\^[0-9]') - CSS_PATH = 'rcon_tui.tcss' + CSS_PATH = 'q3rcon_tui.tcss' + + def __init__(self): + super().__init__() + self.writable = Writable() def compose(self) -> ComposeResult: yield Grid( @@ -168,18 +48,20 @@ class RconApp(App): if event.button.id != 'send': return - if settings.refresh_output: + if not settings.append: self.query_one('#response', RichLog).clear() try: async with Client( settings.host, settings.port, settings.password ) as client: - response = await client.send_command( - self.query_one('#command', Input).value - ) + cmd = self.query_one('#command', Input).value.strip() + if not cmd: + self.app.bell() + return + response = await client.send_command(cmd) self.query_one('#response', RichLog).write( - self.remove_color_codes(response.removeprefix('print\n')) + self.writable.parse(cmd, response) ) except RCONError as e: self.query_one('#response', RichLog).write( @@ -188,10 +70,6 @@ class RconApp(App): self.query_one('#command', Input).value = '' - @staticmethod - def remove_color_codes(s: str) -> str: - return RconApp.RE_COLOR_CODES.sub('', s) - def main(): app = RconApp() diff --git a/src/q3rcon_tui/writable.py b/src/q3rcon_tui/writable.py new file mode 100644 index 0000000..4c81fee --- /dev/null +++ b/src/q3rcon_tui/writable.py @@ -0,0 +1,101 @@ +import re + +from rich.table import Table + +from .settings import settings + + +class Writable: + RE_COLOR_CODES = re.compile(r'\^[0-9]') + RE_PLAYER_FROM_STATUS = re.compile( + r'^\s*(?P[0-9]+)\s+' + r'(?P[0-9-]+)\s+' + r'(?P[0-9]+)\s+' + r'(?P[0-9a-f]+)\s+' + r'(?P.*?)\s+' + r'(?P[0-9]+?)\s*' + r'(?P(?:(?:25[0-5]|2[0-4][0-9]|1[0-9][0-9]|[1-9]?[0-9])\.){3}' + r'(?:25[0-5]|2[0-4][0-9]|1[0-9][0-9]|[1-9]?[0-9])):?' + r'(?P-?[0-9]{1,5})\s*' + r'(?P-?[0-9]{1,5})\s+' + r'(?P[0-9]+)$', + re.IGNORECASE | re.VERBOSE, + ) + RE_CVAR = re.compile( + r'^["](?P[a-z_]+)["]\sis[:]\s' + r'["](?P.*?)\^7["]\s' + r'default[:]\s' + r'["](?P.*?)\^7["]\s' + r'info[:]\s' + r'["](?P.*?)\^7["]$' + ) + + @staticmethod + def remove_color_codes(s: str) -> str: + return Writable.RE_COLOR_CODES.sub('', s) + + def parse(self, cmd, response: str) -> str: + response = response.removeprefix('print\n') + if settings.raw: + return response + + match cmd: + case 'status': + return self.status_table(response) + case _: + match self.RE_CVAR.match(response): + case None: + return self.remove_color_codes(response) + case m: + return self.cvar_table(m) + + def status_table(self, status_response: str) -> Table | str: + table = Table(show_header=True, header_style='bold #88c0d0') + table.add_column('Slot', justify='center') + table.add_column('Score', justify='center') + table.add_column('Ping', justify='center') + table.add_column('GUID', justify='center') + table.add_column('Name', justify='center') + table.add_column('Last', justify='center') + table.add_column('IP', justify='center') + table.add_column('Port', justify='center') + table.add_column('QPort', justify='center') + table.add_column('Rate', justify='center') + + for line in status_response.splitlines(): + match self.RE_PLAYER_FROM_STATUS.match(line): + case None: + continue + case m: + table.add_row( + m.group('slot'), + m.group('score'), + m.group('ping'), + m.group('guid'), + self.remove_color_codes(m.group('name')), + m.group('last'), + m.group('ip'), + m.group('port'), + m.group('qport'), + m.group('rate'), + ) + + if len(table.rows) == 0: + return 'No players connected.' + return table + + def cvar_table(self, m: re.Match) -> Table: + table = Table(show_header=True, header_style='bold #88c0d0') + table.add_column('Name', justify='center') + table.add_column('Value', justify='center') + table.add_column('Default', justify='center') + table.add_column('Info', justify='center') + + table.add_row( + m.group('name'), + self.remove_color_codes(m.group('value')), + self.remove_color_codes(m.group('default')), + self.remove_color_codes(m.group('info')), + ) + + return table