corbz/Spiffo
Archived
1
0
Fork 0
Code Issues 1 Pull Requests Actions Activity

user class and join/leave embeds
All checks were successful
Build and Push Docker Image / build (push) Successful in 12s
Details

Browse Source
This commit is contained in:
Corban-Lee Jones 2024-12-06 12:49:47 +00:00
parent 5e216862a6
commit 21bd1589a4
1 changed files with 68 additions and 23 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

91
cogs/console.py
View File

@ -5,14 +5,34 @@ Reads and handles updates in the server console file.
import re import re
import logging import logging
from pathlib import Path from pathlib import Path
from dataclasses import dataclass
import aiofiles import aiofiles
from discord import Embed, Colour
from discord.ext import commands, tasks from discord.ext import commands, tasks
CONSOLE_FILE_PATH = Path(__file__).parent.parent / "data" / "server-console.txt" CONSOLE_FILE_PATH = Path(__file__).parent.parent / "data" / "server-console.txt"
log = logging.getLogger(__name__) 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_profile_picture_url(self):
return f"https://steamcommunity.com/profiles/{self.steam_id}/profileimage?width=256"
@property
def steam_url(self):
return f"https://steamcommunity.com/profiles/{self.steam_id}"
class ConsoleCog(commands.Cog): class ConsoleCog(commands.Cog):
""" """
Reads and handles the server-console.txt file. Reads and handles the server-console.txt file.
@ -54,6 +74,10 @@ class ConsoleCog(commands.Cog):
self._last_line_number = await file.tell() self._last_line_number = await file.tell()
async def process_console_line(self, line: str): async def process_console_line(self, line: str):
"""
Determine how to handle the given line from the server console.
"""
if "CheckModsNeedUpdate" in line: if "CheckModsNeedUpdate" in line:
await self.handle_mod_needs_update(line) await self.handle_mod_needs_update(line)
@ -67,51 +91,72 @@ class ConsoleCog(commands.Cog):
log.debug("mod update instruction: %s", line) log.debug("mod update instruction: %s", line)
async def handle_player_joined(self, line: str): async def handle_player_joined(self, line: str):
# example for connect: """
# ConnectionManager: [fully-connected] "" connection: guid=1733487070761473 ip=192.168.1.23 steam-id=76561198202697217 access=admin username="corbz" connection-type="UDPRakNet" 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+) ip=([\d\.]+) steam-id=(\d+) access=(\w+) username=\"([^\"]+)\" connection-type=\"([^\"]+)\"" re_pattern = r"guid=(\d+) ip=([\d\.]+) steam-id=(\d+) access=(\w+) username=\"([^\"]+)\" connection-type=\"([^\"]+)\""
re_match = re.search(re_pattern, line) re_match = re.search(re_pattern, line)
if not re_match: if not re_match:
log.warning("failed to parse player data: %s", line) log.warning("failed to parse player data: %s", line)
return return
data = { user = ZomboidUser(
"guid": re_match.group(1), guid=re_match.group(1),
"ip": re_match.group(2), ip=re_match.group(2),
"steam_id": re_match.group(3), steam_id=re_match.group(3),
"access": re_match.group(4), access=re_match.group(4),
"username": re_match.group(5), username=re_match.group(5),
"connection_type": re_match.group(6), connection_type=re_match.group(6),
} )
channel = self.bot.get_channel(self.bot.in_game_channel_id) 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 channel = await self.bot.fetch_channel(self.bot.in_game_channel_id) if not channel else channel
await channel.send(content=f"Player Joined: **{data['username']}**") embed = Embed(
title=user.username,
url=user.steam_url,
description="Player has joined the server",
colour=Colour.brand_green()
)
embed.set_image(user.steam_profile_picture_url)
await channel.send(embed=embed)
async def handle_player_left(self, line: str): async def handle_player_left(self, line: str):
# Example for disconnect: """
# ConnectionManager: [disconnect] "receive-disconnect" connection: guid=1733487070761473 ip=192.168.1.23 steam-id=76561198202697217 access=admin username="corbz" connection-type="Disconnected" 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+) ip=([\d\.]+) steam-id=(\d+) access=(\w+) username=\"([^\"]+)\" connection-type=\"([^\"]+)\"" re_pattern = r"guid=(\d+) ip=([\d\.]+) steam-id=(\d+) access=(\w+) username=\"([^\"]+)\" connection-type=\"([^\"]+)\""
re_match = re.search(re_pattern, line) re_match = re.search(re_pattern, line)
if not re_match: if not re_match:
log.warning("failed to parse player data: %s", line) log.warning("failed to parse player data: %s", line)
return return
data = { user = ZomboidUser(
"guid": re_match.group(1), guid=re_match.group(1),
"ip": re_match.group(2), ip=re_match.group(2),
"steam_id": re_match.group(3), steam_id=re_match.group(3),
"access": re_match.group(4), access=re_match.group(4),
"username": re_match.group(5), username=re_match.group(5),
"connection_type": re_match.group(6), connection_type=re_match.group(6),
} )
channel = self.bot.get_channel(self.bot.in_game_channel_id) 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 channel = await self.bot.fetch_channel(self.bot.in_game_channel_id) if not channel else channel
await channel.send(content=f"Player Left: **{data['username']}**") embed = Embed(
title=user.username,
url=user.steam_url,
description="Player has left the server",
colour=Colour.brand_red()
)
embed.set_image(user.steam_profile_picture_url)
await channel.send(embed=embed)
async def setup(bot: commands.Bot): async def setup(bot: commands.Bot):