Corban-Lee Jones
d495ffc0fa
All checks were successful
Build and Push Docker Image / build (push) Successful in 16s
127 lines
4.1 KiB
Python
127 lines
4.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__)
|
|
|
|
|
|
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 show_player_join_leave_alert(self, line: str, re_pattern: str, embed_title: str):
|
|
"""
|
|
"""
|
|
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 = embed_title
|
|
|
|
await channel.send(embed=embed)
|
|
|
|
async def process_connected_player(self, line: str):
|
|
"""
|
|
"""
|
|
await self.show_player_join_leave_alert(
|
|
line=line,
|
|
re_pattern=r'\[(?P<timestamp>.*?)\] (?P<steam_id>/*?) "(?P<username>.*?)" fully connected \((?P<coordinates>.*?)\)',
|
|
embed_title="Player Has Connected"
|
|
)
|
|
|
|
async def process_disconnected_player(self, line: str):
|
|
"""
|
|
"""
|
|
await self.show_player_join_leave_alert(
|
|
line=line,
|
|
re_pattern=r'\[(?P<timestamp>.*?)\] (?P<steam_id>/*?) "(?P<username>.*?)" disconnected player \((?P<coordinates>.*?)\)',
|
|
embed_title="Player Has Disconnected"
|
|
)
|
|
|
|
|
|
async def setup(bot: commands.Bot):
|
|
cog = PlayersCog(bot)
|
|
await bot.add_cog(cog)
|
|
log.info("Added %s cog", cog.__class__.__name__)
|