Corban-Lee Jones
d5dccdb533
All checks were successful
Build and Push Docker Image / build (push) Successful in 15s
259 lines
8.4 KiB
Python
259 lines
8.4 KiB
Python
"""
|
|
Database schemas.
|
|
"""
|
|
|
|
import logging
|
|
from datetime import datetime, timedelta
|
|
from pytz import timezone
|
|
|
|
import httpx
|
|
from tortoise import fields
|
|
from tortoise.models import Model
|
|
from discord import Embed
|
|
|
|
log = logging.getLogger(__name__)
|
|
utc = timezone("UTC")
|
|
|
|
|
|
class SteamProfileSummary(Model):
|
|
player = fields.ForeignKeyField(
|
|
model_name="models.Player",
|
|
on_delete=fields.CASCADE
|
|
)
|
|
steam_id = fields.CharField(max_length=20, unique=True)
|
|
profile_name = fields.CharField(max_length=32)
|
|
url = fields.CharField(max_length=128)
|
|
avatar_url = fields.CharField(max_length=128)
|
|
last_update = fields.DatetimeField(auto_now=True)
|
|
|
|
class Meta:
|
|
table = "steam_profile_summary"
|
|
|
|
|
|
class Coordinates(Model):
|
|
x = fields.IntField()
|
|
y = fields.IntField()
|
|
z = fields.IntField()
|
|
|
|
class Meta:
|
|
table = "ingame_coordinates"
|
|
|
|
|
|
class PlayerDeath(Model):
|
|
player = fields.ForeignKeyField(
|
|
model_name="models.Player",
|
|
on_delete=fields.CASCADE
|
|
)
|
|
cause = fields.CharField(max_length=32)
|
|
timestamp = fields.DatetimeField(auto_now_add=True)
|
|
coordinates = fields.ForeignKeyField(
|
|
model_name="models.Coordinates",
|
|
on_delete=fields.CASCADE
|
|
)
|
|
|
|
class Meta:
|
|
table = "player_deaths"
|
|
|
|
|
|
class PlayerSession(Model):
|
|
player = fields.ForeignKeyField(
|
|
model_name="models.Player",
|
|
on_delete=fields.CASCADE
|
|
)
|
|
connected_at = fields.DatetimeField()
|
|
disconnected_at = fields.DatetimeField(null=True)
|
|
connected_coords = fields.ForeignKeyField(
|
|
model_name="models.Coordinates",
|
|
related_name="connected_coords",
|
|
on_delete=fields.CASCADE
|
|
)
|
|
disconnected_coords = fields.ForeignKeyField(
|
|
model_name="models.Coordinates",
|
|
related_name="disconnected_coords",
|
|
on_delete=fields.CASCADE,
|
|
null=True
|
|
)
|
|
|
|
class Meta:
|
|
table = "player_session"
|
|
|
|
@property
|
|
def playtime(self) -> timedelta:
|
|
if not self.disconnected_at:
|
|
return datetime.now(tz=utc) - self.connected_at
|
|
|
|
return self.disconnected_at - self.connected_at
|
|
|
|
|
|
class Player(Model):
|
|
"""
|
|
"""
|
|
username = fields.CharField(max_length=20, unique=True)
|
|
is_dead = fields.BooleanField(default=False)
|
|
|
|
class Meta:
|
|
table = "players"
|
|
|
|
async def get_playtime(self) -> timedelta:
|
|
log.info("Getting total playtime for player: %s", self.username)
|
|
sessions = await PlayerSession.filter(player=self)
|
|
total_playtime = timedelta()
|
|
|
|
# I know this is terrible efficiency-wise, but the tortoise docs
|
|
# are so bad and the annotations don't work like Django's models. Deal with it!
|
|
for session in sessions:
|
|
log.debug(
|
|
"playtime info:\nsession start: %s\nsession end: %s\nsession playtime: %s",
|
|
session.connected_at,
|
|
session.disconnected_at,
|
|
session.playtime
|
|
)
|
|
total_playtime += session.playtime
|
|
|
|
return total_playtime
|
|
|
|
async def get_deaths(self) -> list[PlayerDeath]:
|
|
return await PlayerDeath.filter(player=self)
|
|
|
|
async def add_death(
|
|
self,
|
|
coord_x: str | int,
|
|
coord_y: str | int,
|
|
coord_z: str | int,
|
|
cause: str,
|
|
timestamp: datetime
|
|
) -> PlayerDeath:
|
|
"""
|
|
"""
|
|
log.info("Assigning death to player: %s", self.username)
|
|
self.is_dead = True
|
|
await self.save()
|
|
coordinates = await Coordinates.create(x=coord_x, y=coord_y, z=coord_z)
|
|
return await PlayerDeath.create(
|
|
player=self,
|
|
cause=cause,
|
|
timestamp=timestamp,
|
|
coordinates=coordinates
|
|
)
|
|
|
|
async def get_latest_session(self, ignore_closed_sessions: bool = False) -> PlayerSession:
|
|
queryset = PlayerSession.filter(player=self)
|
|
if ignore_closed_sessions:
|
|
queryset = queryset.filter(disconnected_at=None)
|
|
|
|
return await queryset.last()
|
|
|
|
async def open_session(
|
|
self,
|
|
timestamp: datetime,
|
|
coord_x: str | int,
|
|
coord_y: str | int,
|
|
coord_z: str | int,
|
|
) -> PlayerSession:
|
|
log.info("creating session for player: %s", self.username)
|
|
existing_session = await self.get_latest_session(ignore_closed_sessions=True)
|
|
if existing_session:
|
|
log.debug("deleting an unfinished session to open a new one")
|
|
await existing_session.delete()
|
|
|
|
coordinates = await Coordinates.create(x=coord_x, y=coord_y, z=coord_z)
|
|
return await PlayerSession.create(
|
|
player=self,
|
|
connected_at=timestamp,
|
|
connected_coords=coordinates
|
|
)
|
|
|
|
async def close_session(
|
|
self,
|
|
timestamp: datetime,
|
|
coord_x: str | int,
|
|
coord_y: str | int,
|
|
coord_z: str | int,
|
|
) -> PlayerSession:
|
|
log.info("closing session for player: %s", self.username)
|
|
current_session = await self.get_latest_session(ignore_closed_sessions=True)
|
|
if not current_session:
|
|
raise ValueError("Tried to close session that doesn't exist.")
|
|
|
|
coordinates = await Coordinates.create(x=coord_x, y=coord_y, z=coord_z)
|
|
current_session.disconnected_coords = coordinates
|
|
current_session.disconnected_at = timestamp
|
|
await current_session.save()
|
|
|
|
async def get_steam_summary(self) -> SteamProfileSummary | None:
|
|
"""
|
|
Returns the linked steam profile summary or `NoneType` if it doesn't exist.
|
|
"""
|
|
return await SteamProfileSummary.get_or_none(player=self)
|
|
|
|
@staticmethod
|
|
async def fetch_steam_summary_data(steam_id: str | int, steam_api_key: str | int) -> dict:
|
|
"""
|
|
Fetches and returns the raw data of a steam profile summary for the given steam user ID.
|
|
"""
|
|
if not steam_api_key:
|
|
raise ValueError("No Steam API key provided, can't get profile summary.")
|
|
|
|
async with httpx.AsyncClient() as client:
|
|
response = await client.get(url=f"https://api.steampowered.com/ISteamUser/GetPlayerSummaries/v2/?key={steam_api_key}&steamids={steam_id}")
|
|
response.raise_for_status()
|
|
data = response.json()
|
|
|
|
profiles = data.get("response", {}).get("players", [])
|
|
if not profiles:
|
|
raise ValueError("No profiles found in response")
|
|
|
|
return profiles[0]
|
|
|
|
async def update_outdated_steam_summary(
|
|
self,
|
|
steam_id: str | int,
|
|
steam_api_key: str
|
|
) -> SteamProfileSummary:
|
|
"""
|
|
Updates the linked steam profile summary if missing or outdated.
|
|
Returns the resulting summary, regardless of whether it's updated or not.
|
|
"""
|
|
log.debug("Checking steam summary for player: %s", self.username)
|
|
|
|
summary = await self.get_steam_summary()
|
|
|
|
# If the summary exists and isn't outdated then return it, no work to be done!
|
|
if summary and summary.last_update + timedelta(days=1) > datetime.now(tz=utc):
|
|
return summary
|
|
|
|
# Update if summary is NoneType or older than 1 day
|
|
log.info("Steam summary missing or outdated, updating: %s", self.username)
|
|
data = await self.fetch_steam_summary_data(steam_id, steam_api_key)
|
|
summary, created = await SteamProfileSummary.update_or_create(
|
|
steam_id=steam_id,
|
|
defaults={
|
|
"player": self,
|
|
"profile_name": data.get("personaname"),
|
|
"url": data.get("profileurl"),
|
|
"avatar_url": data.get("avatarfull")
|
|
}
|
|
)
|
|
return summary
|
|
|
|
async def get_embed(self) -> Embed:
|
|
log.info("Creating an embed for player: %s", self.username)
|
|
summary = await self.get_steam_summary()
|
|
if not summary:
|
|
raise ValueError("You must fetch the steam_profile_summary before creating an embed.")
|
|
|
|
death_count = len(await self.get_deaths())
|
|
playtime = str(await self.get_playtime()).split(".")[0] # remove the miliseconds
|
|
log.debug("death count is: %s and playtime is: %s", death_count, playtime)
|
|
|
|
embed = Embed(
|
|
title="Player",
|
|
description=(
|
|
f"{self.username} ([{summary.profile_name}]({summary.url}))\n"
|
|
f"Deaths: {death_count}\n"
|
|
f"Playtime: {playtime}"
|
|
)
|
|
)
|
|
embed.set_thumbnail(url=summary.avatar_url)
|
|
return embed
|