#!/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': self._handle_round_over, } 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 = " ^8^3(Warmup)^0" if data.get('WARMUP', False) else "" team_messages = { 'FREE': ' ^7joined the ^8fight^0', 'SPECTATOR': ' ^7joined the ^3Spectators^7', 'RED': ' ^7joined the ^1RED Team^7', 'BLUE': ' ^7joined the ^4BLUE Team^7' } old_team_messages = { 'FREE': 'the ^8fight^0', '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}^8{name}^0{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) # Mark as dead if not data.get('WARMUP', False): import time self.game_state.server_info.dead_players[victim_name] = time.time() victim_prefix = get_team_prefix(victim_name, self.game_state.player_tracker) warmup = " ^8^3(Warmup)^0" if data.get('WARMUP', False) else "" score_prefix = "" # Environmental death (no killer) if 'KILLER' not in data or not data['KILLER']: # -1 for environmental death if not data.get('WARMUP', False): self.game_state.player_tracker.update_score(victim_name, -1) score_prefix = "^8^1[-1]^7^0 " mod = data.get('MOD', 'UNKNOWN') msg_template = DEATH_MESSAGES.get(mod, "%s^8%s^0 ^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"{score_prefix}{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: # -1 for suicide if not data.get('WARMUP', False): self.game_state.player_tracker.update_score(victim_name, -1) score_prefix = "^8^1[-1]^7^0 " weapon = killer.get('WEAPON', 'OTHER_WEAPON') if weapon == 'ROCKET': return f"{score_prefix}{killer_prefix}^8{killer_name}^0 ^7blew herself up.{warmup}\n" elif weapon == 'GRENADE': return f"{score_prefix}{killer_prefix}^8{killer_name}^0 ^7tripped on her own grenade.{warmup}\n" elif weapon == 'PLASMA': return f"{score_prefix}{killer_prefix}^8{killer_name}^0 ^7melted herself.{warmup}\n" else: weapon_name = WEAPON_NAMES.get(weapon, weapon) return f"{score_prefix}{killer_prefix}^8{killer_name}^0 ^7committed suicide with the ^7{weapon_name}{warmup}\n" return None # Regular kill: +1 for killer if not data.get('WARMUP', False): self.game_state.player_tracker.update_score(killer_name, 1) score_prefix = "^8^2[+1]^7^0 " else: score_prefix = "" weapon = killer.get('WEAPON', 'UNKNOWN') weapon_name = WEAPON_KILL_NAMES.get(weapon, f'the {weapon}') return f"{score_prefix}{killer_prefix}^8{killer_name}^0 ^7fragged^7 {victim_prefix}^8{victim_name}^0 ^7with {weapon_name}{warmup}\n" def _handle_round_over(self, data): """Handle ROUND_OVER events for CA""" team_won = data.get('TEAM_WON') round_num = data.get('ROUND', 0) if team_won == 'RED': self.game_state.server_info.red_rounds += 1 logger.info(f"Round {round_num}: RED wins (RED: {self.game_state.server_info.red_rounds}, BLUE: {self.game_state.server_info.blue_rounds})") elif team_won == 'BLUE': self.game_state.server_info.blue_rounds += 1 logger.info(f"Round {round_num}: BLUE wins (RED: {self.game_state.server_info.red_rounds}, BLUE: {self.game_state.server_info.blue_rounds})") import time self.game_state.server_info.round_end_time = time.time() return None # Don't display in chat def _handle_medal(self, data): """Handle PLAYER_MEDAL event""" name = data.get('NAME', 'Unknown') medal = data.get('MEDAL', 'UNKNOWN') warmup = " ^8^3(Warmup)^7^0" if data.get('WARMUP', False) else "" team_prefix = get_team_prefix(name, self.game_state.player_tracker) # RED Medals (^1) if medal in ["FIRSTFRAG", "HUMILIATION", "REVENGE"]: return f"{team_prefix}^8{name}^0 ^7got a medal: ^8^9^1{medal}^0{warmup}\n" # GREEN Medals (^2) elif medal in ["MIDAIR", "PERFECT"]: return f"{team_prefix}^8{name}^0 ^7got a medal: ^8^9^2{medal}^0{warmup}\n" # YELLOW Medals (^3) elif medal in ["EXCELLENT", "HEADSHOT", "RAMPAGE"]: return f"{team_prefix}^8{name}^0 ^7got a medal: ^8^9^3{medal}^0{warmup}\n" # BLUE Medals (^4) elif medal in ["ASSIST", "DEFENSE", "QUADGOD"]: return f"{team_prefix}^8{name}^0 ^7got a medal: ^8^9^4{medal}^0{warmup}\n" # CYAN Medals (^5) elif medal in ["CAPTURE", "COMBOKILL", "IMPRESSIVE"]: return f"{team_prefix}^8{name}^0 ^7got a medal: ^8^9^5{medal}^0{warmup}\n" # PINK Medals (^6) elif medal in ["ACCURACY", "PERFORATED"]: return f"{team_prefix}^8{name}^0 ^7got a medal: ^8^9^6{medal}^0{warmup}\n" else: return f"{team_prefix}^8{name}^0 ^7got a medal: ^8^9^7{medal}^0{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 = "^0 vs. ^8".join(players) return f"^8^3Match has started - ^8^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"^8^5[PLAYER STATS]^7^0 {team_prefix}^8^7{name}^0^7 K/D: {kills}/{deaths} | Best Weapon: {weapon_name} - Acc: {best_accuracy:.2f}% - Kills: {best_weapon_kills}\n"