nonebot-plugin-tetris-stats/nonebot_plugin_tetris_stats/games/tetrio/api/player.py

from enum import Enum
from types import MappingProxyType
from typing import Literal, NamedTuple, cast, overload

from async_lru import alru_cache
from nonebot.compat import type_validate_json

from ....db import anti_duplicate_add
from ....utils.exception import RequestError
from ....utils.request import splice_url
from ..constant import BASE_URL, USER_ID, USER_NAME
from .cache import Cache
from .models import TETRIOHistoricalData
from .schemas.base import FailedModel
from .schemas.records.solo import Solo as SoloRecord
from .schemas.records.solo import SoloSuccessModel as RecordsSoloSuccessModel
from .schemas.summaries import (
    AchievementsSuccessModel,
    SummariesModel,
    ZenithSuccessModel,
    ZenSuccessModel,
)
from .schemas.summaries import (
    SoloSuccessModel as SummariesSoloSuccessModel,
)
from .schemas.summaries.base import User as SummariesUser
from .schemas.user import User
from .schemas.user_info import UserInfo, UserInfoSuccess
from .typing import Records, Summaries


class RecordModeType(str, Enum):
    Sprint = '40l'
    Blitz = 'blitz'


class RecordType(str, Enum):
    Top = 'top'
    Recent = 'recent'
    Progression = 'progression'


class RecordKey(NamedTuple):
    mode_type: RecordModeType
    record_type: RecordType

    def to_records(self) -> Records:
        return cast(Records, f'{self.mode_type.value}_{self.record_type.value}')


class Player:
    __SUMMARIES_MAPPING: MappingProxyType[Summaries, type[SummariesModel]] = MappingProxyType(
        {
            '40l': SummariesSoloSuccessModel,
            'blitz': SummariesSoloSuccessModel,
            'zenith': ZenithSuccessModel,
            'zenithex': ZenithSuccessModel,
            'zen': ZenSuccessModel,
            'achievements': AchievementsSuccessModel,
        }
    )

    @overload
    def __init__(self, *, user_id: str, trust: bool = False): ...
    @overload
    def __init__(self, *, user_name: str, trust: bool = False): ...
    def __init__(self, *, user_id: str | None = None, user_name: str | None = None, trust: bool = False):
        self.user_id = user_id
        self.user_name = user_name
        if not trust:
            if self.user_id is not None:
                if not USER_ID.match(self.user_id):
                    msg = 'Invalid user id'
                    raise ValueError(msg)
            elif self.user_name is not None:
                if not USER_NAME.match(self.user_name):
                    msg = 'Invalid user name'
                    raise ValueError(msg)
            else:
                msg = 'Invalid user'
                raise ValueError(msg)
        self.__user: User | None = None
        self._user_info: UserInfoSuccess | None = None
        self._summaries: dict[Summaries, SummariesModel] = {}
        self._records: dict[RecordKey, RecordsSoloSuccessModel] = {}

    @property
    def _request_user_parameter(self) -> str:
        if self.user_id is not None:
            return self.user_id
        if self.user_name is not None:
            return self.user_name.lower()
        msg = 'Invalid user'
        raise ValueError(msg)

    @property
    async def user(self) -> User:
        if self.__user is not None:
            return self.__user
        if (user := (await self._get_local_summaries_user())) is not None:
            self.__user = User(
                ID=user.id,
                name=user.username,
            )
        else:
            user_info = await self.get_info()
            self.__user = User(
                ID=user_info.data.id,
                name=user_info.data.username,
            )
        self.user_id = self.__user.ID
        self.user_name = self.__user.name
        return self.__user

    async def get_info(self) -> UserInfoSuccess:
        """Get User Info"""
        if self._user_info is None:
            raw_user_info = await Cache.get(splice_url([BASE_URL, 'users/', f'{self._request_user_parameter}']))
            user_info: UserInfo = type_validate_json(UserInfo, raw_user_info)  # type: ignore[arg-type]
            if isinstance(user_info, FailedModel):
                msg = f'用户信息请求错误:\n{user_info.error}'
                raise RequestError(msg)
            self._user_info = user_info
            await anti_duplicate_add(
                TETRIOHistoricalData(
                    user_unique_identifier=(await self.user).unique_identifier,
                    api_type='User Info',
                    data=user_info,
                    update_time=user_info.cache.cached_at,
                ),
            )
        return self._user_info

    @overload
    async def get_summaries(self, summaries_type: Literal['40l', 'blitz']) -> SummariesSoloSuccessModel: ...
    @overload
    async def get_summaries(self, summaries_type: Literal['zenith', 'zenithex']) -> ZenithSuccessModel: ...
    @overload
    async def get_summaries(self, summaries_type: Literal['zen']) -> ZenSuccessModel: ...
    @overload
    async def get_summaries(self, summaries_type: Literal['achievements']) -> AchievementsSuccessModel: ...

    async def get_summaries(self, summaries_type: Summaries) -> SummariesModel:
        if summaries_type not in self._summaries:
            raw_summaries = await Cache.get(
                splice_url([BASE_URL, 'users/', f'{self._request_user_parameter}/', 'summaries/', summaries_type])
            )
            summaries: SummariesModel | FailedModel = type_validate_json(
                self.__SUMMARIES_MAPPING[summaries_type] | FailedModel,  # type: ignore[arg-type]
                raw_summaries,
            )
            if isinstance(summaries, FailedModel):
                msg = f'用户Summaries数据请求错误:\n{summaries.error}'
                raise RequestError(msg)
            self._summaries[summaries_type] = summaries
            await anti_duplicate_add(
                TETRIOHistoricalData(
                    user_unique_identifier=(await self.user).unique_identifier,
                    api_type=summaries_type,
                    data=summaries,
                    update_time=summaries.cache.cached_at,
                ),
            )
        return self._summaries[summaries_type]

    @property
    @alru_cache
    async def sprint(self) -> SummariesSoloSuccessModel:
        return await self.get_summaries('40l')

    @property
    @alru_cache
    async def blitz(self) -> SummariesSoloSuccessModel:
        return await self.get_summaries('blitz')

    @property
    @alru_cache
    async def zen(self) -> ZenSuccessModel:
        return await self.get_summaries('zen')

    async def _get_local_summaries_user(self) -> SummariesUser | None:
        allow_summaries: set[Literal['40l', 'blitz', 'zenith', 'zenithex']] = {
            '40l',
            'blitz',
            'zenith',
            'zenithex',
        }
        if has_summaries := (allow_summaries & self._summaries.keys()):
            for i in has_summaries:
                if (record := (await self.get_summaries(i)).data.record) is not None:
                    return record.user
        return None

    @property
    @alru_cache
    async def avatar_revision(self) -> int | None:
        if self._user_info is not None:
            return self._user_info.data.avatar_revision
        if (user := (await self._get_local_summaries_user())) is not None:
            return user.avatar_revision
        return (await self.get_info()).data.avatar_revision

    @property
    @alru_cache
    async def banner_revision(self) -> int | None:
        if self._user_info is not None:
            return self._user_info.data.banner_revision
        if (user := (await self._get_local_summaries_user())) is not None:
            return user.banner_revision
        return (await self.get_info()).data.banner_revision

    async def get_records(self, mode_type: RecordModeType, records_type: RecordType) -> RecordsSoloSuccessModel:
        if (record_key := RecordKey(mode_type, records_type)) not in self._records:
            raw_records = await Cache.get(
                splice_url(
                    [
                        BASE_URL,
                        'users/',
                        f'{self._request_user_parameter}/',
                        'records/',
                        f'{mode_type}/',
                        records_type,
                    ]
                )
            )
            records: RecordsSoloSuccessModel | FailedModel = type_validate_json(SoloRecord, raw_records)  # type: ignore[arg-type]
            if isinstance(records, FailedModel):
                msg = f'用户Summaries数据请求错误:\n{records.error}'
                raise RequestError(msg)
            self._records[record_key] = records
            await anti_duplicate_add(
                TETRIOHistoricalData(
                    user_unique_identifier=(await self.user).unique_identifier,
                    api_type=record_key.to_records(),
                    data=records,
                    update_time=records.cache.cached_at,
                ),
            )
        return self._records[record_key]