pfl
3d1dc19d9b
Send cvarlist and cmdlist on connect and merge the answer into the autocomplete candidates. Number the server menu from 0, keep suggesting sibling commands while the command name is typed, mask the password in the startup line, ignore venv/, fix README claims.
650 lines
26 KiB
Python
650 lines
26 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
QLPyCon - Quake Live Python Console
|
|
Main entry point
|
|
"""
|
|
|
|
import argparse
|
|
import uuid
|
|
import logging
|
|
import time
|
|
import re
|
|
import curses
|
|
import zmq
|
|
import signal
|
|
import sys
|
|
import os
|
|
|
|
from lib.constants import VERSION, DEFAULT_HOST, POLL_TIMEOUT, QUIT_CONFIRM_TIMEOUT, RESPAWN_DELAY, MAX_COMMAND_HISTORY
|
|
from lib.state import GameState
|
|
from lib.network import RconConnection, StatsConnection
|
|
from lib.parser import EventParser
|
|
from lib.formatter import format_message, format_chat_message, format_powerup_message, strip_color_codes
|
|
from lib.ui import UIManager, select_server
|
|
from lib.settings import ConfigLoader
|
|
from lib.namelist import NameCapture
|
|
|
|
# Pre-compiled regex patterns
|
|
CVAR_RESPONSE_PATTERN = re.compile(r'"([^"]+)"\s+is:"([^"]*)"')
|
|
PORT_PATTERN = re.compile(r'(\d+)')
|
|
BROADCAST_PATTERN = re.compile(r'^broadcast:\s*print\s*"(.+?)(?:\\n)?"\s*$')
|
|
TIMESTAMP_PATTERN = re.compile(r'\^\d\[\^\d[0-9:]+\^\d\]\^\d\s*|\[[0-9:]+\]\s*')
|
|
CONNECT_PATTERN = re.compile(r'^(.+?)\s+connected')
|
|
DISCONNECT_PATTERN = re.compile(r'^(.+?)\s+disconnected')
|
|
KICK_PATTERN = re.compile(r'^(.+?)\s+was kicked')
|
|
INACTIVITY_PATTERN = re.compile(r'^(.+?)\s+Dropped due to inactivity')
|
|
RENAME_PATTERN = re.compile(r'^(.+?)\s+renamed to\s+(.+?)$')
|
|
|
|
# Configure logging
|
|
logger = logging.getLogger('main')
|
|
logger.setLevel(logging.DEBUG)
|
|
|
|
all_json_logger = logging.getLogger('all_json')
|
|
all_json_logger.setLevel(logging.DEBUG)
|
|
|
|
unknown_json_logger = logging.getLogger('unknown_json')
|
|
unknown_json_logger.setLevel(logging.DEBUG)
|
|
|
|
# Ctrl-C state: the signal handler only records the press, the main loop
|
|
# reacts to it, so no screen drawing happens inside a signal handler.
|
|
ctrl_c_pressed = False
|
|
quit_confirm_time = None
|
|
|
|
# Configurable timeouts (set from config in __main__)
|
|
_QUIT_CONFIRM_TIMEOUT = QUIT_CONFIRM_TIMEOUT
|
|
_RESPAWN_DELAY = RESPAWN_DELAY
|
|
|
|
# Global shutdown flag (set by handle_ctrl_c, checked by main_loop)
|
|
shutdown_requested = False
|
|
|
|
def signal_handler(sig, frame):
|
|
"""Record Ctrl+C; force exit if the graceful shutdown already started"""
|
|
global ctrl_c_pressed
|
|
if shutdown_requested:
|
|
# Third Ctrl-C: last-resort forced exit
|
|
curses.endwin()
|
|
os._exit(0)
|
|
ctrl_c_pressed = True
|
|
|
|
|
|
def handle_ctrl_c():
|
|
"""Ask for confirmation on the first Ctrl-C, quit on the second one within the timeout"""
|
|
global ctrl_c_pressed, quit_confirm_time, shutdown_requested
|
|
if not ctrl_c_pressed:
|
|
return
|
|
ctrl_c_pressed = False
|
|
|
|
current_time = time.time()
|
|
if quit_confirm_time is None or (current_time - quit_confirm_time) > _QUIT_CONFIRM_TIMEOUT:
|
|
logger.warning(f"^1^8Press Ctrl-C again within {_QUIT_CONFIRM_TIMEOUT:.0f} seconds to quit^0")
|
|
quit_confirm_time = current_time
|
|
else:
|
|
logger.warning("^1^8Quittin'^0")
|
|
shutdown_requested = True
|
|
|
|
def parse_cvar_response(message, game_state, ui):
|
|
"""
|
|
Parse server cvar responses and update state
|
|
Returns True if message should be suppressed from display
|
|
"""
|
|
# Suppress command echo lines
|
|
if message.startswith('zmq RCON command') and 'from' in message:
|
|
suppress_cmds = ['status', 'roundlimit', 'qlx_serverBrandName', 'g_factoryTitle',
|
|
'mapname', 'timelimit', 'fraglimit', 'capturelimit', 'sv_maxclients']
|
|
if any(f': {cmd}' in message for cmd in suppress_cmds):
|
|
return True
|
|
|
|
# Parse cvar responses (format: "cvar_name" is:"value" default:...)
|
|
cvar_match = CVAR_RESPONSE_PATTERN.search(message)
|
|
if cvar_match:
|
|
cvar_name = cvar_match.group(1)
|
|
value = cvar_match.group(2)
|
|
|
|
if game_state.server_info.update_from_cvar(cvar_name, value):
|
|
ui.update_server_info(game_state)
|
|
return True
|
|
|
|
return False
|
|
|
|
|
|
def handle_stats_connection(message):
|
|
"""
|
|
Handle stats connection info extraction
|
|
Returns (stats_port, stats_password) or (None, None)
|
|
"""
|
|
stats_port = None
|
|
stats_password = None
|
|
|
|
# Extract stats port
|
|
if 'net_port' in message and ' is:' in message and '"net_port"' in message:
|
|
match = CVAR_RESPONSE_PATTERN.search(message)
|
|
if match:
|
|
port_str = match.group(2).strip()
|
|
digit_match = PORT_PATTERN.search(port_str)
|
|
if digit_match:
|
|
stats_port = digit_match.group(1)
|
|
logger.info(f'Got stats port: {stats_port}')
|
|
|
|
# Extract stats password
|
|
if 'zmq_stats_password' in message and ' is:' in message and '"zmq_stats_password"' in message:
|
|
match = CVAR_RESPONSE_PATTERN.search(message)
|
|
if match:
|
|
password_str = match.group(2)
|
|
password_str = strip_color_codes(password_str)
|
|
stats_password = password_str.strip()
|
|
logger.info('Got stats password')
|
|
|
|
return stats_port, stats_password
|
|
|
|
def handle_user_input(ui, rcon):
|
|
"""Read pending keys and send the commands the user submitted"""
|
|
for command in ui.process_input():
|
|
logger.info(f'Sending command: {repr(command.strip())}')
|
|
|
|
# Display command with timestamp
|
|
timestamp = time.strftime('%H:%M:%S')
|
|
ui.print_message(f"^5[^7{timestamp}^5] >>> {command.strip()}^7\n")
|
|
|
|
rcon.send_command(command)
|
|
|
|
|
|
def handle_stats_stream(stats_conn, stats_check_counter, event_parser, ui, game_state):
|
|
"""Poll and process stats stream events"""
|
|
if not stats_conn or not stats_conn.connected:
|
|
return stats_check_counter
|
|
|
|
stats_check_counter += 1
|
|
|
|
if stats_check_counter % 100 == 0:
|
|
logger.debug(f'Stats polling active (check #{stats_check_counter // 100})')
|
|
|
|
stats_msg = stats_conn.recv_message()
|
|
if stats_msg:
|
|
logger.info(f'Stats event received ({len(stats_msg)} bytes)')
|
|
|
|
# Parse game event
|
|
parsed = event_parser.parse_event(stats_msg)
|
|
if parsed:
|
|
# Format with timestamp before displaying
|
|
formatted_msg, attributes = format_message(parsed)
|
|
ui.print_message(formatted_msg)
|
|
ui.update_server_info(game_state)
|
|
|
|
return stats_check_counter
|
|
|
|
|
|
def handle_player_respawns(game_state, ui):
|
|
"""Check and revive dead players after respawn delay"""
|
|
if game_state.server_info.gametype == 'Clan Arena':
|
|
# CA: revive all players after round end
|
|
if game_state.server_info.round_end_time:
|
|
if time.time() - game_state.server_info.round_end_time >= _RESPAWN_DELAY:
|
|
game_state.server_info.dead_players.clear()
|
|
game_state.server_info.round_end_time = None
|
|
ui.update_server_info(game_state)
|
|
else:
|
|
# Other modes: revive individual players after death
|
|
current_time = time.time()
|
|
players_to_revive = [
|
|
name for name, death_time in game_state.server_info.dead_players.items()
|
|
if current_time - death_time >= _RESPAWN_DELAY
|
|
]
|
|
if players_to_revive:
|
|
for name in players_to_revive:
|
|
del game_state.server_info.dead_players[name]
|
|
ui.update_server_info(game_state)
|
|
|
|
|
|
def parse_player_events(message, game_state, ui):
|
|
"""
|
|
Parse connect, disconnect, kick, and rename messages
|
|
Returns True if message should be suppressed
|
|
"""
|
|
try:
|
|
msg = message
|
|
|
|
# Strip broadcast: print "..." wrapper with regex
|
|
broadcast_match = BROADCAST_PATTERN.match(msg)
|
|
if broadcast_match:
|
|
msg = broadcast_match.group(1)
|
|
|
|
# Strip timestamp: [HH:MM:SS] or ^3[^7HH:MM:SS^3]^7
|
|
msg = TIMESTAMP_PATTERN.sub('', msg)
|
|
msg = msg.strip()
|
|
|
|
if not msg:
|
|
return False
|
|
|
|
logger.debug(f'parse_player_events: {repr(msg)}')
|
|
|
|
# Strip color codes for matching
|
|
clean_msg = strip_color_codes(msg)
|
|
|
|
# Match connects: "NAME connected" or "NAME connected with Steam ID"
|
|
connect_match = CONNECT_PATTERN.match(clean_msg)
|
|
if connect_match:
|
|
player_name_match = CONNECT_PATTERN.match(msg)
|
|
player_name = player_name_match.group(1).strip() if player_name_match else connect_match.group(1).strip()
|
|
player_name = strip_color_codes(player_name)
|
|
|
|
logger.info(f'CONNECT: {repr(player_name)}')
|
|
game_state.player_tracker.update_team(player_name, 'SPECTATOR')
|
|
game_state.player_tracker.add_player(player_name)
|
|
ui.update_server_info(game_state)
|
|
|
|
# Only print if this is NOT the Steam ID line
|
|
if 'Steam ID' not in message:
|
|
timestamp = time.strftime('%H:%M:%S')
|
|
ui.print_message(f"^3[^7{timestamp}^3]^7 ^8{player_name} ^9^2connected\n")
|
|
|
|
return True
|
|
|
|
# Regular disconnect
|
|
disconnect_match = DISCONNECT_PATTERN.match(clean_msg)
|
|
if disconnect_match:
|
|
original_match = DISCONNECT_PATTERN.match(msg)
|
|
player_name = original_match.group(1).strip() if original_match else disconnect_match.group(1).strip()
|
|
player_name = strip_color_codes(player_name)
|
|
|
|
logger.info(f'DISCONNECT: {repr(player_name)}')
|
|
game_state.player_tracker.remove_player(player_name)
|
|
ui.update_server_info(game_state)
|
|
|
|
timestamp = time.strftime('%H:%M:%S')
|
|
ui.print_message(f"^3[^7{timestamp}^3]^7 ^8{player_name} ^9^1disconnected\n")
|
|
return True
|
|
|
|
# Kick
|
|
kick_match = KICK_PATTERN.match(clean_msg)
|
|
if kick_match:
|
|
original_match = KICK_PATTERN.match(msg)
|
|
player_name = original_match.group(1).strip() if original_match else kick_match.group(1).strip()
|
|
player_name = strip_color_codes(player_name)
|
|
|
|
logger.info(f'KICK: {repr(player_name)}')
|
|
game_state.player_tracker.remove_player(player_name)
|
|
ui.update_server_info(game_state)
|
|
|
|
timestamp = time.strftime('%H:%M:%S')
|
|
ui.print_message(f"^3[^7{timestamp}^3]^7 ^8{player_name} ^1was kicked\n")
|
|
return True
|
|
|
|
# Inactivity
|
|
inactivity_match = INACTIVITY_PATTERN.match(clean_msg)
|
|
if inactivity_match:
|
|
original_match = INACTIVITY_PATTERN.match(msg)
|
|
player_name = original_match.group(1).strip() if original_match else inactivity_match.group(1).strip()
|
|
player_name = strip_color_codes(player_name)
|
|
|
|
logger.info(f'INACTIVITY DROP: {repr(player_name)}')
|
|
game_state.player_tracker.remove_player(player_name)
|
|
ui.update_server_info(game_state)
|
|
|
|
timestamp = time.strftime('%H:%M:%S')
|
|
ui.print_message(f"^3[^7{timestamp}^3]^7 ^8{player_name}^0 ^3dropped due to inactivity^7\n")
|
|
return True
|
|
|
|
# Match renames: "OldName renamed to NewName"
|
|
rename_match = RENAME_PATTERN.match(clean_msg)
|
|
if rename_match:
|
|
# Extract from original message
|
|
original_match = RENAME_PATTERN.match(msg)
|
|
if original_match:
|
|
old_name = original_match.group(1).strip()
|
|
new_name = original_match.group(2).strip()
|
|
else:
|
|
old_name = rename_match.group(1).strip()
|
|
new_name = rename_match.group(2).strip()
|
|
|
|
# Remove color codes from both names
|
|
old_name = strip_color_codes(old_name)
|
|
new_name = strip_color_codes(new_name)
|
|
old_name = old_name.rstrip('\n\r') # Remove trailing newline
|
|
new_name = new_name.rstrip('\n\r') # Remove trailing newline
|
|
|
|
logger.info(f'RENAME: {repr(old_name)} -> {repr(new_name)}')
|
|
game_state.player_tracker.rename_player(old_name, new_name)
|
|
ui.update_server_info(game_state)
|
|
timestamp = time.strftime('%H:%M:%S')
|
|
ui.print_message(f"^3[^7{timestamp}^3]^7 ^8{old_name} ^6renamed to^7 ^8{new_name}\n")
|
|
return True
|
|
|
|
except Exception as e:
|
|
logger.error(f'Error in parse_player_events: {e}')
|
|
|
|
return False
|
|
|
|
def main_loop(screen, args):
|
|
"""
|
|
Main application loop.
|
|
Runs single-threaded: ZMQ sockets and the terminal (stdin) are polled together,
|
|
so all curses drawing happens here and never from another thread.
|
|
"""
|
|
|
|
# Setup signal handler for Ctrl+C with confirmation
|
|
signal.signal(signal.SIGINT, signal_handler)
|
|
|
|
# Set logging level
|
|
if args.verbose == 0:
|
|
logger.setLevel(args.config_log_level if args.config_log_level is not None else logging.WARNING)
|
|
elif args.verbose == 1:
|
|
logger.setLevel(logging.INFO)
|
|
else:
|
|
logger.setLevel(logging.DEBUG)
|
|
|
|
# Setup file logging for unknown events
|
|
unknown_handler = logging.FileHandler(args.unknown_log, mode='a')
|
|
unknown_formatter = logging.Formatter('%(asctime)s - %(message)s', '%Y-%m-%d %H:%M:%S')
|
|
unknown_handler.setFormatter(unknown_formatter)
|
|
unknown_json_logger.addHandler(unknown_handler)
|
|
unknown_json_logger.propagate = False
|
|
|
|
# Initialize components
|
|
game_state = GameState()
|
|
name_capture = NameCapture() # cvar/command names learned from the server for autocomplete
|
|
ui = UIManager(screen, args.title, args.max_history,
|
|
player_names_provider=lambda: game_state.player_tracker.get_player_names(),
|
|
learned_names_provider=lambda: name_capture.names)
|
|
|
|
# Setup logging to output window
|
|
log_handler = ui.setup_logging()
|
|
logger.addHandler(log_handler)
|
|
|
|
# Attach UI log handler to lib module loggers
|
|
for lib_name in ('network', 'parser', 'state', 'namelist'):
|
|
lib_logger = logging.getLogger(lib_name)
|
|
if log_handler not in lib_logger.handlers:
|
|
lib_logger.addHandler(log_handler)
|
|
lib_logger.setLevel(logger.level)
|
|
|
|
# Display startup messages
|
|
ui.print_message(f"*** QL pyCon Version {VERSION} starting ***\n")
|
|
ui.print_message(f"zmq python bindings {zmq.__version__}, libzmq version {zmq.zmq_version()}\n")
|
|
|
|
# Initialize network connections
|
|
password_display = '***' if args.password else '(none)'
|
|
ui.print_message(f"Connecting with host={args.host} password={password_display}\n")
|
|
rcon = RconConnection(args.host, args.password, args.identity)
|
|
rcon.connect()
|
|
|
|
# One poller wakes the loop on RCON data, connection events or a key press
|
|
poller = zmq.Poller()
|
|
poller.register(rcon.socket, zmq.POLLIN)
|
|
poller.register(rcon.monitor, zmq.POLLIN)
|
|
poller.register(sys.stdin, zmq.POLLIN)
|
|
|
|
stats_conn = None
|
|
stats_port = None
|
|
stats_password = None
|
|
stats_check_counter = 0
|
|
|
|
# Timer refresh tracking (update UI once per second)
|
|
last_ui_update = 0
|
|
|
|
# Setup JSON logging if requested
|
|
json_logger = None
|
|
if args.json_log:
|
|
json_handler = logging.FileHandler(args.json_log, mode='a')
|
|
json_formatter = logging.Formatter('%(asctime)s - %(message)s', '%Y-%m-%d %H:%M:%S')
|
|
json_handler.setFormatter(json_formatter)
|
|
all_json_logger.addHandler(json_handler)
|
|
all_json_logger.propagate = False
|
|
json_logger = all_json_logger
|
|
|
|
# Create event parser
|
|
event_parser = EventParser(game_state, json_logger, unknown_json_logger)
|
|
|
|
# Main event loop with resource cleanup
|
|
try:
|
|
while not shutdown_requested:
|
|
# Wait for RCON data, a connection event, a stats event or a key press
|
|
ready = dict(poller.poll(POLL_TIMEOUT))
|
|
|
|
handle_ctrl_c()
|
|
|
|
# Check monitor for connection events
|
|
monitor_event = rcon.check_monitor()
|
|
if monitor_event and monitor_event[0] == zmq.EVENT_CONNECTED:
|
|
ui.print_message("Connected to server\n")
|
|
rcon.send_command(b'register')
|
|
logger.info('Registration message sent')
|
|
|
|
ui.print_message("Requesting connection info...\n")
|
|
rcon.send_command(b'zmq_stats_password')
|
|
rcon.send_command(b'net_port')
|
|
name_capture.request(rcon)
|
|
elif monitor_event and monitor_event[0] == zmq.EVENT_DISCONNECTED:
|
|
timestamp = time.strftime('%H:%M:%S')
|
|
ui.print_message(f"^3[^7{timestamp}^3] ^8^1Disconnected from server - waiting for reconnect...^0^7\n")
|
|
if stats_conn is not None:
|
|
poller.unregister(stats_conn.socket)
|
|
stats_conn.close()
|
|
stats_conn = None
|
|
stats_port = None
|
|
stats_password = None
|
|
|
|
# Handle user input
|
|
handle_user_input(ui, rcon)
|
|
|
|
# Poll stats stream if connected
|
|
stats_check_counter = handle_stats_stream(stats_conn, stats_check_counter, event_parser, ui, game_state)
|
|
|
|
# Check if we need to revive players
|
|
handle_player_respawns(game_state, ui)
|
|
|
|
# Process RCON messages
|
|
if rcon.socket in ready:
|
|
logger.debug('Socket has data available')
|
|
msg_count = 0
|
|
|
|
while True:
|
|
message = rcon.recv_message()
|
|
if message is None:
|
|
if msg_count > 0:
|
|
logger.debug(f'Read {msg_count} message(s)')
|
|
break
|
|
|
|
msg_count += 1
|
|
|
|
if len(message) == 0:
|
|
logger.debug('Received empty message (keepalive)')
|
|
continue
|
|
|
|
logger.debug(f'Received message ({len(message)} bytes): {repr(message[:100])}')
|
|
|
|
# Check for player connect/disconnect/rename events
|
|
if parse_player_events(message, game_state, ui):
|
|
continue
|
|
|
|
if 'Game Initialization' in message:
|
|
logger.info('Game initialization detected - refreshing server info')
|
|
|
|
timestamp = time.strftime('%H:%M:%S')
|
|
ui.print_message(f"^3[^7{timestamp}^3] ^8^3Game initialized - Refreshing server info^0^7\n")
|
|
|
|
rcon.send_command(b'qlx_serverBrandName')
|
|
rcon.send_command(b'g_factoryTitle')
|
|
rcon.send_command(b'mapname')
|
|
rcon.send_command(b'timelimit')
|
|
rcon.send_command(b'fraglimit')
|
|
rcon.send_command(b'roundlimit')
|
|
rcon.send_command(b'capturelimit')
|
|
rcon.send_command(b'sv_maxclients')
|
|
|
|
# Clear player dict since map changed
|
|
game_state.server_info.players = {}
|
|
game_state.player_tracker.player_teams = {}
|
|
game_state.server_info.reset_round_scores()
|
|
game_state.server_info.dead_players.clear()
|
|
ui.update_server_info(game_state)
|
|
|
|
# Try to parse as cvar response
|
|
if parse_cvar_response(message, game_state, ui):
|
|
logger.debug('Suppressed cvar response')
|
|
continue
|
|
|
|
# Check for stats connection info
|
|
port, password = handle_stats_connection(message)
|
|
if port:
|
|
stats_port = port
|
|
if password:
|
|
stats_password = password
|
|
|
|
# Connect to stats if we have both credentials
|
|
if stats_port and stats_password and stats_conn is None:
|
|
try:
|
|
ui.print_message("Connecting to stats stream...\n")
|
|
host_ip = args.host.split('//')[1].split(':')[0]
|
|
|
|
stats_conn = StatsConnection(host_ip, stats_port, stats_password)
|
|
stats_conn.connect()
|
|
poller.register(stats_conn.socket, zmq.POLLIN)
|
|
|
|
ui.print_message("Stats stream connected - ready for game events\n")
|
|
|
|
# Request initial server info
|
|
logger.info('Sending initial server info queries')
|
|
rcon.send_command(b'qlx_serverBrandName')
|
|
rcon.send_command(b'g_factoryTitle')
|
|
rcon.send_command(b'mapname')
|
|
rcon.send_command(b'timelimit')
|
|
rcon.send_command(b'fraglimit')
|
|
rcon.send_command(b'roundlimit')
|
|
rcon.send_command(b'capturelimit')
|
|
rcon.send_command(b'sv_maxclients')
|
|
|
|
if args.json_log:
|
|
ui.print_message(f"*** JSON capture enabled: {args.json_log} ***\n")
|
|
|
|
except Exception as e:
|
|
timestamp = time.strftime('%H:%M:%S')
|
|
ui.print_message(f"^1[^7{timestamp}^1] Error: Stats connection failed: {e}^7\n")
|
|
logger.error(f'Stats connection failed: {e}')
|
|
|
|
# Hide the cvarlist/cmdlist output while learning names from it.
|
|
# Runs after the cvar handling above so the stats credentials and
|
|
# server info responses are never swallowed.
|
|
if name_capture.handle(message):
|
|
continue
|
|
|
|
# Try to parse as game event
|
|
parsed_event = event_parser.parse_event(message)
|
|
if parsed_event:
|
|
ui.print_message(parsed_event)
|
|
continue
|
|
|
|
# Check if it looks like JSON but wasn't parsed
|
|
stripped = message.strip()
|
|
if stripped and stripped[0] in ('{', '['):
|
|
logger.debug('Unparsed JSON event')
|
|
continue
|
|
|
|
# Try powerup message formatting
|
|
powerup_msg = format_powerup_message(message, game_state.player_tracker)
|
|
if powerup_msg:
|
|
ui.print_message(powerup_msg)
|
|
continue
|
|
|
|
# Filter bot debug messages in default mode
|
|
is_bot_debug = (' entered ' in message and
|
|
any(x in message for x in [' seek ', ' battle ', ' chase', ' fight']))
|
|
if is_bot_debug and args.verbose == 0:
|
|
logger.debug(f'Filtered bot debug: {message[:50]}')
|
|
continue
|
|
|
|
# Check if it's a chat message
|
|
if ':' in message and not message.startswith(('print', 'broadcast', 'zmq')):
|
|
message = format_chat_message(message, game_state.player_tracker)
|
|
|
|
# Format and display message
|
|
formatted_msg, attributes = format_message(message)
|
|
ui.print_message(formatted_msg)
|
|
|
|
# Update server info panel every second (for live timer display)
|
|
current_time = time.time()
|
|
if current_time - last_ui_update >= 1.0:
|
|
ui.update_server_info(game_state)
|
|
last_ui_update = current_time
|
|
|
|
finally:
|
|
# Clean up resources
|
|
logger.info("Shutting down...")
|
|
if rcon:
|
|
logger.debug("Closing RCON connection")
|
|
rcon.close()
|
|
if stats_conn:
|
|
logger.debug("Closing stats connection")
|
|
stats_conn.close()
|
|
logger.info("Shutdown complete")
|
|
|
|
if __name__ == '__main__':
|
|
# Load config
|
|
config = ConfigLoader()
|
|
config.load()
|
|
|
|
# Read settings from config ([logging]/[ui]/[behavior])
|
|
_QUIT_CONFIRM_TIMEOUT = config.get_quit_timeout(QUIT_CONFIRM_TIMEOUT)
|
|
_RESPAWN_DELAY = config.get_respawn_delay(RESPAWN_DELAY)
|
|
|
|
# Parse arguments
|
|
parser = argparse.ArgumentParser(description='Quake Live Python Console')
|
|
parser.add_argument('server', nargs='?', default=None,
|
|
help='Named server from qlpycon.conf (e.g. ffa, duel)')
|
|
parser.add_argument('--host', default=None,
|
|
help='ZMQ URI to connect to (e.g. tcp://1.2.3.4:28960)')
|
|
parser.add_argument('--password', default=None,
|
|
help='RCON password')
|
|
parser.add_argument('--list', action='store_true',
|
|
help='List configured servers and exit')
|
|
parser.add_argument('--identity', default=uuid.uuid1().hex,
|
|
help='Socket identity (random UUID by default)')
|
|
parser.add_argument('-v', '--verbose', action='count', default=0,
|
|
help='Increase verbosity (-v INFO, -vv DEBUG)')
|
|
parser.add_argument('--unknown-log', default='unknown_events.log',
|
|
help='File to log unknown JSON events')
|
|
parser.add_argument('-j', '--json', dest='json_log', default=None,
|
|
help='File to log all JSON events')
|
|
args = parser.parse_args()
|
|
|
|
# Apply config settings to args
|
|
args.config_log_level = config.get_log_level()
|
|
args.max_history = config.get_max_history(MAX_COMMAND_HISTORY)
|
|
|
|
# Handle --list
|
|
if args.list:
|
|
servers = config.get_servers()
|
|
if not servers:
|
|
print('No servers configured in qlpycon.conf')
|
|
else:
|
|
print('Configured servers:')
|
|
for name, host in servers.items():
|
|
print(f' {name:<12} {host}')
|
|
sys.exit(0)
|
|
|
|
# No server name and no --host: let the user pick from the configured servers
|
|
if args.server is None and args.host is None:
|
|
servers = config.get_servers()
|
|
if servers:
|
|
args.server = curses.wrapper(select_server, servers)
|
|
if args.server is None:
|
|
sys.exit(0)
|
|
|
|
# Resolve host and password
|
|
if args.server:
|
|
host, password = config.get_server(args.server)
|
|
if host is None:
|
|
print(f"Error: server '{args.server}' not found in qlpycon.conf")
|
|
print("Use 'qlpycon --list' to see configured servers.")
|
|
sys.exit(1)
|
|
args.host = args.host or host
|
|
args.password = args.password or password
|
|
args.title = f'Quake Live PyCon: {args.server} ({args.host})'
|
|
else:
|
|
args.host = args.host or config.get_host() or DEFAULT_HOST
|
|
if not args.host.startswith('tcp://'):
|
|
args.host = f'tcp://{args.host}'
|
|
args.password = args.password or config.get_password()
|
|
args.title = f'Quake Live PyCon: {args.host}'
|
|
|
|
curses.wrapper(main_loop, args)
|