Spiffo/utils/models.py

"""
Database schemas.
"""

import logging
from datetime import datetime, timedelta

import httpx
from tortoise import fields
from tortoise.functions import Sum
from tortoise.expressions import F
from tortoise.models import Model, Q
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 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=False)
    connected_coords = fields.ForeignKeyField(
        model_name="models.Coordinates",
        on_delete=fields.CASCADE
    )
    disconnected_coords = fields.ForeignKeyField(
        model_name="models.Coordinates",
        on_delete=fields.CASCADE,
        null=True
    )

    class Meta:
        table = "player_session"

    @property
    def playtime(self) -> timedelta:
        if not self.disconnected_at:
            return datetime.now() - 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:
        playtime = await PlayerSession.filter(player=self).exclude(disconnected_at=None).annotate(
            total_playtime=Sum(F("disconnected_at") - F("connected_at"))
        ).values()
        return timedelta(seconds=playtime[0]["total_playtime"].total_seconds() if playtime else timedelta())

    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.debug("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.debug("creating session for player: %s", self.username)
        existing_session = await self.get_latest_session(ignore_closed_sessions=True)
        if existing_session:
            raise ValueError("Tried to open session while an open one exists.")

        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.debug("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:
        """
        """
        return await 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())
        playtime = str(await self.get_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