""" 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(null=True) last_disconnection = fields.DatetimeField(null=True) play_time_seconds = fields.IntField(default=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 = await 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 = len(await self.get_deaths()) 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