✨ TETR.IO 添加 list 命令

nonebot_plugin_tetris_stats/db/__init__.py

@@ -11,7 +11,7 @@ from nonebot_plugin_orm import AsyncSession, get_session
 from nonebot_plugin_user import User  # type: ignore[import-untyped]
 from sqlalchemy import select
 
-from ..utils.typing import CommandType, GameType
+from ..utils.typing import AllCommandType, BaseCommandType, GameType, TETRIOCommandType
 from .models import Bind, TriggerHistoricalData
 
 UTC = timezone.utc
@@ -92,7 +92,7 @@ async def anti_duplicate_add(cls: type[T], model: T) -> None:
 async def trigger(
     session_persist_id: int,
     game_platform: Literal['IO'],
-    command_type: CommandType | Literal['rank', 'config', 'record'],
+    command_type: TETRIOCommandType,
     command_args: list[str],
 ) -> AsyncGenerator:
     yield
@@ -103,7 +103,7 @@ async def trigger(
 async def trigger(
     session_persist_id: int,
     game_platform: GameType,
-    command_type: CommandType,
+    command_type: BaseCommandType,
     command_args: list[str],
 ) -> AsyncGenerator:
     yield
@@ -113,7 +113,7 @@ async def trigger(
 async def trigger(
     session_persist_id: int,
     game_platform: GameType,
-    command_type: CommandType | Literal['rank', 'config', 'record'],
+    command_type: AllCommandType,
     command_args: list[str],
 ) -> AsyncGenerator:
     trigger_time = datetime.now(UTC)

nonebot_plugin_tetris_stats/db/models.py

@@ -1,6 +1,6 @@
 from collections.abc import Callable, Sequence
 from datetime import datetime
-from typing import Any, Literal
+from typing import Any
 
 from nonebot.compat import PYDANTIC_V2, type_validate_json
 from nonebot_plugin_orm import Model
@@ -9,7 +9,7 @@ from sqlalchemy import JSON, DateTime, Dialect, String, TypeDecorator
 from sqlalchemy.orm import Mapped, MappedAsDataclass, mapped_column
 from typing_extensions import override
 
-from ..utils.typing import CommandType, GameType
+from ..utils.typing import AllCommandType, GameType
 
 
 class PydanticType(TypeDecorator):
@@ -76,6 +76,6 @@ class TriggerHistoricalData(MappedAsDataclass, Model):
     trigger_time: Mapped[datetime] = mapped_column(DateTime)
     session_persist_id: Mapped[int]
     game_platform: Mapped[GameType] = mapped_column(String(32), index=True)
-    command_type: Mapped[CommandType | Literal['rank', 'config', 'record']] = mapped_column(String(16), index=True)
+    command_type: Mapped[AllCommandType] = mapped_column(String(16), index=True)
     command_args: Mapped[list[str]] = mapped_column(JSON)
     finish_time: Mapped[datetime] = mapped_column(DateTime)

nonebot_plugin_tetris_stats/games/tetrio/__init__.py

@@ -82,6 +82,14 @@ command.add(
                 ),
             ),
         ),
+        Subcommand(
+            'list',
+            Option('--max-tr', Arg('max_tr', float), help_text='TR的上限'),
+            Option('--min-tr', Arg('min_tr', float), help_text='TR的下限'),
+            Option('--limit', Arg('limit', int), help_text='查询数量'),
+            Option('--country', Arg('country', str), help_text='国家代码'),
+            help_text='查询 TETR.IO 段位排行榜',
+        ),
         Subcommand(
             'rank',
             Args(Arg('rank', ValidRank, notice='TETR.IO 段位')),
@@ -155,11 +163,12 @@ alc.shortcut(
 
 add_block_handlers(alc.assign('TETRIO.query'))
 
-from . import bind, config, query, rank, record  # noqa: E402
+from . import bind, config, list, query, rank, record  # noqa: E402
 
 __all__ = [
     'bind',
     'config',
+    'list',
     'query',
     'rank',
     'record',

nonebot_plugin_tetris_stats/games/tetrio/api/schemas/tetra_league.py

@@ -10,7 +10,7 @@ class _User(BaseModel):
     username: str
     role: str
     xp: float
-    supporter: bool
+    supporter: bool | None = None
     verified: bool
     country: str | None = None
 

nonebot_plugin_tetris_stats/games/tetrio/api/tetra_league.py

@@ -1,4 +1,5 @@
-from typing import Literal, NamedTuple, overload
+from typing import Literal, NamedTuple, TypedDict, overload
+from urllib.parse import urlencode
 
 from nonebot.compat import type_validate_json
 
@@ -10,6 +11,24 @@ from .schemas.base import FailedModel
 from .schemas.tetra_league import TetraLeague, TetraLeagueSuccess
 
 
+class Parameter(TypedDict, total=False):
+    after: float
+    before: float
+    limit: int
+    country: str
+
+
+async def leaderboard(parameter: Parameter | None = None) -> TetraLeagueSuccess:
+    league: TetraLeague = type_validate_json(
+        TetraLeague,  # type: ignore[arg-type]
+        (await Cache.get(splice_url([BASE_URL, 'users/lists/league', f'?{urlencode(parameter or {})}']))),
+    )
+    if isinstance(league, FailedModel):
+        msg = f'排行榜数据请求错误:\n{league.error}'
+        raise RequestError(msg)
+    return league
+
+
 class FullExport(NamedTuple):
     model: TetraLeagueSuccess
     original: bytes

nonebot_plugin_tetris_stats/games/tetrio/list.py

@@ -0,0 +1,74 @@
+from nonebot_plugin_alconna.uniseg import UniMessage
+from nonebot_plugin_session import EventSession  # type: ignore[import-untyped]
+from nonebot_plugin_session_orm import get_session_persist_id  # type: ignore[import-untyped]
+
+from ...db import trigger
+from ...utils.host import HostPage, get_self_netloc
+from ...utils.metrics import get_metrics
+from ...utils.render import render
+from ...utils.render.schemas.tetrio.tetrio_user_list_v2 import List, TetraLeague, User
+from ...utils.screenshot import screenshot
+from .. import alc
+from .api.schemas.tetra_league import ValidLeague
+from .api.tetra_league import Parameter, leaderboard
+from .constant import GAME_TYPE
+
+
+@alc.assign('TETRIO.list')
+async def _(
+    event_session: EventSession,
+    max_tr: float | None = None,
+    min_tr: float | None = None,
+    limit: int | None = None,
+    country: str | None = None,
+):
+    async with trigger(
+        session_persist_id=await get_session_persist_id(event_session),
+        game_platform=GAME_TYPE,
+        command_type='list',
+        command_args=[],
+    ):
+        parameter: Parameter = {}
+        if max_tr is not None:
+            parameter['after'] = max_tr
+        if min_tr is not None:
+            parameter['before'] = min_tr
+        if limit is not None:
+            parameter['limit'] = limit
+        if country is not None:
+            parameter['country'] = country
+        league = await leaderboard(parameter)
+        async with HostPage(
+            await render(
+                'v2/tetrio/user/list',
+                List(
+                    show_index=True,
+                    users=[
+                        User(
+                            id=i.id,
+                            name=i.username.upper(),
+                            avatar=f'https://tetr.io/user-content/avatars/{i.id}.jpg',
+                            country=i.country,
+                            verified=i.verified,
+                            tetra_league=TetraLeague(
+                                rank=i.league.rank,
+                                tr=round(i.league.rating, 2),
+                                glicko=round(i.league.glicko, 2),
+                                rd=round(i.league.rd, 2),
+                                decaying=i.league.decaying,
+                                pps=(metrics := get_metrics(pps=i.league.pps, apm=i.league.apm, vs=i.league.vs)).pps,
+                                apm=metrics.apm,
+                                apl=metrics.apl,
+                                vs=metrics.vs,
+                                adpl=metrics.adpl,
+                            ),
+                            xp=i.xp,
+                            join_at=None,
+                        )
+                        for i in league.data.users
+                        if isinstance(i.league, ValidLeague)
+                    ],
+                ),
+            )
+        ) as page_hash:
+            await UniMessage.image(raw=await screenshot(f'http://{get_self_netloc()}/host/{page_hash}.html')).finish()

nonebot_plugin_tetris_stats/utils/render/__init__.py

@@ -9,6 +9,7 @@ from .schemas.tetrio.tetrio_info import Info as TETRIOInfo
 from .schemas.tetrio.tetrio_record_blitz import Record as TETRIORecordBlitz
 from .schemas.tetrio.tetrio_record_sprint import Record as TETRIORecordSprint
 from .schemas.tetrio.tetrio_user_info_v2 import Info as TETRIOUserInfoV2
+from .schemas.tetrio.tetrio_user_list_v2 import List as TETRIOUserListV2
 from .schemas.top_info import Info as TOPInfo
 from .schemas.tos_info import Info as TOSInfo
 
@@ -37,6 +38,10 @@ async def render(render_type: Literal['v1/tos/info'], data: TOSInfo) -> str: ...
 async def render(render_type: Literal['v2/tetrio/user/info'], data: TETRIOUserInfoV2) -> str: ...
 
 
+@overload
+async def render(render_type: Literal['v2/tetrio/user/list'], data: TETRIOUserListV2) -> str: ...
+
+
 @overload
 async def render(render_type: Literal['v2/tetrio/record/40l'], data: TETRIORecordSprint) -> str: ...
 
@@ -52,10 +57,18 @@ async def render(
         'v1/top/info',
         'v1/tos/info',
         'v2/tetrio/user/info',
+        'v2/tetrio/user/list',
         'v2/tetrio/record/40l',
         'v2/tetrio/record/blitz',
     ],
-    data: Bind | TETRIOInfo | TOPInfo | TOSInfo | TETRIOUserInfoV2 | TETRIORecordSprint | TETRIORecordBlitz,
+    data: Bind
+    | TETRIOInfo
+    | TOPInfo
+    | TOSInfo
+    | TETRIOUserInfoV2
+    | TETRIOUserListV2
+    | TETRIORecordSprint
+    | TETRIORecordBlitz,
 ) -> str:
     if PYDANTIC_V2:
         return await env.get_template('index.html').render_async(

nonebot_plugin_tetris_stats/utils/render/schemas/tetrio/tetrio_user_list_v2.py

@@ -0,0 +1,37 @@
+from datetime import datetime
+
+from pydantic import BaseModel
+
+from .....games.tetrio.api.typing import Rank
+from ....typing import Number
+from ..base import Avatar
+
+
+class TetraLeague(BaseModel):
+    rank: Rank
+    tr: Number
+
+    glicko: Number | None
+    rd: Number | None
+    decaying: bool
+    pps: Number
+    apm: Number
+    apl: Number
+    vs: Number | None
+    adpl: Number | None
+
+
+class User(BaseModel):
+    id: str
+    name: str
+    avatar: str | Avatar
+    country: str | None
+    verified: bool
+    tetra_league: TetraLeague
+    xp: Number
+    join_at: datetime | None
+
+
+class List(BaseModel):
+    show_index: bool
+    users: list[User]

nonebot_plugin_tetris_stats/utils/typing.py

@@ -2,7 +2,9 @@ from typing import Literal
 
 Number = float | int
 GameType = Literal['IO', 'TOP', 'TOS']
-CommandType = Literal['bind', 'query']
+BaseCommandType = Literal['bind', 'query']
+TETRIOCommandType = BaseCommandType | Literal['rank', 'config', 'list', 'record']
+AllCommandType = BaseCommandType | TETRIOCommandType
 Me = Literal[
     '我',
     '自己',