#!/usr/bin/env python3 """ cs_discord_bridge.py — Two-way Discord ↔ Counter-Strike 1.6 bridge Features: - Chat relay (say, say_team) with tombstone for dead players - Join / leave (posts leave messages always) - Kill events (killer → victim, with flags, headshots) - Suicide / world kills - Bomb events (spawned, dropped, got, planted, defused, bombed) with 💣 emoji - Discord → CS chat via asynchronous RCON client - Consistent GeoIP flags (🤖 for bots, 🏳️ for failed human lookups) - Enhanced debugging for shell command execution - Round start/end and team win messages with scores - Map change messages suppressed until human detected, re-suppressed when all humans leave - Messages sent when human player (non-BOT SteamID) detected in logs - Message suppression for non-disconnect events when no humans Listen port set to 8008 for CS logs. """ import asyncio import datetime import os import re from typing import Optional from io import BytesIO try: import tomllib # py311+ except ModuleNotFoundError: import tomli as tomllib import discord from discord.ext import commands import geoip2.database def cc_to_flag(cc: str) -> str: if not cc or len(cc) != 2: return "🏳️" # White flag for failed GeoIP lookups for humans base = 127397 return chr(ord(cc[0].upper()) + base) + chr(ord(cc[1].upper()) + base) # --- Asynchronous RCON Client Class (UDP Version) --- class RconClient: def __init__(self, host, port, password): self.host = host self.port = port self.password = password self.packet_size = 1024 async def get_challenge(self): loop = asyncio.get_running_loop() on_done = loop.create_future() message = b'\xFF\xFF\xFF\xFFgetchallenge\n' class ChallengeProtocol(asyncio.DatagramProtocol): def __init__(self): self.transport = None self.future = on_done def connection_made(self, transport): self.transport = transport self.transport.sendto(message) def datagram_received(self, data, addr): try: challenge = data[5:].decode().split(' ')[1].strip() self.future.set_result(challenge) except (IndexError, UnicodeDecodeError) as e: self.future.set_exception(ValueError(f"Invalid challenge response: {e}")) finally: self.transport.close() def error_received(self, exc): self.future.set_exception(exc) self.transport.close() try: transport, protocol = await loop.create_datagram_endpoint( ChallengeProtocol, remote_addr=(self.host, self.port) ) return await asyncio.wait_for(on_done, timeout=5.0) except asyncio.TimeoutError: print(f"[{now_iso()}] RCON challenge request timed out.") except Exception as e: print(f"[{now_iso()}] [ERROR] Getting RCON challenge: {e}") return None async def send_command(self, command: str): challenge = await self.get_challenge() if not challenge: print(f"[{now_iso()}] Cannot send command, failed to get RCON challenge.") return loop = asyncio.get_running_loop() on_done = loop.create_future() message_buffer = BytesIO() message_buffer.write(b'\xFF\xFF\xFF\xFF') message_buffer.write('rcon '.encode()) message_buffer.write(challenge.encode()) message_buffer.write(b' ') message_buffer.write(self.password.encode()) message_buffer.write(b' ') # New line message_buffer.write(command.encode()) # New line message_buffer.write(b'\n') message_bytes = message_buffer.getvalue() class CommandProtocol(asyncio.DatagramProtocol): def __init__(self): self.transport = None self.future = on_done def connection_made(self, transport): self.transport = transport self.transport.sendto(message_bytes) self.future.set_result(None) self.transport.close() def error_received(self, exc): self.future.set_exception(exc) self.transport.close() try: transport, protocol = await loop.create_datagram_endpoint( CommandProtocol, remote_addr=(self.host, self.port) ) await asyncio.wait_for(on_done, timeout=5.0) print(f"[{now_iso()}] Sent RCON command: {command}") except asyncio.TimeoutError: print(f"[{now_iso()}] RCON command timed out.") except Exception as e: print(f"[{now_iso()}] [ERROR] Sending RCON command: {e}") # --- Regexes --- HL_SAY_REGEXES = [ re.compile(r'L \d{2}/\d{2}/\d{4} - \d{2}:\d{2}:\d{2}: "(.+?)<\d+><(.*?)><(.*?)>" say "(.*)"(?: \(dead\))?$'), re.compile(r'L \d{2}/\d{2}/\d{4} - \d{2}:\d{2}:\d{2}: "(.+?)<\d+><(.*?)><(.*?)>" say_team "(.*)"(?: \(dead\))?$'), ] CONNECT_REGEX = re.compile(r'L \d{2}/\d{2}/\d{4} - \d{2}:\d{2}:\d{2}: "(.+?)<\d+><(.*?)><(.*?)>" connected, address "(.+?):(\d+)"$') JOIN_LEAVE_REGEXES = [ re.compile(r'L \d{2}/\d{2}/\d{4} - \d{2}:\d{2}:\d{2}: "(.+?)<\d+><(.*?)><(.*?)>" entered the game$'), re.compile(r'L \d{2}/\d{2}/\d{4} - \d{2}:\d{2}:\d{2}: "(.+?)<\d+><(.*?)><(.*?)>" disconnected(?: \(reason "(.*?)"\))?$'), re.compile(r'L \d{2}/\d{2}/\d{4} - \d{2}:\d{2}:\d{2}: "(.+?)<\d+><(.*?)><(.*?)>" joined team "(.*)"$'), ] KILL_REGEX = re.compile( r'L \d{2}/\d{2}/\d{4} - \d{2}:\d{2}:\d{2}: "(.+?)<\d+><(.*?)><(.*?)>" killed "(.+?)<\d+><(.*?)><(.*?)>" with "(.*?)"(?: \(headshot\))?$' ) SUICIDE_REGEXES = [ re.compile(r'L \d{2}/\d{2}/\d{4} - \d{2}:\d{2}:\d{2}: "(.+?)<\d+><(.*?)><(.*?)>" committed suicide with "(.*?)"$'), re.compile(r'L \d{2}/\d{2}/\d{4} - \d{2}:\d{2}:\d{2}: "(.+?)<\d+><(.*?)><(.*?)>" killed self with "(.*?)"$'), ] BOMB_REGEX = re.compile( r'L \d{2}/\d{2}/\d{4} - \d{2}:\d{2}:\d{2}: "(.+?)<\d+><(.*?)><(.*?)>" triggered "(Spawned_With_The_Bomb|Dropped_The_The_Bomb|Got_The_Bomb|Planted_The_Bomb)"' ) TEAM_BOMB_REGEX = re.compile( r'L \d{2}/\d{2}/\d{4} - \d{2}:\d{2}:\d{2}: Team "(CT|TERRORIST)" triggered "(Bomb_Defused|Target_Bombed)" \(CT "(\d+)"\) \(T "(\d+)"\)' ) STEAM_VALIDATE_REGEX = re.compile(r'L \d{2}/\d{2}/\d{4} - \d{2}:\d{2}:\d{2}: "(.+?)<\d+><(.*?)><(.*?)>" STEAM USERID validated$') RCON_REGEX = re.compile(r'L \d{2}/\d{2}/\d{4} - \d{2}:\d{2}:\d{2}: Rcon: "rcon (.+)" from "(.+)"$') MAPCHANGE_REGEX = re.compile(r'L \d{2}/\d{2}/\d{4} - \d{2}:\d{2}:\d{2}: Started map "(.+)" \(CRC ".+"') STARTED_MAP_REGEX = re.compile(r'L \d{2}/\d{2}/\d{4} - \d{2}:\d{2}:\d{2}: Started map "(.+?)"') ROUND_START_REGEX = re.compile(r'L \d{2}/\d{2}/\d{4} - \d{2}:\d{2}:\d{2}: World triggered "Round_Start"$') ROUND_END_REGEX = re.compile(r'L \d{2}/\d{2}/\d{4} - \d{2}:\d{2}:\d{2}: World triggered "Round_End"$') TEAM_WIN_REGEX = re.compile(r'L \d{2}/\d{2}/\d{4} - \d{2}:\d{2}:\d{2}: Team "(CT|TERRORIST)" triggered "(CTs_Win|Terrorists_Win)" \(CT "(\d+)"\) \(T "(\d+)"\)$') def now_iso(): return datetime.datetime.now().astimezone().isoformat(timespec="seconds") class HLLogUDP(asyncio.DatagramProtocol): def __init__(self, on_line): super().__init__() self.on_line = on_line def datagram_received(self, data: bytes, addr): try: text = data.decode('utf-8', errors='replace').strip() except Exception: text = repr(data) for line in text.splitlines(): asyncio.create_task(self.on_line(line, addr)) class CSDiscordBridge: def __init__(self, cfg: dict): self.cfg = cfg intents = discord.Intents.default() intents.message_content = True intents.messages = True # Ensure message intents are enabled self.bot = commands.Bot(command_prefix=cfg.get("discord_prefix", "!"), intents=intents) self.channel: Optional[discord.TextChannel] = None self.geoip_reader = None db_path = self.cfg.get("geoip_db", "/home/csserver/discord/GeoLite2-Country.mmdb") try: self.geoip_reader = geoip2.database.Reader(db_path) except Exception as e: print(f"[{now_iso()}] [WARN] GeoIP not available: {e}") self.player_flags = {} self.connected_players = {} self.has_human_player = False self.allow_bot_messages = cfg.get("allow_bot_messages", False) self.pending_map_change = None self.suppress_join_messages = False self.bot.event(self.on_ready) self.bot.event(self.on_message) self.rcon_client = RconClient( self.cfg["cs_host"], int(self.cfg["cs_port"]), self.cfg["cs_rcon_password"] ) async def on_ready(self): ch_id = int(self.cfg["discord_channel_id"]) try: self.channel = self.bot.get_channel(ch_id) or await self.bot.fetch_channel(ch_id) print(f"[{now_iso()}] Logged in as {self.bot.user}. Bridging to #{self.channel.name} ({self.channel.id})") except Exception as e: print(f"[{now_iso()}] [ERROR] Failed to fetch Discord channel {ch_id}: {e}") return await self._send_startup_commands() listen_host = self.cfg.get("listen_host", "0.0.0.0") listen_port = int(self.cfg.get("listen_port", 8008)) loop = asyncio.get_running_loop() print(f"[{now_iso()}] Listening for HL logs on {listen_host}:{listen_port}") try: await loop.create_datagram_endpoint( lambda: HLLogUDP(self.handle_hl_line), local_addr=(listen_host, listen_port) ) except Exception as e: print(f"[{now_iso()}] [ERROR] Failed to start UDP listener: {e}") async def _send_startup_commands(self): print(f"[{now_iso()}] Sending RCON command to disable chat prefixes...") await self.rcon_client.send_command('amx_chat_prefix ""') await self.rcon_client.send_command('cs_chat_prefix ""') await self.rcon_client.send_command('sv_say_prefix ""') async def on_message(self, message: discord.Message): print(f"[{now_iso()}] [DEBUG] Received Discord message: author={message.author}, channel={message.channel.id}, content={message.content}") if message.author.bot: print(f"[{now_iso()}] [DEBUG] Ignoring message from bot: {message.author}") return if not self.channel or message.channel.id != self.channel.id: print(f"[{now_iso()}] [DEBUG] Message from wrong channel: {message.channel.id}, expected {self.channel.id if self.channel else 'None'}") return prefix = self.cfg.get("discord_prefix", "!") if message.content.startswith(prefix): print(f"[{now_iso()}] [DEBUG] Ignoring command with prefix: {message.content}") return content = self._sanitize_discord(message.content) author = message.author.display_name line = f'say (DISCORD) {author}: {content}' print(f"[{now_iso()}] [DEBUG] Processing Discord message: {line} (has_human_player={self.has_human_player})") await self._send_rcon_say(line) async def _send_rcon_say(self, command: str): limit = self.cfg.get("say_length_limit", 190) if len(command) > limit: command = command[:limit-1] + "…" print(f"[{now_iso()}] [DEBUG] Sending RCON command to CS: {command}") await self.rcon_client.send_command(command) async def handle_hl_line(self, line: str, addr): if "Server cvar" in line: return print(f"[{now_iso()}] Received log: {line}") # Strip potential garbage prefixes like ����log cleaned_line = line if line.startswith('\ufffd\ufffd\ufffd\ufffdlog'): cleaned_line = line[8:].strip() print(f"[{now_iso()}] Stripped garbage prefix, cleaned log: {cleaned_line}") log_start_match = re.search(r'L \d{2}/\d{2}/\d{4} - \d{2}:\d{2}:\d{2}:', cleaned_line) if not log_start_match: print(f"[{now_iso()}] Unmatched log (no valid start): {cleaned_line}") if "connected, address" in cleaned_line and self._parse_connected(cleaned_line): return return cleaned_line = cleaned_line[log_start_match.start():].strip() print(f"[{now_iso()}] Cleaned log: {cleaned_line}") map_change = self._parse_mapchange(cleaned_line) if map_change: print(f"[{now_iso()}] Detected map change: {map_change}. Storing for later.") self.pending_map_change = map_change self.suppress_join_messages = True await self._post_to_discord(map_change, force=True) return if self._parse_steam_validated(cleaned_line): return if self._parse_rcon(cleaned_line): return if self._parse_connected(cleaned_line): return joinleave = self._parse_join_leave(cleaned_line) if joinleave: print(f"[{now_iso()}] Parsed join/leave: {joinleave} (has_human_player={self.has_human_player})") await self._post_to_discord(joinleave) return say, team_flag, payload = self._parse_say(cleaned_line) if say: print(f"[{now_iso()}] Parsed say (team={team_flag}): {payload} (has_human_player={self.has_human_player})") await self._post_to_discord(payload) return kill = self._parse_kill(cleaned_line) if kill: print(f"[{now_iso()}] Parsed kill: {kill} (has_human_player={self.has_human_player})") await self._post_to_discord(kill) return suicide = self._parse_suicide(cleaned_line) if suicide: print(f"[{now_iso()}] Parsed suicide: {suicide} (has_human_player={self.has_human_player})") await self._post_to_discord(suicide) return bomb = self._parse_bomb(cleaned_line) if bomb: print(f"[{now_iso()}] Parsed bomb event: {bomb} (has_human_player={self.has_human_player})") await self._post_to_discord(bomb) return if ROUND_START_REGEX.search(cleaned_line): payload = "🔄 Round started!" print(f"[{now_iso()}] Parsed round start: {payload} (has_human_player={self.has_human_player})") self.suppress_join_messages = False await self._post_to_discord(payload) return if ROUND_END_REGEX.search(cleaned_line): payload = "🏁 Round ended!" print(f"[{now_iso()}] Parsed round end: {payload} (has_human_player={self.has_human_player})") await self._post_to_discord(payload) return m = TEAM_WIN_REGEX.search(cleaned_line) if m: team, _, ct_score, t_score = m.groups() if team == "CT": payload = f"🔵 **Counter-Terrorists Win!** (CT {ct_score}, T {t_score})" else: payload = f"🔴 **Terrorists Win!** (CT {ct_score}, T {t_score})" print(f"[{now_iso()}] Parsed team win: {payload} (has_human_player={self.has_human_player})") await self._post_to_discord(payload) return print(f"[{now_iso()}] Unmatched log: {cleaned_line} (has_human_player={self.has_human_player})") def _parse_mapchange(self, line: str) -> Optional[str]: m_mapchange = re.compile( r'L \d{2}/\d{2}/\d{4} - \d{2}:\d{2}:\d{2}: -------- Mapchange to (.+?) --------$' ).search(line) m_started_map = re.compile( r'L \d{2}/\d{2}/\d{4} - \d{2}:\d{2}:\d{2}: Started map "(.+?)"' ).search(line) map_name = None if m_mapchange: map_name = m_mapchange.group(1) elif m_started_map: map_name = m_started_map.group(1) if map_name: # snapshot current state before reset had_humans = self.has_human_player # reset for the new map self.has_human_player = False self.connected_players = {} self.suppress_join_messages = True # Only announce if humans were present or bot messages are allowed if had_humans or self.allow_bot_messages: return f"🔄 Map changed to `{map_name}`" else: print(f"[{now_iso()}] Suppressing mapchange announcement ({map_name}) — no human players were online.") return None return None def _parse_steam_validated(self, line: str) -> bool: m = STEAM_VALIDATE_REGEX.search(line) if m: name, steamid, team = m.groups() print(f"[{now_iso()}] Handling 'STEAM USERID validated' for {name} ({steamid}).") if steamid != "BOT": if "STEAM_ID_PENDING" in self.connected_players: pending_name = self.connected_players.pop("STEAM_ID_PENDING") self.connected_players[steamid] = pending_name if "STEAM_ID_PENDING" in self.player_flags: self.player_flags[steamid] = self.player_flags.pop("STEAM_ID_PENDING") print(f"[{now_iso()}] Transferred flag for {pending_name} from STEAM_ID_PENDING to {steamid}: {self.player_flags[steamid]}") return True return False def _parse_rcon(self, line: str) -> bool: m = RCON_REGEX.search(line) if m: command, _ = m.groups() if not command.strip(): # Ignore incomplete RCON commands print(f"[{now_iso()}] [DEBUG] Ignored incomplete RCON command: {line}") return True print(f"[{now_iso()}] Handled RCON command log: {command}") return True return False def _parse_connected(self, line: str) -> bool: m = CONNECT_REGEX.search(line) if m: name, steamid, team, ip, port = m.groups() print(f"[{now_iso()}] [DEBUG] Matched CONNECT_REGEX for: {name} ({steamid}, IP {ip})") if steamid == "BOT": print(f"[{now_iso()}] Bot connected: {name} ({steamid})") self.player_flags[steamid] = "🤖" print(f"[{now_iso()}] Assigned flag code for {name} ({steamid}): 'None' -> '🤖'") return True self.connected_players[steamid] = name self.has_human_player = True flag = "🏳️" country_code = None if self.geoip_reader: try: response = self.geoip_reader.country(ip) country_code = response.country.iso_code if country_code: flag = cc_to_flag(country_code) print(f"[{now_iso()}] GeoIP lookup for {name} ({steamid}, IP {ip}): country code '{country_code}' -> flag '{flag}'") else: print(f"[{now_iso()}] [WARN] No country code found for IP {ip}") except Exception as e: print(f"[{now_iso()}] [ERROR] GeoIP lookup failed for IP {ip}: {e}") else: print(f"[{now_iso()}] [ERROR] GeoIP reader not initialized for {name} ({steamid})") self.player_flags[steamid] = flag print(f"[{now_iso()}] Assigned flag code for {name} ({steamid}): '{country_code or 'None'}' -> '{flag}'") print(f"[{now_iso()}] [DEBUG] GeoIP database path: {self.cfg.get('geoip_db', '/home/csserver/discord/GeoLite2-Country.mmdb')}") print(f"[{now_iso()}] [DEBUG] Player flags dictionary: {self.player_flags}") return True else: print(f"[{now_iso()}] [DEBUG] CONNECT_REGEX failed to match: {line}") return False def _parse_join_leave(self, line: str) -> Optional[str]: # Ignore raw connection lines like: " from "64.188.91.127:53453" if ' from "' in line: print(f"[{now_iso()}] Ignoring connection info line: {line.strip()}") return None entered_the_game_regex = re.compile( r'L \d{2}/\d{2}/\d{4} - \d{2}:\d{2}:\d{2}: "(.+?)<\d+><(.*?)><(.*?)?>" entered the game$' ) m = entered_the_game_regex.search(line) if m: name, steamid, team = m.groups()[:3] if steamid == "BOT": self.player_flags[steamid] = "🤖" print(f"[{now_iso()}] Stored flag for {name} ({steamid}): 🤖") print(f"[{now_iso()}] Ignoring bot join message for {name}.") return None entered_the_game_regex = re.compile( r'L \d{2}/\d{2}/\d{4} - \d{2}:\d{2}:\d{2}: "(.+?)<\d+><(.*?)><(.*?)?>" entered the game$' ) m = entered_the_game_regex.search(line) if m: name, steamid, team = m.groups()[:3] if steamid == "BOT": self.player_flags[steamid] = "🤖" print(f"[{now_iso()}] Stored flag for {name} ({steamid}): 🤖") print(f"[{now_iso()}] Ignoring bot join message for {name}.") return None if self.suppress_join_messages: print(f"[{now_iso()}] Suppressing join message for {name} ({steamid}) during map change.") self.connected_players[steamid] = name self.has_human_player = True return None self.connected_players[steamid] = name self.has_human_player = True name = self._sanitize_name(name) flag = self.player_flags.get(steamid, "🏳️") if flag == "🏳️" and steamid != "BOT": print(f"[{now_iso()}] [WARN] No flag found for {name} ({steamid}), attempting GeoIP lookup") m_connect = CONNECT_REGEX.search(line) if m_connect: _, _, _, ip, _ = m_connect.groups() if self.geoip_reader and ip: try: response = self.geoip_reader.country(ip) country_code = response.country.iso_code if country_code: flag = cc_to_flag(country_code) self.player_flags[steamid] = flag print(f"[{now_iso()}] Fallback GeoIP lookup for {name} ({steamid}, IP {ip}): country code '{country_code}' -> flag '{flag}'") else: print(f"[{now_iso()}] [WARN] No country code found for IP {ip} in fallback lookup") except Exception as e: print(f"[{now_iso()}] [ERROR] Fallback GeoIP lookup failed for IP {ip}: {e}") print(f"[{now_iso()}] Human player joined: {name} ({steamid}), flag={flag}, has_human_player={self.has_human_player}") return f'👋 {flag} `{name}` joined the server' for rx in JOIN_LEAVE_REGEXES: m = rx.search(line) if m: if "disconnected" in line: name, steamid, _, reason = m.groups() if steamid == "BOT": print(f"[{now_iso()}] Ignoring bot leave message for {name}.") return None if self.suppress_join_messages: print(f"[{now_iso()}] Suppressing leave message for {name} ({steamid}) during map change window") if steamid in self.connected_players: del self.connected_players[steamid] return None name = self._sanitize_name(name) flag = self.player_flags.get(steamid, "🏳️") leave_message = f'❌ {flag} `{name}` left the server' + (f' ({reason})' if reason else '') if steamid in self.connected_players: del self.connected_players[steamid] if not any(sid != "BOT" for sid in self.connected_players): self.has_human_player = False print(f"[{now_iso()}] No human players remain, has_human_player={self.has_human_player}") print(f"[{now_iso()}] Human player left: {name} ({steamid}), flag={flag}, has_human_player={self.has_human_player}") return leave_message elif "joined team" in line: name, steamid, team, _ = m.groups() if steamid == "BOT": self.player_flags[steamid] = "🤖" print(f"[{now_iso()}] Stored flag for {name} ({steamid}): 🤖") print(f"[{now_iso()}] Ignoring bot joined team message.") return None self.has_human_player = True print(f"[{now_iso()}] Human detected in joined team: {name} ({steamid}), has_human_player={self.has_human_player}") return None return None def _parse_say(self, line: str): for rx in HL_SAY_REGEXES: m = rx.search(line) if m: name, steamid, team, msg = m.groups() team_flag = "say_team" in rx.pattern or "say_team" in line name = self._sanitize_name(name) msg = self._sanitize_cs(msg) flag = self.player_flags.get(steamid, "🤖" if steamid == "BOT" else "🏳️") if steamid != "BOT": self.has_human_player = True print(f"[{now_iso()}] Human detected in say: {name} ({steamid}), has_human_player={self.has_human_player}") team_icon = "🔵" if team.upper().startswith("CT") else "🔴" if team.upper().startswith("T") else "⚪" dead_indicator = " 🪦" if "(dead)" in line else "" payload = f'💬 {flag} {team_icon} `{name}`{dead_indicator}: {msg}' return True, team_flag, payload return False, False, None def _parse_kill(self, line: str) -> Optional[str]: m = KILL_REGEX.search(line) if not m: return None killer_name, killer_id, killer_team, victim_name, victim_id, victim_team, weapon = m.groups() if killer_id == "BOT" and victim_id == "BOT": print(f"[{now_iso()}] Ignoring kill message (bot killed bot).") return None if killer_id != "BOT" or victim_id != "BOT": self.has_human_player = True print(f"[{now_iso()}] Human detected in kill: {killer_name} ({killer_id}) or {victim_name} ({victim_id}), has_human_player={self.has_human_player}") killer_flag = self.player_flags.get(killer_id, "🤖" if killer_id == "BOT" else "🏳️") victim_flag = self.player_flags.get(victim_id, "🤖" if victim_id == "BOT" else "🏳️") killer_name = self._sanitize_name(killer_name) victim_name = self._sanitize_name(victim_name) headshot = " 💥" if "(headshot)" in line else "" killer_team_icon = "🔵" if killer_team.upper().startswith("CT") else "🔴" if killer_team.upper().startswith("T") else "⚪" victim_team_icon = "🔵" if victim_team.upper().startswith("CT") else "🔴" if victim_team.upper().startswith("T") else "⚪" # Determine article vowel_sounds = ('a', 'e', 'i', 'o', 'u') exceptions = {"tmp", "usp", "ump", "uzi"} # extend if needed if weapon.lower() in exceptions: article = "a" else: article = "an" if weapon.lower().startswith(vowel_sounds) else "a" return f'{killer_flag} {killer_team_icon} `{killer_name}` killed {victim_flag} {victim_team_icon} `{victim_name}` with {article} *{weapon}*{headshot}' def _parse_suicide(self, line: str) -> Optional[str]: for rx in SUICIDE_REGEXES: m = rx.search(line) if m: name, steamid, _, weapon = m.groups() if steamid != "BOT": self.has_human_player = True print(f"[{now_iso()}] Human detected in suicide: {name} ({steamid}), has_human_player={self.has_human_player}") name = self._sanitize_name(name) flag = self.player_flags.get(steamid, "🤖" if steamid == "BOT" else "🏳️") return f'💀 {flag} `{name}` died ({weapon})' return None def _parse_bomb(self, line: str) -> Optional[str]: m = BOMB_REGEX.search(line) if m: name, steamid, _, event = m.groups() name = self._sanitize_name(name) flag = self.player_flags.get(steamid, "🤖" if steamid == "BOT" else "🏳️") if steamid != "BOT": self.has_human_player = True print(f"[{now_iso()}] Human detected in bomb event: {name} ({steamid}), has_human_player={self.has_human_player}") if event == "Planted_The_Bomb": return '💣 *The bomb has been planted!*' m = TEAM_BOMB_REGEX.search(line) if m: team, event, ct_score, t_score = m.groups() if event == "Bomb_Defused": return '💣 *The bomb has been defused!*' elif event == "Target_Bombed": return '💣 *Target successfully bombed!*' return None async def _post_to_discord(self, content: str, force=False): if not self.channel: print(f"[{now_iso()}] [ERROR] Discord channel not set, cannot post: {content}") return # Always allow join/leave messages through (mapchange suppression already handled in _parse_join_leave) if force or content.startswith(("👋", "❌")) or self.has_human_player or self.allow_bot_messages: try: await self.channel.send(content) print(f"[{now_iso()}] Successfully posted to Discord: {content} (has_human_player={self.has_human_player})") except Exception as e: print(f"[{now_iso()}] [ERROR] Failed to send to Discord: {e} (content={content})") else: print(f"[{now_iso()}] Suppressing Discord message due to no human players: {content} (has_human_player={self.has_human_player})") def _sanitize_discord(self, s: str) -> str: s = s.replace('"', '').strip() return ''.join(ch for ch in s if 31 < ord(ch) < 127) def _sanitize_cs(self, s: str) -> str: s = s.replace('"', "'") return ''.join(ch for ch in s if 31 < ord(ch) < 127) def _sanitize_name(self, s: str) -> str: s = s.strip() if len(s) > 24: s = s[:23] + "…" return self._sanitize_cs(s) async def run(self): listen_host = self.cfg.get("listen_host", "0.0.0.0") listen_port = int(self.cfg.get("listen_port", 8008)) loop = asyncio.get_running_loop() print(f"[{now_iso()}] Listening for HL logs on {listen_host}:{listen_port}") try: await loop.create_datagram_endpoint( lambda: HLLogUDP(self.handle_hl_line), local_addr=(listen_host, listen_port) ) except Exception as e: print(f"[{now_iso()}] [ERROR] Failed to start UDP listener: {e}") await self.bot.start(self.cfg["discord_token"]) def load_config(path: str) -> dict: try: with open(path, "rb") as f: cfg = tomllib.load(f) except Exception as e: print(f"[{now_iso()}] [ERROR] Failed to load config {path}: {e}") raise SystemExit(f"Failed to load config: {e}") required = ["discord_token", "discord_channel_id", "cs_host", "cs_port", "cs_rcon_password"] for key in required: if key not in cfg: print(f"[{now_iso()}] [ERROR] Missing required config key: {key}") raise SystemExit(f"Missing required config key: {key}") return cfg async def amain(cfg_path: str): cfg = load_config(cfg_path) bridge = CSDiscordBridge(cfg) await bridge.run() def main(): cfg_path = os.environ.get("CSBRIDGE_CONFIG", "config.toml") try: asyncio.run(amain(cfg_path)) except KeyboardInterrupt: print(f"[{now_iso()}] Exiting...") except Exception as e: print(f"[{now_iso()}] [ERROR] Fatal error: {e}") if __name__ == "__main__": main()