Corban-Lee Jones
8fecd43f31
All checks were successful
Build and Push Docker Image / build (push) Successful in 15s
150 lines
4.3 KiB
Python
150 lines
4.3 KiB
Python
"""
|
|
Database schemas.
|
|
"""
|
|
|
|
import logging
|
|
from datetime import datetime
|
|
|
|
import httpx
|
|
from tortoise import fields
|
|
from tortoise.queryset import QuerySet
|
|
from tortoise.models import Model
|
|
from discord import Embed
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
|
|
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 PlayerDeath(Model):
|
|
player = fields.ForeignKeyField(
|
|
model_name="models.Player",
|
|
on_delete=fields.CASCADE
|
|
)
|
|
coordinate_x = fields.IntField()
|
|
coordinate_y = fields.IntField()
|
|
coordinate_z = fields.IntField()
|
|
cause = fields.CharField(max_length=32)
|
|
timestamp = fields.DatetimeField(auto_now_add=True)
|
|
|
|
class Meta:
|
|
table = "player_deaths"
|
|
|
|
|
|
class Player(Model):
|
|
"""
|
|
"""
|
|
username = fields.CharField(max_length=20, unique=True)
|
|
last_connection = fields.DatetimeField()
|
|
last_disconnection = fields.DatetimeField()
|
|
play_time_seconds = fields.DecimalField(default=0.0)
|
|
|
|
class Meta:
|
|
table = "players"
|
|
|
|
@property
|
|
def is_online(self) -> bool:
|
|
"""
|
|
"""
|
|
if not self.last_connection:
|
|
return False
|
|
|
|
return (self.last_connection and not self.last_disconnection) \
|
|
or self.last_connection > self.last_disconnection
|
|
|
|
async def get_deaths(self) -> QuerySet[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.debug("Assigning death to player: %s", self.username)
|
|
return await PlayerDeath.create(
|
|
player=self,
|
|
coordinate_x=coord_x,
|
|
coordinate_y=coord_y,
|
|
coordinate_z=coord_z,
|
|
cause=cause,
|
|
timestamp=timestamp
|
|
)
|
|
|
|
async def get_steam_summary(self) -> SteamProfileSummary | None:
|
|
"""
|
|
"""
|
|
return SteamProfileSummary.get_or_none(player=self)
|
|
|
|
async def update_steam_summary(self, steam_id: str | int, steam_api_key: str) -> SteamProfileSummary:
|
|
"""
|
|
"""
|
|
log.debug("Updating Steam summary for player: %s", self.username)
|
|
|
|
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")
|
|
|
|
profile = profiles[0]
|
|
|
|
summary, created = SteamProfileSummary.get_or_create(
|
|
steam_id=steam_id,
|
|
defaults={
|
|
"player": self,
|
|
"profile_name": profile.get("personaname"),
|
|
"url": profile.get("profileurl"),
|
|
"avatar_url": profile.get("avatarfull")
|
|
}
|
|
)
|
|
|
|
if not created:
|
|
summary.profile_name = profile.get("personaname")
|
|
summary.url = profile.get("profileurl")
|
|
summary.avatar_url = profile.get("avatarfull")
|
|
await summary.save()
|
|
|
|
return summary
|
|
|
|
async def get_embed(self) -> Embed:
|
|
summary = await self.get_steam_summary()
|
|
if not summary:
|
|
raise ValueError("You must fetch the steam_profile_summary before creating an embed.")
|
|
|
|
death_count = (await self.get_deaths()).count()
|
|
|
|
embed = Embed(
|
|
title="Player",
|
|
description=(
|
|
f"{self.username} ([{summary.profile_name}]({summary.url}))\n"
|
|
f"Deaths: {death_count}"
|
|
f"Playtime ???"
|
|
)
|
|
)
|
|
embed.set_thumbnail(summary.avatar_url)
|
|
return embed
|