mirror of
https://github.com/onyx-and-iris/q3rcon-cli.git
synced 2026-03-23 18:29:18 +00:00
Compare commits
8 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 60afda97dc | |||
| 968cd2dc8a | |||
| d7ea4d9fad | |||
| d0519bb8e9 | |||
| 3a7cc3eacb | |||
| fce5858a93 | |||
| a9aff25c4e | |||
| 7af5caad89 |
@ -1,4 +1,4 @@
|
||||
# SPDX-FileCopyrightText: 2026-present onyx-and-iris <code@onyxandiris.online>
|
||||
#
|
||||
# SPDX-License-Identifier: MIT
|
||||
__version__ = '0.2.3'
|
||||
__version__ = '0.2.5'
|
||||
|
||||
@ -1,6 +1,6 @@
|
||||
import clypi
|
||||
from aioq3rcon import Client, IncorrectPasswordError
|
||||
from clypi import Command, arg
|
||||
from clypi import Command, Spinner, arg
|
||||
from typing_extensions import override
|
||||
|
||||
from . import console
|
||||
@ -72,30 +72,36 @@ class Q3rconCli(Command):
|
||||
clypi.style('to quit.', fg='blue'),
|
||||
)
|
||||
|
||||
DEFAULT_TIMEOUT = 2
|
||||
DEFAULT_FRAGMENT_READ_TIMEOUT = 0.25
|
||||
while command := input(clypi.style('cmd: ', fg='green')):
|
||||
if command.lower() == 'q':
|
||||
break
|
||||
|
||||
fragment_read_timeout = None
|
||||
if command in (
|
||||
'status',
|
||||
'fast_restart',
|
||||
'map_restart',
|
||||
'map',
|
||||
'map_rotate',
|
||||
):
|
||||
fragment_read_timeout = 1
|
||||
CMD_CONFIG = {
|
||||
'status': (2, 1, False),
|
||||
'fast_restart': (3, 1, True),
|
||||
'map_restart': (3, 1, True),
|
||||
'map': (3, 1, True),
|
||||
'map_rotate': (3, 1, True),
|
||||
}
|
||||
timeout, fragment_read_timeout, interpret = CMD_CONFIG.get(
|
||||
command.split()[0].lower(),
|
||||
(DEFAULT_TIMEOUT, DEFAULT_FRAGMENT_READ_TIMEOUT, False),
|
||||
)
|
||||
|
||||
async with Spinner(f"Sending command: '{command}'", suffix='...'):
|
||||
async with Client(
|
||||
self.host,
|
||||
self.port,
|
||||
self.password,
|
||||
fragment_read_timeout=fragment_read_timeout
|
||||
or DEFAULT_FRAGMENT_READ_TIMEOUT,
|
||||
timeout=timeout,
|
||||
fragment_read_timeout=fragment_read_timeout,
|
||||
) as client:
|
||||
try:
|
||||
if response := await client.send_command(command):
|
||||
if response := await client.send_command(
|
||||
command, interpret=interpret
|
||||
):
|
||||
console.out.print_response(response)
|
||||
except TimeoutError:
|
||||
console.err.print(
|
||||
|
||||
@ -1,5 +1,5 @@
|
||||
from aioq3rcon import Client
|
||||
from clypi import Command, arg
|
||||
from clypi import Command, Spinner, arg
|
||||
from typing_extensions import override
|
||||
|
||||
from q3rcon_cli import console
|
||||
@ -14,6 +14,8 @@ class Fastrestart(Command):
|
||||
|
||||
@override
|
||||
async def run(self):
|
||||
async with Spinner('Executing fast restart', suffix='...'):
|
||||
async with Client(self.host, self.port, self.password) as client:
|
||||
if response := await client.send_command('fast_restart'):
|
||||
response = await client.send_command('fast_restart', interpret=True)
|
||||
|
||||
console.out.print_response(response)
|
||||
|
||||
@ -23,21 +23,24 @@ class Gametype(Command):
|
||||
|
||||
@override
|
||||
async def run(self):
|
||||
if not Gametype.new_gametype:
|
||||
if not self.new_gametype:
|
||||
async with Spinner('Fetching current gametype', suffix='...'):
|
||||
async with Client(self.host, self.port, self.password) as client:
|
||||
if response := await client.send_command('g_gametype'):
|
||||
response = await client.send_command('g_gametype')
|
||||
console.out.print_cvar(response)
|
||||
return
|
||||
|
||||
async with Spinner(f'Changing gametype to {self.new_gametype}', suffix='...'):
|
||||
async with Client(self.host, self.port, self.password) as client:
|
||||
DEFAULT_FRAGMENT_READ_TIMEOUT = client.fragment_read_timeout
|
||||
|
||||
await client.send_command(f'g_gametype {self.new_gametype}')
|
||||
|
||||
if self.force:
|
||||
async with Spinner('Forcing gametype change...'):
|
||||
async with Spinner('Forcing gametype change', suffix='...'):
|
||||
async with Client(self.host, self.port, self.password) as client:
|
||||
client.timeout = 3
|
||||
client.fragment_read_timeout = 1
|
||||
await client.send_command('map_restart')
|
||||
|
||||
client.fragment_read_timeout = DEFAULT_FRAGMENT_READ_TIMEOUT
|
||||
if response := await client.send_command('g_gametype'):
|
||||
console.out.print_cvar(response)
|
||||
console.out.print(
|
||||
f'Gametype changed successfully to {self.new_gametype}.', style='green'
|
||||
)
|
||||
|
||||
@ -1,5 +1,5 @@
|
||||
from aioq3rcon import Client
|
||||
from clypi import Command, Positional, arg
|
||||
from clypi import Command, Positional, Spinner, arg
|
||||
from typing_extensions import override
|
||||
|
||||
from q3rcon_cli import console
|
||||
@ -19,12 +19,14 @@ class Hostname(Command):
|
||||
@override
|
||||
async def run(self):
|
||||
if not self.new_hostname:
|
||||
async with Spinner('Fetching current hostname', suffix='...'):
|
||||
async with Client(self.host, self.port, self.password) as client:
|
||||
if response := await client.send_command('sv_hostname'):
|
||||
response = await client.send_command('sv_hostname')
|
||||
console.out.print_cvar(response)
|
||||
return
|
||||
|
||||
async with Spinner(f'Changing hostname to {self.new_hostname}', suffix='...'):
|
||||
async with Client(self.host, self.port, self.password) as client:
|
||||
await client.send_command(f'sv_hostname {self.new_hostname}')
|
||||
if response := await client.send_command('sv_hostname'):
|
||||
console.out.print_cvar(response)
|
||||
|
||||
console.out.print(f'Hostname changed to: {self.new_hostname}', style='green')
|
||||
|
||||
@ -19,15 +19,18 @@ class Map(Command):
|
||||
@override
|
||||
async def run(self):
|
||||
if not self.new_map:
|
||||
async with Spinner('Getting current map', suffix='...'):
|
||||
async with Client(self.host, self.port, self.password) as client:
|
||||
if response := await client.send_command('mapname'):
|
||||
response = await client.send_command('mapname')
|
||||
console.out.print_cvar(response)
|
||||
return
|
||||
|
||||
async with Spinner('Changing map...'):
|
||||
async with Spinner('Changing map', suffix='...'):
|
||||
async with Client(
|
||||
self.host, self.port, self.password, fragment_read_timeout=1
|
||||
) as client:
|
||||
await client.send_command(f'map mp_{self.new_map.removeprefix("mp_")}')
|
||||
if response := await client.send_command('mapname'):
|
||||
console.out.print_cvar(response)
|
||||
|
||||
console.out.print(
|
||||
f'Map changed to {self.new_map.removeprefix("mp_")}', style='green'
|
||||
)
|
||||
|
||||
@ -1,5 +1,5 @@
|
||||
from aioq3rcon import Client
|
||||
from clypi import Command, arg
|
||||
from clypi import Command, Spinner, arg
|
||||
from typing_extensions import override
|
||||
|
||||
from q3rcon_cli import console
|
||||
@ -14,6 +14,8 @@ class Mapname(Command):
|
||||
|
||||
@override
|
||||
async def run(self):
|
||||
async with Spinner('Getting map name...'):
|
||||
async with Client(self.host, self.port, self.password) as client:
|
||||
if response := await client.send_command('mapname'):
|
||||
response = await client.send_command('mapname')
|
||||
|
||||
console.out.print_cvar(response)
|
||||
|
||||
@ -14,9 +14,10 @@ class Maprestart(Command):
|
||||
|
||||
@override
|
||||
async def run(self):
|
||||
async with Spinner('Restarting map...'):
|
||||
async with Spinner('Restarting map', suffix='...'):
|
||||
async with Client(
|
||||
self.host, self.port, self.password, fragment_read_timeout=1
|
||||
self.host, self.port, self.password, timeout=3, fragment_read_timeout=1
|
||||
) as client:
|
||||
if response := await client.send_command('map_restart'):
|
||||
response = await client.send_command('map_restart', interpret=True)
|
||||
|
||||
console.out.print_response(response)
|
||||
|
||||
@ -14,9 +14,14 @@ class Maprotate(Command):
|
||||
|
||||
@override
|
||||
async def run(self):
|
||||
async with Spinner('Rotating map...'):
|
||||
async with Spinner('Rotating map', suffix='...'):
|
||||
async with Client(
|
||||
self.host, self.port, self.password, fragment_read_timeout=1
|
||||
self.host,
|
||||
self.port,
|
||||
self.password,
|
||||
timeout=3,
|
||||
fragment_read_timeout=1,
|
||||
) as client:
|
||||
if response := await client.send_command('map_rotate'):
|
||||
response = await client.send_command('map_rotate', interpret=True)
|
||||
|
||||
console.out.print_response(response)
|
||||
|
||||
@ -1,5 +1,5 @@
|
||||
from aioq3rcon import Client
|
||||
from clypi import Command, arg
|
||||
from clypi import Command, Spinner, arg
|
||||
from typing_extensions import override
|
||||
|
||||
from q3rcon_cli import console
|
||||
@ -14,6 +14,8 @@ class Plugins(Command):
|
||||
|
||||
@override
|
||||
async def run(self):
|
||||
async with Spinner('Fetching plugins...'):
|
||||
async with Client(self.host, self.port, self.password) as client:
|
||||
if response := await client.send_command('plugins'):
|
||||
response = await client.send_command('plugins')
|
||||
|
||||
console.out.print_response(response)
|
||||
|
||||
@ -14,9 +14,10 @@ class Status(Command):
|
||||
|
||||
@override
|
||||
async def run(self):
|
||||
async with Spinner('Fetching status...'):
|
||||
async with Spinner('Fetching status', suffix='...'):
|
||||
async with Client(
|
||||
self.host, self.port, self.password, fragment_read_timeout=0.5
|
||||
self.host, self.port, self.password, fragment_read_timeout=1
|
||||
) as client:
|
||||
if response := await client.send_command('status'):
|
||||
response = await client.send_command('status')
|
||||
|
||||
console.out.print_status(response)
|
||||
|
||||
@ -1,7 +1,7 @@
|
||||
import re
|
||||
|
||||
import clypi
|
||||
from clypi import cprint
|
||||
from clypi import Boxes, cprint
|
||||
|
||||
|
||||
class Console:
|
||||
@ -33,6 +33,8 @@ class OutConsole(Console):
|
||||
r'(?P<rate>[0-9]+)$',
|
||||
re.IGNORECASE | re.VERBOSE,
|
||||
)
|
||||
STATUS_MAP_REGEX = re.compile(r'^map: (?P<mapname>mp_[a-z_]+)$')
|
||||
|
||||
CVAR_REGEX = re.compile(
|
||||
r'^["](?P<name>[a-z_]+)["]\sis[:]\s'
|
||||
r'["](?P<value>.*?)["]\s'
|
||||
@ -69,40 +71,68 @@ class OutConsole(Console):
|
||||
_guids.append(m.group('guid'))
|
||||
_names.append(self._remove_colour_codes(m.group('name')))
|
||||
_ips.append(m.group('ip'))
|
||||
elif m := OutConsole.STATUS_MAP_REGEX.match(line):
|
||||
cprint(f'Current map: {m.group("mapname")}', fg=self.style)
|
||||
|
||||
if not _slots:
|
||||
cprint('\nNo players connected.\n', fg=self.style)
|
||||
cprint('No players connected.', fg='red')
|
||||
return
|
||||
|
||||
slots = clypi.boxed(_slots, title='Slot', width=10, align='center')
|
||||
scores = clypi.boxed(_scores, title='Score', width=10, align='center')
|
||||
pings = clypi.boxed(_pings, title='Ping', width=10, align='center')
|
||||
guids = clypi.boxed(_guids, title='GUID', width=len(max(_guids, key=len)) + 4)
|
||||
names = clypi.boxed(_names, title='Name', width=len(max(_names, key=len)) + 4)
|
||||
ips = clypi.boxed(_ips, title='IP', width=len(max(_ips, key=len)) + 4)
|
||||
print(f'\n{clypi.stack(slots, scores, pings, guids, names, ips, padding=0)}')
|
||||
slots = clypi.boxed(
|
||||
_slots, title='Slot', width=10, align='center', style=Boxes.ROUNDED
|
||||
)
|
||||
scores = clypi.boxed(
|
||||
_scores, title='Score', width=10, align='center', style=Boxes.ROUNDED
|
||||
)
|
||||
pings = clypi.boxed(
|
||||
_pings, title='Ping', width=10, align='center', style=Boxes.ROUNDED
|
||||
)
|
||||
guids = clypi.boxed(
|
||||
_guids,
|
||||
title='GUID',
|
||||
width=len(max(_guids, key=len)) + 4,
|
||||
style=Boxes.ROUNDED,
|
||||
)
|
||||
names = clypi.boxed(
|
||||
_names,
|
||||
title='Name',
|
||||
width=len(max(_names, key=len)) + 4,
|
||||
style=Boxes.ROUNDED,
|
||||
)
|
||||
ips = clypi.boxed(
|
||||
_ips, title='IP', width=len(max(_ips, key=len)) + 4, style=Boxes.ROUNDED
|
||||
)
|
||||
print(f'{clypi.stack(slots, scores, pings, guids, names, ips, padding=0)}')
|
||||
|
||||
def print_cvar(self, response: str):
|
||||
response = self._remove_colour_codes(response).removeprefix('print\n')
|
||||
|
||||
if m := self.CVAR_REGEX.match(response):
|
||||
name = clypi.boxed(
|
||||
[m.group('name')], title='Name', width=max(len(m.group('name')) + 4, 15)
|
||||
[m.group('name')],
|
||||
title='Name',
|
||||
width=max(len(m.group('name')) + 4, 15),
|
||||
style=Boxes.ROUNDED,
|
||||
)
|
||||
value = clypi.boxed(
|
||||
[m.group('value')],
|
||||
title='Value',
|
||||
width=max(len(m.group('value')) + 4, 15),
|
||||
style=Boxes.ROUNDED,
|
||||
)
|
||||
default = clypi.boxed(
|
||||
[m.group('default')],
|
||||
title='Default',
|
||||
width=max(len(m.group('default')) + 4, 15),
|
||||
style=Boxes.ROUNDED,
|
||||
)
|
||||
info = clypi.boxed(
|
||||
[m.group('info')], title='Info', width=max(len(m.group('info')) + 4, 15)
|
||||
[m.group('info')],
|
||||
title='Info',
|
||||
width=max(len(m.group('info')) + 4, 15),
|
||||
style=Boxes.ROUNDED,
|
||||
)
|
||||
print(f'\n{clypi.stack(name, value, default, info, padding=0)}')
|
||||
print(f'{clypi.stack(name, value, default, info, padding=0)}')
|
||||
|
||||
|
||||
out = OutConsole()
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user