#!/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, STATUS_OUTPUT_WINDOW 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+(.+?)$') STATUS_HEAD_PATTERN = re.compile(r'^\s*\d+\s+-?\d+\s+(bot|\d+)\s+\S', re.MULTILINE) STATUS_MAP_PATTERN = re.compile(r'^map:\s+\S+\s*$', re.MULTILINE) STEAMID_PATTERN = re.compile(r'\b\d{16,}\b') # 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 """ # 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 is_command_echo(message): """True for the server's echo lines announcing that an RCON client sent a command""" return message.startswith('zmq RCON command') def is_command_output(message): """ Detect console output produced by RCON commands: cvar dumps and status tables. These are hidden; the 'zmq RCON command' echo lines still show which client ran what. Shape checks run on color-stripped text because the server sprinkles Quake color codes into status rows. Rows also arrive broadcast-wrapped or split across several messages, so the checks are containment-based and a row fragment (head without steamid, tail without row prefix) is caught too. JSON stats events are exempt: they contain 16+ digit STEAM_IDs and are handled by the event parser. """ clean = strip_color_codes(message) stripped = clean.lstrip() if stripped[:1] in ('{', '['): return False if CVAR_RESPONSE_PATTERN.search(clean): return True if 'num score ping' in clean or '--- -----' in clean: return True if STEAMID_PATTERN.search(clean) or STATUS_HEAD_PATTERN.search(clean): return True if STATUS_MAP_PATTERN.search(clean): return True return False CHAT_SHAPE_PATTERN = re.compile(r'^[\w(][^:]*:\s') def is_real_content(message): """ True for messages that are definitely not status table debris: JSON, chat-shaped lines and game initialization. Used to end the status suppression window early; everything else inside the window is treated as table debris, because the server splits rows into arbitrary chunks (names, flag columns, '0 16384', bare steamids) that no shape check can distinguish from real content. The status table's own 'map:' line is chat-shaped and explicitly excluded, or it would end the window on the first chunk. """ clean = strip_color_codes(message).strip() if not clean: return False if clean[:1] in ('{', '['): return True if clean.startswith('zmq'): return False if clean.startswith('map:'): return False if clean.startswith('broadcast:'): clean = clean[10:].lstrip() if clean.startswith('print "'): clean = clean[7:] if clean.endswith('"'): clean = clean[:-1] if 'Game Initialization' in clean: return True if CHAT_SHAPE_PATTERN.match(clean): 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 (parse_event returns a display-ready string) parsed = event_parser.parse_event(stats_msg) if parsed: ui.print_message(parsed) 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)' timestamp = time.strftime('%H:%M:%S') ui.print_message(f"^3[^7{timestamp}^3]^7 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 status_suppress_until = 0.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: timestamp = time.strftime('%H:%M:%S') ui.print_message(f"^3[^7{timestamp}^3]^7 Connected to server ({monitor_event[2]}) " f"as rcon client {rcon.identity}\n") rcon.send_command(b'register') logger.info('Registration message sent') timestamp = time.strftime('%H:%M:%S') ui.print_message(f"^3[^7{timestamp}^3]^7 Requesting connection info...\n") rcon.send_command(b'zmq_stats_password') rcon.send_command(b'net_port') if args.learn_names: 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 ({monitor_event[2]}) " f"as rcon client {rcon.identity} - 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 # Command echoes only show at -v or higher verbosity; a # status echo opens a window that hides its chunked output if is_command_echo(message): if message.rstrip().endswith(': status'): status_suppress_until = time.time() + STATUS_OUTPUT_WINDOW if args.verbose == 0: logger.debug('Suppressed RCON command echo') continue # Suppress everything after a 'status' echo until real # content arrives (chat, JSON, game init) or the window closes if status_suppress_until: if time.time() >= status_suppress_until: status_suppress_until = 0.0 elif is_real_content(message): status_suppress_until = 0.0 else: logger.debug('Suppressed status output chunk') 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: timestamp = time.strftime('%H:%M:%S') ui.print_message(f"^3[^7{timestamp}^3]^7 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) timestamp = time.strftime('%H:%M:%S') ui.print_message(f"^3[^7{timestamp}^3]^7 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 output of RCON commands (cvar dumps, status tables) if is_command_output(message): logger.debug('Suppressed RCON command output') continue # 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, _ = 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) args.learn_names = config.get_bool('behavior', 'learn_names', False) # 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)