From 3e6ef3546b151d4f0e4e2cad35dc937ef20894f8 Mon Sep 17 00:00:00 2001 From: devRaGonSa Date: Tue, 19 May 2026 12:24:39 +0200 Subject: [PATCH] feat: add rcon admin log parser --- backend/app/rcon_admin_log_parser.py | 248 ++++++++++++++++++++ backend/tests/test_rcon_admin_log_parser.py | 107 +++++++++ 2 files changed, 355 insertions(+) create mode 100644 backend/app/rcon_admin_log_parser.py create mode 100644 backend/tests/test_rcon_admin_log_parser.py diff --git a/backend/app/rcon_admin_log_parser.py b/backend/app/rcon_admin_log_parser.py new file mode 100644 index 0000000..6a5e433 --- /dev/null +++ b/backend/app/rcon_admin_log_parser.py @@ -0,0 +1,248 @@ +"""Parser for Hell Let Loose RCON admin log messages.""" + +from __future__ import annotations + +import re +from dataclasses import asdict, dataclass +from typing import Literal + + +RconAdminLogEventType = Literal[ + "match_start", + "match_end", + "kill", + "team_switch", + "connected", + "disconnected", + "chat", + "kick", + "ban", + "message", + "unknown", +] + + +_PREFIX_RE = re.compile( + r"^\[(?P.+?)\s+\((?P\d+)\)\]\s+(?P.*)$", + re.DOTALL, +) + +MATCH_START_RE = re.compile( + r"^MATCH START\s+(?P.+?)\s+(?P[A-Za-z]+)\s*$", + re.DOTALL, +) + +MATCH_END_RE = re.compile( + r"^MATCH ENDED\s+`(?P.+?)`\s+ALLIED\s+\((?P\d+)\s*-\s*(?P\d+)\)\s+AXIS\s*$", + re.DOTALL, +) + +KILL_RE = re.compile( + r"^KILL:\s+" + r"(?P.+?)" + r"\((?PAllies|Axis|None)/(?P[^)]*)\)" + r"\s+->\s+" + r"(?P.+?)" + r"\((?PAllies|Axis|None)/(?P[^)]*)\)" + r"\s+with\s+(?P.+?)\s*$", + re.DOTALL, +) + +TEAM_SWITCH_RE = re.compile( + r"^TEAMSWITCH\s+(?P.+?)\s+\((?P[^>]*)\s+>\s+(?P[^)]*)\)\s*$", + re.DOTALL, +) + +CONNECTED_RE = re.compile( + r"^CONNECTED\s+(?P.+?)\s+\((?P[^)]*)\)\s*$", + re.DOTALL, +) + +DISCONNECTED_RE = re.compile( + r"^DISCONNECTED\s+(?P.+?)\s+\((?P[^)]*)\)\s*$", + re.DOTALL, +) + +CHAT_RE = re.compile( + r"^CHAT\[(?P[^\]]+)\]\[(?P.+?)\((?PAllies|Axis|None)/(?P[^)]*)\)\]:\s*(?P.*)$", + re.DOTALL, +) + +KICK_RE = re.compile( + r"^KICK:\s+\[(?P.+?)\]\s+has been kicked\.\s+\[(?P.*)\]\s*$", + re.DOTALL, +) + +MESSAGE_RE = re.compile( + r"^MESSAGE:\s+player\s+\[(?P.+?)\((?P[^)]*)\)\],\s+content\s+\[(?P.*)\]\s*$", + re.DOTALL, +) + + +@dataclass(frozen=True, slots=True) +class ParsedRconAdminLogEvent: + event_type: RconAdminLogEventType + raw_message: str + relative_time: str | None = None + server_time: int | None = None + map_name: str | None = None + game_mode: str | None = None + allied_score: int | None = None + axis_score: int | None = None + winner: str | None = None + killer_name: str | None = None + killer_team: str | None = None + killer_id: str | None = None + victim_name: str | None = None + victim_team: str | None = None + victim_id: str | None = None + weapon: str | None = None + player_name: str | None = None + player_id: str | None = None + from_team: str | None = None + to_team: str | None = None + chat_scope: str | None = None + chat_team: str | None = None + content: str | None = None + reason: str | None = None + + +def parse_rcon_admin_log_message(message: str) -> ParsedRconAdminLogEvent: + raw_message = str(message or "") + prefix_match = _PREFIX_RE.match(raw_message) + relative_time = None + server_time = None + body = raw_message + + if prefix_match: + relative_time = prefix_match.group("relative") + server_time = _coerce_int(prefix_match.group("server_time")) + body = prefix_match.group("body") + + parser_payload = { + "raw_message": raw_message, + "relative_time": relative_time, + "server_time": server_time, + } + + if match := MATCH_START_RE.match(body): + return ParsedRconAdminLogEvent( + event_type="match_start", + map_name=_clean(match.group("map_name")), + game_mode=_clean(match.group("game_mode")), + **parser_payload, + ) + + if match := MATCH_END_RE.match(body): + allied_score = _coerce_int(match.group("allied_score")) + axis_score = _coerce_int(match.group("axis_score")) + return ParsedRconAdminLogEvent( + event_type="match_end", + map_name=_clean(match.group("map_name")), + allied_score=allied_score, + axis_score=axis_score, + winner=_resolve_winner(allied_score, axis_score), + **parser_payload, + ) + + if match := KILL_RE.match(body): + return ParsedRconAdminLogEvent( + event_type="kill", + killer_name=_clean(match.group("killer_name")), + killer_team=_clean(match.group("killer_team")), + killer_id=_clean(match.group("killer_id")), + victim_name=_clean(match.group("victim_name")), + victim_team=_clean(match.group("victim_team")), + victim_id=_clean(match.group("victim_id")), + weapon=_clean(match.group("weapon")), + **parser_payload, + ) + + if match := TEAM_SWITCH_RE.match(body): + return ParsedRconAdminLogEvent( + event_type="team_switch", + player_name=_clean(match.group("player_name")), + from_team=_clean(match.group("from_team")), + to_team=_clean(match.group("to_team")), + **parser_payload, + ) + + if match := CONNECTED_RE.match(body): + return ParsedRconAdminLogEvent( + event_type="connected", + player_name=_clean(match.group("player_name")), + player_id=_clean(match.group("player_id")), + **parser_payload, + ) + + if match := DISCONNECTED_RE.match(body): + return ParsedRconAdminLogEvent( + event_type="disconnected", + player_name=_clean(match.group("player_name")), + player_id=_clean(match.group("player_id")), + **parser_payload, + ) + + if match := CHAT_RE.match(body): + return ParsedRconAdminLogEvent( + event_type="chat", + player_name=_clean(match.group("player_name")), + player_id=_clean(match.group("player_id")), + chat_scope=_clean(match.group("scope")), + chat_team=_clean(match.group("team")), + content=_clean(match.group("content")), + **parser_payload, + ) + + if match := KICK_RE.match(body): + return ParsedRconAdminLogEvent( + event_type="kick", + player_name=_clean(match.group("player_name")), + reason=_clean(match.group("reason")), + **parser_payload, + ) + + if body.upper().startswith("BAN"): + return ParsedRconAdminLogEvent(event_type="ban", content=_clean(body), **parser_payload) + + if match := MESSAGE_RE.match(body): + return ParsedRconAdminLogEvent( + event_type="message", + player_name=_clean(match.group("player_name")), + player_id=_clean(match.group("player_id")), + content=_clean(match.group("content")), + **parser_payload, + ) + + return ParsedRconAdminLogEvent(event_type="unknown", content=_clean(body), **parser_payload) + + +def parse_rcon_admin_log_entry(entry: dict[str, object]) -> dict[str, object]: + parsed = parse_rcon_admin_log_message(str(entry.get("message") or "")) + payload = asdict(parsed) + payload["timestamp"] = entry.get("timestamp") + return payload + + +def _clean(value: str | None) -> str | None: + if value is None: + return None + normalized = value.strip() + return normalized or None + + +def _coerce_int(value: object) -> int | None: + try: + return int(value) + except (TypeError, ValueError): + return None + + +def _resolve_winner(allied_score: int | None, axis_score: int | None) -> str | None: + if allied_score is None or axis_score is None: + return None + if allied_score > axis_score: + return "allied" + if axis_score > allied_score: + return "axis" + return "draw" diff --git a/backend/tests/test_rcon_admin_log_parser.py b/backend/tests/test_rcon_admin_log_parser.py new file mode 100644 index 0000000..a301df0 --- /dev/null +++ b/backend/tests/test_rcon_admin_log_parser.py @@ -0,0 +1,107 @@ +from app.rcon_admin_log_parser import parse_rcon_admin_log_message + + +def test_parse_match_start(): + parsed = parse_rcon_admin_log_message( + "[2:09:15 hours (1779178245)] MATCH START UTAH BEACH Warfare" + ) + + assert parsed.event_type == "match_start" + assert parsed.server_time == 1779178245 + assert parsed.map_name == "UTAH BEACH" + assert parsed.game_mode == "Warfare" + + +def test_parse_match_end(): + parsed = parse_rcon_admin_log_message( + "[20:36:53 hours (1779111786)] MATCH ENDED `ST MARIE DU MONT Warfare` ALLIED (5 - 0) AXIS " + ) + + assert parsed.event_type == "match_end" + assert parsed.map_name == "ST MARIE DU MONT Warfare" + assert parsed.allied_score == 5 + assert parsed.axis_score == 0 + assert parsed.winner == "allied" + + +def test_parse_kill(): + parsed = parse_rcon_admin_log_message( + "[1:20:19 hours (1779181181)] KILL: AntonioPruna(Allies/76561198000000000) -> " + "[7DV] NEⓇA TACTICAL FEMB✡Y(Axis/76561199000000000) with M1 GARAND" + ) + + assert parsed.event_type == "kill" + assert parsed.killer_name == "AntonioPruna" + assert parsed.killer_team == "Allies" + assert parsed.killer_id == "76561198000000000" + assert parsed.victim_name == "[7DV] NEⓇA TACTICAL FEMB✡Y" + assert parsed.victim_team == "Axis" + assert parsed.victim_id == "76561199000000000" + assert parsed.weapon == "M1 GARAND" + + +def test_parse_team_switch(): + parsed = parse_rcon_admin_log_message( + "[21:34:19 hours (1779108340)] TEAMSWITCH Ekenef (None > Allies)" + ) + + assert parsed.event_type == "team_switch" + assert parsed.player_name == "Ekenef" + assert parsed.from_team == "None" + assert parsed.to_team == "Allies" + + +def test_parse_connected(): + parsed = parse_rcon_admin_log_message( + "[21:34:22 hours (1779108337)] CONNECTED Ekenef (76561198109813520)" + ) + + assert parsed.event_type == "connected" + assert parsed.player_name == "Ekenef" + assert parsed.player_id == "76561198109813520" + + +def test_parse_disconnected(): + parsed = parse_rcon_admin_log_message( + "[21:10:53 hours (1779109746)] DISCONNECTED [BxB] Rab◯l◯k◯ (76561198111111111)" + ) + + assert parsed.event_type == "disconnected" + assert parsed.player_name == "[BxB] Rab◯l◯k◯" + assert parsed.player_id == "76561198111111111" + + +def test_parse_chat(): + parsed = parse_rcon_admin_log_message( + "[18:38:35 hours (1779118884)] CHAT[Team][BXB Ivanxu(Axis/6215e24a1f05c5815ed9e8bf185f94fd)]: !vip" + ) + + assert parsed.event_type == "chat" + assert parsed.chat_scope == "Team" + assert parsed.player_name == "BXB Ivanxu" + assert parsed.chat_team == "Axis" + assert parsed.player_id == "6215e24a1f05c5815ed9e8bf185f94fd" + assert parsed.content == "!vip" + + +def test_parse_kick(): + parsed = parse_rcon_admin_log_message( + "[2:09:10 hours (1779178249)] KICK: [[7DV] NEⓇA TACTICAL FEMB✡Y] has been kicked. " + "[Making free spaces for members of the Spanish Discord community.]" + ) + + assert parsed.event_type == "kick" + assert parsed.player_name == "[7DV] NEⓇA TACTICAL FEMB✡Y" + assert "Making free spaces" in parsed.reason + + +def test_parse_message_profile(): + parsed = parse_rcon_admin_log_message( + "[21:34:19 hours (1779108340)] MESSAGE: player [Ekenef(76561198109813520)], " + "content [─ Ekenef ─\\n▒ Totales ▒\\nbajas : 141 (6 TKs)\\nmuertes : 268 (5 TKs)]" + ) + + assert parsed.event_type == "message" + assert parsed.player_name == "Ekenef" + assert parsed.player_id == "76561198109813520" + assert "bajas : 141" in parsed.content