Corban-Lee Jones
7bb2c8dbef
All checks were successful
Build and Push Docker Image / build (push) Successful in 14s
153 lines
5.0 KiB
Python
153 lines
5.0 KiB
Python
"""
|
|
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__)
|