#!/usr/bin/env python3 """ JSON game event parsing for QLPyCon Parses events from Quake Live stats stream """ import json import logging from config import WEAPON_NAMES, WEAPON_KILL_NAMES, DEATH_MESSAGES from formatter import get_team_prefix, strip_color_codes logger = logging.getLogger('parser') def calculate_weapon_accuracies(weapon_data): """Calculate accuracy percentages for all weapons""" accuracies = {} for weapon, stats in weapon_data.items(): shots_fired = int(stats.get('S', 0)) shots_hit = int(stats.get('H', 0)) accuracy = shots_hit / shots_fired if shots_fired > 0 else 0 accuracies[weapon] = accuracy return accuracies class EventParser: """Parses JSON game events into formatted messages""" def __init__(self, game_state, json_logger=None, unknown_logger=None): self.game_state = game_state self.json_logger = json_logger self.unknown_logger = unknown_logger def parse_event(self, message): """ Parse JSON event and return formatted message string Returns None if event should not be displayed """ try: event = json.loads(message) # Log all JSON if logger is configured if self.json_logger: self.json_logger.info('JSON Event received:') self.json_logger.info(json.dumps(event, indent=2)) self.json_logger.info('---') if 'TYPE' not in event or 'DATA' not in event: logger.debug('JSON missing TYPE or DATA') return None event_type = event['TYPE'] data = event['DATA'] if 'WARMUP' in data: self.game_state.server_info.warmup = data['WARMUP'] # Route to appropriate handler handler_map = { 'PLAYER_SWITCHTEAM': self._handle_switchteam, 'PLAYER_DEATH': self._handle_death, 'PLAYER_KILL': self._handle_death, # Same handler 'PLAYER_MEDAL': self._handle_medal, 'MATCH_STARTED': self._handle_match_started, 'MATCH_REPORT': self._handle_match_report, 'PLAYER_STATS': self._handle_player_stats, 'PLAYER_CONNECT': lambda d: None, 'PLAYER_DISCONNECT': lambda d: None, 'ROUND_OVER': lambda d: None, } handler = handler_map.get(event_type) if handler: return handler(data) else: # Unknown event logger.debug(f'Unknown event type: {event_type}') if self.unknown_logger: self.unknown_logger.info(f'Unknown event type: {event_type}') self.unknown_logger.info(f'Full JSON: {json.dumps(event, indent=2)}') return None except json.JSONDecodeError as e: logger.debug(f'JSON decode error: {e}') return None except (KeyError, TypeError) as e: logger.debug(f'Error parsing event: {e}') return None def _handle_switchteam(self, data): """Handle PLAYER_SWITCHTEAM event""" if 'KILLER' not in data: return None killer = data['KILLER'] name = killer.get('NAME', 'Unknown') team = killer.get('TEAM', '') old_team = killer.get('OLD_TEAM', '') # Update player team self.game_state.player_tracker.update_team(name, team) self.game_state.player_tracker.add_player(name) if team == old_team: return None warmup = " ^0^3(Warmup)^9" if data.get('WARMUP', False) else "" team_messages = { 'FREE': ' ^7joined the ^0fight^9', 'SPECTATOR': ' ^7joined the ^3Spectators^7', 'RED': ' ^7joined the ^1RED Team^7', 'BLUE': ' ^7joined the ^4BLUE Team^7' } old_team_messages = { 'FREE': 'the ^0fight^9', 'SPECTATOR': 'the ^3Spectators^7', 'RED': '^7the ^1RED Team^7', 'BLUE': '^7the ^4BLUE Team^7' } team_msg = team_messages.get(team, f' ^7joined team {team}^7') old_team_msg = old_team_messages.get(old_team, f'team {old_team}') team_prefix = get_team_prefix(name, self.game_state.player_tracker) return f"{team_prefix}^0{name}^9{team_msg} from {old_team_msg}{warmup}\n" def _handle_death(self, data): """Handle PLAYER_DEATH and PLAYER_KILL events""" if 'VICTIM' not in data: return None victim = data['VICTIM'] victim_name = victim.get('NAME', 'Unknown') # Check for duplicate time_val = data.get('TIME', 0) killer_name = data.get('KILLER', {}).get('NAME', '') if data.get('KILLER') else '' if self.game_state.event_deduplicator.is_duplicate('PLAYER_DEATH', time_val, killer_name, victim_name): return None # Update victim team if 'TEAM' in victim: self.game_state.player_tracker.update_team(victim_name, victim['TEAM']) self.game_state.player_tracker.add_player(victim_name) victim_prefix = get_team_prefix(victim_name, self.game_state.player_tracker) warmup = " ^0^3(Warmup)^9" if data.get('WARMUP', False) else "" # Environmental death (no killer) if 'KILLER' not in data or not data['KILLER']: mod = data.get('MOD', 'UNKNOWN') msg_template = DEATH_MESSAGES.get(mod, "%s^0%s^9 ^1DIED FROM %s^7") if mod in DEATH_MESSAGES: msg = msg_template % (victim_prefix, victim_name) else: msg = msg_template % (victim_prefix, victim_name, mod) return f"{msg}{warmup}\n" # Player killed by another player killer = data['KILLER'] killer_name = killer.get('NAME', 'Unknown') # Update killer team if 'TEAM' in killer: self.game_state.player_tracker.update_team(killer_name, killer['TEAM']) self.game_state.player_tracker.add_player(killer_name) killer_prefix = get_team_prefix(killer_name, self.game_state.player_tracker) # Suicide if killer_name == victim_name: weapon = killer.get('WEAPON', 'OTHER_WEAPON') if weapon != 'OTHER_WEAPON': weapon_name = WEAPON_NAMES.get(weapon, weapon) return f"{killer_prefix}^0{killer_name}^9 ^7committed suicide with the ^7{weapon_name}{warmup}\n" return None # Regular kill weapon = killer.get('WEAPON', 'UNKNOWN') weapon_name = WEAPON_KILL_NAMES.get(weapon, f'the {weapon}') return f"{killer_prefix}^0{killer_name}^9 ^7fragged^7 {victim_prefix}^0{victim_name}^9 ^7with {weapon_name}{warmup}\n" def _handle_medal(self, data): """Handle PLAYER_MEDAL event""" name = data.get('NAME', 'Unknown') medal = data.get('MEDAL', 'UNKNOWN') warmup = " ^0^3(Warmup)^7^9" if data.get('WARMUP', False) else "" team_prefix = get_team_prefix(name, self.game_state.player_tracker) return f"{team_prefix}^0{name}^9 ^7got a medal: ^6{medal}{warmup}\n" def _handle_match_started(self, data): """Handle MATCH_STARTED event""" if self.game_state.server_info.is_team_mode(): return None players = [] for player in data.get('PLAYERS', []): name = player.get('NAME', 'Unknown') players.append(name) if players: formatted = "^9 vs. ^0".join(players) return f"^0^3Match has started - ^0^7{formatted}\n" return None def _handle_match_report(self, data): """Handle MATCH_REPORT event""" if not self.game_state.server_info.is_team_mode(): return None red_score = int(data.get('TSCORE0', '0')) blue_score = int(data.get('TSCORE1', '0')) if red_score > blue_score: return f"^1RED TEAM ^7WINS by a score of {red_score} to {blue_score}\n" elif blue_score > red_score: return f"^4BLUE TEAM ^7WINS by a score of {blue_score} to {red_score}\n" else: return f"^7The match is a TIE with a score of {red_score} to {blue_score}\n" def _handle_player_stats(self, data): """Handle PLAYER_STATS event""" name = data.get('NAME', 'Unknown') team_prefix = get_team_prefix(name, self.game_state.player_tracker) kills = int(data.get('KILLS', '0')) deaths = int(data.get('DEATHS', '0')) weapon_data = data.get('WEAPONS', {}) accuracies = calculate_weapon_accuracies(weapon_data) if not accuracies: return None best_weapon = max(accuracies, key=accuracies.get) best_accuracy = accuracies[best_weapon] * 100 weapon_stats = weapon_data.get(best_weapon, {}) best_weapon_kills = int(weapon_stats.get('K', 0)) weapon_name = WEAPON_NAMES.get(best_weapon, best_weapon) return f"{team_prefix}^0^7{name}^9^7 K/D: {kills}/{deaths} | Best Weapon: {weapon_name} - Acc: {best_accuracy:.2f}% - Kills: {best_weapon_kills}\n"