f75adac97a
- Add pre-compiled regex pattern for color codes - Add SPECIAL_CHAR constant (chr(25)) for clarity - Move time import to module level in parser - Centralize color code stripping via strip_color_codes() - Make qlpycon.bash fully configurable (workdir, serverip) - Add validation checks for workdir and venv - Fix port numbers in help text (28960-28969)
360 lines
14 KiB
Python
360 lines
14 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
JSON game event parsing for QLPyCon
|
|
Parses events from Quake Live stats stream
|
|
"""
|
|
|
|
import json
|
|
import logging
|
|
import time
|
|
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):
|
|
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 = int(killer.get('HEALTH', 0))
|
|
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})")
|
|
|
|
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"
|