From 5b62e491de14c5050be4bfa05845e43333568fb7 Mon Sep 17 00:00:00 2001 From: shoucandanghehe Date: Fri, 25 Oct 2024 13:31:10 +0800 Subject: [PATCH 1/3] =?UTF-8?q?=E2=9C=A8=20=E5=AE=9E=E7=8E=B0=E8=8E=B7?= =?UTF-8?q?=E5=8F=96=20leagueflow?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../games/tetrio/api/player.py | 14 +++++++ .../tetrio/api/schemas/labs/leagueflow.py | 38 +++++++++++++++++++ 2 files changed, 52 insertions(+) create mode 100644 nonebot_plugin_tetris_stats/games/tetrio/api/schemas/labs/leagueflow.py diff --git a/nonebot_plugin_tetris_stats/games/tetrio/api/player.py b/nonebot_plugin_tetris_stats/games/tetrio/api/player.py index 2ded6e01..0c199904 100644 --- a/nonebot_plugin_tetris_stats/games/tetrio/api/player.py +++ b/nonebot_plugin_tetris_stats/games/tetrio/api/player.py @@ -11,6 +11,7 @@ from .cache import Cache from .models import TETRIOHistoricalData from .schemas.base import FailedModel +from .schemas.labs.leagueflow import LeagueFlow, LeagueFlowSuccess from .schemas.records.solo import Solo as SoloRecord from .schemas.records.solo import SoloSuccessModel as RecordsSoloSuccessModel from .schemas.summaries import ( @@ -84,6 +85,7 @@ def __init__(self, *, user_id: str | None = None, user_name: str | None = None, self._user_info: UserInfoSuccess | None = None self._summaries: dict[Summaries, SummariesModel] = {} self._records: dict[RecordKey, RecordsSoloSuccessModel] = {} + self._leagueflow: LeagueFlowSuccess | None = None @property def _request_user_parameter(self) -> str: @@ -161,6 +163,18 @@ async def get_summaries(self, summaries_type: Summaries) -> SummariesModel: ) return self._summaries[summaries_type] + async def get_leagueflow(self) -> LeagueFlowSuccess: + if self._leagueflow is None: + leagueflow: LeagueFlow = type_validate_json( + LeagueFlow, # type: ignore[arg-type] + await Cache.get(BASE_URL / 'labs/leagueflow' / self._request_user_parameter), + ) + if isinstance(leagueflow, FailedModel): + msg = f'League 历史记录请求错误:\n{leagueflow.error}' + raise RequestError(msg) + self._leagueflow = leagueflow + return self._leagueflow + @property async def sprint(self) -> SummariesSoloSuccessModel: return await self.get_summaries('40l') diff --git a/nonebot_plugin_tetris_stats/games/tetrio/api/schemas/labs/leagueflow.py b/nonebot_plugin_tetris_stats/games/tetrio/api/schemas/labs/leagueflow.py new file mode 100644 index 00000000..591856fd --- /dev/null +++ b/nonebot_plugin_tetris_stats/games/tetrio/api/schemas/labs/leagueflow.py @@ -0,0 +1,38 @@ +from datetime import datetime +from enum import IntEnum +from typing import NamedTuple + +from pydantic import BaseModel, Field + +from ..base import FailedModel +from ..base import SuccessModel as BaseSuccessModel + + +class Result(IntEnum): + VICTORY = 1 + DEFEAT = 2 + VICTORY_BY_DISQUALIFICATION = 3 + DEFEAT_BY_DISQUALIFICATION = 4 + TIE = 5 + NO_CONTEST = 6 + MATCH_NULLIFIED = 7 + + +class Point(NamedTuple): + timestamp_offset: int + result: Result + post_match_tr: int + opponent_pre_match_tr: int + """If the opponent was unranked, same as post_match_tr.""" + + +class Data(BaseModel): + start_time: datetime = Field(..., alias='startTime') + points: list[Point] + + +class LeagueFlowSuccess(BaseSuccessModel): + data: Data + + +LeagueFlow = LeagueFlowSuccess | FailedModel From 8c011859ddd2527f553d1d2171d927137e6c60cf Mon Sep 17 00:00:00 2001 From: shoucandanghehe Date: Sat, 26 Oct 2024 17:41:54 +0800 Subject: [PATCH 2/3] =?UTF-8?q?=E2=9C=A8=20TETR.IO=20=E9=80=82=E9=85=8D=20?= =?UTF-8?q?v1=20=E6=A8=A1=E6=9D=BF?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../games/tetrio/query/__init__.py | 134 ++++++++++++++ .../games/tetrio/query/tools.py | 52 ++++++ .../games/tetrio/query/v1.py | 172 ++++++++++++++++++ .../games/tetrio/{query.py => query/v2.py} | 157 ++-------------- 4 files changed, 375 insertions(+), 140 deletions(-) create mode 100644 nonebot_plugin_tetris_stats/games/tetrio/query/__init__.py create mode 100644 nonebot_plugin_tetris_stats/games/tetrio/query/tools.py create mode 100644 nonebot_plugin_tetris_stats/games/tetrio/query/v1.py rename nonebot_plugin_tetris_stats/games/tetrio/{query.py => query/v2.py} (50%) diff --git a/nonebot_plugin_tetris_stats/games/tetrio/query/__init__.py b/nonebot_plugin_tetris_stats/games/tetrio/query/__init__.py new file mode 100644 index 00000000..115ace16 --- /dev/null +++ b/nonebot_plugin_tetris_stats/games/tetrio/query/__init__.py @@ -0,0 +1,134 @@ +from datetime import timezone + +from arclet.alconna import Arg, ArgFlag +from nonebot import get_driver +from nonebot.adapters import Event +from nonebot.matcher import Matcher +from nonebot_plugin_alconna import Args, At, Option, Subcommand +from nonebot_plugin_alconna.uniseg import UniMessage +from nonebot_plugin_orm import get_session +from nonebot_plugin_session import EventSession +from nonebot_plugin_session_orm import get_session_persist_id # type: ignore[import-untyped] +from nonebot_plugin_user import User as NBUser +from nonebot_plugin_user import get_user +from sqlalchemy import select + +from ....db import query_bind_info, trigger +from ....utils.exception import FallbackError +from ....utils.typing import Me +from ... import add_block_handlers, alc +from ...constant import CANT_VERIFY_MESSAGE +from .. import command, get_player +from ..api import Player +from ..constant import GAME_TYPE +from ..models import TETRIOUserConfig +from ..typing import Template +from .v1 import make_query_image_v1 +from .v2 import make_query_image_v2 + +UTC = timezone.utc + +driver = get_driver() + +command.add( + Subcommand( + 'query', + Args( + Arg( + 'target', + At | Me, + notice='@想要查询的人 / 自己', + flags=[ArgFlag.HIDDEN, ArgFlag.OPTIONAL], + ), + Arg( + 'account', + get_player, + notice='TETR.IO 用户名 / ID', + flags=[ArgFlag.HIDDEN, ArgFlag.OPTIONAL], + ), + ), + Option( + '--template', + Arg('template', Template), + alias=['-T'], + help_text='要使用的查询模板', + ), + help_text='查询 TETR.IO 游戏信息', + ), +) + +alc.shortcut( + '(?i:io)(?i:查询|查|query|stats)', + command='tstats TETR.IO query', + humanized='io查', +) +alc.shortcut( + 'fkosk', + command='tstats TETR.IO query', + arguments=['我'], + fuzzy=False, + humanized='An Easter egg!', +) + +add_block_handlers(alc.assign('TETRIO.query')) + + +async def make_query_result(player: Player, template: Template) -> UniMessage: + if template == 'v1': + try: + return UniMessage.image(raw=await make_query_image_v1(player)) + except FallbackError: + template = 'v2' + if template == 'v2': + return UniMessage.image(raw=await make_query_image_v2(player)) + return None + + +@alc.assign('TETRIO.query') +async def _( # noqa: PLR0913 + user: NBUser, + event: Event, + matcher: Matcher, + target: At | Me, + event_session: EventSession, + template: Template | None = None, +): + async with trigger( + session_persist_id=await get_session_persist_id(event_session), + game_platform=GAME_TYPE, + command_type='query', + command_args=[f'--template {template}'] if template is not None else [], + ): + async with get_session() as session: + bind = await query_bind_info( + session=session, + user=await get_user( + event_session.platform, target.target if isinstance(target, At) else event.get_user_id() + ), + game_platform=GAME_TYPE, + ) + if template is None: + template = await session.scalar( + select(TETRIOUserConfig.query_template).where(TETRIOUserConfig.id == user.id) + ) + if bind is None: + await matcher.finish('未查询到绑定信息') + message = UniMessage(CANT_VERIFY_MESSAGE) + player = Player(user_id=bind.game_account, trust=True) + await (message + await make_query_result(player, template or 'v1')).finish() + + +@alc.assign('TETRIO.query') +async def _(user: NBUser, account: Player, event_session: EventSession, template: Template | None = None): + async with trigger( + session_persist_id=await get_session_persist_id(event_session), + game_platform=GAME_TYPE, + command_type='query', + command_args=[f'--template {template}'] if template is not None else [], + ): + async with get_session() as session: + if template is None: + template = await session.scalar( + select(TETRIOUserConfig.query_template).where(TETRIOUserConfig.id == user.id) + ) + await (await make_query_result(account, template or 'v1')).finish() diff --git a/nonebot_plugin_tetris_stats/games/tetrio/query/tools.py b/nonebot_plugin_tetris_stats/games/tetrio/query/tools.py new file mode 100644 index 00000000..8fb029f3 --- /dev/null +++ b/nonebot_plugin_tetris_stats/games/tetrio/query/tools.py @@ -0,0 +1,52 @@ +from collections.abc import Callable +from datetime import timedelta +from typing import TypeVar, overload +from zoneinfo import ZoneInfo + +from ....utils.exception import FallbackError +from ....utils.render.schemas.tetrio.user.base import TetraLeagueHistoryData +from ..api.schemas.labs.leagueflow import LeagueFlowSuccess +from ..api.schemas.summaries.league import LeagueSuccessModel, NeverPlayedData, NeverRatedData, RatedData + + +def flow_to_history( + leagueflow: LeagueFlowSuccess, + handle: Callable[[list[TetraLeagueHistoryData]], list[TetraLeagueHistoryData]] | None = None, +) -> list[TetraLeagueHistoryData]: + start_time = leagueflow.data.start_time.astimezone(ZoneInfo('Asia/Shanghai')) + ret = [ + TetraLeagueHistoryData( + record_at=start_time + timedelta(milliseconds=i.timestamp_offset), + tr=i.post_match_tr, + ) + for i in leagueflow.data.points + if start_time + timedelta(milliseconds=i.timestamp_offset) + ] + return ret if handle is None else handle(ret) + + +N = TypeVar('N', int, float) + + +def handling_special_value(value: N) -> N | None: + return value if value != -1 else None + + +L = TypeVar('L', NeverPlayedData, NeverRatedData, RatedData) + + +@overload +def get_league_data(user_info: LeagueSuccessModel, league_type: type[L]) -> L: ... +@overload +def get_league_data( + user_info: LeagueSuccessModel, league_type: None = None +) -> NeverPlayedData | NeverRatedData | RatedData: ... +def get_league_data( + user_info: LeagueSuccessModel, league_type: type[L] | None = None +) -> L | NeverPlayedData | NeverRatedData | RatedData: + league = user_info.data + if league_type is None: + return league + if isinstance(league, league_type): + return league + raise FallbackError diff --git a/nonebot_plugin_tetris_stats/games/tetrio/query/v1.py b/nonebot_plugin_tetris_stats/games/tetrio/query/v1.py new file mode 100644 index 00000000..b9b3a9db --- /dev/null +++ b/nonebot_plugin_tetris_stats/games/tetrio/query/v1.py @@ -0,0 +1,172 @@ +from asyncio import gather +from datetime import datetime, timedelta +from hashlib import md5 +from math import ceil, floor +from zoneinfo import ZoneInfo + +from yarl import URL + +from ....utils.exception import FallbackError +from ....utils.host import HostPage, get_self_netloc +from ....utils.render import render +from ....utils.render.schemas.base import Avatar, Ranking +from ....utils.render.schemas.tetrio.user.base import TetraLeagueHistoryData +from ....utils.render.schemas.tetrio.user.info_v1 import Info, Radar, TetraLeague, TetraLeagueHistory, User +from ....utils.screenshot import screenshot +from ..api import Player +from ..api.schemas.summaries.league import RatedData +from ..constant import TR_MAX, TR_MIN +from .tools import flow_to_history, get_league_data + + +def get_value_bounds(values: list[int | float]) -> tuple[int, int]: + value_max = 10 * ceil(max(values) / 10) + value_min = 10 * floor(min(values) / 10) + return value_max, value_min + + +def get_split(value_max: int, value_min: int) -> tuple[int, int]: + offset = 0 + overflow = 0 + + while True: + if (new_max_value := value_max + offset + overflow) > TR_MAX: + overflow -= 1 + continue + if (new_min_value := value_min - offset + overflow) < TR_MIN: + overflow += 1 + continue + if ((new_max_value - new_min_value) / 40).is_integer(): + split_value = int((value_max + offset - (value_min - offset)) / 4) + break + offset += 1 + return split_value, offset + overflow + + +def get_specified_point( + previous_point: TetraLeagueHistoryData, + behind_point: TetraLeagueHistoryData, + point_time: datetime, +) -> TetraLeagueHistoryData: + """根据给出的 previous_point 和 behind_point, 推算 point_time 点处的数据 + + Args: + previous_point (Data): 前面的数据点 + behind_point (Data): 后面的数据点 + point_time (datetime): 要推算的点的位置 + + Returns: + Data: 要推算的点的数据 + """ + # 求两个点的斜率 + slope = (behind_point.tr - previous_point.tr) / ( + datetime.timestamp(behind_point.record_at) - datetime.timestamp(previous_point.record_at) + ) + return TetraLeagueHistoryData( + record_at=point_time, + tr=previous_point.tr + slope * (datetime.timestamp(point_time) - datetime.timestamp(previous_point.record_at)), + ) + + +def handle_history_data(data: list[TetraLeagueHistoryData]) -> list[TetraLeagueHistoryData]: + data.sort(key=lambda x: x.record_at) + + right_border = datetime.now(ZoneInfo('Asia/Shanghai')).replace(hour=0, minute=0, second=0, microsecond=0) + left_border = right_border - timedelta(days=9) + + lefts: list[TetraLeagueHistoryData] = [] + in_border: list[TetraLeagueHistoryData] = [] + rights: list[TetraLeagueHistoryData] = [] + for i in data: + if i.record_at < left_border: + lefts.append(i) + elif i.record_at < right_border: + in_border.append(i) + else: + rights.append(i) + ret: list[TetraLeagueHistoryData] = [] + if lefts: + ret.append(get_specified_point(lefts[-1], in_border[0], left_border)) + else: + ret.append(TetraLeagueHistoryData(tr=in_border[0].tr, record_at=left_border)) + ret.extend(in_border) + if rights: + ret.append(get_specified_point(in_border[-1], rights[0], right_border.replace(microsecond=1000))) + else: + ret.append(TetraLeagueHistoryData(tr=in_border[-1].tr, record_at=right_border.replace(microsecond=1000))) + return ret + + +async def make_query_image_v1(player: Player) -> bytes: + ( + (user, user_info, league, sprint, blitz, leagueflow), + (avatar_revision,), + ) = await gather( + gather(player.user, player.get_info(), player.league, player.sprint, player.blitz, player.get_leagueflow()), + gather(player.avatar_revision), + ) + league_data = get_league_data(league, RatedData) + if league_data.vs is None: + raise FallbackError + histories = flow_to_history(leagueflow, handle_history_data) + value_max, value_min = get_value_bounds([i.tr for i in histories]) + split_value, offset = get_split(value_max, value_min) + if sprint.data.record is not None: + duration = timedelta(milliseconds=sprint.data.record.results.stats.finaltime).total_seconds() + sprint_value = f'{duration:.3f}s' if duration < 60 else f'{duration // 60:.0f}m {duration % 60:.3f}s' # noqa: PLR2004 + else: + sprint_value = 'N/A' + blitz_value = f'{blitz.data.record.results.stats.score:,}' if blitz.data.record is not None else 'N/A' + netloc = get_self_netloc() + async with HostPage( + page=await render( + 'v1/tetrio/info', + Info( + user=User( + avatar=str( + URL(f'http://{netloc}/host/resource/tetrio/avatars/{user.ID}') % {'revision': avatar_revision} + ) + if avatar_revision is not None and avatar_revision != 0 + else Avatar( + type='identicon', + hash=md5(user.ID.encode()).hexdigest(), # noqa: S324 + ), + name=user.name.upper(), + bio=user_info.data.bio, + ), + ranking=Ranking( + rating=round(league_data.glicko, 2), + rd=round(league_data.rd, 2), + ), + tetra_league=TetraLeague( + rank=league_data.rank, + tr=round(league_data.tr, 2), + global_rank=league_data.standing, + pps=league_data.pps, + lpm=round(lpm := (league_data.pps * 24), 2), + apm=league_data.apm, + apl=round(league_data.apm / lpm, 2), + vs=league_data.vs, + adpm=round(adpm := (league_data.vs * 0.6), 2), + adpl=round(adpm / lpm, 2), + ), + tetra_league_history=TetraLeagueHistory( + data=histories, + split_interval=split_value, + min_tr=value_min, + max_tr=value_max, + offset=offset, + ), + radar=Radar( + app=(app := (league_data.apm / (60 * league_data.pps))), + dsps=(dsps := ((league_data.vs / 100) - (league_data.apm / 60))), + dspp=(dspp := (dsps / league_data.pps)), + ci=150 * dspp - 125 * app + 50 * (league_data.vs / league_data.apm) - 25, + ge=2 * ((app * dsps) / league_data.pps), + ), + sprint=sprint_value, + blitz=blitz_value, + ), + ) + ) as page_hash: + return await screenshot(f'http://{netloc}/host/{page_hash}.html') diff --git a/nonebot_plugin_tetris_stats/games/tetrio/query.py b/nonebot_plugin_tetris_stats/games/tetrio/query/v2.py similarity index 50% rename from nonebot_plugin_tetris_stats/games/tetrio/query.py rename to nonebot_plugin_tetris_stats/games/tetrio/query/v2.py index 88bfe000..1b8e57cd 100644 --- a/nonebot_plugin_tetris_stats/games/tetrio/query.py +++ b/nonebot_plugin_tetris_stats/games/tetrio/query/v2.py @@ -1,160 +1,37 @@ from asyncio import gather -from datetime import datetime, timedelta, timezone +from datetime import datetime, timedelta from hashlib import md5 -from typing import TypeVar -from arclet.alconna import Arg, ArgFlag -from nonebot import get_driver -from nonebot.adapters import Event -from nonebot.matcher import Matcher -from nonebot_plugin_alconna import Args, At, Option, Subcommand -from nonebot_plugin_alconna.uniseg import UniMessage -from nonebot_plugin_orm import get_session -from nonebot_plugin_session import EventSession -from nonebot_plugin_session_orm import get_session_persist_id # type: ignore[import-untyped] -from nonebot_plugin_user import User as NBUser -from nonebot_plugin_user import get_user -from sqlalchemy import select from yarl import URL -from ...db import query_bind_info, trigger -from ...utils.host import HostPage, get_self_netloc -from ...utils.metrics import get_metrics -from ...utils.render import render -from ...utils.render.schemas.base import Avatar -from ...utils.render.schemas.tetrio.user.info_v2 import ( +from ....utils.host import HostPage, get_self_netloc +from ....utils.metrics import get_metrics +from ....utils.render import render +from ....utils.render.schemas.base import Avatar +from ....utils.render.schemas.tetrio.user.info_v2 import ( Badge, Blitz, + Info, Sprint, Statistic, TetraLeague, TetraLeagueStatistic, + User, Zen, ) -from ...utils.render.schemas.tetrio.user.info_v2 import Info as V2TemplateInfo -from ...utils.render.schemas.tetrio.user.info_v2 import User as V2TemplateUser -from ...utils.screenshot import screenshot -from ...utils.typing import Me -from .. import add_block_handlers, alc -from ..constant import CANT_VERIFY_MESSAGE -from . import command, get_player -from .api import Player -from .api.schemas.summaries.league import NeverPlayedData, NeverRatedData -from .constant import GAME_TYPE -from .models import TETRIOUserConfig -from .typing import Template - -UTC = timezone.utc - -driver = get_driver() - -command.add( - Subcommand( - 'query', - Args( - Arg( - 'target', - At | Me, - notice='@想要查询的人 / 自己', - flags=[ArgFlag.HIDDEN, ArgFlag.OPTIONAL], - ), - Arg( - 'account', - get_player, - notice='TETR.IO 用户名 / ID', - flags=[ArgFlag.HIDDEN, ArgFlag.OPTIONAL], - ), - ), - Option( - '--template', - Arg('template', Template), - alias=['-T'], - help_text='要使用的查询模板', - ), - help_text='查询 TETR.IO 游戏信息', - ), -) - -alc.shortcut( - '(?i:io)(?i:查询|查|query|stats)', - command='tstats TETR.IO query', - humanized='io查', -) -alc.shortcut( - 'fkosk', - command='tstats TETR.IO query', - arguments=['我'], - fuzzy=False, - humanized='An Easter egg!', -) - -add_block_handlers(alc.assign('TETRIO.query')) - - -@alc.assign('TETRIO.query') -async def _( # noqa: PLR0913 - user: NBUser, - event: Event, - matcher: Matcher, - target: At | Me, - event_session: EventSession, - template: Template | None = None, -): - async with trigger( - session_persist_id=await get_session_persist_id(event_session), - game_platform=GAME_TYPE, - command_type='query', - command_args=[f'--template {template}'] if template is not None else [], - ): - async with get_session() as session: - bind = await query_bind_info( - session=session, - user=await get_user( - event_session.platform, target.target if isinstance(target, At) else event.get_user_id() - ), - game_platform=GAME_TYPE, - ) - if template is None: - template = await session.scalar( - select(TETRIOUserConfig.query_template).where(TETRIOUserConfig.id == user.id) - ) - if bind is None: - await matcher.finish('未查询到绑定信息') - message = UniMessage(CANT_VERIFY_MESSAGE) - player = Player(user_id=bind.game_account, trust=True) - await (message + UniMessage.image(raw=await make_query_image_v2(player))).finish() - - -@alc.assign('TETRIO.query') -async def _(user: NBUser, account: Player, event_session: EventSession, template: Template | None = None): - async with trigger( - session_persist_id=await get_session_persist_id(event_session), - game_platform=GAME_TYPE, - command_type='query', - command_args=[f'--template {template}'] if template is not None else [], - ): - async with get_session() as session: - if template is None: - template = await session.scalar( - select(TETRIOUserConfig.query_template).where(TETRIOUserConfig.id == user.id) - ) - await (UniMessage.image(raw=await make_query_image_v2(account))).finish() - - -N = TypeVar('N', int, float) - - -def handling_special_value(value: N) -> N | None: - return value if value != -1 else None +from ....utils.screenshot import screenshot +from ..api import Player +from ..api.schemas.summaries.league import NeverPlayedData, NeverRatedData +from .tools import flow_to_history, handling_special_value async def make_query_image_v2(player: Player) -> bytes: ( (user, user_info, league, sprint, blitz, zen), - (avatar_revision, banner_revision), + (avatar_revision, banner_revision, leagueflow), ) = await gather( gather(player.user, player.get_info(), player.league, player.sprint, player.blitz, player.zen), - gather(player.avatar_revision, player.banner_revision), + gather(player.avatar_revision, player.banner_revision, player.get_leagueflow()), ) if sprint.data.record is not None: duration = timedelta(milliseconds=sprint.data.record.results.stats.finaltime).total_seconds() @@ -176,8 +53,8 @@ async def make_query_image_v2(player: Player) -> bytes: async with HostPage( await render( 'v2/tetrio/user/info', - V2TemplateInfo( - user=V2TemplateUser( + Info( + user=User( id=user.ID, name=user.name.upper(), bio=user_info.data.bio, @@ -227,7 +104,7 @@ async def make_query_image_v2(player: Player) -> bytes: adpl=metrics.adpl, statistic=TetraLeagueStatistic(total=league.data.gamesplayed, wins=league.data.gameswon), decaying=league.data.decaying, - history=None, + history=flow_to_history(leagueflow), ) if not isinstance(league.data, NeverPlayedData) else None, From 6a129d2ebbc65218d1d172e72bf549a71aee3a53 Mon Sep 17 00:00:00 2001 From: shoucandanghehe Date: Sat, 26 Oct 2024 18:03:23 +0800 Subject: [PATCH 3/3] =?UTF-8?q?=E2=9C=A8=20=E9=99=90=E5=88=B6=20v2=20histo?= =?UTF-8?q?ry=20=E6=95=B0=E9=87=8F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../games/tetrio/query/v2.py | 154 +++++++++--------- 1 file changed, 79 insertions(+), 75 deletions(-) diff --git a/nonebot_plugin_tetris_stats/games/tetrio/query/v2.py b/nonebot_plugin_tetris_stats/games/tetrio/query/v2.py index 1b8e57cd..7a5926ad 100644 --- a/nonebot_plugin_tetris_stats/games/tetrio/query/v2.py +++ b/nonebot_plugin_tetris_stats/games/tetrio/query/v2.py @@ -50,84 +50,88 @@ async def make_query_image_v2(player: Player) -> bytes: else: play_time = game_time netloc = get_self_netloc() - async with HostPage( - await render( - 'v2/tetrio/user/info', - Info( - user=User( - id=user.ID, - name=user.name.upper(), - bio=user_info.data.bio, - banner=str( - URL(f'http://{netloc}/host/resource/tetrio/banners/{user.ID}') % {'revision': banner_revision} + async with ( + HostPage( + await render( + 'v2/tetrio/user/info', + Info( + user=User( + id=user.ID, + name=user.name.upper(), + bio=user_info.data.bio, + banner=str( + URL(f'http://{netloc}/host/resource/tetrio/banners/{user.ID}') + % {'revision': banner_revision} + ) + if banner_revision is not None and banner_revision != 0 + else None, + avatar=str( + URL(f'http://{netloc}/host/resource/tetrio/avatars/{user.ID}') + % {'revision': avatar_revision} + ) + if avatar_revision is not None and avatar_revision != 0 + else Avatar( + type='identicon', + hash=md5(user.ID.encode()).hexdigest(), # noqa: S324 + ), + badges=[ + Badge( + id=i.id, + description=i.label, + group=i.group, + receive_at=i.ts if isinstance(i.ts, datetime) else None, + ) + for i in user_info.data.badges + ], + country=user_info.data.country, + role=user_info.data.role, + xp=user_info.data.xp, + friend_count=user_info.data.friend_count, + supporter_tier=user_info.data.supporter_tier, + bad_standing=user_info.data.badstanding or False, + playtime=play_time, + join_at=user_info.data.ts, + ), + tetra_league=TetraLeague( + rank=league.data.rank, + highest_rank='z' if isinstance(league.data, NeverRatedData) else league.data.bestrank, + tr=round(league.data.tr, 2), + glicko=round(league.data.glicko, 2), + rd=round(league.data.rd, 2), + global_rank=league.data.standing, + country_rank=league.data.standing_local, + pps=(metrics := get_metrics(pps=league.data.pps, apm=league.data.apm, vs=league.data.vs)).pps, + apm=metrics.apm, + apl=metrics.apl, + vs=metrics.vs, + adpl=metrics.adpl, + statistic=TetraLeagueStatistic(total=league.data.gamesplayed, wins=league.data.gameswon), + decaying=league.data.decaying, + history=flow_to_history(leagueflow, lambda x: x[-100:]), ) - if banner_revision is not None and banner_revision != 0 + if not isinstance(league.data, NeverPlayedData) else None, - avatar=str( - URL(f'http://{netloc}/host/resource/tetrio/avatars/{user.ID}') % {'revision': avatar_revision} - ) - if avatar_revision is not None and avatar_revision != 0 - else Avatar( - type='identicon', - hash=md5(user.ID.encode()).hexdigest(), # noqa: S324 + statistic=Statistic( + total=handling_special_value(user_info.data.gamesplayed), + wins=handling_special_value(user_info.data.gameswon), ), - badges=[ - Badge( - id=i.id, - description=i.label, - group=i.group, - receive_at=i.ts if isinstance(i.ts, datetime) else None, - ) - for i in user_info.data.badges - ], - country=user_info.data.country, - role=user_info.data.role, - xp=user_info.data.xp, - friend_count=user_info.data.friend_count, - supporter_tier=user_info.data.supporter_tier, - bad_standing=user_info.data.badstanding or False, - playtime=play_time, - join_at=user_info.data.ts, - ), - tetra_league=TetraLeague( - rank=league.data.rank, - highest_rank='z' if isinstance(league.data, NeverRatedData) else league.data.bestrank, - tr=round(league.data.tr, 2), - glicko=round(league.data.glicko, 2), - rd=round(league.data.rd, 2), - global_rank=league.data.standing, - country_rank=league.data.standing_local, - pps=(metrics := get_metrics(pps=league.data.pps, apm=league.data.apm, vs=league.data.vs)).pps, - apm=metrics.apm, - apl=metrics.apl, - vs=metrics.vs, - adpl=metrics.adpl, - statistic=TetraLeagueStatistic(total=league.data.gamesplayed, wins=league.data.gameswon), - decaying=league.data.decaying, - history=flow_to_history(leagueflow), - ) - if not isinstance(league.data, NeverPlayedData) - else None, - statistic=Statistic( - total=handling_special_value(user_info.data.gamesplayed), - wins=handling_special_value(user_info.data.gameswon), + sprint=Sprint( + time=sprint_value, + global_rank=sprint.data.rank, + play_at=sprint.data.record.ts, + ) + if sprint.data.record is not None + else None, + blitz=Blitz( + score=blitz.data.record.results.stats.score, + global_rank=blitz.data.rank, + play_at=blitz.data.record.ts, + ) + if blitz.data.record is not None + else None, + zen=Zen(level=zen.data.level, score=zen.data.score), ), - sprint=Sprint( - time=sprint_value, - global_rank=sprint.data.rank, - play_at=sprint.data.record.ts, - ) - if sprint.data.record is not None - else None, - blitz=Blitz( - score=blitz.data.record.results.stats.score, - global_rank=blitz.data.rank, - play_at=blitz.data.record.ts, - ) - if blitz.data.record is not None - else None, - zen=Zen(level=zen.data.level, score=zen.data.score), ), - ), - ) as page_hash: + ) as page_hash + ): return await screenshot(f'http://{netloc}/host/{page_hash}.html')