Corban-Lee Jones
67bb52c767
All checks were successful
Build and Push Docker Image / build (push) Successful in 16s
234 lines
8.1 KiB
Python
234 lines
8.1 KiB
Python
"""
|
|
Handles tasks related to in-game players, such as connect/disconnect alerts.
|
|
"""
|
|
|
|
import re
|
|
import logging
|
|
from os import getenv
|
|
from pathlib import Path
|
|
from dataclasses import dataclass
|
|
from datetime import datetime
|
|
|
|
import httpx
|
|
from discord import Embed, Colour
|
|
from discord.ext import commands, tasks
|
|
from rcon.source import rcon
|
|
|
|
from utils.reader import LogFileReader
|
|
from utils.models import Player, PlayerDeath, create_or_update_player
|
|
|
|
ZOMBOID_FOLDER_PATH = Path(getenv("SPIFFO__ZOMBOID_FOLDER_PATH"))
|
|
LOGS_FOLDER_PATH = ZOMBOID_FOLDER_PATH / "Logs"
|
|
USER_LOG_FILE_PATH = None
|
|
for path in LOGS_FOLDER_PATH.iterdir():
|
|
if path.stem.endswith("_user.txt"):
|
|
USER_LOG_FILE_PATH = path
|
|
break
|
|
|
|
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["avatarfull"]
|
|
# )
|
|
|
|
# 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"
|
|
# "Deaths: ???\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.
|
|
"""
|
|
file_handler: LogFileReader
|
|
|
|
def __init__(self, bot: commands.Bot):
|
|
self.bot = bot
|
|
self.file_handler = LogFileReader(USER_LOG_FILE_PATH)
|
|
self.listen_for_changes.start()
|
|
|
|
@tasks.loop(seconds=3)
|
|
async def listen_for_changes(self):
|
|
for line in await self.file_handler.read():
|
|
await self.process_log_line(line)
|
|
|
|
async def process_log_line(self, line: str):
|
|
log.debug("processing log line")
|
|
|
|
if "died" in line:
|
|
await self.process_player_death(line)
|
|
elif "fully connected" in line:
|
|
await self.process_connected_player(line)
|
|
elif "disconnected player" in line:
|
|
await self.process_disconnected_player(line)
|
|
|
|
async def process_player_death(line: str):
|
|
re_pattern = r"\[(?P<timestamp>[\d\-:\.]+)\] user (?P<username>.+?) died at \((?P<x>\d+),(?P<y>\d+),(?P<z>\d+)\) \((?P<cause>.+?)\)"
|
|
re_match = re.search(re_pattern, line)
|
|
if not re_match:
|
|
log.warning("failed to parse player death log: %s", line)
|
|
return
|
|
|
|
username = re_match.group("username")
|
|
player = await Player.get_or_none(username=username)
|
|
if not player:
|
|
log.warning("Player returned none, cannot add death: %s", username)
|
|
return
|
|
|
|
await player.add_death(
|
|
coord_x=re_match.group("x"),
|
|
coord_y=re_match.group("y"),
|
|
coord_z=re_match.group("z"),
|
|
cause=re_match.group("cause"),
|
|
timestamp=datetime.strptime(
|
|
re_match.group("timestamp"),
|
|
"%m-%d-%y %H:%M:%S.%f"
|
|
)
|
|
)
|
|
await player.save()
|
|
|
|
log.debug("successfully registered player death to %s", re_match.group("username"))
|
|
|
|
async def process_connected_player(self, line: str):
|
|
"""
|
|
"""
|
|
re_pattern = r'\[(?P<timestamp>.*?)\] (?P<steam_id>/*?) "(?P<username>.*?)" fully connected \((?P<coordinates>.*?)\)'
|
|
re_match = re.search(re_pattern, line)
|
|
|
|
if not re_match:
|
|
log.warning("Failed to parse player data: %s", line)
|
|
return
|
|
|
|
player = await Player.get_or_create(username=re_match.group("username"))
|
|
await player.update_steam_summary(self.bot.steam_api_key, re_match.group("steam_id"))
|
|
|
|
channel = self.bot.get_channel(self.bot.in_game_channel_id)
|
|
channel = channel or await self.bot.fetch_channel(self.bot.in_game_channel_id)
|
|
|
|
embed = await player.get_embed()
|
|
embed.title = "Player Has Connected"
|
|
|
|
await channel.send(embed=embed)
|
|
|
|
|
|
async def process_disconnected_player(self, line: str):
|
|
pass
|
|
|
|
|
|
# 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(self.bot.steam_api_key)
|
|
|
|
# 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(self.bot.steam_api_key)
|
|
|
|
# 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__)
|