mirror of
https://github.com/onyx-and-iris/q3rcon-tui.git
synced 2026-02-26 03:09:09 +00:00
add Writable class for creating RenderableTypes
This commit is contained in:
parent
8d11f60201
commit
82fb9b9fc4
@ -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()
|
||||
|
||||
101
src/q3rcon_tui/writable.py
Normal file
101
src/q3rcon_tui/writable.py
Normal file
@ -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<slot>[0-9]+)\s+'
|
||||
r'(?P<score>[0-9-]+)\s+'
|
||||
r'(?P<ping>[0-9]+)\s+'
|
||||
r'(?P<guid>[0-9a-f]+)\s+'
|
||||
r'(?P<name>.*?)\s+'
|
||||
r'(?P<last>[0-9]+?)\s*'
|
||||
r'(?P<ip>(?:(?: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<port>-?[0-9]{1,5})\s*'
|
||||
r'(?P<qport>-?[0-9]{1,5})\s+'
|
||||
r'(?P<rate>[0-9]+)$',
|
||||
re.IGNORECASE | re.VERBOSE,
|
||||
)
|
||||
RE_CVAR = re.compile(
|
||||
r'^["](?P<name>[a-z_]+)["]\sis[:]\s'
|
||||
r'["](?P<value>.*?)\^7["]\s'
|
||||
r'default[:]\s'
|
||||
r'["](?P<default>.*?)\^7["]\s'
|
||||
r'info[:]\s'
|
||||
r'["](?P<info>.*?)\^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
|
||||
Loading…
x
Reference in New Issue
Block a user