DebianandSisyphus
71bd740d78
Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent) Co-authored-by: Sisyphus <[email protected]>
120 lines
4.4 KiB
Python
120 lines
4.4 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Cvar and command names learned from the server, for autocomplete.
|
|
|
|
After connecting, main.py sends `cvarlist` and `cmdlist`. The server answers
|
|
with one line per name and a count line at the end:
|
|
cvarlist: S A sv_hostname "My Server" ... 1234 total cvars
|
|
cmdlist: addbot ... 321 commands
|
|
NameCapture collects the names from those lines and hides the listing from
|
|
the output window while it is running.
|
|
"""
|
|
|
|
import logging
|
|
import re
|
|
import time
|
|
|
|
logger = logging.getLogger('namelist')
|
|
|
|
# Capture stops when no listing output arrived for this many seconds.
|
|
# Guards against the count line looking different than expected, which would
|
|
# otherwise hide all later output forever. Measured from the last message, not
|
|
# from the request: a large cvarlist (5000+ messages) takes seconds to process.
|
|
CAPTURE_TIMEOUT = 5.0
|
|
|
|
# Absolute cap on a capture, measured from the request. The timeout above is
|
|
# refreshed by every incoming message, so regular server output can keep a
|
|
# stuck capture (count line never matches) alive indefinitely without this bound.
|
|
CAPTURE_MAX_DURATION = 30.0
|
|
|
|
# Cvar line: the name is the word directly before the quoted value.
|
|
# Example: 'S A sv_hostname "My Server"' -> sv_hostname
|
|
CVARLIST_LINE = re.compile(r'([A-Za-z_]\w*)\s+"')
|
|
|
|
# Command line: one bare word.
|
|
# Example: 'addbot' -> addbot
|
|
CMDLIST_LINE = re.compile(r'^\s*([A-Za-z_]\w*)\s*$')
|
|
|
|
# Count line that ends a listing.
|
|
# Examples: '1234 total cvars', '321 commands', '321 total commands'
|
|
END_LINE = re.compile(r'^\s*\d+\s+(total cvars|total commands|commands)\s*$')
|
|
|
|
|
|
def unwrap(message):
|
|
"""
|
|
Turn one RCON message into plain text with real newlines.
|
|
Example: 'print "addbot\\n"' -> 'addbot\n'
|
|
"""
|
|
text = message
|
|
if text.startswith('print "'):
|
|
text = text[7:]
|
|
if text.endswith('"'):
|
|
text = text[:-1]
|
|
return text.replace('\\n', '\n')
|
|
|
|
|
|
class NameCapture:
|
|
"""Collects cvar and command names from the server's cvarlist and cmdlist output"""
|
|
|
|
def __init__(self):
|
|
self.names = set()
|
|
self.listings_pending = 0 # count lines still expected; 0 = not capturing
|
|
self.deadline = 0
|
|
self.deadline_max = 0
|
|
self.partial_line = ''
|
|
|
|
def request(self, rcon):
|
|
"""Send cvarlist and cmdlist; their output is captured until both count lines arrived"""
|
|
self.listings_pending = 2
|
|
self.deadline = time.time() + CAPTURE_TIMEOUT
|
|
self.deadline_max = time.time() + CAPTURE_MAX_DURATION
|
|
self.partial_line = ''
|
|
rcon.send_command(b'cvarlist')
|
|
rcon.send_command(b'cmdlist')
|
|
logger.info('Requested cvarlist and cmdlist for autocomplete')
|
|
|
|
def handle(self, message):
|
|
"""
|
|
Feed one RCON message.
|
|
Returns True when the message belongs to a running listing (hide it from the output).
|
|
"""
|
|
if self.listings_pending == 0:
|
|
return False
|
|
now = time.time()
|
|
if now > self.deadline_max:
|
|
logger.warning(f'Capture exceeded {CAPTURE_MAX_DURATION:.0f}s, '
|
|
f'stopping capture with {len(self.names)} names')
|
|
self.listings_pending = 0
|
|
return False
|
|
if now > self.deadline:
|
|
logger.warning(f'No listing output for {CAPTURE_TIMEOUT:.0f}s, '
|
|
f'stopping capture with {len(self.names)} names')
|
|
self.listings_pending = 0
|
|
return False
|
|
self.deadline = now + CAPTURE_TIMEOUT
|
|
|
|
# A line can arrive in several messages (the server sends each flag column of
|
|
# a cvarlist row as its own 1-byte message), so keep the unfinished tail
|
|
text = self.partial_line + unwrap(message)
|
|
lines = text.split('\n')
|
|
self.partial_line = lines.pop()
|
|
for line in lines:
|
|
self._learn_line(line)
|
|
|
|
# The last count line may come without a trailing newline
|
|
if END_LINE.match(self.partial_line):
|
|
self._learn_line(self.partial_line)
|
|
self.partial_line = ''
|
|
return True
|
|
|
|
def _learn_line(self, line):
|
|
if END_LINE.match(line):
|
|
self.listings_pending -= 1
|
|
if self.listings_pending == 0:
|
|
logger.info(f'Learned {len(self.names)} cvar and command names from the server')
|
|
return
|
|
|
|
match = CVARLIST_LINE.search(line) or CMDLIST_LINE.match(line)
|
|
if match:
|
|
self.names.add(match.group(1))
|