✨ 历史tr图表 (#499)

* ✨ 实现获取 leagueflow

* ✨ TETR.IO 适配 v1 模板

* ✨ 限制 v2 history 数量

nonebot_plugin_tetris_stats/games/tetrio/api/player.py

@@ -11,6 +11,7 @@ from ..constant import BASE_URL, USER_ID, USER_NAME
 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 @@ class Player:
         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 @@ class Player:
             )
         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')

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

nonebot_plugin_tetris_stats/games/tetrio/query.py

@@ -1,256 +0,0 @@
-from asyncio import gather
-from datetime import datetime, timedelta, timezone
-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 (
-    Badge,
-    Blitz,
-    Sprint,
-    Statistic,
-    TetraLeague,
-    TetraLeagueStatistic,
-    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
-
-
-async def make_query_image_v2(player: Player) -> bytes:
-    (
-        (user, user_info, league, sprint, blitz, zen),
-        (avatar_revision, banner_revision),
-    ) = await gather(
-        gather(player.user, player.get_info(), player.league, player.sprint, player.blitz, player.zen),
-        gather(player.avatar_revision, player.banner_revision),
-    )
-    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'
-
-    play_time: str | None
-    if (game_time := handling_special_value(user_info.data.gametime)) is not None:
-        if game_time // 3600 > 0:
-            play_time = f'{game_time//3600:.0f}h {game_time % 3600 // 60:.0f}m {game_time % 60:.0f}s'
-        elif game_time // 60 > 0:
-            play_time = f'{game_time//60:.0f}m {game_time % 60:.0f}s'
-        else:
-            play_time = f'{game_time:.0f}s'
-    else:
-        play_time = game_time
-    netloc = get_self_netloc()
-    async with HostPage(
-        await render(
-            'v2/tetrio/user/info',
-            V2TemplateInfo(
-                user=V2TemplateUser(
-                    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=None,
-                )
-                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),
-            ),
-        ),
-    ) as page_hash:
-        return await screenshot(f'http://{netloc}/host/{page_hash}.html')

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()

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

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')

nonebot_plugin_tetris_stats/games/tetrio/query/v2.py

@@ -0,0 +1,137 @@
+from asyncio import gather
+from datetime import datetime, timedelta
+from hashlib import md5
+
+from yarl import URL
+
+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.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, leagueflow),
+    ) = await gather(
+        gather(player.user, player.get_info(), player.league, player.sprint, player.blitz, player.zen),
+        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()
+        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'
+
+    play_time: str | None
+    if (game_time := handling_special_value(user_info.data.gametime)) is not None:
+        if game_time // 3600 > 0:
+            play_time = f'{game_time//3600:.0f}h {game_time % 3600 // 60:.0f}m {game_time % 60:.0f}s'
+        elif game_time // 60 > 0:
+            play_time = f'{game_time//60:.0f}m {game_time % 60:.0f}s'
+        else:
+            play_time = f'{game_time:.0f}s'
+    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}
+                        )
+                        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 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),
+                ),
+            ),
+        ) as page_hash
+    ):
+        return await screenshot(f'http://{netloc}/host/{page_hash}.html')