#!/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""" # Get Match Time if 'TIME' in data: self.game_state.server_info.match_time = int(data['TIME']) 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"^8^5[SWITCH]^7 {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""" # Get Match Time if 'TIME' in data: self.game_state.server_info.match_time = int(data['TIME']) 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}') hp_left = killer.get('HEALTH', '0 HP') hp_left_colored = "" if hp_left <= 0: # from the grave hp_left_colored = f"^8^5From the Grave^0" elif hp_left < 25: # red hp_left_colored = f"^8^1{hp_left}^0 ^7HP" elif hp_left < 80: # yellow hp_left_colored = f"^8^3{hp_left}^0 ^7HP" elif hp_left < 126: # white hp_left_colored = f"^8^7{hp_left}^0 ^7HP" else: # green hp_left_colored = f"^8^2{hp_left}^0 ^7HP" return f"{score_prefix}{killer_prefix}^8{killer_name}^0 ^7fragged^7 {victim_prefix}^8{victim_name}^0 ^7with {weapon_name}^0 ^7({hp_left_colored}^7){warmup}\n" def _handle_round_over(self, data): """Handle ROUND_OVER events for CA""" # Get Match Time if 'TIME' in data: self.game_state.server_info.match_time = int(data['TIME']) 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""" # Get Match Time if 'TIME' in data: self.game_state.server_info.match_time = int(data['TIME']) name = data.get('NAME', 'Unknown') medal = data.get('MEDAL', 'UNKNOWN') warmup = " ^8^3(Warmup)^7^0" if data.get('WARMUP', False) else "" medal_prefix = "^8^6[MEDAL]^7^0 " team_prefix = get_team_prefix(name, self.game_state.player_tracker) # RED Medals (^1) if medal in ["FIRSTFRAG", "HUMILIATION", "REVENGE"]: return f"{medal_prefix}{team_prefix}^8{name}^0 ^7got ^8^1{medal}^0{warmup}\n" # GREEN Medals (^2) elif medal in ["MIDAIR", "PERFECT"]: return f"{medal_prefix}{team_prefix}^8{name}^0 ^7got ^8^2{medal}^0{warmup}\n" # YELLOW Medals (^3) elif medal in ["EXCELLENT", "HEADSHOT", "RAMPAGE"]: return f"{medal_prefix}{team_prefix}^8{name}^0 ^7got ^8^3{medal}^0{warmup}\n" # BLUE Medals (^4) elif medal in ["ASSIST", "DEFENSE", "QUADGOD"]: return f"{medal_prefix}{team_prefix}^8{name}^0 ^7got ^8^4{medal}^0{warmup}\n" # CYAN Medals (^5) elif medal in ["CAPTURE", "COMBOKILL", "IMPRESSIVE"]: return f"{medal_prefix}{team_prefix}^8{name}^0 ^7got ^8^5{medal}^0{warmup}\n" # PINK Medals (^6) elif medal in ["ACCURACY", "PERFORATED"]: return f"{medal_prefix}{team_prefix}^8{name}^0 ^7got ^8^6{medal}^0{warmup}\n" else: return f"{medal_prefix}{team_prefix}^8{name}^0 ^7got ^8^7{medal}^0{warmup}\n" def _handle_match_started(self, data): """Handle MATCH_STARTED event""" # Get Match Time if 'TIME' in data: self.game_state.server_info.match_time = int(data['TIME']) if self.game_state.server_info.is_team_mode(): return f"^8^2[GAME ON]^0 ^7Match has started - ^1^8RED ^0^7vs. ^4^8BLUE\n" 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^2[GAME ON]^0 ^7Match has started - ^8^7{formatted}\n" return None def _handle_match_report(self, data): """Handle MATCH_REPORT event""" # Get Match Time if 'TIME' in data: self.game_state.server_info.match_time = int(data['TIME']) 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')) report_prefix = "^8^1[GAME OVER]" if red_score > blue_score: return f"{report_prefix} ^7The ^1RED TEAM ^7WINS^0 by a score of ^8^1{red_score} ^0^7to ^8^4{blue_score}\n" elif blue_score > red_score: return f"{report_prefix} ^7The ^4BLUE TEAM ^7WINS^0 by a score of ^8^4{blue_score} ^0^7to ^8^1{red_score}\n" else: return f"{report_prefix} ^7The match is a TIE^0 with a score of ^8^1{red_score} ^0^7to ^8^4{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"