✨ TOS 查数据使用图片回复

nonebot_plugin_tetris_stats/games/tos/query.py

@@ -1,22 +1,33 @@
 from asyncio import gather
-from dataclasses import dataclass
-from typing import Literal
+from datetime import timedelta
+from http import HTTPStatus
+from typing import Literal, NamedTuple
+from urllib.parse import urlunparse
 
 from nonebot.adapters import Bot, Event
 from nonebot.matcher import Matcher
 from nonebot_plugin_alconna import At
 from nonebot_plugin_alconna.uniseg import UniMessage
 from nonebot_plugin_orm import get_session
-from nonebot_plugin_session import EventSession  # type: ignore[import-untyped]
+from nonebot_plugin_session import EventSession  # type: ignore[import-untyped]  # type: ignore[import-untyped]
 from nonebot_plugin_session_orm import get_session_persist_id  # type: ignore[import-untyped]
+from nonebot_plugin_userinfo import EventUserInfo, UserInfo  # type: ignore[import-untyped]
 
 from ...db import query_bind_info, trigger
+from ...utils.avatar import get_avatar
+from ...utils.exception import RequestError
+from ...utils.host import HostPage, get_self_netloc
 from ...utils.metrics import TetrisMetricsProWithLPMADPM, get_metrics
 from ...utils.platform import get_platform
-from ...utils.typing import Me
+from ...utils.render import render
+from ...utils.render.schemas.base import People, Ranking
+from ...utils.render.schemas.tos_info import Info, Multiplayer, Radar
+from ...utils.screenshot import screenshot
+from ...utils.typing import Me, Number
 from ..constant import CANT_VERIFY_MESSAGE
 from . import alc
 from .api import Player
+from .api.schemas.user_info import UserInfoSuccess
 from .constant import GAME_TYPE
 
 
@@ -24,7 +35,12 @@ def add_special_handlers(
     teaid_prefix: Literal['onebot-', 'kook-', 'discord-', 'qqguild-'], match_event: type[Event]
 ) -> None:
     @alc.assign('query')
-    async def _(event: Event, target: At | Me, event_session: EventSession):
+    async def _(
+        event: Event,
+        target: At | Me,
+        event_session: EventSession,
+        event_user_info: UserInfo = EventUserInfo(),  # noqa: B008
+    ):
         if isinstance(event, match_event):
             async with trigger(
                 session_persist_id=await get_session_persist_id(event_session),
@@ -32,16 +48,23 @@ def add_special_handlers(
                 command_type='query',
                 command_args=[],
             ):
-                await (
-                    await make_query_text(
-                        Player(
-                            teaid=f'{teaid_prefix}{target.target}'
-                            if isinstance(target, At)
-                            else f'{teaid_prefix}{event.get_user_id()}',
-                            trust=True,
-                        )
-                    )
-                ).finish()
+                player = Player(
+                    teaid=f'{teaid_prefix}{target.target}'
+                    if isinstance(target, At)
+                    else f'{teaid_prefix}{event.get_user_id()}',
+                    trust=True,
+                )
+                try:
+                    user_info, game_data = await gather(player.get_info(), get_game_data(player))
+                    if game_data is not None:
+                        await UniMessage.image(
+                            raw=await make_query_image(user_info, game_data, event_user_info)
+                        ).finish()
+                    await (await make_query_text(user_info, game_data)).finish()
+                except RequestError as e:
+                    if e.status_code == HTTPStatus.BAD_REQUEST and '未找到此用户' in e.message:
+                        return
+                    raise
 
 
 try:
@@ -76,7 +99,14 @@ except ImportError:
 
 
 @alc.assign('query')
-async def _(bot: Bot, event: Event, matcher: Matcher, target: At | Me, event_session: EventSession):
+async def _(  # noqa: PLR0913
+    bot: Bot,
+    event: Event,
+    matcher: Matcher,
+    target: At | Me,
+    event_session: EventSession,
+    event_user_info: UserInfo = EventUserInfo(),  # noqa: B008
+):
     async with trigger(
         session_persist_id=await get_session_persist_id(event_session),
         game_platform=GAME_TYPE,
@@ -93,24 +123,35 @@ async def _(bot: Bot, event: Event, matcher: Matcher, target: At | Me, event_ses
         if bind is None:
             await matcher.finish('未查询到绑定信息')
         message = CANT_VERIFY_MESSAGE
-        await (message + await make_query_text(Player(teaid=bind.game_account, trust=True))).finish()
+        player = Player(teaid=bind.game_account, trust=True)
+        user_info, game_data = await gather(player.get_info(), get_game_data(player))
+        if game_data is not None:
+            await (
+                message + UniMessage.image(raw=await make_query_image(user_info, game_data, event_user_info))
+            ).finish()
+        await (message + await make_query_text(user_info, game_data)).finish()
 
 
 @alc.assign('query')
-async def _(account: Player, event_session: EventSession):
+async def _(account: Player, event_session: EventSession, event_user_info: UserInfo = EventUserInfo()):  # noqa: B008
     async with trigger(
         session_persist_id=await get_session_persist_id(event_session),
         game_platform=GAME_TYPE,
         command_type='query',
         command_args=[],
     ):
-        await (await make_query_text(account)).finish()
+        user_info, game_data = await gather(account.get_info(), get_game_data(account))
+        if game_data is not None:
+            await UniMessage.image(raw=await make_query_image(user_info, game_data, event_user_info)).finish()
+        await (await make_query_text(user_info, game_data)).finish()
 
 
-@dataclass
-class GameData:
+class GameData(NamedTuple):
     game_num: int
     metrics: TetrisMetricsProWithLPMADPM
+    OR: Number
+    dspp: Number
+    ge: Number
 
 
 async def get_game_data(player: Player, query_num: int = 50) -> GameData | None:
@@ -119,7 +160,7 @@ async def get_game_data(player: Player, query_num: int = 50) -> GameData | None:
     if user_profile.data == []:
         return None
     weighted_total_lpm = weighted_total_apm = weighted_total_adpm = total_time = 0.0
-    num = 0
+    total_attack = total_dig = total_offset = total_pieses = total_receive = num = 0
     for i in user_profile.data:
         # 排除单人局和时间为0的游戏
         # 茶: 不计算没挖掘的局, 即使apm和lpm也如此
@@ -133,6 +174,11 @@ async def get_game_data(player: Player, query_num: int = 50) -> GameData | None:
         weighted_total_lpm += lpm * time
         weighted_total_apm += apm * time
         weighted_total_adpm += adpm * time
+        total_attack += i.attack
+        total_dig += i.dig
+        total_offset += i.offset
+        total_pieses += i.pieces
+        total_receive += i.receive
         total_time += time
         num += 1
         if num >= query_num:
@@ -143,14 +189,51 @@ async def get_game_data(player: Player, query_num: int = 50) -> GameData | None:
     metrics = get_metrics(
         lpm=weighted_total_lpm / total_time, apm=weighted_total_apm / total_time, adpm=weighted_total_adpm / total_time
     )
-    lpm = weighted_total_lpm / total_time
-    apm = weighted_total_apm / total_time
-    adpm = weighted_total_adpm / total_time
-    return GameData(game_num=num, metrics=metrics)
+    return GameData(
+        game_num=num,
+        metrics=metrics,
+        OR=total_offset / total_receive * 100,
+        dspp=total_dig / total_pieses,
+        ge=2 * ((total_attack * total_dig) / total_pieses**2),
+    )
 
 
-async def make_query_text(player: Player) -> UniMessage:
-    user_info, game_data = await gather(player.get_info(), get_game_data(player))
+async def make_query_image(user_info: UserInfoSuccess, game_data: GameData, event_user_info: UserInfo) -> bytes:
+    metrics = game_data.metrics
+    duration = timedelta(milliseconds=float(user_info.data.pb_sprint)).total_seconds()
+    sprint_value = f'{duration:.1f}s' if duration < 60 else f'{duration // 60:.0f}m {duration % 60:.1f}s'  # noqa: PLR2004
+    async with HostPage(
+        await render(
+            'tos/info',
+            Info(
+                user=People(avatar=await get_avatar(event_user_info, 'Data URI', None), name=user_info.data.name),
+                ranking=Ranking(rating=float(user_info.data.ranking), rd=round(float(user_info.data.rd_now), 2)),
+                multiplayer=Multiplayer(
+                    pps=metrics.pps,
+                    lpm=metrics.lpm,
+                    apm=metrics.apm,
+                    apl=metrics.apl,
+                    vs=metrics.vs,
+                    adpm=metrics.adpm,
+                    adpl=metrics.adpl,
+                ),
+                radar=Radar(
+                    app=(app := (metrics.apm / (60 * metrics.pps))),
+                    OR=game_data.OR,
+                    dspp=game_data.dspp,
+                    ci=150 * game_data.dspp - 125 * app + 50 * (metrics.vs / metrics.apm) - 25,
+                    ge=game_data.ge,
+                ),
+                sprint=sprint_value,
+                challenge=f'{int(user_info.data.pb_challenge):,}' if user_info.data.pb_challenge != '0' else 'N/A',
+                marathon=f'{int(user_info.data.pb_marathon):,}' if user_info.data.pb_marathon != '0' else 'N/A',
+            ),
+        )
+    ) as page_hash:
+        return await screenshot(urlunparse(('http', get_self_netloc(), f'/host/{page_hash}.html', '', '', '')))
+
+
+async def make_query_text(user_info: UserInfoSuccess, game_data: GameData | None) -> UniMessage:
     user_data = user_info.data
     message = f'用户 {user_data.name} ({user_data.teaid}) '
     if user_data.ranked_games == '0':