diff --git a/cogs/console.py b/cogs/console.py index 2267e08..5a60ee7 100644 --- a/cogs/console.py +++ b/cogs/console.py @@ -18,33 +18,6 @@ CONSOLE_FILE_PATH = Path(__file__).parent.parent / "data" / "server-console.txt" log = logging.getLogger(__name__) -@dataclass(slots=True) -class ZomboidUser: - guid: str - ip: str - steam_id: str - access: str - username: str - connection_type: str - - @property - def steam_url(self): - return f"https://steamcommunity.com/profiles/{self.steam_id}" - - async def get_steam_profile_picture(self, steam_api_key: str): - if not steam_api_key: - log.warning("No steam API key, can't get profile picture.") - return - - async with httpx.AsyncClient() as client: - response = await client.get(url=f"https://api.steampowered.com/ISteamUser/GetPlayerSummaries/v2/?key={steam_api_key}&steamids={self.steam_id}") - response.raise_for_status() - - data = response.json() - avatar_url = data["response"]["players"][0]["avatarfull"] - return avatar_url - - class ConsoleCog(commands.Cog): """ Reads and handles the server-console.txt file. @@ -94,10 +67,12 @@ class ConsoleCog(commands.Cog): await self.handle_mod_needs_update(line) elif "ConnectionManager: [fully-connected]" in line: - await self.handle_player_joined(line) + cog = self.bot.get_cog("PlayersCog") + await cog.handle_player_connected(line) elif "ConnectionManager: [disconnect]" in line: - await self.handle_player_left(line) + cog = self.bot.get_cog("PlayersCog") + await cog.handle_player_disconnected(line) async def alert_and_wait_for_restart(self, intervals_ms: list[int], reason: str): for interval_ms in intervals_ms: @@ -143,75 +118,6 @@ class ConsoleCog(commands.Cog): await channel.send(content="The server is currently being restarted for mod updates.") - async def handle_player_joined(self, line: str): - """ - Report when a user has joined the server into a specified Discord channel. - Example of line: - ConnectionManager: [fully-connected] "" connection: guid=*** ip=*** steam-id=*** access=admin username="corbz" connection-type="UDPRakNet" - """ - re_pattern = r"guid=(\d+)\s+ip=([\d\.]+)\s+steam-id=(\d+)\s+access=(\w*)\s+username=\"([^\"]+)\"\s+connection-type=\"([^\"]+)\"" - re_match = re.search(re_pattern, line) - if not re_match: - log.warning("failed to parse player data: %s", line) - return - - user = ZomboidUser( - guid=re_match.group(1), - ip=re_match.group(2), - steam_id=re_match.group(3), - access=re_match.group(4), - username=re_match.group(5), - connection_type=re_match.group(6) - ) - - channel = self.bot.get_channel(self.bot.in_game_channel_id) - channel = await self.bot.fetch_channel(self.bot.in_game_channel_id) if not channel else channel - - embed = Embed( - title=user.username, - url=user.steam_url, - description="Player has joined the server", - colour=Colour.brand_green() - ) - embed.set_thumbnail(url=await user.get_steam_profile_picture(self.bot.steam_api_key)) - - await channel.send(embed=embed) - - async def handle_player_left(self, line: str): - """ - Report when a user has left the server into a specified Discord channel. - Example of line: - ConnectionManager: [disconnect] "receive-disconnect" connection: guid=*** ip=*** steam-id=*** access=admin username="corbz" connection-type="Disconnected" - """ - re_pattern = r"guid=(\d+)\s+ip=([\d\.]+)\s+steam-id=(\d+)\s+access=(\w*)\s+username=\"([^\"]+)\"\s+connection-type=\"([^\"]+)\"" - re_match = re.search(re_pattern, line) - if not re_match: - log.warning("failed to parse player data: %s", line) - return - - user = ZomboidUser( - guid=re_match.group(1), - ip=re_match.group(2), - steam_id=re_match.group(3), - access=re_match.group(4), - username=re_match.group(5), - connection_type=re_match.group(6) - ) - - channel = self.bot.get_channel(self.bot.in_game_channel_id) - channel = await self.bot.fetch_channel(self.bot.in_game_channel_id) if not channel else channel - - embed = Embed( - title=user.username, - url=user.steam_url, - description="Player has left the server", - colour=Colour.brand_red() - ) - embed.set_thumbnail(url=await user.get_steam_profile_picture(self.bot.steam_api_key)) - - await channel.send(embed=embed) - - async def setup(bot: commands.Bot): cog = ConsoleCog(bot) await bot.add_cog(cog) diff --git a/cogs/players.py b/cogs/players.py new file mode 100644 index 0000000..148041c --- /dev/null +++ b/cogs/players.py @@ -0,0 +1,152 @@ +""" +Handles tasks related to in-game players, such as connect/disconnect alerts. +""" + +import re +import logging +from dataclasses import dataclass + +import httpx +from discord import Embed, Colour +from discord.ext import commands, tasks +from rcon.source import rcon + +log = logging.getLogger(__name__) + + +@dataclass(slots=True) +class SteamProfileSummary: + steam_id: int + profile_name: str + url: str + avatar_url: str + + +@dataclass(slots=True) +class ZomboidUser: + guid: str + ip: str + steam_id: str + access: str + username: str + connection_type: str + steam_profile_summary: SteamProfileSummary | None = None + + async def fetch_steam_profile(self, steam_api_key: str): + if not steam_api_key: + log.warning("No steam API key, can't get profile summary.") + return + + async with httpx.AsyncClient() as client: + response = await client.get(url=f"https://api.steampowered.com/ISteamUser/GetPlayerSummaries/v2/?key={steam_api_key}&steamids={self.steam_id}") + response.raise_for_status() + + all_data = response.json() + user_data = all_data["response"]["players"][0] + + log.debug("fetched user data for: %s", self.steam_id) + + self.steam_profile_summary = SteamProfileSummary( + steam_id=user_data["steamid"], + profile_name=user_data["personaname"], + url=user_data["profileurl"], + avatar_url=user_data["avatarmedium"] + ) + + log.debug("successfully parsed steam profile summary for: %s", self.steam_id) + + @property + def embed(self) -> Embed: + if not self.steam_profile_summary: + raise ValueError("You must fetch the steam_profile_summary before creating an embed.") + + embed = Embed( + title="Player", + description=( + f"{self.username} ([{self.steam_profile_summary.profile_name}]({self.steam_profile_summary.url}))\n" + "kills: ???\n" + "Playtime: ???" + ) + ) + embed.set_thumbnail(url=self.steam_profile_summary.avatar_url) + return embed + + +class PlayersCog(commands.Cog): + """ + Handles tasks related to in-game players. + """ + def __init__(self, bot: commands.Bot): + self.bot = bot + + @commands.Cog.listener() + async def on_ready(self): + if not self.update_activity.is_running(): + self.update_activity.start() + + async def handle_player_connected(self, line: str): + """ + Report when a user has joined the server into a specified Discord channel. + Example of line: + ConnectionManager: [fully-connected] "" connection: guid=*** ip=*** steam-id=*** access=admin username="corbz" connection-type="UDPRakNet" + """ + re_pattern = r"guid=(\d+)\s+ip=([\d\.]+)\s+steam-id=(\d+)\s+access=(\w*)\s+username=\"([^\"]+)\"\s+connection-type=\"([^\"]+)\"" + re_match = re.search(re_pattern, line) + if not re_match: + log.warning("failed to parse player data: %s", line) + return + + user = ZomboidUser( + guid=re_match.group(1), + ip=re_match.group(2), + steam_id=re_match.group(3), + access=re_match.group(4), + username=re_match.group(5), + connection_type=re_match.group(6) + ) + await user.fetch_steam_profile() + + channel = self.bot.get_channel(self.bot.in_game_channel_id) + channel = await self.bot.fetch_channel(self.bot.in_game_channel_id) if not channel else channel + + embed = user.embed + embed.title = "Player Has Connected" + embed.colour = Colour.brand_green() + + await channel.send(embed=embed) + + async def handle_player_disconnected(self, line: str): + """ + Report when a user has left the server into a specified Discord channel. + Example of line: + ConnectionManager: [disconnect] "receive-disconnect" connection: guid=*** ip=*** steam-id=*** access=admin username="corbz" connection-type="Disconnected" + """ + re_pattern = r"guid=(\d+)\s+ip=([\d\.]+)\s+steam-id=(\d+)\s+access=(\w*)\s+username=\"([^\"]+)\"\s+connection-type=\"([^\"]+)\"" + re_match = re.search(re_pattern, line) + if not re_match: + log.warning("failed to parse player data: %s", line) + return + + user = ZomboidUser( + guid=re_match.group(1), + ip=re_match.group(2), + steam_id=re_match.group(3), + access=re_match.group(4), + username=re_match.group(5), + connection_type=re_match.group(6) + ) + await user.fetch_steam_profile() + + channel = self.bot.get_channel(self.bot.in_game_channel_id) + channel = await self.bot.fetch_channel(self.bot.in_game_channel_id) if not channel else channel + + embed = user.embed + embed.title = "Player Has Disconnected" + embed.colour = Colour.brand_red() + + await channel.send(embed=embed) + +async def setup(bot: commands.Bot): + cog = PlayersCog(bot) + await bot.add_cog(cog) + log.info("Added %s cog", cog.__class__.__name__)