🔖 1.4.5

Bumps [nonebot-plugin-alconna](https://github.com/nonebot/plugin-alconna) from 0.50.3 to 0.51.0.
- [Release notes](https://github.com/nonebot/plugin-alconna/releases)
- [Commits](https://github.com/nonebot/plugin-alconna/compare/v0.50.3...v0.51.0)

---
updated-dependencies:
- dependency-name: nonebot-plugin-alconna
  dependency-type: direct:production
  update-type: version-update:semver-minor
...

Signed-off-by: dependabot[bot] <support@github.com>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>

.github/workflows/TypeCheck.yml

@@ -0,0 +1,27 @@
+name: TypeCheck
+
+on:
+    push:
+
+jobs:
+    Mypy:
+        runs-on: ubuntu-latest
+        steps:
+            - uses: actions/checkout@v4
+
+            - name: Install poetry
+              run: pipx install poetry
+              shell: bash
+
+            - uses: actions/setup-python@v5
+              with:
+                  python-version: '3.10'
+                  cache: 'poetry'
+
+            - run: poetry install
+              shell: bash
+
+            - name: Run Mypy
+              shell: bash
+              run: |
+                  poetry run mypy ./nonebot_plugin_tetris_stats

.gitignore

@@ -20,3 +20,5 @@ bot.py
 TODO
 *.fish
 extracted_skin_mino_*
+sample_*
+TODO*

.pre-commit-config.yaml

@@ -0,0 +1,22 @@
+default_install_hook_types: [pre-commit, prepare-commit-msg]
+ci:
+    autofix_commit_msg: ':rotating_light: auto fix by pre-commit hooks'
+    autofix_prs: true
+    autoupdate_branch: master
+    autoupdate_schedule: monthly
+    autoupdate_commit_msg: ':arrow_up: auto update by pre-commit hooks'
+repos:
+    - repo: https://github.com/astral-sh/ruff-pre-commit
+      rev: v0.4.10
+      hooks:
+          - id: ruff
+            args: [--fix, --exit-non-zero-on-fix]
+            stages: [commit]
+          - id: ruff-format
+            stages: [commit]
+
+    - repo: https://github.com/nonebot/nonemoji
+      rev: v0.1.4
+      hooks:
+          - id: nonemoji
+            stages: [prepare-commit-msg]

README.md

@@ -87,3 +87,13 @@ pip install nonebot-plugin-tetris-stats
 ## 📝 开源
 
 本项目使用 [AGPL-3.0](https://github.com/shoucandanghehe/nonebot-plugin-tetris-stats/blob/main/LICENSE) 许可证开源
+
+## 🤓☝ 给个 star 吧
+
+<a href="https://star-history.com/#A-Minos/nonebot-plugin-tetris-stats&Date">
+ <picture>
+   <source media="(prefers-color-scheme: dark)" srcset="https://api.star-history.com/svg?repos=A-Minos/nonebot-plugin-tetris-stats&type=Date&theme=dark" />
+   <source media="(prefers-color-scheme: light)" srcset="https://api.star-history.com/svg?repos=A-Minos/nonebot-plugin-tetris-stats&type=Date" />
+   <img alt="Star History Chart" src="https://api.star-history.com/svg?repos=A-Minos/nonebot-plugin-tetris-stats&type=Date" />
+ </picture>
+</a>

nonebot_plugin_tetris_stats/__init__.py

@@ -1,14 +1,19 @@
 from nonebot import require
-from nonebot.plugin import PluginMetadata
+from nonebot.plugin import PluginMetadata, inherit_supported_adapters
 
-require('nonebot_plugin_alconna')
-require('nonebot_plugin_apscheduler')
-require('nonebot_plugin_localstore')
-require('nonebot_plugin_orm')
-require('nonebot_plugin_session_orm')
-require('nonebot_plugin_session')
-require('nonebot_plugin_user')
-require('nonebot_plugin_userinfo')
+require_plugins = {
+    'nonebot_plugin_alconna',
+    'nonebot_plugin_apscheduler',
+    'nonebot_plugin_localstore',
+    'nonebot_plugin_orm',
+    'nonebot_plugin_session_orm',
+    'nonebot_plugin_session',
+    'nonebot_plugin_user',
+    'nonebot_plugin_userinfo',
+}
+
+for i in require_plugins:
+    require(i)
 
 from nonebot_plugin_alconna import namespace  # noqa: E402
 
@@ -16,6 +21,7 @@ with namespace('tetris_stats') as ns:
     ns.enable_message_cache = False
 
 from .config import migrations  # noqa: E402
+from .config.config import Config  # noqa: E402
 
 __plugin_meta__ = PluginMetadata(
     name='Tetris Stats',
@@ -23,6 +29,8 @@ __plugin_meta__ = PluginMetadata(
     usage='发送 tstats --help 查询使用方法',
     type='application',
     homepage='https://github.com/A-minos/nonebot-plugin-tetris-stats',
+    config=Config,
+    supported_adapters=inherit_supported_adapters(*require_plugins),
     extra={
         'orm_version_location': migrations,
     },

nonebot_plugin_tetris_stats/config/config.py

@@ -1,6 +1,6 @@
 from pathlib import Path
 
-from nonebot_plugin_localstore import get_cache_dir  # type: ignore[import-untyped]
+from nonebot_plugin_localstore import get_cache_dir
 from pydantic import BaseModel
 
 CACHE_PATH: Path = get_cache_dir('nonebot_plugin_tetris_stats')
@@ -10,3 +10,4 @@ class Config(BaseModel):
     """配置类"""
 
     tetris_req_timeout: float = 30.0
+    tetris_screenshot_quality: float = 2

nonebot_plugin_tetris_stats/config/migrations/9cd1647db502_merge_old_db.py

@@ -37,14 +37,10 @@ def migrate_old_data(connection: Connection) -> None:
     Bind = Base.classes.nonebot_plugin_tetris_stats_bind  # noqa: N806
 
     def non_empty(obj: str) -> bool:
-        if obj != '' and not obj.isspace():
-            return True
-        return False
+        return bool(obj != '' and not obj.isspace())
 
     def is_int(obj: int | str) -> bool:
-        if isinstance(obj, int) or obj.isdigit():
-            return True
-        return False
+        return bool(isinstance(obj, int) or obj.isdigit())
 
     bind_list = [
         Bind(chat_platform='OneBot V11', chat_account=int(row.QQ), game_platform='IO', game_account=row.USER)

nonebot_plugin_tetris_stats/config/migrations/f5b4a6d1325b_tetr_io_new_season.py

@@ -0,0 +1,91 @@
+"""TETR.IO new season
+
+迁移 ID: f5b4a6d1325b
+父迁移: a1195e989cc6
+创建时间: 2024-08-01 20:44:48.644912
+
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+import sqlalchemy as sa
+from alembic import op
+
+if TYPE_CHECKING:
+    from collections.abc import Sequence
+
+revision: str = 'f5b4a6d1325b'
+down_revision: str | Sequence[str] | None = 'a1195e989cc6'
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade(name: str = '') -> None:
+    if name:
+        return
+    with op.batch_alter_table('nonebot_plugin_tetris_stats_iorank', schema=None) as batch_op:
+        batch_op.drop_index('ix_nonebot_plugin_tetris_stats_iorank_file_hash')
+        batch_op.drop_index('ix_nonebot_plugin_tetris_stats_iorank_rank')
+        batch_op.drop_index('ix_nonebot_plugin_tetris_stats_iorank_update_time')
+
+    op.drop_table('nonebot_plugin_tetris_stats_iorank')
+
+    with op.batch_alter_table('nonebot_plugin_tetris_stats_tetriohistoricaldata', schema=None) as batch_op:
+        batch_op.drop_index('ix_nonebot_plugin_tetris_stats_tetriohistoricaldata_api_type')
+        batch_op.drop_index('ix_nonebot_plugin_tetris_stats_tetriohistoricaldata_update_time')
+        batch_op.drop_index('ix_nonebot_plugin_tetris_stats_tetriohistoricaldata_user_unique_identifier')
+
+    op.drop_table('nonebot_plugin_tetris_stats_tetriohistoricaldata')
+
+    op.create_table(
+        'nonebot_plugin_tetris_stats_tetriohistoricaldata',
+        sa.Column('id', sa.Integer(), nullable=False),
+        sa.Column('user_unique_identifier', sa.String(length=24), nullable=False),
+        sa.Column('api_type', sa.String(length=16), nullable=False),
+        sa.Column('data', sa.JSON(), nullable=False),
+        sa.Column('update_time', sa.DateTime(), nullable=False),
+        sa.PrimaryKeyConstraint('id', name=op.f('pk_nonebot_plugin_tetris_stats_tetriohistoricaldata')),
+        info={'bind_key': 'nonebot_plugin_tetris_stats'},
+    )
+    with op.batch_alter_table('nonebot_plugin_tetris_stats_tetriohistoricaldata', schema=None) as batch_op:
+        batch_op.create_index(
+            batch_op.f('ix_nonebot_plugin_tetris_stats_tetriohistoricaldata_api_type'), ['api_type'], unique=False
+        )
+        batch_op.create_index(
+            batch_op.f('ix_nonebot_plugin_tetris_stats_tetriohistoricaldata_update_time'), ['update_time'], unique=False
+        )
+        batch_op.create_index(
+            batch_op.f('ix_nonebot_plugin_tetris_stats_tetriohistoricaldata_user_unique_identifier'),
+            ['user_unique_identifier'],
+            unique=False,
+        )
+
+
+def downgrade(name: str = '') -> None:
+    if name:
+        return
+    op.create_table(
+        'nonebot_plugin_tetris_stats_iorank',
+        sa.Column('id', sa.INTEGER(), nullable=False),
+        sa.Column('rank', sa.VARCHAR(length=2), nullable=False),
+        sa.Column('tr_line', sa.FLOAT(), nullable=False),
+        sa.Column('player_count', sa.INTEGER(), nullable=False),
+        sa.Column('low_pps', sa.JSON(), nullable=False),
+        sa.Column('low_apm', sa.JSON(), nullable=False),
+        sa.Column('low_vs', sa.JSON(), nullable=False),
+        sa.Column('avg_pps', sa.FLOAT(), nullable=False),
+        sa.Column('avg_apm', sa.FLOAT(), nullable=False),
+        sa.Column('avg_vs', sa.FLOAT(), nullable=False),
+        sa.Column('high_pps', sa.JSON(), nullable=False),
+        sa.Column('high_apm', sa.JSON(), nullable=False),
+        sa.Column('high_vs', sa.JSON(), nullable=False),
+        sa.Column('update_time', sa.DATETIME(), nullable=False),
+        sa.Column('file_hash', sa.VARCHAR(length=128), nullable=True),
+        sa.PrimaryKeyConstraint('id', name='pk_nonebot_plugin_tetris_stats_iorank'),
+    )
+    with op.batch_alter_table('nonebot_plugin_tetris_stats_iorank', schema=None) as batch_op:
+        batch_op.create_index('ix_nonebot_plugin_tetris_stats_iorank_update_time', ['update_time'], unique=False)
+        batch_op.create_index('ix_nonebot_plugin_tetris_stats_iorank_rank', ['rank'], unique=False)
+        batch_op.create_index('ix_nonebot_plugin_tetris_stats_iorank_file_hash', ['file_hash'], unique=False)

nonebot_plugin_tetris_stats/db/__init__.py

@@ -8,10 +8,10 @@ from typing import TYPE_CHECKING, Literal, TypeVar, overload
 from nonebot.exception import FinishedException
 from nonebot.log import logger
 from nonebot_plugin_orm import AsyncSession, get_session
-from nonebot_plugin_user import User  # type: ignore[import-untyped]
+from nonebot_plugin_user import User
 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/__init__.py

@@ -9,15 +9,17 @@ from nonebot_plugin_alconna import AlcMatches, Alconna, At, CommandMeta, on_alco
 from .. import ns
 from ..utils.exception import MessageFormatError, NeedCatchError
 
-alc = on_alconna(
-    Alconna(
-        ['tetris-stats', 'tstats'],
-        namespace=ns,
-        meta=CommandMeta(
-            description='俄罗斯方块相关游戏数据查询',
-            fuzzy_match=True,
-        ),
+command: Alconna = Alconna(
+    ['tetris-stats', 'tstats'],
+    namespace=ns,
+    meta=CommandMeta(
+        description='俄罗斯方块相关游戏数据查询',
+        fuzzy_match=True,
     ),
+)
+
+alc = on_alconna(
+    command=command,
     skip_for_unmatch=False,
     auto_send_output=True,
     use_origin=True,

nonebot_plugin_tetris_stats/games/constant.py

@@ -1,3 +1 @@
-BIND_COMMAND: list[str] = ['绑定', 'bind']
-QUERY_COMMAND: list[str] = ['查', '查询', 'query', 'stats']
 CANT_VERIFY_MESSAGE = '* 由于无法验证绑定信息, 不能保证查询到的用户为本人\n'

nonebot_plugin_tetris_stats/games/tetrio/__init__.py

@@ -1,13 +1,10 @@
-from arclet.alconna import Arg, ArgFlag, Args, Option, Subcommand
-from nonebot_plugin_alconna import At
+from nonebot_plugin_alconna import Subcommand
 
 from ...utils.exception import MessageFormatError
-from ...utils.typing import Me
-from .. import add_block_handlers, alc
+from .. import alc
+from .. import command as main_command
 from .api import Player
-from .api.typing import ValidRank
 from .constant import USER_ID, USER_NAME
-from .typing import Template
 
 
 def get_player(user_id_or_name: str) -> Player | MessageFormatError:
@@ -18,149 +15,22 @@ def get_player(user_id_or_name: str) -> Player | MessageFormatError:
     return MessageFormatError('用户名/ID不合法')
 
 
-alc.command.add(
-    Subcommand(
-        'TETR.IO',
-        Subcommand(
-            'bind',
-            Args(
-                Arg(
-                    'account',
-                    get_player,
-                    notice='TETR.IO 用户名 / ID',
-                    flags=[ArgFlag.HIDDEN],
-                )
-            ),
-            help_text='绑定 TETR.IO 账号',
-        ),
-        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 游戏信息',
-        ),
-        Subcommand(
-            'record',
-            Option(
-                '--40l',
-                dest='sprint',
-            ),
-            Option(
-                '--blitz',
-                dest='blitz',
-            ),
-            Args(
-                Arg(
-                    'target',
-                    At | Me,
-                    notice='@想要查询的人 / 自己',
-                    flags=[ArgFlag.HIDDEN, ArgFlag.OPTIONAL],
-                ),
-                Arg(
-                    'account',
-                    get_player,
-                    notice='TETR.IO 用户名 / ID',
-                    flags=[ArgFlag.HIDDEN, ArgFlag.OPTIONAL],
-                ),
-            ),
-        ),
-        Subcommand(
-            'rank',
-            Args(Arg('rank', ValidRank, notice='TETR.IO 段位')),
-            help_text='查询 TETR.IO 段位信息',
-        ),
-        Subcommand(
-            'config',
-            Option(
-                '--default-template',
-                Arg('template', Template),
-                alias=['-DT', 'DefaultTemplate'],
-            ),
-        ),
-        dest='TETRIO',
-        help_text='TETR.IO 游戏相关指令',
-    )
+command = Subcommand(
+    'TETR.IO',
+    alias=['TETRIO', 'tetr.io', 'tetrio', 'io'],
+    dest='TETRIO',
+    help_text='TETR.IO 游戏相关指令',
 )
 
-alc.shortcut(
-    '(?i:io)(?i:绑定|绑|bind)',
-    {
-        'command': 'tstats TETR.IO bind',
-        'humanized': 'io绑定',
-    },
-)
-alc.shortcut(
-    '(?i:io)(?i:查询|查|query|stats)',
-    {
-        'command': 'tstats TETR.IO query',
-        'humanized': 'io查',
-    },
-)
-alc.shortcut(
-    '(?i:io)(?i:记录|record)(?i:40l)',
-    {
-        'command': 'tstats TETR.IO record --40l',
-        'humanized': 'io记录40l',
-    },
-)
-alc.shortcut(
-    '(?i:io)(?i:记录|record)(?i:blitz)',
-    {
-        'command': 'tstats TETR.IO record --blitz',
-        'humanized': 'io记录blitz',
-    },
-)
-alc.shortcut(
-    '(?i:io)(?i:段位|段|rank)',
-    {
-        'command': 'tstats TETR.IO rank',
-        'humanized': 'iorank',
-    },
-)
-alc.shortcut(
-    '(?i:io)(?i:配置|配|config)',
-    {
-        'command': 'tstats TETR.IO config',
-        'humanized': 'io配置',
-    },
-)
 
-alc.shortcut(
-    'fkosk',
-    {
-        'command': 'tstats TETR.IO query',
-        'args': ['我'],
-        'fuzzy': False,
-        'humanized': 'An Easter egg!',
-    },
-)
+from . import bind, config, query, record  # noqa: E402
 
-add_block_handlers(alc.assign('TETRIO.query'))
-
-from . import bind, config, query, rank, record  # noqa: E402
+main_command.add(command)
 
 __all__ = [
+    'alc',
     'bind',
     'config',
     'query',
-    'rank',
     'record',
 ]

nonebot_plugin_tetris_stats/games/tetrio/api/models.py

@@ -7,11 +7,12 @@ from sqlalchemy.orm import Mapped, MappedAsDataclass, mapped_column
 
 from ....db.models import PydanticType
 from .schemas.base import SuccessModel
+from .typing import Summaries
 
 
 class TETRIOHistoricalData(MappedAsDataclass, Model):
     id: Mapped[int] = mapped_column(init=False, primary_key=True)
     user_unique_identifier: Mapped[str] = mapped_column(String(24), index=True)
-    api_type: Mapped[Literal['User Info', 'User Records']] = mapped_column(String(16), index=True)
+    api_type: Mapped[Literal['User Info', Summaries]] = mapped_column(String(16), index=True)
     data: Mapped[SuccessModel] = mapped_column(PydanticType(get_model=[SuccessModel.__subclasses__], models=set()))
     update_time: Mapped[datetime] = mapped_column(DateTime, index=True)

nonebot_plugin_tetris_stats/games/tetrio/api/player.py

@@ -1,5 +1,7 @@
-from typing import overload
+from types import MappingProxyType
+from typing import Literal, overload
 
+from async_lru import alru_cache
 from nonebot.compat import type_validate_json
 
 from ....db import anti_duplicate_add
@@ -9,12 +11,31 @@ from ..constant import BASE_URL, USER_ID, USER_NAME
 from .cache import Cache
 from .models import TETRIOHistoricalData
 from .schemas.base import FailedModel
+from .schemas.summaries import (
+    AchievementsSuccessModel,
+    SoloSuccessModel,
+    SummariesModel,
+    ZenithSuccessModel,
+    ZenSuccessModel,
+)
+from .schemas.summaries.base import User as SummariesUser
 from .schemas.user import User
 from .schemas.user_info import UserInfo, UserInfoSuccess
-from .schemas.user_records import SoloModeRecord, UserRecords, UserRecordsSuccess, Zen
+from .typing import Summaries
 
 
 class Player:
+    __SUMMARIES_MAPPING: MappingProxyType[Summaries, type[SummariesModel]] = MappingProxyType(
+        {
+            '40l': SoloSuccessModel,
+            'blitz': SoloSuccessModel,
+            'zenith': ZenithSuccessModel,
+            'zenithex': ZenithSuccessModel,
+            'zen': ZenSuccessModel,
+            'achievements': AchievementsSuccessModel,
+        }
+    )
+
     @overload
     def __init__(self, *, user_id: str, trust: bool = False): ...
     @overload
@@ -36,7 +57,7 @@ class Player:
                 raise ValueError(msg)
         self.__user: User | None = None
         self._user_info: UserInfoSuccess | None = None
-        self._user_records: UserRecordsSuccess | None = None
+        self._summaries: dict[Summaries, SummariesModel] = {}
 
     @property
     def _request_user_parameter(self) -> str:
@@ -49,14 +70,21 @@ class Player:
 
     @property
     async def user(self) -> User:
-        if self.__user is None:
+        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.user.id,
-                name=user_info.data.user.username,
+                ID=user_info.data.id,
+                name=user_info.data.username,
             )
-            self.user_id = user_info.data.user.id
-            self.user_name = user_info.data.user.username
+        self.user_id = self.__user.ID
+        self.user_name = self.__user.name
         return self.__user
 
     async def get_info(self) -> UserInfoSuccess:
@@ -79,36 +107,81 @@ class Player:
             )
         return self._user_info
 
-    async def get_records(self) -> UserRecordsSuccess:
-        """Get User Records"""
-        if self._user_records is None:
-            raw_user_records = await Cache.get(
-                splice_url([BASE_URL, 'users/', f'{self._request_user_parameter}/', 'records'])
+    @overload
+    async def get_summaries(self, summaries_type: Literal['40l', 'blitz']) -> SoloSuccessModel: ...
+    @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])
             )
-            user_records: UserRecords = type_validate_json(UserRecords, raw_user_records)  # type: ignore[arg-type]
-            if isinstance(user_records, FailedModel):
-                msg = f'用户Solo数据请求错误:\n{user_records.error}'
+            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._user_records = user_records
+            self._summaries[summaries_type] = summaries
             await anti_duplicate_add(
                 TETRIOHistoricalData,
                 TETRIOHistoricalData(
                     user_unique_identifier=(await self.user).unique_identifier,
-                    api_type='User Records',
-                    data=user_records,
-                    update_time=user_records.cache.cached_at,
+                    api_type=summaries_type,
+                    data=summaries,
+                    update_time=summaries.cache.cached_at,
                 ),
             )
-        return self._user_records
+        return self._summaries[summaries_type]
 
     @property
-    async def sprint(self) -> SoloModeRecord:
-        return (await self.get_records()).data.records.sprint
+    @alru_cache
+    async def sprint(self) -> SoloSuccessModel:
+        return await self.get_summaries('40l')
 
     @property
-    async def blitz(self) -> SoloModeRecord:
-        return (await self.get_records()).data.records.blitz
+    @alru_cache
+    async def blitz(self) -> SoloSuccessModel:
+        return await self.get_summaries('blitz')
 
     @property
-    async def zen(self) -> Zen:
-        return (await self.get_records()).data.zen
+    @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

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

@@ -4,6 +4,12 @@ from typing import Literal
 from pydantic import BaseModel
 
 
+class P(BaseModel):  # what is P
+    pri: float
+    sec: float
+    ter: float
+
+
 class Cache(BaseModel):
     status: str
     cached_at: datetime

nonebot_plugin_tetris_stats/games/tetrio/api/schemas/leaderboards/ar.py

@@ -0,0 +1,27 @@
+from pydantic import BaseModel, Field
+
+from ..base import SuccessModel
+from .base import Entry as BaseEntry
+
+
+class ArCounts(BaseModel):
+    bronze: int | None = Field(None, alias='1')
+    silver: int | None = Field(None, alias='2')
+    gold: int | None = Field(None, alias='3')
+    platinum: int | None = Field(None, alias='4')
+    diamond: int | None = Field(None, alias='5')
+    issued: int | None = Field(None, alias='100')
+    top10: int | None = Field(None, alias='t10')
+
+
+class Entry(BaseEntry):
+    ar: int
+    ar_counts: ArCounts
+
+
+class Data(BaseModel):
+    entries: list[Entry]
+
+
+class ArSuccessModel(SuccessModel):
+    data: Data

nonebot_plugin_tetris_stats/games/tetrio/api/schemas/leaderboards/base.py

@@ -0,0 +1,30 @@
+from datetime import datetime
+
+from pydantic import BaseModel, Field
+
+from ...typing import Rank
+from ..base import P
+
+
+class League(BaseModel):
+    gamesplayed: int
+    gameswon: int
+    rating: int
+    rank: Rank
+    decaying: bool
+
+
+class Entry(BaseModel):
+    id: str = Field(..., alias='_id')
+    username: str
+    role: str
+    xp: float
+    league: League
+    supporter: bool | None = None
+    verified: bool
+    country: str | None = None
+    ts: datetime
+    gamesplayed: int
+    gameswon: int
+    gametime: float
+    p: P

nonebot_plugin_tetris_stats/games/tetrio/api/schemas/leaderboards/solo.py

@@ -0,0 +1,12 @@
+from pydantic import BaseModel
+
+from ..base import SuccessModel
+from ..summaries.solo import Record
+
+
+class Data(BaseModel):
+    entries: list[Record]
+
+
+class SoloSuccessModel(SuccessModel):
+    data: Data

nonebot_plugin_tetris_stats/games/tetrio/api/schemas/leaderboards/xp.py

@@ -0,0 +1,12 @@
+from pydantic import BaseModel
+
+from ..base import SuccessModel
+from .base import Entry
+
+
+class Data(BaseModel):
+    entries: list[Entry]
+
+
+class XpSuccessModel(SuccessModel):
+    data: Data

nonebot_plugin_tetris_stats/games/tetrio/api/schemas/leaderboards/zenith.py

@@ -0,0 +1,12 @@
+from pydantic import BaseModel
+
+from ..base import SuccessModel
+from ..summaries.zenith import Record
+
+
+class Data(BaseModel):
+    entries: list[Record]
+
+
+class ZenithSuccessModel(SuccessModel):
+    data: Data

nonebot_plugin_tetris_stats/games/tetrio/api/schemas/summaries/__init__.py

@@ -0,0 +1,20 @@
+from .achievements import Achievements, AchievementsSuccessModel
+from .solo import Blitz, SoloSuccessModel, Sprint
+from .zen import Zen, ZenSuccessModel
+from .zenith import Zenith, ZenithEx, ZenithSuccessModel
+
+SummariesModel = AchievementsSuccessModel | SoloSuccessModel | ZenSuccessModel | ZenithSuccessModel
+
+__all__ = [
+    'Achievements',
+    'AchievementsSuccessModel',
+    'Blitz',
+    'Sprint',
+    'SoloSuccessModel',
+    'Zen',
+    'ZenSuccessModel',
+    'Zenith',
+    'ZenithEx',
+    'ZenithSuccessModel',
+    'SummariesModel',
+]

nonebot_plugin_tetris_stats/games/tetrio/api/schemas/summaries/achievements.py

@@ -0,0 +1,29 @@
+from typing import TypeAlias
+
+from pydantic import BaseModel
+
+from ..base import FailedModel, SuccessModel
+
+
+class Achievement(BaseModel):
+    # 这**都是些啥
+    k: int
+    o: int
+    rt: int
+    vt: int
+    min: int
+    deci: int
+    name: str
+    object: str
+    category: str
+    hidden: bool
+    desc: str
+    n: str
+    stub: bool
+
+
+class AchievementsSuccessModel(SuccessModel):
+    data: list[Achievement]
+
+
+Achievements: TypeAlias = AchievementsSuccessModel | FailedModel

nonebot_plugin_tetris_stats/games/tetrio/api/schemas/summaries/base.py

@@ -0,0 +1,23 @@
+from pydantic import BaseModel
+
+
+class User(BaseModel):
+    id: str
+    username: str
+    avatar_revision: int | None
+    banner_revision: int | None
+    country: str | None
+    verified: int
+    supporter: int
+
+
+class AggregateStats(BaseModel):
+    apm: float
+    pps: float
+    vsscore: float
+
+
+class Finesse(BaseModel):
+    combo: int
+    faults: int
+    perfectpieces: int

nonebot_plugin_tetris_stats/games/tetrio/api/schemas/summaries/solo.py

@@ -0,0 +1,103 @@
+from datetime import datetime
+from typing import Literal, TypeAlias
+
+from pydantic import BaseModel, Field
+
+from ..base import FailedModel, P, SuccessModel
+from .base import AggregateStats, Finesse, User
+
+
+class Time(BaseModel):
+    start: int
+    zero: bool
+    locked: bool
+    prev: int
+    frameoffset: int
+
+
+class Clears(BaseModel):
+    singles: int
+    doubles: int
+    triples: int
+    quads: int
+    realtspins: int
+    minitspins: int
+    minitspinsingles: int
+    tspinsingles: int
+    minitspindoubles: int
+    tspindoubles: int
+    tspintriples: int
+    tspinquads: int
+    allclear: int
+
+
+class Garbage(BaseModel):
+    sent: int
+    received: int
+    attack: int | None
+    cleared: int
+
+
+class Stats(BaseModel):
+    seed: int | None = None  # ?: 不知道是之后都没有了还是还会有
+    lines: int
+    level_lines: int
+    level_lines_needed: int
+    inputs: int
+    holds: int
+    time: Time | None = None  # ?: 不知道是之后都没有了还是还会有
+    score: int
+    zenlevel: int
+    zenprogress: int
+    level: int
+    combo: int
+    currentcombopower: int | None = None
+    topcombo: int
+    btb: int
+    topbtb: int
+    currentbtbchainpower: int | None = None
+    tspins: int
+    piecesplaced: int
+    clears: Clears
+    garbage: Garbage
+    kills: int
+    finesse: Finesse
+    finaltime: float
+
+
+class Results(BaseModel):
+    aggregatestats: AggregateStats
+    stats: Stats
+    gameoverreason: str
+
+
+class Record(BaseModel):
+    id: str = Field(..., alias='_id')
+    replayid: str
+    stub: bool
+    gamemode: Literal['40l', 'blitz']
+    pb: bool
+    oncepb: bool
+    ts: datetime
+    revolution: None
+    user: User
+    otherusers: list
+    leaderboards: list[str]
+    results: Results
+    extras: dict
+    disputed: bool
+    p: P
+
+
+class Data(BaseModel):
+    record: Record | None
+    rank: int
+    rank_local: int
+
+
+class SoloSuccessModel(SuccessModel):
+    data: Data
+
+
+Sprint: TypeAlias = SoloSuccessModel | FailedModel
+Blitz: TypeAlias = SoloSuccessModel | FailedModel

nonebot_plugin_tetris_stats/games/tetrio/api/schemas/summaries/zen.py

@@ -0,0 +1,17 @@
+from typing import TypeAlias
+
+from pydantic import BaseModel
+
+from ..base import FailedModel, SuccessModel
+
+
+class Data(BaseModel):
+    level: int
+    score: int
+
+
+class ZenSuccessModel(SuccessModel):
+    data: Data
+
+
+Zen: TypeAlias = ZenSuccessModel | FailedModel

nonebot_plugin_tetris_stats/games/tetrio/api/schemas/summaries/zenith.py

@@ -0,0 +1,131 @@
+from datetime import datetime
+from typing import Literal, TypeAlias
+
+from pydantic import BaseModel, Field
+
+from ..base import FailedModel, P, SuccessModel
+from .base import AggregateStats, Finesse, User
+
+
+class Clears(BaseModel):
+    singles: int
+    doubles: int
+    triples: int
+    quads: int
+    pentas: int
+    realtspins: int
+    minitspins: int
+    minitspinsingles: int
+    tspinsingles: int
+    minitspindoubles: int
+    tspindoubles: int
+    minitspintriples: int
+    tspintriples: int
+    minitspinquads: int
+    tspinquads: int
+    tspinpentas: int
+    allclear: int
+
+
+class Garbage(BaseModel):
+    sent: int
+    sent_nomult: int
+    maxspike: int
+    maxspike_nomult: int
+    received: int
+    attack: int
+    cleared: int
+
+
+class _Zenith(BaseModel):
+    altitude: float
+    rank: float
+    peakrank: float
+    avgrankpts: float
+    floor: int
+    targetingfactor: float
+    targetinggrace: float
+    totalbonus: float
+    revives: int
+    revives_total: int = Field(..., alias='revivesTotal')
+    speedrun: bool
+    speedrun_seen: bool
+    splits: list[int]
+
+
+class Stats(BaseModel):
+    lines: int
+    level_lines: int
+    level_lines_needed: int
+    inputs: int
+    holds: int
+    score: int
+    zenlevel: int
+    zenprogress: int
+    level: int
+    combo: int
+    topcombo: int
+    combopower: int
+    btb: int
+    topbtb: int
+    btbpower: int
+    tspins: int
+    piecesplaced: int
+    clears: Clears
+    garbage: Garbage
+    kills: int
+    finesse: Finesse
+    zenith: _Zenith
+    finaltime: float
+
+
+class Results(BaseModel):
+    aggregatestats: AggregateStats
+    stats: Stats
+    gameoverreason: str
+
+
+class ExtrasZenith(BaseModel):
+    mods: list[str]
+
+
+class Extras(BaseModel):
+    zenith: ExtrasZenith
+
+
+class Record(BaseModel):
+    id: str = Field(..., alias='_id')
+    replayid: str
+    stub: bool
+    gamemode: Literal['zenith', 'zenithex']
+    pb: bool
+    oncepb: bool
+    ts: datetime
+    revolution: None
+    user: User
+    otherusers: list
+    leaderboards: list[str]
+    results: Results
+    extras: Extras
+    disputed: bool
+    p: P
+
+
+class Best(BaseModel):
+    record: None  # WTF
+    rank: int
+
+
+class Data(BaseModel):
+    record: Record | None
+    rank: int
+    rank_local: int
+    best: Best
+
+
+class ZenithSuccessModel(SuccessModel):
+    data: Data
+
+
+Zenith: TypeAlias = ZenithSuccessModel | FailedModel
+ZenithEx: TypeAlias = ZenithSuccessModel | FailedModel

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/schemas/user_info.py

@@ -3,7 +3,6 @@ from typing import Literal
 
 from pydantic import BaseModel, Field
 
-from ..typing import Rank
 from .base import FailedModel
 from .base import SuccessModel as BaseSuccessModel
 
@@ -15,67 +14,6 @@ class Badge(BaseModel):
     ts: datetime | Literal[False] | None = None
 
 
-class MetaLeague(BaseModel):
-    decaying: bool
-
-
-class NeverPlayedLeague(MetaLeague):
-    gamesplayed: Literal[0]
-    gameswon: Literal[0]
-    rating: Literal[-1]
-    rank: Literal['z']
-    standing: Literal[-1]
-    standing_local: Literal[-1]
-    next_rank: None
-    prev_rank: None
-    next_at: Literal[-1]
-    prev_at: Literal[-1]
-    percentile: Literal[-1]
-    percentile_rank: Literal['z']
-    apm: None = None
-    pps: None = None
-    vs: None = None
-
-
-class NeverRatedLeague(MetaLeague):
-    gamesplayed: Literal[1, 2, 3, 4, 5, 6, 7, 8, 9]
-    gameswon: int
-    rating: Literal[-1]
-    rank: Literal['z']
-    standing: Literal[-1]
-    standing_local: Literal[-1]
-    next_rank: None
-    prev_rank: None
-    next_at: Literal[-1]
-    prev_at: Literal[-1]
-    percentile: Literal[-1]
-    percentile_rank: Literal['z']
-    apm: float
-    pps: float
-    vs: float | None = None
-
-
-class RatedLeague(MetaLeague):
-    gamesplayed: int
-    gameswon: int
-    rating: float
-    rank: Rank
-    bestrank: Rank
-    standing: int
-    standing_local: int
-    next_rank: Rank | None = None
-    prev_rank: Rank | None = None
-    next_at: int
-    prev_at: int
-    percentile: float
-    percentile_rank: str
-    glicko: float
-    rd: float
-    apm: float
-    pps: float
-    vs: float | None = None
-
-
 class Discord(BaseModel):
     id: str
     username: str
@@ -89,7 +27,7 @@ class Distinguishment(BaseModel):
     type: str
 
 
-class User(BaseModel):
+class Data(BaseModel):
     id: str = Field(..., alias='_id')
     username: str
     role: Literal['anon', 'user', 'bot', 'halfmod', 'mod', 'admin', 'sysop', 'banned']
@@ -105,7 +43,6 @@ class User(BaseModel):
     supporter: bool | None = None  # osk说是必有, 但实际上不是 fkosk
     supporter_tier: int
     verified: bool
-    league: NeverPlayedLeague | NeverRatedLeague | RatedLeague
     avatar_revision: int | None = None
     """This user's avatar ID. Get their avatar at
 
@@ -122,10 +59,6 @@ class User(BaseModel):
     distinguishment: Distinguishment | None = None
 
 
-class Data(BaseModel):
-    user: User
-
-
 class UserInfoSuccess(BaseSuccessModel):
     data: Data
 

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/api/typing.py

@@ -21,3 +21,13 @@ ValidRank = Literal[
 ]
 
 Rank = ValidRank | Literal['z']  # 未定级
+
+Summaries = Literal[
+    '40l',
+    'blitz',
+    'zenith',
+    'zenithex',
+    # 'league',  # 等待正式赛季开始
+    'zen',
+    'achievements',
+]

nonebot_plugin_tetris_stats/games/tetrio/bind.py

@@ -1,13 +1,14 @@
-from asyncio import gather
 from hashlib import md5
 from urllib.parse import urlencode
 
+from arclet.alconna import Arg, ArgFlag
+from nonebot_plugin_alconna import Args, Subcommand
 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
 from nonebot_plugin_session_orm import get_session_persist_id  # type: ignore[import-untyped]
-from nonebot_plugin_user import User  # type: ignore[import-untyped]
-from nonebot_plugin_userinfo import BotUserInfo, UserInfo  # type: ignore[import-untyped]
+from nonebot_plugin_user import User
+from nonebot_plugin_userinfo import BotUserInfo, UserInfo
 
 from ...db import BindStatus, create_or_update_bind, trigger
 from ...utils.host import HostPage, get_self_netloc
@@ -15,10 +16,31 @@ from ...utils.image import get_avatar
 from ...utils.render import Bind, render
 from ...utils.render.schemas.base import Avatar, People
 from ...utils.screenshot import screenshot
-from . import alc
+from . import alc, command, get_player
 from .api import Player
 from .constant import GAME_TYPE
 
+command.add(
+    Subcommand(
+        'bind',
+        Args(
+            Arg(
+                'account',
+                get_player,
+                notice='TETR.IO 用户名 / ID',
+                flags=[ArgFlag.HIDDEN],
+            )
+        ),
+        help_text='绑定 TETR.IO 账号',
+    )
+)
+
+alc.shortcut(
+    '(?i:io)(?i:绑定|绑|bind)',
+    command='tstats TETR.IO bind',
+    humanized='io绑定',
+)
+
 
 @alc.assign('TETRIO.bind')
 async def _(nb_user: User, account: Player, event_session: EventSession, bot_info: UserInfo = BotUserInfo()):  # noqa: B008
@@ -28,7 +50,7 @@ async def _(nb_user: User, account: Player, event_session: EventSession, bot_inf
         command_type='bind',
         command_args=[],
     ):
-        user, user_info = await gather(account.user, account.get_info())
+        user = await account.user
         async with get_session() as session:
             bind_status = await create_or_update_bind(
                 session=session,
@@ -45,11 +67,10 @@ async def _(nb_user: User, account: Player, event_session: EventSession, bot_inf
                         platform='TETR.IO',
                         status='unknown',
                         user=People(
-                            avatar=f'http://{netloc}/host/resource/tetrio/avatars/{user.ID}?{urlencode({"revision": user_info.data.user.avatar_revision})}'
-                            if user_info.data.user.avatar_revision is not None
-                            and user_info.data.user.avatar_revision != 0
-                            else Avatar(type='identicon', hash=md5(user_info.data.user.id.encode()).hexdigest()),  # noqa: S324
-                            name=user_info.data.user.username.upper(),
+                            avatar=f'http://{netloc}/host/resource/tetrio/avatars/{user.ID}?{urlencode({"revision": avatar_revision})}'
+                            if (avatar_revision := (await account.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(),
                         ),
                         bot=People(
                             avatar=await get_avatar(bot_info, 'Data URI', '../../static/logo/logo.svg'),

nonebot_plugin_tetris_stats/games/tetrio/config.py

@@ -1,20 +1,51 @@
+from arclet.alconna import Arg
+from nonebot_plugin_alconna import Option, Subcommand
 from nonebot_plugin_alconna.uniseg import UniMessage
 from nonebot_plugin_orm import async_scoped_session
-from nonebot_plugin_user import User  # type: ignore[import-untyped]
+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
 from sqlalchemy import select
 
-from . import alc
+from ...db import trigger
+from . import alc, command
+from .constant import GAME_TYPE
 from .models import TETRIOUserConfig
 from .typing import Template
 
+command.add(
+    Subcommand(
+        'config',
+        Option(
+            '--default-template',
+            Arg('template', Template, notice='模板版本'),
+            alias=['-DT', 'DefaultTemplate'],
+            help_text='设置默认查询模板',
+        ),
+        help_text='TETR.IO 查询个性化配置',
+    ),
+)
+
+alc.shortcut(
+    '(?i:io)(?i:配置|配|config)',
+    command='tstats TETR.IO config',
+    humanized='io配置',
+)
+
 
 @alc.assign('TETRIO.config')
-async def _(user: User, session: async_scoped_session, template: Template):
-    config = (await session.scalars(select(TETRIOUserConfig).where(TETRIOUserConfig.id == user.id))).one_or_none()
-    if config is None:
-        config = TETRIOUserConfig(id=user.id, query_template=template)
-        session.add(config)
-    else:
-        config.query_template = template
-    await session.commit()
-    await UniMessage('配置成功').finish()
+async def _(user: User, session: async_scoped_session, event_session: EventSession, template: Template):
+    async with trigger(
+        session_persist_id=await get_session_persist_id(event_session),
+        game_platform=GAME_TYPE,
+        command_type='config',
+        command_args=[f'--default-template {template}'],
+    ):
+        config = (await session.scalars(select(TETRIOUserConfig).where(TETRIOUserConfig.id == user.id))).one_or_none()
+        if config is None:
+            config = TETRIOUserConfig(id=user.id, query_template=template)
+            session.add(config)
+        else:
+            config.query_template = template
+        await session.commit()
+        await UniMessage('配置成功').finish()

nonebot_plugin_tetris_stats/games/tetrio/constant.py

@@ -1,13 +1,13 @@
 from re import compile
 from typing import Literal
 
-from .api.typing import Rank
+from .api.typing import ValidRank
 
 GAME_TYPE: Literal['IO'] = 'IO'
 
 BASE_URL = 'https://ch.tetr.io/api/'
 
-RANK_PERCENTILE: dict[Rank, float] = {
+RANK_PERCENTILE: dict[ValidRank, float] = {
     'x': 1,
     'u': 5,
     'ss': 11,

nonebot_plugin_tetris_stats/games/tetrio/models.py

@@ -1,34 +1,10 @@
-from datetime import datetime
-
 from nonebot_plugin_orm import Model
-from sqlalchemy import JSON, DateTime, String
+from sqlalchemy import String
 from sqlalchemy.orm import Mapped, MappedAsDataclass, mapped_column
 
-from .api.typing import Rank
 from .typing import Template
 
 
-class IORank(MappedAsDataclass, Model):
-    id: Mapped[int] = mapped_column(init=False, primary_key=True)
-    rank: Mapped[Rank] = mapped_column(String(2), index=True)
-    tr_line: Mapped[float]
-    player_count: Mapped[int]
-    low_pps: Mapped[tuple[dict[str, str], float]] = mapped_column(JSON)
-    low_apm: Mapped[tuple[dict[str, str], float]] = mapped_column(JSON)
-    low_vs: Mapped[tuple[dict[str, str], float]] = mapped_column(JSON)
-    avg_pps: Mapped[float]
-    avg_apm: Mapped[float]
-    avg_vs: Mapped[float]
-    high_pps: Mapped[tuple[dict[str, str], float]] = mapped_column(JSON)
-    high_apm: Mapped[tuple[dict[str, str], float]] = mapped_column(JSON)
-    high_vs: Mapped[tuple[dict[str, str], float]] = mapped_column(JSON)
-    update_time: Mapped[datetime] = mapped_column(
-        DateTime,
-        index=True,
-    )
-    file_hash: Mapped[str | None] = mapped_column(String(128), index=True)
-
-
 class TETRIOUserConfig(MappedAsDataclass, Model):
     id: Mapped[int] = mapped_column(primary_key=True)
     query_template: Mapped[Template] = mapped_column(String(2))

nonebot_plugin_tetris_stats/games/tetrio/query.py

@@ -1,58 +1,90 @@
 from asyncio import gather
-from collections import defaultdict
-from datetime import date, datetime, timedelta, timezone
+from datetime import datetime, timedelta, timezone
 from hashlib import md5
-from math import ceil, floor
-from typing import ClassVar, TypeVar, overload
+from typing import TYPE_CHECKING, TypeVar
 from urllib.parse import urlencode
-from zoneinfo import ZoneInfo
 
-from aiofiles import open
+from arclet.alconna import Arg, ArgFlag
 from nonebot import get_driver
 from nonebot.adapters import Event
-from nonebot.compat import type_validate_json
 from nonebot.matcher import Matcher
-from nonebot_plugin_alconna import At
+from nonebot_plugin_alconna import Args, At, Option, Subcommand
 from nonebot_plugin_alconna.uniseg import UniMessage
-from nonebot_plugin_apscheduler import scheduler  # type: ignore[import-untyped]
-from nonebot_plugin_localstore import get_data_file  # type: ignore[import-untyped]
 from nonebot_plugin_orm import get_session
-from nonebot_plugin_session import EventSession  # type: ignore[import-untyped]
+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  # type: ignore[import-untyped]
-from nonebot_plugin_user import get_user  # type: ignore[import-untyped]
+from nonebot_plugin_user import User as NBUser
+from nonebot_plugin_user import get_user
 from sqlalchemy import select
-from zstandard import ZstdDecompressor
 
 from ...db import query_bind_info, trigger
-from ...utils.exception import FallbackError
 from ...utils.host import HostPage, get_self_netloc
-from ...utils.metrics import TetrisMetricsProWithPPSVS, get_metrics
 from ...utils.render import render
-from ...utils.render.schemas.base import Avatar, Ranking
-from ...utils.render.schemas.tetrio_info import Data, Radar, TetraLeague, TetraLeagueHistory
-from ...utils.render.schemas.tetrio_info import Info as V1TemplateInfo
-from ...utils.render.schemas.tetrio_info import User as V1TemplateUser
-from ...utils.render.schemas.tetrio_info_v2 import Badge, Blitz, Sprint, Statistic, TetraLeagueStatistic
-from ...utils.render.schemas.tetrio_info_v2 import Info as V2TemplateInfo
-from ...utils.render.schemas.tetrio_info_v2 import TetraLeague as V2TemplateTetraLeague
-from ...utils.render.schemas.tetrio_info_v2 import User as V2TemplateUser
+from ...utils.render.schemas.base import Avatar
+from ...utils.render.schemas.tetrio.user.info_v2 import Badge, Blitz, Sprint, Statistic, 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, Number
+from ...utils.typing import Me
+from .. import add_block_handlers, alc
 from ..constant import CANT_VERIFY_MESSAGE
-from . import alc
-from .api import Player, User, UserInfoSuccess
-from .api.models import TETRIOHistoricalData
-from .api.schemas.tetra_league import TetraLeagueSuccess
-from .api.schemas.user_info import NeverPlayedLeague, NeverRatedLeague, RatedLeague
-from .constant import GAME_TYPE, TR_MAX, TR_MIN
-from .models import IORank, TETRIOUserConfig
+from . import command, get_player
+from .api import Player
+from .constant import GAME_TYPE
+from .models import TETRIOUserConfig
 from .typing import Template
 
+if TYPE_CHECKING:
+    from .api.schemas.summaries import SoloSuccessModel, ZenSuccessModel
+    from .api.schemas.user import User
+    from .api.schemas.user_info import UserInfoSuccess
+
 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
@@ -67,7 +99,7 @@ async def _(  # noqa: PLR0913
         session_persist_id=await get_session_persist_id(event_session),
         game_platform=GAME_TYPE,
         command_type='query',
-        command_args=[],
+        command_args=[f'--default-template {template}'] if template is not None else [],
     ):
         async with get_session() as session:
             bind = await query_bind_info(
@@ -85,7 +117,7 @@ async def _(  # noqa: PLR0913
             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()
+        await (message + UniMessage.image(raw=await make_query_image_v2(player))).finish()
 
 
 @alc.assign('TETRIO.query')
@@ -94,220 +126,14 @@ async def _(user: NBUser, account: Player, event_session: EventSession, template
         session_persist_id=await get_session_persist_id(event_session),
         game_platform=GAME_TYPE,
         command_type='query',
-        command_args=[],
+        command_args=[f'--default-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()
-
-
-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: Data,
-    behind_point: Data,
-    point_time: datetime,
-) -> Data:
-    """根据给出的 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 Data(
-        record_at=point_time,
-        tr=previous_point.tr + slope * (datetime.timestamp(point_time) - datetime.timestamp(previous_point.record_at)),
-    )
-
-
-async def query_historical_data(user: User, user_info: UserInfoSuccess) -> list[Data]:
-    today = datetime.now(ZoneInfo('Asia/Shanghai')).replace(hour=0, minute=0, second=0, microsecond=0)
-    forward = timedelta(days=9)
-    start_time = (today - forward).astimezone(UTC)
-    async with get_session() as session:
-        historical_data = (
-            await session.scalars(
-                select(TETRIOHistoricalData)
-                .where(TETRIOHistoricalData.update_time >= start_time)
-                .where(TETRIOHistoricalData.user_unique_identifier == user.unique_identifier)
-                .where(TETRIOHistoricalData.api_type == 'User Info')
-            )
-        ).all()
-        if historical_data:
-            extra = (
-                await session.scalars(
-                    select(TETRIOHistoricalData)
-                    .where(TETRIOHistoricalData.user_unique_identifier == user.unique_identifier)
-                    .where(TETRIOHistoricalData.api_type == 'User Info')
-                    .order_by(TETRIOHistoricalData.id.desc())
-                    .where(TETRIOHistoricalData.id < min([i.id for i in historical_data]))
-                    .limit(1)
-                )
-            ).one_or_none()
-            if extra is not None:
-                historical_data = list(historical_data)
-                historical_data.append(extra)
-    full_export_data = FullExport.get_data(user.unique_identifier)
-    if not historical_data and not full_export_data:
-        return [
-            Data(record_at=today - forward, tr=user_info.data.user.league.rating),
-            Data(record_at=today.replace(microsecond=1000), tr=user_info.data.user.league.rating),
-        ]
-    histories = [
-        Data(
-            record_at=i.update_time.astimezone(ZoneInfo('Asia/Shanghai')),
-            tr=i.data.data.user.league.rating,
-        )
-        for i in historical_data
-        if isinstance(i.data, UserInfoSuccess) and isinstance(i.data.data.user.league, RatedLeague)
-    ] + full_export_data
-
-    # 按照时间排序
-    histories = sorted(histories, key=lambda x: x.record_at)
-    for index, value in enumerate(histories):
-        # 在历史记录里找有没有今天0点后的数据, 并且至少要有两个数据点
-        if value.record_at > today and len(histories) >= 2:  # noqa: PLR2004
-            histories = histories[:index] + [
-                get_specified_point(histories[index - 1], histories[index], today.replace(microsecond=1000))
-            ]
-            break
-    else:
-        histories.append(
-            get_specified_point(
-                histories[-1],
-                Data(record_at=user_info.cache.cached_at, tr=user_info.data.user.league.rating),
-                today.replace(microsecond=1000),
-            )
-        )
-    if histories[0].record_at < (today - forward):
-        histories[0] = get_specified_point(
-            histories[0],
-            histories[1],
-            today - forward,
-        )
-    else:
-        histories.insert(0, Data(record_at=today - forward, tr=histories[0].tr))
-    return histories
-
-
-L = TypeVar('L', NeverPlayedLeague, NeverRatedLeague, RatedLeague)
-
-
-@overload
-def get_league(user_info: UserInfoSuccess, league_type: type[L]) -> L: ...
-@overload
-def get_league(
-    user_info: UserInfoSuccess, league_type: None = None
-) -> NeverPlayedLeague | NeverRatedLeague | RatedLeague: ...
-def get_league(
-    user_info: UserInfoSuccess, league_type: type[L] | None = None
-) -> L | NeverPlayedLeague | NeverRatedLeague | RatedLeague:
-    league = user_info.data.user.league
-    if league_type is None:
-        return league
-    if isinstance(league, league_type):
-        return league
-    raise FallbackError
-
-
-async def make_query_image_v1(player: Player) -> bytes:
-    user, user_info, sprint, blitz = await gather(player.user, player.get_info(), player.sprint, player.blitz)
-    league = get_league(user_info, RatedLeague)
-    if league.vs is None:
-        raise FallbackError
-    histories = await query_historical_data(user, user_info)
-    value_max, value_min = get_value_bounds([i.tr for i in histories])
-    split_value, offset = get_split(value_max, value_min)
-    if sprint.record is not None:
-        duration = timedelta(milliseconds=sprint.record.endcontext.final_time).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.record.endcontext.score:,}' if blitz.record is not None else 'N/A'
-    netloc = get_self_netloc()
-    async with HostPage(
-        page=await render(
-            'v1/tetrio/info',
-            V1TemplateInfo(
-                user=V1TemplateUser(
-                    avatar=f'http://{netloc}/host/resource/tetrio/avatars/{user.ID}?{urlencode({"revision": user_info.data.user.avatar_revision})}'
-                    if user_info.data.user.avatar_revision is not None and user_info.data.user.avatar_revision != 0
-                    else Avatar(
-                        type='identicon',
-                        hash=md5(user_info.data.user.id.encode()).hexdigest(),  # noqa: S324
-                    ),
-                    name=user.name.upper(),
-                    bio=user_info.data.user.bio,
-                ),
-                ranking=Ranking(
-                    rating=round(league.glicko, 2),
-                    rd=round(league.rd, 2),
-                ),
-                tetra_league=TetraLeague(
-                    rank=league.rank,
-                    tr=round(league.rating, 2),
-                    global_rank=league.standing,
-                    pps=league.pps,
-                    lpm=round(lpm := (league.pps * 24), 2),
-                    apm=league.apm,
-                    apl=round(league.apm / lpm, 2),
-                    vs=league.vs,
-                    adpm=round(adpm := (league.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.apm / (60 * league.pps))),
-                    dsps=(dsps := ((league.vs / 100) - (league.apm / 60))),
-                    dspp=(dspp := (dsps / league.pps)),
-                    ci=150 * dspp - 125 * app + 50 * (league.vs / league.apm) - 25,
-                    ge=2 * ((app * dsps) / league.pps),
-                ),
-                sprint=sprint_value,
-                blitz=blitz_value,
-            ),
-        )
-    ) as page_hash:
-        return await screenshot(f'http://{netloc}/host/{page_hash}.html')
+        await (UniMessage.image(raw=await make_query_image_v2(account))).finish()
 
 
 N = TypeVar('N', int, float)
@@ -318,19 +144,32 @@ def handling_special_value(value: N) -> N | None:
 
 
 async def make_query_image_v2(player: Player) -> bytes:
-    user, user_info, sprint, blitz, zen = await gather(
-        player.user, player.get_info(), player.sprint, player.blitz, player.zen
+    user: User
+    user_info: UserInfoSuccess
+    sprint: SoloSuccessModel
+    blitz: SoloSuccessModel
+    zen: ZenSuccessModel
+    avatar_revision: int | None
+    banner_revision: int | None
+    # [todo) 有没有什么办法能让这类型推导成功)
+    user, user_info, sprint, blitz, zen, avatar_revision, banner_revision = await gather(  # type: ignore[assignment]
+        player.user,
+        player.get_info(),
+        player.sprint,
+        player.blitz,
+        player.zen,
+        player.avatar_revision,
+        player.banner_revision,
     )
-    league = get_league(user_info)
 
-    if sprint.record is not None:
-        duration = timedelta(milliseconds=sprint.record.endcontext.final_time).total_seconds()
+    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.user.gametime)) is not 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:
@@ -342,20 +181,20 @@ async def make_query_image_v2(player: Player) -> bytes:
     netloc = get_self_netloc()
     async with HostPage(
         await render(
-            'v2/tetrio/info',
+            'v2/tetrio/user/info',
             V2TemplateInfo(
                 user=V2TemplateUser(
                     id=user.ID,
                     name=user.name.upper(),
-                    bio=user_info.data.user.bio,
-                    banner=f'http://{netloc}/host/resource/tetrio/banners/{user.ID}?{urlencode({"revision": user_info.data.user.avatar_revision})}'
-                    if user_info.data.user.banner_revision is not None and user_info.data.user.banner_revision != 0
+                    bio=user_info.data.bio,
+                    banner=f'http://{netloc}/host/resource/tetrio/banners/{user.ID}?{urlencode({"revision": banner_revision})}'
+                    if banner_revision is not None and banner_revision != 0
                     else None,
-                    avatar=f'http://{netloc}/host/resource/tetrio/avatars/{user.ID}?{urlencode({"revision": user_info.data.user.avatar_revision})}'
-                    if user_info.data.user.avatar_revision is not None and user_info.data.user.avatar_revision != 0
+                    avatar=f'http://{netloc}/host/resource/tetrio/avatars/{user.ID}?{urlencode({"revision": avatar_revision})}'
+                    if avatar_revision is not None and avatar_revision != 0
                     else Avatar(
                         type='identicon',
-                        hash=md5(user_info.data.user.id.encode()).hexdigest(),  # noqa: S324
+                        hash=md5(user.ID.encode()).hexdigest(),  # noqa: S324
                     ),
                     badges=[
                         Badge(
@@ -364,188 +203,39 @@ async def make_query_image_v2(player: Player) -> bytes:
                             group=i.group,
                             receive_at=i.ts if isinstance(i.ts, datetime) else None,
                         )
-                        for i in user_info.data.user.badges
+                        for i in user_info.data.badges
                     ],
-                    country=user_info.data.user.country,
-                    xp=user_info.data.user.xp,
-                    friend_count=user_info.data.user.friend_count or 0,
-                    supporter_tier=user_info.data.user.supporter_tier,
-                    bad_standing=user_info.data.user.badstanding or False,
-                    verified=user_info.data.user.verified,
+                    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,
+                    verified=user_info.data.verified,
                     playtime=play_time,
-                    join_at=user_info.data.user.ts,
+                    join_at=user_info.data.ts,
                 ),
-                tetra_league=V2TemplateTetraLeague(
-                    rank=league.rank,
-                    highest_rank=league.bestrank,
-                    tr=round(league.rating, 2),
-                    glicko=round(league.glicko, 2),
-                    rd=round(league.rd, 2),
-                    global_rank=handling_special_value(league.standing),
-                    country_rank=handling_special_value(league.standing_local),
-                    pps=(
-                        metrics := get_metrics(pps=league.pps, apm=league.apm, vs=league.vs)
-                        if league.vs is not None
-                        else get_metrics(pps=league.pps, apm=league.apm)
-                    ).pps,
-                    apm=metrics.apm,
-                    apl=metrics.apl,
-                    vs=metrics.vs if isinstance(metrics, TetrisMetricsProWithPPSVS) else None,
-                    adpl=metrics.adpl if isinstance(metrics, TetrisMetricsProWithPPSVS) else None,
-                    statistic=TetraLeagueStatistic(
-                        total=league.gamesplayed,
-                        wins=league.gameswon,
-                    ),
-                    decaying=league.decaying,
-                )
-                if isinstance(league, RatedLeague)
-                else None,
+                tetra_league=None,
                 statistic=Statistic(
-                    total=handling_special_value(user_info.data.user.gamesplayed),
-                    wins=handling_special_value(user_info.data.user.gameswon),
+                    total=handling_special_value(user_info.data.gamesplayed),
+                    wins=handling_special_value(user_info.data.gameswon),
                 ),
                 sprint=Sprint(
                     time=sprint_value,
-                    global_rank=sprint.rank,
-                    play_at=sprint.record.ts,
+                    global_rank=sprint.data.rank,
+                    play_at=sprint.data.record.ts,
                 )
-                if sprint.record is not None
+                if sprint.data.record is not None
                 else None,
                 blitz=Blitz(
-                    score=blitz.record.endcontext.score,
-                    global_rank=blitz.rank,
-                    play_at=blitz.record.ts,
+                    score=blitz.data.record.results.stats.score,
+                    global_rank=blitz.data.rank,
+                    play_at=blitz.data.record.ts,
                 )
-                if blitz.record is not None
+                if blitz.data.record is not None
                 else None,
-                zen=zen,
+                zen=Zen(level=zen.data.level, score=zen.data.score),
             ),
         ),
     ) as page_hash:
         return await screenshot(f'http://{netloc}/host/{page_hash}.html')
-
-
-async def make_query_text(player: Player) -> UniMessage:
-    user, user_info, sprint, blitz = await gather(player.user, player.get_info(), player.sprint, player.blitz)
-    league = get_league(user_info)
-
-    user_name = user.name.upper()
-
-    message = ''
-    if isinstance(league, NeverPlayedLeague):
-        message += f'用户 {user_name} 没有排位统计数据'
-    else:
-        if isinstance(league, NeverRatedLeague):
-            message += f'用户 {user_name} 暂未完成定级赛, 最近十场的数据:'
-        else:
-            if league.rank == 'z':
-                message += f'用户 {user_name} 暂无段位, {round(league.rating,2)} TR'
-            else:
-                message += f'{league.rank.upper()} 段用户 {user_name} {round(league.rating,2)} TR (#{league.standing})'
-            message += f', 段位分 {round(league.glicko,2)}±{round(league.rd,2)}, 最近十场的数据:'
-        metrics = (
-            get_metrics(pps=league.pps, apm=league.apm, vs=league.vs)
-            if league.vs is not None
-            else get_metrics(pps=league.pps, apm=league.apm)
-        )
-        message += f"\nL'PM: {metrics.lpm} ( {metrics.pps} pps )"
-        message += f'\nAPM: {metrics.apm} ( x{metrics.apl} )'
-        if isinstance(metrics, TetrisMetricsProWithPPSVS):
-            message += f'\nADPM: {metrics.adpm} ( x{metrics.adpl} ) ( {metrics.vs}vs )'
-    if sprint.record is not None:
-        message += f'\n40L: {round(sprint.record.endcontext.final_time/1000,2)}s'
-        message += f' ( #{sprint.rank} )' if sprint.rank is not None else ''
-    if blitz.record is not None:
-        message += f'\nBlitz: {blitz.record.endcontext.score}'
-        message += f' ( #{blitz.rank} )' if blitz.rank is not None else ''
-    return UniMessage(message)
-
-
-async def make_query_result(player: Player, template: Template) -> UniMessage:
-    try:
-        if template == 'v1':
-            return UniMessage.image(raw=await make_query_image_v1(player))
-        if template == 'v2':
-            return UniMessage.image(raw=await make_query_image_v2(player))
-    except FallbackError:
-        ...
-    return await make_query_text(player)
-
-
-class FullExport:
-    cache: ClassVar[defaultdict[str, set[tuple[datetime, Number]]]] = defaultdict(set)
-    latest_update: ClassVar[date | None] = None
-
-    @classmethod
-    async def init(cls) -> None:
-        async with get_session() as session:
-            full_exports = (await session.scalars(select(IORank).where(IORank.update_time >= cls.start_time()))).all()
-        await gather(
-            *[
-                cls._load(update_time, file_hash)
-                for file_hash, update_time in {
-                    i.file_hash: i.update_time for i in full_exports if i.file_hash is not None
-                }.items()
-            ]
-        )
-
-    @classmethod
-    async def update(cls) -> None:
-        if cls.latest_update == datetime.now(tz=ZoneInfo('Asia/Shanghai')).date():
-            return
-        start_time = cls.start_time()
-        for i in cls.cache:
-            cls.cache[i] = {j for j in cls.cache[i] if j[0] >= start_time}
-        latest_time = max(cls.cache)
-        async with get_session() as session:
-            full_exports = (await session.scalars(select(IORank).where(IORank.update_time > latest_time))).all()
-        await gather(
-            *[
-                cls._load(update_time, file_hash)
-                for file_hash, update_time in {
-                    i.file_hash: i.update_time for i in full_exports if i.file_hash is not None
-                }.items()
-            ]
-        )
-        cls.latest_update = datetime.now(tz=ZoneInfo('Asia/Shanghai')).date()
-
-    @classmethod
-    def get_data(cls, unique_identifier: str) -> list[Data]:
-        return [Data(record_at=i[0], tr=i[1]) for i in cls.cache[unique_identifier]]
-
-    @classmethod
-    def start_time(cls) -> datetime:
-        return (
-            datetime.now(ZoneInfo('Asia/Shanghai')).replace(hour=0, minute=0, second=0, microsecond=0)
-            - timedelta(days=9)
-        ).astimezone(UTC)
-
-    @classmethod
-    async def _load(cls, update_time: datetime, file_hash: str) -> None:
-        try:
-            users = type_validate_json(TetraLeagueSuccess, await cls.decompress(file_hash)).data.users
-        except FileNotFoundError:
-            await cls.clear_invalid(file_hash)
-            return
-        update_time = update_time.astimezone(ZoneInfo('Asia/Shanghai'))
-        for i in users:
-            cls.cache[i.id].add((update_time, i.league.rating))
-
-    @classmethod
-    async def decompress(cls, file_hash: str) -> bytes:
-        async with open(get_data_file('nonebot_plugin_tetris_stats', f'{file_hash}.json.zst'), mode='rb') as file:
-            return ZstdDecompressor().decompress(await file.read())
-
-    @classmethod
-    async def clear_invalid(cls, file_hash: str) -> None:
-        async with get_session() as session:
-            full_exports = (await session.scalars(select(IORank).where(IORank.file_hash == file_hash))).all()
-            for i in full_exports:
-                i.file_hash = None
-            await session.commit()
-
-
-@driver.on_startup
-async def _():
-    await FullExport.init()
-    scheduler.add_job(FullExport.update, 'interval', hours=1)

nonebot_plugin_tetris_stats/games/tetrio/rank.py

@@ -1,174 +0,0 @@
-from collections import defaultdict
-from collections.abc import Callable
-from datetime import datetime, timedelta, timezone
-from hashlib import sha512
-from math import floor
-from statistics import mean
-from zoneinfo import ZoneInfo
-
-from aiofiles import open
-from nonebot import get_driver
-from nonebot.compat import model_dump
-from nonebot.matcher import Matcher
-from nonebot.utils import run_sync
-from nonebot_plugin_apscheduler import scheduler  # type: ignore[import-untyped]
-from nonebot_plugin_localstore import get_data_file  # type: ignore[import-untyped]
-from nonebot_plugin_orm import get_session
-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 sqlalchemy import func, select
-from zstandard import ZstdCompressor
-
-from ...db import trigger
-from ...utils.exception import RequestError
-from ...utils.metrics import get_metrics
-from ...utils.retry import retry
-from . import alc
-from .api.schemas.base import FailedModel
-from .api.schemas.tetra_league import ValidUser
-from .api.schemas.user import User
-from .api.tetra_league import full_export
-from .api.typing import Rank
-from .constant import GAME_TYPE, RANK_PERCENTILE
-from .models import IORank
-
-UTC = timezone.utc
-
-driver = get_driver()
-
-
-@alc.assign('TETRIO.rank')
-async def _(matcher: Matcher, rank: Rank, event_session: EventSession):
-    async with trigger(
-        session_persist_id=await get_session_persist_id(event_session),
-        game_platform=GAME_TYPE,
-        command_type='rank',
-        command_args=[],
-    ):
-        if rank == 'z':
-            await matcher.finish('暂不支持查询未知段位')
-        async with get_session() as session:
-            latest_data = (
-                await session.scalars(select(IORank).where(IORank.rank == rank).order_by(IORank.id.desc()).limit(1))
-            ).one()
-            compare_data = (
-                await session.scalars(
-                    select(IORank)
-                    .where(IORank.rank == rank)
-                    .order_by(
-                        func.abs(
-                            func.julianday(IORank.update_time)
-                            - func.julianday(latest_data.update_time - timedelta(hours=24))
-                        )
-                    )
-                    .limit(1)
-                )
-            ).one()
-        message = ''
-        if (datetime.now(UTC) - latest_data.update_time.replace(tzinfo=UTC)) > timedelta(hours=7):
-            message += 'Warning: 数据超过7小时未更新, 请联系Bot主人查看后台\n'
-        message += f'{rank.upper()} 段 分数线 {latest_data.tr_line:.2f} TR, {latest_data.player_count} 名玩家\n'
-        if compare_data.id != latest_data.id:
-            message += f'对比 {(latest_data.update_time-compare_data.update_time).total_seconds()/3600:.2f} 小时前趋势: {f"↑{difference:.2f}" if (difference:=latest_data.tr_line-compare_data.tr_line) > 0 else f"↓{-difference:.2f}" if difference < 0 else "→"}'
-        else:
-            message += '暂无对比数据'
-        avg = get_metrics(pps=latest_data.avg_pps, apm=latest_data.avg_apm, vs=latest_data.avg_vs)
-        low_pps = get_metrics(pps=latest_data.low_pps[1])
-        low_vs = get_metrics(vs=latest_data.low_vs[1])
-        max_pps = get_metrics(pps=latest_data.high_pps[1])
-        max_vs = get_metrics(vs=latest_data.high_vs[1])
-        message += (
-            '\n'
-            '平均数据:\n'
-            f"L'PM: {avg.lpm} ( {avg.pps} pps )\n"
-            f'APM: {avg.apm} ( x{avg.apl} )\n'
-            f'ADPM: {avg.adpm} ( x{avg.adpl} ) ( {avg.vs}vs )\n'
-            '\n'
-            '最低数据:\n'
-            f"L'PM: {low_pps.lpm} ( {low_pps.pps} pps ) By: {latest_data.low_pps[0]['name'].upper()}\n"
-            f'APM: {latest_data.low_apm[1]} By: {latest_data.low_apm[0]["name"].upper()}\n'
-            f'ADPM: {low_vs.adpm} ( {low_vs.vs}vs ) By: {latest_data.low_vs[0]["name"].upper()}\n'
-            '\n'
-            '最高数据:\n'
-            f"L'PM: {max_pps.lpm} ( {max_pps.pps} pps ) By: {latest_data.high_pps[0]['name'].upper()}\n"
-            f'APM: {latest_data.high_apm[1]} By: {latest_data.high_apm[0]["name"].upper()}\n'
-            f'ADPM: {max_vs.adpm} ( {max_vs.vs}vs ) By: {latest_data.high_vs[0]["name"].upper()}\n'
-            '\n'
-            f'数据更新时间: {latest_data.update_time.replace(tzinfo=UTC).astimezone(ZoneInfo("Asia/Shanghai")).strftime("%Y-%m-%d %H:%M:%S")}'
-        )
-        await matcher.finish(message)
-
-
-@scheduler.scheduled_job('cron', hour='0,6,12,18', minute=0)
-@retry(exception_type=RequestError, delay=timedelta(minutes=15))
-async def get_tetra_league_data() -> None:
-    league, original = await full_export(with_original=True)
-    if isinstance(league, FailedModel):
-        msg = f'排行榜数据请求错误:\n{league.error}'
-        raise RequestError(msg)
-
-    def pps(user: ValidUser) -> float:
-        return user.league.pps
-
-    def apm(user: ValidUser) -> float:
-        return user.league.apm
-
-    def vs(user: ValidUser) -> float:
-        return user.league.vs
-
-    def _min(users: list[ValidUser], field: Callable[[ValidUser], float]) -> ValidUser:
-        return min(users, key=field)
-
-    def _max(users: list[ValidUser], field: Callable[[ValidUser], float]) -> ValidUser:
-        return max(users, key=field)
-
-    def build_extremes_data(
-        users: list[ValidUser],
-        field: Callable[[ValidUser], float],
-        sort: Callable[[list[ValidUser], Callable[[ValidUser], float]], ValidUser],
-    ) -> tuple[dict[str, str], float]:
-        user = sort(users, field)
-        return model_dump(User(ID=user.id, name=user.username)), field(user)
-
-    data_hash: str | None = await run_sync((await run_sync(sha512)(original)).hexdigest)()
-    async with open(get_data_file('nonebot_plugin_tetris_stats', f'{data_hash}.json.zst'), mode='wb') as file:
-        await file.write(await run_sync(ZstdCompressor(level=12, threads=-1).compress)(original))
-
-    users = [i for i in league.data.users if isinstance(i, ValidUser)]
-    rank_to_users: defaultdict[Rank, list[ValidUser]] = defaultdict(list)
-    for i in users:
-        rank_to_users[i.league.rank].append(i)
-    rank_info: list[IORank] = []
-    for rank, percentile in RANK_PERCENTILE.items():
-        offset = floor((percentile / 100) * len(users)) - 1
-        tr_line = users[offset].league.rating
-        rank_users = rank_to_users[rank]
-        rank_info.append(
-            IORank(
-                rank=rank,
-                tr_line=tr_line,
-                player_count=len(rank_users),
-                low_pps=(build_extremes_data(rank_users, pps, _min)),
-                low_apm=(build_extremes_data(rank_users, apm, _min)),
-                low_vs=(build_extremes_data(rank_users, vs, _min)),
-                avg_pps=mean({i.league.pps for i in rank_users}),
-                avg_apm=mean({i.league.apm for i in rank_users}),
-                avg_vs=mean({i.league.vs for i in rank_users}),
-                high_pps=(build_extremes_data(rank_users, pps, _max)),
-                high_apm=(build_extremes_data(rank_users, apm, _max)),
-                high_vs=(build_extremes_data(rank_users, vs, _max)),
-                update_time=league.cache.cached_at,
-                file_hash=data_hash,
-            )
-        )
-    async with get_session() as session:
-        session.add_all(rank_info)
-        await session.commit()
-
-
-@driver.on_startup
-async def _() -> None:
-    async with get_session() as session:
-        latest_time = await session.scalar(select(IORank.update_time).order_by(IORank.id.desc()).limit(1))
-    if latest_time is None or datetime.now(tz=UTC) - latest_time.replace(tzinfo=UTC) > timedelta(hours=6):
-        await get_tetra_league_data()

nonebot_plugin_tetris_stats/games/tetrio/record/__init__.py

@@ -1,4 +1,31 @@
-from . import blitz, sprint
+from arclet.alconna import Arg, ArgFlag
+from nonebot_plugin_alconna import Args, At, Subcommand
+
+from ....utils.typing import Me
+from .. import command as base_command
+from .. import get_player
+
+command = Subcommand(
+    'record',
+    Args(
+        Arg(
+            'target',
+            At | Me,
+            notice='@想要查询的人 / 自己',
+            flags=[ArgFlag.HIDDEN, ArgFlag.OPTIONAL],
+        ),
+        Arg(
+            'account',
+            get_player,
+            notice='TETR.IO 用户名 / ID',
+            flags=[ArgFlag.HIDDEN, ArgFlag.OPTIONAL],
+        ),
+    ),
+)
+
+from . import blitz, sprint  # noqa: E402
+
+base_command.add(command)
 
 __all__ = [
     'blitz',

nonebot_plugin_tetris_stats/games/tetrio/record/blitz.py

@@ -5,26 +5,36 @@ from urllib.parse import urlencode
 
 from nonebot.adapters import Event
 from nonebot.matcher import Matcher
-from nonebot_plugin_alconna import At
+from nonebot_plugin_alconna import At, Option
 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_user import get_user  # type: ignore[import-untyped]
+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 get_user
 
-from ....db import query_bind_info
+from ....db import query_bind_info, trigger
 from ....utils.exception import RecordNotFoundError
 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_record_base import Finesse, Max, Mini, Tspins, User
-from ....utils.render.schemas.tetrio_record_blitz import Record, Statistic
+from ....utils.render.schemas.tetrio.record.base import Finesse, Max, Mini, Tspins, User
+from ....utils.render.schemas.tetrio.record.blitz import Record, Statistic
 from ....utils.screenshot import screenshot
 from ....utils.typing import Me
 from ...constant import CANT_VERIFY_MESSAGE
 from .. import alc
 from ..api.player import Player
 from ..constant import GAME_TYPE
+from . import command
+
+command.add(Option('--blitz', dest='blitz'))
+
+alc.shortcut(
+    '(?i:io)(?i:记录|record)(?i:blitz)',
+    command='tstats TETR.IO record --blitz',
+    humanized='io记录blitz',
+)
 
 
 @alc.assign('TETRIO.record.blitz')
@@ -34,64 +44,81 @@ async def _(
     target: At | Me,
     event_session: EventSession,
 ):
-    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 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_blitz_image(player))).finish()
+    async with trigger(
+        session_persist_id=await get_session_persist_id(event_session),
+        game_platform=GAME_TYPE,
+        command_type='record',
+        command_args=['--blitz'],
+    ):
+        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 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_blitz_image(player))).finish()
 
 
 @alc.assign('TETRIO.record.blitz')
-async def _(account: Player):
-    await UniMessage.image(raw=await make_blitz_image(account)).finish()
+async def _(account: Player, event_session: EventSession):
+    async with trigger(
+        session_persist_id=await get_session_persist_id(event_session),
+        game_platform=GAME_TYPE,
+        command_type='record',
+        command_args=['--blitz'],
+    ):
+        await UniMessage.image(raw=await make_blitz_image(account)).finish()
 
 
 async def make_blitz_image(player: Player) -> bytes:
-    user, user_info, blitz = await gather(player.user, player.get_info(), player.blitz)
-    if blitz.record is None:
-        msg = f'未找到用户 {user.name.upper()} 的 40L 记录'
+    user, blitz = await gather(player.user, player.blitz)
+    if blitz.data.record is None:
+        msg = f'未找到用户 {user.name.upper()} 的 Blitz 记录'
         raise RecordNotFoundError(msg)
-    endcontext = blitz.record.endcontext
-    clears = endcontext.clears
-    duration = timedelta(milliseconds=endcontext.final_time).total_seconds()
-    metrics = get_metrics(pps=endcontext.piecesplaced / duration)
+    stats = blitz.data.record.results.stats
+    clears = stats.clears
+    duration = timedelta(milliseconds=stats.finaltime).total_seconds()
+    metrics = get_metrics(pps=stats.piecesplaced / duration)
     netloc = get_self_netloc()
     async with HostPage(
         page=await render(
             'v2/tetrio/record/blitz',
             Record(
+                type='best',
                 user=User(
                     id=user.ID,
                     name=user.name.upper(),
-                    avatar=f'http://{netloc}/host/resource/tetrio/avatars/{user.ID}?{urlencode({"revision": user_info.data.user.avatar_revision})}'
-                    if user_info.data.user.avatar_revision is not None and user_info.data.user.avatar_revision != 0
+                    avatar=f'http://{netloc}/host/resource/tetrio/avatars/{user.ID}?{urlencode({"revision": avatar_revision})}'
+                    if (avatar_revision := (await player.avatar_revision)) is not None and avatar_revision != 0
                     else Avatar(
                         type='identicon',
                         hash=md5(user.ID.encode()).hexdigest(),  # noqa: S324
                     ),
                 ),
-                replay_id=blitz.record.replayid,
-                rank=blitz.rank,
+                replay_id=blitz.data.record.replayid,
+                rank=blitz.data.rank,
+                personal_rank=1,
                 statistic=Statistic(
-                    keys=endcontext.inputs,
-                    kpp=round(endcontext.inputs / endcontext.piecesplaced, 2),
-                    kps=round(endcontext.inputs / duration, 2),
-                    max=Max(combo=endcontext.combo, btb=endcontext.btb),
-                    pieces=endcontext.piecesplaced,
+                    keys=stats.inputs,
+                    kpp=round(stats.inputs / stats.piecesplaced, 2),
+                    kps=round(stats.inputs / duration, 2),
+                    max=Max(
+                        combo=max((0, stats.topcombo - 1)),
+                        btb=max((0, stats.topbtb - 1)),
+                    ),
+                    pieces=stats.piecesplaced,
                     pps=metrics.pps,
-                    lines=endcontext.lines,
+                    lines=stats.lines,
                     lpm=metrics.lpm,
-                    holds=endcontext.holds,
-                    score=endcontext.score,
-                    spp=round(endcontext.score / metrics._pps, 2),  # noqa: SLF001 获取最高精度
+                    holds=stats.holds,
+                    score=stats.score,
+                    spp=round(stats.score / stats.piecesplaced, 2),
                     single=clears.singles,
                     double=clears.doubles,
                     triple=clears.triples,
@@ -100,7 +127,7 @@ async def make_blitz_image(player: Player) -> bytes:
                         total=clears.realtspins,
                         single=clears.tspinsingles,
                         double=clears.tspindoubles,
-                        triple=clears.triples,
+                        triple=clears.tspintriples,
                         mini=Mini(
                             total=clears.minitspins,
                             single=clears.minitspinsingles,
@@ -109,12 +136,12 @@ async def make_blitz_image(player: Player) -> bytes:
                     ),
                     all_clear=clears.allclear,
                     finesse=Finesse(
-                        faults=endcontext.finesse.faults,
-                        accuracy=round(endcontext.finesse.perfectpieces / endcontext.piecesplaced * 100, 2),
+                        faults=stats.finesse.faults,
+                        accuracy=round(stats.finesse.perfectpieces / stats.piecesplaced * 100, 2),
                     ),
-                    level=endcontext.level,
+                    level=stats.level,
                 ),
-                play_at=blitz.record.ts,
+                play_at=blitz.data.record.ts,
             ),
         )
     ) as page_hash:

nonebot_plugin_tetris_stats/games/tetrio/record/sprint.py

@@ -5,26 +5,36 @@ from urllib.parse import urlencode
 
 from nonebot.adapters import Event
 from nonebot.matcher import Matcher
-from nonebot_plugin_alconna import At
+from nonebot_plugin_alconna import At, Option
 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_user import get_user  # type: ignore[import-untyped]
+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 get_user
 
-from ....db import query_bind_info
+from ....db import query_bind_info, trigger
 from ....utils.exception import RecordNotFoundError
 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_record_base import Finesse, Max, Mini, Tspins, User
-from ....utils.render.schemas.tetrio_record_sprint import Record, Statistic
+from ....utils.render.schemas.tetrio.record.base import Finesse, Max, Mini, Statistic, Tspins, User
+from ....utils.render.schemas.tetrio.record.sprint import Record
 from ....utils.screenshot import screenshot
 from ....utils.typing import Me
 from ...constant import CANT_VERIFY_MESSAGE
 from .. import alc
 from ..api.player import Player
 from ..constant import GAME_TYPE
+from . import command
+
+command.add(Option('--40l', dest='sprint'))
+
+alc.shortcut(
+    '(?i:io)(?i:记录|record)(?i:40l)',
+    command='tstats TETR.IO record --40l',
+    humanized='io记录40l',
+)
 
 
 @alc.assign('TETRIO.record.sprint')
@@ -34,65 +44,82 @@ async def _(
     target: At | Me,
     event_session: EventSession,
 ):
-    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 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_sprint_image(player))).finish()
+    async with trigger(
+        session_persist_id=await get_session_persist_id(event_session),
+        game_platform=GAME_TYPE,
+        command_type='record',
+        command_args=['--40l'],
+    ):
+        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 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_sprint_image(player))).finish()
 
 
 @alc.assign('TETRIO.record.sprint')
-async def _(account: Player):
-    await UniMessage.image(raw=await make_sprint_image(account)).finish()
+async def _(account: Player, event_session: EventSession):
+    async with trigger(
+        session_persist_id=await get_session_persist_id(event_session),
+        game_platform=GAME_TYPE,
+        command_type='record',
+        command_args=['--40l'],
+    ):
+        await UniMessage.image(raw=await make_sprint_image(account)).finish()
 
 
 async def make_sprint_image(player: Player) -> bytes:
-    user, user_info, sprint = await gather(player.user, player.get_info(), player.sprint)
-    if sprint.record is None:
+    user, sprint = await gather(player.user, player.sprint)
+    if sprint.data.record is None:
         msg = f'未找到用户 {user.name.upper()} 的 40L 记录'
         raise RecordNotFoundError(msg)
-    endcontext = sprint.record.endcontext
-    clears = endcontext.clears
-    duration = timedelta(milliseconds=endcontext.final_time).total_seconds()
+    stats = sprint.data.record.results.stats
+    clears = stats.clears
+    duration = timedelta(milliseconds=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
-    metrics = get_metrics(pps=endcontext.piecesplaced / duration)
+    metrics = get_metrics(pps=stats.piecesplaced / duration)
     netloc = get_self_netloc()
     async with HostPage(
         page=await render(
             'v2/tetrio/record/40l',
             Record(
+                type='best',
                 user=User(
                     id=user.ID,
                     name=user.name.upper(),
-                    avatar=f'http://{netloc}/host/resource/tetrio/avatars/{user.ID}?{urlencode({"revision": user_info.data.user.avatar_revision})}'
-                    if user_info.data.user.avatar_revision is not None and user_info.data.user.avatar_revision != 0
+                    avatar=f'http://{netloc}/host/resource/tetrio/avatars/{user.ID}?{urlencode({"revision": avatar_revision})}'
+                    if (avatar_revision := (await player.avatar_revision)) is not None and avatar_revision != 0
                     else Avatar(
                         type='identicon',
                         hash=md5(user.ID.encode()).hexdigest(),  # noqa: S324
                     ),
                 ),
                 time=sprint_value,
-                replay_id=sprint.record.replayid,
-                rank=sprint.rank,
+                replay_id=sprint.data.record.replayid,
+                rank=sprint.data.rank,
+                personal_rank=1,
                 statistic=Statistic(
-                    keys=endcontext.inputs,
-                    kpp=round(endcontext.inputs / endcontext.piecesplaced, 2),
-                    kps=round(endcontext.inputs / duration, 2),
-                    max=Max(combo=endcontext.combo, btb=endcontext.btb),
-                    pieces=endcontext.piecesplaced,
+                    keys=stats.inputs,
+                    kpp=round(stats.inputs / stats.piecesplaced, 2),
+                    kps=round(stats.inputs / duration, 2),
+                    max=Max(
+                        combo=max((0, stats.topcombo - 1)),
+                        btb=max((0, stats.topbtb - 1)),
+                    ),
+                    pieces=stats.piecesplaced,
                     pps=metrics.pps,
-                    lines=endcontext.lines,
+                    lines=stats.lines,
                     lpm=metrics.lpm,
-                    holds=endcontext.holds,
-                    score=endcontext.score,
+                    holds=stats.holds,
+                    score=stats.score,
                     single=clears.singles,
                     double=clears.doubles,
                     triple=clears.triples,
@@ -101,7 +128,7 @@ async def make_sprint_image(player: Player) -> bytes:
                         total=clears.realtspins,
                         single=clears.tspinsingles,
                         double=clears.tspindoubles,
-                        triple=clears.triples,
+                        triple=clears.tspintriples,
                         mini=Mini(
                             total=clears.minitspins,
                             single=clears.minitspinsingles,
@@ -110,11 +137,11 @@ async def make_sprint_image(player: Player) -> bytes:
                     ),
                     all_clear=clears.allclear,
                     finesse=Finesse(
-                        faults=endcontext.finesse.faults,
-                        accuracy=round(endcontext.finesse.perfectpieces / endcontext.piecesplaced * 100, 2),
+                        faults=stats.finesse.faults,
+                        accuracy=round(stats.finesse.perfectpieces / stats.piecesplaced * 100, 2),
                     ),
                 ),
-                play_at=sprint.record.ts,
+                play_at=sprint.data.record.ts,
             ),
         )
     ) as page_hash:

nonebot_plugin_tetris_stats/games/top/__init__.py

@@ -1,9 +1,9 @@
-from arclet.alconna import Arg, ArgFlag, Args, Subcommand
-from nonebot_plugin_alconna import At
+from arclet.alconna import Arg, ArgFlag
+from nonebot_plugin_alconna import Args, At, Subcommand
 
 from ...utils.exception import MessageFormatError
 from ...utils.typing import Me
-from .. import add_block_handlers, alc
+from .. import add_block_handlers, alc, command
 from .api import Player
 from .constant import USER_NAME
 
@@ -14,7 +14,7 @@ def get_player(name: str) -> Player | MessageFormatError:
     return MessageFormatError('用户名/ID不合法')
 
 
-alc.command.add(
+command.add(
     Subcommand(
         'TOP',
         Subcommand(

nonebot_plugin_tetris_stats/games/top/bind.py

@@ -1,9 +1,9 @@
 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
 from nonebot_plugin_session_orm import get_session_persist_id  # type: ignore[import-untyped]
-from nonebot_plugin_user import User  # type: ignore[import-untyped]
-from nonebot_plugin_userinfo import BotUserInfo, EventUserInfo, UserInfo  # type: ignore[import-untyped]
+from nonebot_plugin_user import User
+from nonebot_plugin_userinfo import BotUserInfo, EventUserInfo, UserInfo
 
 from ...db import BindStatus, create_or_update_bind, trigger
 from ...utils.host import HostPage, get_self_netloc

nonebot_plugin_tetris_stats/games/top/query.py

@@ -3,9 +3,9 @@ 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
 from nonebot_plugin_session_orm import get_session_persist_id  # type: ignore[import-untyped]
-from nonebot_plugin_user import get_user  # type: ignore[import-untyped]
+from nonebot_plugin_user import get_user
 
 from ...db import query_bind_info, trigger
 from ...utils.metrics import get_metrics

nonebot_plugin_tetris_stats/games/tos/__init__.py

@@ -1,9 +1,9 @@
-from arclet.alconna import Arg, ArgFlag, Args, Subcommand
-from nonebot_plugin_alconna import At
+from arclet.alconna import Arg, ArgFlag
+from nonebot_plugin_alconna import Args, At, Subcommand
 
 from ...utils.exception import MessageFormatError
 from ...utils.typing import Me
-from .. import add_block_handlers, alc
+from .. import add_block_handlers, alc, command
 from .api import Player
 from .constant import USER_NAME
 
@@ -19,7 +19,7 @@ def get_player(teaid_or_name: str) -> Player | MessageFormatError:
     return MessageFormatError('用户名/ID不合法')
 
 
-alc.command.add(
+command.add(
     Subcommand(
         'TOS',
         Subcommand(

nonebot_plugin_tetris_stats/games/tos/bind.py

@@ -1,9 +1,9 @@
 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
 from nonebot_plugin_session_orm import get_session_persist_id  # type: ignore[import-untyped]
-from nonebot_plugin_user import User  # type: ignore[import-untyped]
-from nonebot_plugin_userinfo import BotUserInfo, EventUserInfo, UserInfo  # type: ignore[import-untyped]
+from nonebot_plugin_user import User
+from nonebot_plugin_userinfo import BotUserInfo, EventUserInfo, UserInfo
 
 from ...db import BindStatus, create_or_update_bind, trigger
 from ...utils.host import HostPage, get_self_netloc

nonebot_plugin_tetris_stats/games/tos/constant.py

@@ -6,9 +6,6 @@ GAME_TYPE: Literal['TOS'] = 'TOS'
 BASE_URL = {
     'https://teatube.cn:8888/',
     'http://cafuuchino1.studio26f.org:19970',
-    'http://cafuuchino2.studio26f.org:19970',
-    'http://cafuuchino3.studio26f.org:19970',
-    'http://cafuuchino4.studio26f.org:19970',
 }
 
 USER_NAME = compile(

nonebot_plugin_tetris_stats/games/tos/query.py

@@ -8,10 +8,10 @@ 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
 from nonebot_plugin_session_orm import get_session_persist_id  # type: ignore[import-untyped]
-from nonebot_plugin_user import get_user  # type: ignore[import-untyped]
-from nonebot_plugin_userinfo import EventUserInfo, UserInfo  # type: ignore[import-untyped]
+from nonebot_plugin_user import get_user
+from nonebot_plugin_userinfo import EventUserInfo, UserInfo
 
 from ...db import query_bind_info, trigger
 from ...utils.exception import RequestError

nonebot_plugin_tetris_stats/utils/host.py

@@ -11,7 +11,7 @@ from nonebot.log import logger
 from ..config.config import CACHE_PATH
 from .image import img_to_png
 from .request import Request
-from .templates import templates_dir
+from .templates import TEMPLATES_DIR
 
 if TYPE_CHECKING:
     from pydantic import IPvAnyAddress
@@ -48,7 +48,7 @@ class HostPage:
 def _():
     app.mount(
         '/host/assets',
-        StaticFiles(directory=templates_dir / 'assets'),
+        StaticFiles(directory=TEMPLATES_DIR / 'assets'),
         name='assets',
     )
     logger.success('assets mounted')

nonebot_plugin_tetris_stats/utils/image.py

@@ -2,7 +2,7 @@ from base64 import b64encode
 from io import BytesIO
 from typing import Literal, overload
 
-from nonebot_plugin_userinfo import UserInfo  # type: ignore[import-untyped]
+from nonebot_plugin_userinfo import UserInfo
 from PIL import Image
 
 

nonebot_plugin_tetris_stats/utils/render/__init__.py

@@ -3,59 +3,73 @@ from typing import Literal, overload
 from jinja2 import Environment, FileSystemLoader
 from nonebot.compat import PYDANTIC_V2
 
-from ..templates import templates_dir
+from ..templates import TEMPLATES_DIR
 from .schemas.bind import Bind
-from .schemas.tetrio_info import Info as TETRIOInfo
-from .schemas.tetrio_info_v2 import Info as TETRIOInfoV2
-from .schemas.tetrio_record_blitz import Record as TETRIORecordBlitz
-from .schemas.tetrio_record_sprint import Record as TETRIORecordSprint
+from .schemas.tetrio.rank.detail import Data as TETRIORankDetailData
+from .schemas.tetrio.rank.v1 import Data as TETRIORankDataV1
+from .schemas.tetrio.rank.v2 import Data as TETRIORankDataV2
+from .schemas.tetrio.record.blitz import Record as TETRIORecordBlitz
+from .schemas.tetrio.record.sprint import Record as TETRIORecordSprint
+from .schemas.tetrio.user.info_v1 import Info as TETRIOUserInfoV1
+from .schemas.tetrio.user.info_v2 import Info as TETRIOUserInfoV2
+from .schemas.tetrio.user.list_v2 import List as TETRIOUserListV2
 from .schemas.top_info import Info as TOPInfo
 from .schemas.tos_info import Info as TOSInfo
 
 env = Environment(
-    loader=FileSystemLoader(templates_dir), autoescape=True, trim_blocks=True, lstrip_blocks=True, enable_async=True
+    loader=FileSystemLoader(TEMPLATES_DIR), autoescape=True, trim_blocks=True, lstrip_blocks=True, enable_async=True
 )
 
 
 @overload
 async def render(render_type: Literal['v1/binding'], data: Bind) -> str: ...
-
-
 @overload
-async def render(render_type: Literal['v1/tetrio/info'], data: TETRIOInfo) -> str: ...
-
-
+async def render(render_type: Literal['v1/tetrio/info'], data: TETRIOUserInfoV1) -> str: ...
+@overload
+async def render(render_type: Literal['v1/tetrio/rank'], data: TETRIORankDataV1) -> str: ...
 @overload
 async def render(render_type: Literal['v1/top/info'], data: TOPInfo) -> str: ...
-
-
 @overload
 async def render(render_type: Literal['v1/tos/info'], data: TOSInfo) -> str: ...
-
-
 @overload
-async def render(render_type: Literal['v2/tetrio/info'], data: TETRIOInfoV2) -> 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: ...
-
-
 @overload
 async def render(render_type: Literal['v2/tetrio/record/blitz'], data: TETRIORecordBlitz) -> str: ...
+@overload
+async def render(render_type: Literal['v2/tetrio/rank'], data: TETRIORankDataV2) -> str: ...
+@overload
+async def render(render_type: Literal['v2/tetrio/rank/detail'], data: TETRIORankDetailData) -> str: ...
 
 
 async def render(
     render_type: Literal[
         'v1/binding',
         'v1/tetrio/info',
+        'v1/tetrio/rank',
         'v1/top/info',
         'v1/tos/info',
-        'v2/tetrio/info',
+        'v2/tetrio/user/info',
+        'v2/tetrio/user/list',
         'v2/tetrio/record/40l',
         'v2/tetrio/record/blitz',
+        'v2/tetrio/rank',
+        'v2/tetrio/rank/detail',
     ],
-    data: Bind | TETRIOInfo | TOPInfo | TOSInfo | TETRIOInfoV2 | TETRIORecordSprint | TETRIORecordBlitz,
+    data: Bind
+    | TETRIOUserInfoV1
+    | TETRIORankDataV1
+    | TOPInfo
+    | TOSInfo
+    | TETRIOUserInfoV2
+    | TETRIOUserListV2
+    | TETRIORecordSprint
+    | TETRIORecordBlitz
+    | TETRIORankDataV2
+    | TETRIORankDetailData,
 ) -> str:
     if PYDANTIC_V2:
         return await env.get_template('index.html').render_async(

nonebot_plugin_tetris_stats/utils/render/schemas/tetrio/rank/detail.py

@@ -0,0 +1,31 @@
+from datetime import datetime
+
+from pydantic import BaseModel
+
+from ......games.tetrio.api.typing import ValidRank
+
+
+class SpecialData(BaseModel):
+    apm: float
+    pps: float
+    lpm: float
+    vs: float
+    adpm: float
+    apl: float | None = None
+    adpl: float | None = None
+    apm_holder: str | None = None
+    pps_holder: str | None = None
+    vs_holder: str | None = None
+
+
+class Data(BaseModel):
+    name: ValidRank
+    trending: float
+    require_tr: float
+    players: int
+
+    minimum_data: SpecialData
+    average_data: SpecialData
+    maximum_data: SpecialData
+
+    updated_at: datetime

nonebot_plugin_tetris_stats/utils/render/schemas/tetrio/rank/v1.py

@@ -0,0 +1,16 @@
+from datetime import datetime
+
+from pydantic import BaseModel
+
+from ......games.tetrio.api.typing import ValidRank
+
+
+class ItemData(BaseModel):
+    trending: float
+    require_tr: float
+    players: int
+
+
+class Data(BaseModel):
+    items: dict[ValidRank, ItemData]
+    updated_at: datetime

nonebot_plugin_tetris_stats/utils/render/schemas/tetrio/rank/v2.py

@@ -0,0 +1,25 @@
+from datetime import datetime
+
+from pydantic import BaseModel
+
+from ......games.tetrio.api.typing import ValidRank
+
+
+class AverageData(BaseModel):
+    pps: float
+    apm: float
+    apl: float
+    vs: float
+    adpl: float
+
+
+class ItemData(BaseModel):
+    require_tr: float
+    trending: float
+    average_data: AverageData
+    players: int
+
+
+class Data(BaseModel):
+    items: dict[ValidRank, ItemData]
+    updated_at: datetime

nonebot_plugin_tetris_stats/utils/render/schemas/tetrio/record/base.py

@@ -1,6 +1,9 @@
+from datetime import datetime
+from typing import Literal
+
 from pydantic import BaseModel
 
-from .base import People
+from ...base import People
 
 
 class User(People):
@@ -32,7 +35,7 @@ class Finesse(BaseModel):
     accuracy: float
 
 
-class RecordStatistic(BaseModel):
+class Statistic(BaseModel):
     keys: int
     kpp: float
     kps: float
@@ -56,3 +59,17 @@ class RecordStatistic(BaseModel):
     all_clear: int
 
     finesse: Finesse
+
+
+class Record(BaseModel):
+    type: Literal['best', 'personal_best', 'recent', 'disputed']
+
+    user: User
+
+    replay_id: str
+    rank: int | None
+    personal_rank: int | None
+
+    statistic: Statistic
+
+    play_at: datetime

nonebot_plugin_tetris_stats/utils/render/schemas/tetrio/record/blitz.py

@@ -0,0 +1,12 @@
+from .base import Record as BaseRecord
+from .base import Statistic as BaseStatistic
+
+
+class Statistic(BaseStatistic):
+    spp: float
+
+    level: int
+
+
+class Record(BaseRecord):
+    statistic: Statistic

nonebot_plugin_tetris_stats/utils/render/schemas/tetrio/record/sprint.py

@@ -0,0 +1,5 @@
+from .base import Record as BaseRecord
+
+
+class Record(BaseRecord):
+    time: str

nonebot_plugin_tetris_stats/utils/render/schemas/tetrio/user/base.py

@@ -0,0 +1,10 @@
+from datetime import datetime
+
+from pydantic import BaseModel
+
+from .....typing import Number
+
+
+class TetraLeagueHistoryData(BaseModel):
+    record_at: datetime
+    tr: Number

nonebot_plugin_tetris_stats/utils/render/schemas/tetrio/user/info_v1.py

@@ -0,0 +1,49 @@
+from pydantic import BaseModel
+
+from ......games.tetrio.api.typing import Rank
+from .....typing import Number
+from ...base import People, Ranking
+from .base import TetraLeagueHistoryData
+
+
+class User(People):
+    bio: str | None
+
+
+class TetraLeague(BaseModel):
+    rank: Rank
+    tr: Number
+    global_rank: Number
+    pps: Number
+    lpm: Number
+    apm: Number
+    apl: Number
+    vs: Number
+    adpm: Number
+    adpl: Number
+
+
+class TetraLeagueHistory(BaseModel):
+    data: list[TetraLeagueHistoryData]
+    split_interval: Number
+    min_tr: Number
+    max_tr: Number
+    offset: Number
+
+
+class Radar(BaseModel):
+    app: Number
+    dsps: Number
+    dspp: Number
+    ci: Number
+    ge: Number
+
+
+class Info(BaseModel):
+    user: User
+    ranking: Ranking
+    tetra_league: TetraLeague
+    tetra_league_history: TetraLeagueHistory
+    radar: Radar
+    sprint: str
+    blitz: str

nonebot_plugin_tetris_stats/utils/render/schemas/tetrio/user/info_v2.py

@@ -1,11 +1,13 @@
 from datetime import datetime
+from typing import Literal
 
 from pydantic import BaseModel
 
-from ....games.tetrio.api.schemas.user_records import Zen
-from ....games.tetrio.api.typing import Rank
-from ...typing import Number
-from .base import Avatar
+from ......games.tetrio.api.schemas.user_records import Zen
+from ......games.tetrio.api.typing import Rank, ValidRank
+from .....typing import Number
+from ...base import Avatar
+from .base import TetraLeagueHistoryData
 
 
 class Badge(BaseModel):
@@ -20,12 +22,14 @@ class User(BaseModel):
     name: str
     country: str | None
 
+    role: Literal['anon', 'user', 'bot', 'halfmod', 'mod', 'admin', 'sysop', 'banned']
+
     avatar: str | Avatar
     banner: str | None
 
     bio: str | None
 
-    friend_count: int
+    friend_count: int | None
     supporter_tier: int
 
     verified: bool
@@ -50,20 +54,20 @@ class TetraLeagueStatistic(BaseModel):
 
 class TetraLeague(BaseModel):
     rank: Rank
-    highest_rank: Rank
+    highest_rank: ValidRank
 
     tr: Number
 
-    glicko: Number
-    rd: Number
+    glicko: Number | None
+    rd: Number | None
 
     global_rank: int | None
     country_rank: int | None
 
-    pps: Number
+    pps: Number | None
 
-    apm: Number
-    apl: Number
+    apm: Number | None
+    apl: Number | None
 
     vs: Number | None
     adpl: Number | None
@@ -72,6 +76,8 @@ class TetraLeague(BaseModel):
 
     decaying: bool
 
+    history: list[TetraLeagueHistoryData] | None
+
 
 class Sprint(BaseModel):
     time: str
@@ -91,4 +97,4 @@ class Info(BaseModel):
     statistic: Statistic | None
     sprint: Sprint | None
     blitz: Blitz | None
-    zen: Zen
+    zen: Zen | None

nonebot_plugin_tetris_stats/utils/render/schemas/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/render/schemas/tetrio_info.py

@@ -1,72 +0,0 @@
-from datetime import datetime
-from typing import Annotated, ClassVar
-
-from nonebot.compat import PYDANTIC_V2
-from pydantic import BaseModel
-
-from ....games.tetrio.api.typing import Rank
-from ...typing import Number
-from .base import People, Ranking
-
-if PYDANTIC_V2:
-    from pydantic import PlainSerializer
-
-
-def format_datetime_to_timestamp(dt: datetime) -> int:
-    return int(dt.timestamp() * 1000)
-
-
-class User(People):
-    bio: str | None
-
-
-class TetraLeague(BaseModel):
-    rank: Rank
-    tr: Number
-    global_rank: Number
-    pps: Number
-    lpm: Number
-    apm: Number
-    apl: Number
-    vs: Number
-    adpm: Number
-    adpl: Number
-
-
-class Data(BaseModel):
-    if PYDANTIC_V2:
-        record_at: Annotated[datetime, PlainSerializer(format_datetime_to_timestamp, return_type=int)]
-    else:
-        record_at: datetime  # type: ignore[no-redef]
-    tr: Number
-
-
-class TetraLeagueHistory(BaseModel):
-    data: list[Data]
-    split_interval: Number
-    min_tr: Number
-    max_tr: Number
-    offset: Number
-
-
-class Radar(BaseModel):
-    app: Number
-    dsps: Number
-    dspp: Number
-    ci: Number
-    ge: Number
-
-
-class Info(BaseModel):
-    user: User
-    ranking: Ranking
-    tetra_league: TetraLeague
-    tetra_league_history: TetraLeagueHistory
-    radar: Radar
-    sprint: str
-    blitz: str
-
-    if not PYDANTIC_V2:
-
-        class Config:
-            json_encoders: ClassVar[dict] = {datetime: format_datetime_to_timestamp}

nonebot_plugin_tetris_stats/utils/render/schemas/tetrio_record_blitz.py

@@ -1,22 +0,0 @@
-from datetime import datetime
-
-from pydantic import BaseModel
-
-from .tetrio_record_base import RecordStatistic, User
-
-
-class Statistic(RecordStatistic):
-    spp: float
-
-    level: int
-
-
-class Record(BaseModel):
-    user: User
-
-    replay_id: str
-    rank: int | None
-
-    statistic: Statistic
-
-    play_at: datetime

nonebot_plugin_tetris_stats/utils/render/schemas/tetrio_record_sprint.py

@@ -1,18 +0,0 @@
-from datetime import datetime
-
-from pydantic import BaseModel
-
-from .tetrio_record_base import RecordStatistic as Statistic
-from .tetrio_record_base import User
-
-
-class Record(BaseModel):
-    user: User
-
-    time: str
-    replay_id: str
-    rank: int | None
-
-    statistic: Statistic
-
-    play_at: datetime

nonebot_plugin_tetris_stats/utils/screenshot.py

@@ -1,15 +1,31 @@
-from playwright.async_api import TimeoutError
+from nonebot import get_plugin_config
+from playwright.async_api import TimeoutError, ViewportSize
 
+from ..config.config import Config
 from .browser import BrowserManager
 from .retry import retry
+from .time_it import time_it
+
+config = get_plugin_config(Config)
 
 
 @retry(exception_type=TimeoutError, reply='截图失败, 重试中')
+@time_it
 async def screenshot(url: str) -> bytes:
     browser = await BrowserManager.get_browser()
     async with (
-        await browser.new_page(viewport={'width': 3000, 'height': 3000}) as page,
+        await browser.new_page(device_scale_factor=config.tetris_screenshot_quality) as page,
     ):
         await page.goto(url)
         await page.wait_for_load_state('networkidle')
+        size: ViewportSize = await page.evaluate("""
+            () => {
+                const element = document.querySelector('#content');
+                return {
+                    width: element.offsetWidth,
+                    height: element.offsetHeight,
+                };
+            };
+        """)
+        await page.set_viewport_size(size)
         return await page.locator('id=content').screenshot(timeout=5000, type='png')

nonebot_plugin_tetris_stats/utils/templates.py

@@ -1,71 +1,130 @@
-from asyncio.subprocess import PIPE, create_subprocess_exec
+from hashlib import sha256
+from http import HTTPStatus
+from pathlib import Path
 from shutil import rmtree
+from time import time_ns
+from zipfile import ZipFile
 
+from aiofiles import open
+from httpx import AsyncClient
 from nonebot import get_driver
 from nonebot.log import logger
 from nonebot.permission import SUPERUSER
-from nonebot_plugin_alconna import on_alconna
-from nonebot_plugin_localstore import get_data_dir  # type: ignore[import-untyped]
+from nonebot_plugin_alconna import Alconna, Args, Option, on_alconna
+from nonebot_plugin_localstore import get_cache_file, get_data_dir
+from rich.progress import Progress
 
 driver = get_driver()
 
-templates_dir = get_data_dir('nonebot_plugin_tetris_stats') / 'templates'
+TEMPLATES_DIR = get_data_dir('nonebot_plugin_tetris_stats') / 'templates'
 
-alc = on_alconna('更新模板', permission=SUPERUSER)
+alc = on_alconna(Alconna('更新模板', Option('--revision', Args['revision', str], alias={'-R'})), permission=SUPERUSER)
+
+
+async def download_templates(tag: str) -> Path:
+    logger.info(f'开始下载模板 {tag}')
+    async with AsyncClient() as client:
+        if tag == 'latest':
+            logger.info('目标为 latest, 正在获取最新版本号')
+            tag = (
+                (
+                    await client.get(
+                        'https://github.com/A-Minos/tetris-stats-templates/releases/latest', follow_redirects=True
+                    )
+                )
+                .url.path.strip('/')
+                .rsplit('/', 1)[-1]
+            )
+            logger.success(f'获取到的最新版本号: {tag}')
+        path = get_cache_file('nonebot_plugin_tetris_stats', f'dist_{time_ns()}.zip')
+        with Progress() as progress:
+            task_id = progress.add_task('[red]Downloading...', total=None)
+            async with (
+                client.stream(
+                    'GET',
+                    f'https://github.com/A-Minos/tetris-stats-templates/releases/download/{tag}/dist.zip',
+                    follow_redirects=True,
+                ) as response,
+                open(path, 'wb') as file,
+            ):
+                response.raise_for_status()
+                progress.update(task_id, total=int(response.headers.get('content-length', 0)) or None)
+                async for chunk in response.aiter_bytes():
+                    await file.write(chunk)
+                    progress.update(task_id, advance=len(chunk))
+        logger.success('模板下载完成')
+        return path
+
+
+async def unzip_templates(zip_path: Path) -> Path:
+    logger.info('开始解压模板')
+    temp_path = TEMPLATES_DIR.parent / f'temp_{time_ns()}'
+    with ZipFile(zip_path) as zip_file:
+        zip_file.extractall(temp_path)
+    zip_path.unlink()
+    logger.success('模板解压完成')
+    return temp_path
+
+
+async def check_hash(hash_file_path: Path) -> bool:
+    logger.info('开始校验模板哈希值')
+    for i in hash_file_path.read_text().splitlines():
+        file_sha256, file_relative_path = i.split(maxsplit=1)
+        file_path = hash_file_path.parent / file_relative_path
+        hasher = sha256()
+        if not file_path.is_file():
+            logger.error(f'{file_path.name} 不存在或不是文件')
+            return False
+        async with open(file_path, 'rb') as file:
+            while True:
+                chunk = await file.read(65535)
+                if not chunk:
+                    break
+                hasher.update(chunk)
+        if hasher.hexdigest() != file_sha256:
+            logger.error(f'{file_path.name} hash 不匹配')
+            return False
+        logger.debug(f'{file_path.name} hash 匹配成功')
+    logger.success('模板哈希值校验成功')
+    return True
+
+
+async def init_templates(tag: str) -> bool:
+    logger.info(f'开始初始化模板 {tag}')
+    temp_path = await unzip_templates(await download_templates(tag))
+    if not await check_hash(temp_path / 'hash.sha256'):
+        rmtree(temp_path)
+        return False
+    if TEMPLATES_DIR.exists():
+        logger.info('清除旧模板文件')
+        rmtree(TEMPLATES_DIR)
+    temp_path.rename(TEMPLATES_DIR)
+    logger.info('模板初始化完成')
+    return True
+
+
+async def check_tag(tag: str) -> bool:
+    async with AsyncClient() as client:
+        return (
+            await client.get(f'https://github.com/A-Minos/tetris-stats-templates/releases/tag/{tag}')
+        ).status_code != HTTPStatus.NOT_FOUND
 
 
 @driver.on_startup
-async def init_templates() -> None:
-    try:
-        await create_subprocess_exec('git', '--version', stdout=PIPE)
-    except FileNotFoundError as e:
-        msg = '未找到 git, 请确保 git 已安装并在环境变量中\n安装步骤请参阅: https://git-scm.com/book/zh/v2/%E8%B5%B7%E6%AD%A5-%E5%AE%89%E8%A3%85-Git'
-        raise RuntimeError(msg) from e
-    if not templates_dir.exists():
-        logger.info('模板仓库不存在, 正在尝试初始化...')
-        proc = await create_subprocess_exec(
-            'git',
-            'clone',
-            '-b',
-            'gh-pages',
-            'https://github.com/A-Minos/tetris-stats-templates',
-            templates_dir,
-            '--depth=1',
-            stdout=PIPE,
-            stderr=PIPE,
-        )
-        stdout, stderr = await proc.communicate()
-        if proc.returncode != 0:
-            for i in stderr.decode().splitlines():
-                logger.error(i)
-            msg = '初始化模板仓库失败'
-            raise RuntimeError(msg)
-        logger.success('模板仓库初始化成功')
+async def _():
+    if (path := (TEMPLATES_DIR / 'hash.sha256')).is_file() and await check_hash(path):
+        logger.success('模板验证成功')
         return
-    proc = await create_subprocess_exec(
-        'git', 'rev-parse', '--is-inside-work-tree', stdout=PIPE, stderr=PIPE, cwd=templates_dir
-    )
-    stdout, stderr = await proc.communicate()
-    if proc.returncode != 0:
-        for i in stderr.decode().splitlines():
-            logger.error(i)
-        logger.warning('模板仓库状态异常, 尝试重新初始化')
-        rmtree(templates_dir)
-        await init_templates()
-        return
-    logger.info('正在更新模板仓库...')
-    proc = await create_subprocess_exec('git', 'pull', stdout=PIPE, stderr=PIPE, cwd=templates_dir)
-    stdout, stderr = await proc.communicate()
-    logger.info(stdout.decode().strip())
-    if proc.returncode != 0:
-        for i in stderr.decode().splitlines():
-            logger.error(i)
-        msg = '更新模板仓库失败'
+    if not await init_templates('latest'):
+        msg = '模板初始化失败'
         raise RuntimeError(msg)
-    logger.success('模板仓库更新成功')
 
 
 @alc.handle()
-async def _():
-    await init_templates()
-    await alc.finish('模板仓库更新成功')
+async def _(revision: str | None = None):
+    if revision is not None and not await check_tag(revision):
+        await alc.finish(f'{revision} 不是模板仓库中的有效标签')
+    logger.info('开始更新模板')
+    if await init_templates(revision or 'latest'):
+        await alc.finish('更新模板成功')
+    await alc.finish('更新模板失败')

nonebot_plugin_tetris_stats/utils/time_it.py

@@ -0,0 +1,21 @@
+from collections.abc import Callable, Coroutine
+from functools import wraps
+from time import time_ns
+from typing import Any, ParamSpec, TypeVar
+
+from nonebot.log import logger
+
+T = TypeVar('T')
+P = ParamSpec('P')
+
+
+def time_it(func: Callable[P, Coroutine[Any, Any, T]]) -> Callable[P, Coroutine[Any, Any, T]]:
+    @wraps(func)
+    async def wrapper(*args: P.args, **kwargs: P.kwargs) -> T:
+        start = time_ns()
+        try:
+            return await func(*args, **kwargs)
+        finally:
+            logger.debug(f'{func.__name__} took {(time_ns() - start) / 1e6}ms')
+
+    return wrapper

nonebot_plugin_tetris_stats/utils/typing.py

@@ -1,10 +1,10 @@
-from collections.abc import Awaitable, Callable
-from typing import Any, Literal
+from typing import Literal
 
 Number = float | int
 GameType = Literal['IO', 'TOP', 'TOS']
-CommandType = Literal['bind', 'query']
-AsyncCallable = Callable[..., Awaitable[Any]]
+BaseCommandType = Literal['bind', 'query']
+TETRIOCommandType = BaseCommandType | Literal['rank', 'config', 'list', 'record']
+AllCommandType = BaseCommandType | TETRIOCommandType
 Me = Literal[
     '我',
     '自己',

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = 'nonebot-plugin-tetris-stats'
-version = '1.3.0'
+version = '1.4.5'
 description = '一款基于 NoneBot2 的用于查询 Tetris 相关游戏数据的插件'
 authors = ['scdhh <wallfjjd@gmail.com>']
 readme = 'README.md'
@@ -11,16 +11,17 @@ license = 'AGPL-3.0'
 [tool.poetry.dependencies]
 python = '^3.10'
 nonebot2 = { extras = ["fastapi"], version = "^2.3.0" }
-nonebot-plugin-alconna = ">=0.40"
-nonebot-plugin-apscheduler = "^0.4.0"
-nonebot-plugin-localstore = "^0.6.0"
+nonebot-plugin-alconna = ">=0.48.0"
+nonebot-plugin-apscheduler = ">=0.4,<0.6"
+nonebot-plugin-localstore = ">=0.6,<0.8"
 nonebot-plugin-orm = ">=0.1.1,<0.8.0"
 nonebot-plugin-session = "^0.3.1"
 nonebot-plugin-session-orm = "^0.2.0"
-nonebot-plugin-user = "^0.2.0"
+nonebot-plugin-user = ">=0.2,<0.4"
 nonebot-plugin-userinfo = "^0.2.4"
 aiocache = "^0.12.2"
-aiofiles = "^23.2.1"
+aiofiles = ">=23.2.1,<25.0.0"
+async-lru = "^2.0.4"
 httpx = "^0.27.0"
 jinja2 = "^3.1.3"
 lxml = '^5.1.0'
@@ -29,12 +30,12 @@ pillow = "^10.3.0"
 playwright = '^1.41.2'
 rich = "^13.7.1"
 ujson = '^5.9.0'
-zstandard = "^0.22.0"
+zstandard = ">=0.22,<0.24"
 
 [tool.poetry.group.dev.dependencies]
 mypy = '>=1.9'
 ruff = '>=0.3.0'
-types-aiofiles = "^23.2.0.20240106"
+types-aiofiles = ">=23.2.0.20240106,<25.0.0.0"
 types-lxml = "^2024.2.9"
 types-pillow = "^10.2.0.20240423"
 types-ujson = '^5.9.0'