🔖 1.3.5

Bumps [nonebot-plugin-apscheduler](https://github.com/nonebot/plugin-apscheduler) from 0.4.0 to 0.5.0.
- [Release notes](https://github.com/nonebot/plugin-apscheduler/releases)
- [Commits](https://github.com/nonebot/plugin-apscheduler/compare/v0.4.0...v0.5.0)

---
updated-dependencies:
- dependency-name: nonebot-plugin-apscheduler
  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: Run mypy type check
+
+on:
+    push:
+
+jobs:
+    release:
+        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.11'
+                  cache: 'poetry'
+
+            - run: poetry install
+              shell: bash
+
+            - name: Run Mypy
+              shell: bash
+              run: |
+                  poetry run mypy ./nonebot_plugin_tetris_stats

.gitignore

@@ -19,3 +19,4 @@ package-lock.json
 bot.py
 TODO
 *.fish
+extracted_skin_mino_*

.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]

nonebot_plugin_tetris_stats/__init__.py

@@ -5,15 +5,22 @@ require('nonebot_plugin_alconna')
 require('nonebot_plugin_apscheduler')
 require('nonebot_plugin_localstore')
 require('nonebot_plugin_orm')
-require('nonebot_plugin_session')
 require('nonebot_plugin_session_orm')
+require('nonebot_plugin_session')
+require('nonebot_plugin_user')
+require('nonebot_plugin_userinfo')
 
-from .config.config import migrations  # noqa: E402
+from nonebot_plugin_alconna import namespace  # noqa: E402
+
+with namespace('tetris_stats') as ns:
+    ns.enable_message_cache = False
+
+from .config import migrations  # noqa: E402
 
 __plugin_meta__ = PluginMetadata(
     name='Tetris Stats',
     description='一个用于查询 Tetris 相关游戏玩家数据的插件',
-    usage='发送 {游戏名} --help 查询使用方法',
+    usage='发送 tstats --help 查询使用方法',
     type='application',
     homepage='https://github.com/A-minos/nonebot-plugin-tetris-stats',
     extra={
@@ -21,5 +28,4 @@ __plugin_meta__ = PluginMetadata(
     },
 )
 
-from . import game_data_processor  # noqa: F401, E402
-from .utils import host  # noqa: F401, E402
+from . import games  # noqa: F401, E402

nonebot_plugin_tetris_stats/config/config.py

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

nonebot_plugin_tetris_stats/config/migrations/8a91210ce14d_correct_the_data_in_historicaldata.py

@@ -46,7 +46,9 @@ def upgrade(name: str = '') -> None:  # noqa: C901
         TimeRemainingColumn,
     )
 
-    from nonebot_plugin_tetris_stats.game_data_processor.schemas import BaseProcessedData  # type: ignore[attr-defined]
+    from nonebot_plugin_tetris_stats.game_data_processor.schemas import (  # type: ignore[import-untyped]
+        BaseProcessedData,
+    )
 
     Base = automap_base()  # noqa: N806
     Base.prepare(autoload_with=op.get_bind())

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/a1195e989cc6_add_tetrio_user_configuration.py

@@ -0,0 +1,44 @@
+"""Add TETRIO user configuration
+
+迁移 ID: a1195e989cc6
+父迁移: b15844837693
+创建时间: 2024-06-09 04:20:07.819194
+
+"""
+
+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 = 'a1195e989cc6'
+down_revision: str | Sequence[str] | None = 'b15844837693'
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade(name: str = '') -> None:
+    if name:
+        return
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_table(
+        'nonebot_plugin_tetris_stats_tetriouserconfig',
+        sa.Column('id', sa.Integer(), nullable=False),
+        sa.Column('query_template', sa.String(length=2), nullable=False),
+        sa.PrimaryKeyConstraint('id', name=op.f('pk_nonebot_plugin_tetris_stats_tetriouserconfig')),
+        info={'bind_key': 'nonebot_plugin_tetris_stats'},
+    )
+    # ### end Alembic commands ###
+
+
+def downgrade(name: str = '') -> None:
+    if name:
+        return
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_table('nonebot_plugin_tetris_stats_tetriouserconfig')
+    # ### end Alembic commands ###

nonebot_plugin_tetris_stats/config/migrations/b15844837693_migrate_to_nonobot_plugin_user.py

@@ -0,0 +1,71 @@
+"""Migrate to nonobot-plugin-user
+
+迁移 ID: b15844837693
+父迁移: 3c25a5a8c050
+创建时间: 2024-06-08 02:27:35.227596
+
+"""
+
+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 = 'b15844837693'
+down_revision: str | Sequence[str] | None = '3c25a5a8c050'
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade(name: str = '') -> None:
+    if name:
+        return
+    # ### commands auto generated by Alembic - please adjust! ###
+    with op.batch_alter_table('nonebot_plugin_tetris_stats_bind', schema=None) as batch_op:
+        batch_op.drop_index('ix_nonebot_plugin_tetris_stats_bind_chat_account')
+        batch_op.drop_index('ix_nonebot_plugin_tetris_stats_bind_chat_platform')
+
+    op.drop_table('nonebot_plugin_tetris_stats_bind')
+
+    op.create_table(
+        'nonebot_plugin_tetris_stats_bind',
+        sa.Column('id', sa.Integer(), nullable=False),
+        sa.Column('user_id', sa.Integer(), nullable=False),
+        sa.Column('game_platform', sa.String(length=32), nullable=False),
+        sa.Column('game_account', sa.String(), nullable=False),
+        sa.PrimaryKeyConstraint('id', name=op.f('pk_nonebot_plugin_tetris_stats_bind')),
+        info={'bind_key': 'nonebot_plugin_tetris_stats'},
+    )
+    with op.batch_alter_table('nonebot_plugin_tetris_stats_bind', schema=None) as batch_op:
+        batch_op.create_index(batch_op.f('ix_nonebot_plugin_tetris_stats_bind_user_id'), ['user_id'], unique=False)
+    # ### end Alembic commands ###
+
+
+def downgrade(name: str = '') -> None:
+    if name:
+        return
+    # ### commands auto generated by Alembic - please adjust! ###
+    with op.batch_alter_table('nonebot_plugin_tetris_stats_bind', schema=None) as batch_op:
+        batch_op.drop_index(batch_op.f('ix_nonebot_plugin_tetris_stats_bind_user_id'))
+
+    op.drop_table('nonebot_plugin_tetris_stats_bind')
+
+    op.create_table(
+        'nonebot_plugin_tetris_stats_bind',
+        sa.Column('id', sa.INTEGER(), nullable=False),
+        sa.Column('chat_platform', sa.VARCHAR(length=32), nullable=False),
+        sa.Column('chat_account', sa.VARCHAR(), nullable=False),
+        sa.Column('game_platform', sa.VARCHAR(length=32), nullable=False),
+        sa.Column('game_account', sa.VARCHAR(), nullable=False),
+        sa.PrimaryKeyConstraint('id', name='pk_nonebot_plugin_tetris_stats_bind'),
+    )
+    with op.batch_alter_table('nonebot_plugin_tetris_stats_bind', schema=None) as batch_op:
+        batch_op.create_index('ix_nonebot_plugin_tetris_stats_bind_chat_platform', ['chat_platform'], unique=False)
+        batch_op.create_index('ix_nonebot_plugin_tetris_stats_bind_chat_account', ['chat_account'], unique=False)
+
+    # ### end Alembic commands ###

nonebot_plugin_tetris_stats/config/migrations/b7fbdafc339a_add_user_unique_identifier_field_to_.py

@@ -46,7 +46,7 @@ def upgrade(name: str = '') -> None:
     from sqlalchemy.ext.automap import automap_base
     from sqlalchemy.orm import Session
 
-    from nonebot_plugin_tetris_stats.game_data_processor.schemas import BaseUser
+    from nonebot_plugin_tetris_stats.game_data_processor.schemas import BaseUser  # type: ignore[import-untyped]
 
     with op.batch_alter_table('nonebot_plugin_tetris_stats_historicaldata', schema=None) as batch_op:
         batch_op.add_column(sa.Column('user_unique_identifier', sa.String(length=32), nullable=True))

nonebot_plugin_tetris_stats/db/__init__.py

@@ -1,3 +1,4 @@
+from asyncio import Lock
 from collections.abc import AsyncGenerator
 from contextlib import asynccontextmanager
 from datetime import datetime, timezone
@@ -7,17 +8,18 @@ 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 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
 
 if TYPE_CHECKING:
-    from ..game_data_processor.io_data_processor.api.models import TETRIOHistoricalData
-    from ..game_data_processor.top_data_processor.api.models import TOPHistoricalData
-    from ..game_data_processor.tos_data_processor.api.models import TOSHistoricalData
+    from ..games.tetrio.api.models import TETRIOHistoricalData
+    from ..games.top.api.models import TOPHistoricalData
+    from ..games.tos.api.models import TOSHistoricalData
 
 
 class BindStatus(Enum):
@@ -27,37 +29,28 @@ class BindStatus(Enum):
 
 async def query_bind_info(
     session: AsyncSession,
-    chat_platform: str,
-    chat_account: str,
+    user: User,
     game_platform: GameType,
 ) -> Bind | None:
     return (
-        await session.scalars(
-            select(Bind)
-            .where(Bind.chat_platform == chat_platform)
-            .where(Bind.chat_account == chat_account)
-            .where(Bind.game_platform == game_platform)
-        )
+        await session.scalars(select(Bind).where(Bind.user_id == user.id).where(Bind.game_platform == game_platform))
     ).one_or_none()
 
 
 async def create_or_update_bind(
     session: AsyncSession,
-    chat_platform: str,
-    chat_account: str,
+    user: User,
     game_platform: GameType,
     game_account: str,
 ) -> BindStatus:
     bind = await query_bind_info(
         session=session,
-        chat_platform=chat_platform,
-        chat_account=chat_account,
+        user=user,
         game_platform=game_platform,
     )
     if bind is None:
         bind = Bind(
-            chat_platform=chat_platform,
-            chat_account=chat_account,
+            user_id=user.id,
             game_platform=game_platform,
             game_account=game_account,
         )
@@ -72,9 +65,11 @@ async def create_or_update_bind(
 
 T = TypeVar('T', 'TETRIOHistoricalData', 'TOPHistoricalData', 'TOSHistoricalData')
 
+lock = Lock()
+
 
 async def anti_duplicate_add(cls: type[T], model: T) -> None:
-    async with get_session() as session:
+    async with lock, get_session() as session:
         result = (
             await session.scalars(
                 select(cls)
@@ -88,8 +83,8 @@ async def anti_duplicate_add(cls: type[T], model: T) -> None:
                 if i.data == model.data:
                     logger.debug('Anti duplicate successfully')
                     return
-            session.add(model)
-            await session.commit()
+        session.add(model)
+        await session.commit()
 
 
 @asynccontextmanager
@@ -97,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'],
+    command_type: TETRIOCommandType,
     command_args: list[str],
 ) -> AsyncGenerator:
     yield
@@ -108,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
@@ -118,15 +113,13 @@ async def trigger(
 async def trigger(
     session_persist_id: int,
     game_platform: GameType,
-    command_type: CommandType | Literal['rank'],
+    command_type: AllCommandType,
     command_args: list[str],
 ) -> AsyncGenerator:
-    logger.debug('running')
     trigger_time = datetime.now(UTC)
     try:
         yield
     except FinishedException:
-        logger.debug('yield')
         async with get_session() as session:
             session.add(
                 TriggerHistoricalData(

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):
@@ -23,9 +23,8 @@ class PydanticType(TypeDecorator):
         *args: Any,
         **kwargs: Any,
     ):
-        for i in get_model:
-            models.update(i())
-        self.models = models
+        self.get_model = get_model
+        self._models = models
         super().__init__(*args, **kwargs)
 
     if PYDANTIC_V2:
@@ -33,7 +32,7 @@ class PydanticType(TypeDecorator):
         @override
         def process_bind_param(self, value: Any | None, dialect: Dialect) -> str:
             # 将 Pydantic 模型实例转换为 JSON
-            if isinstance(value, tuple(self.pydantic_models)):
+            if isinstance(value, tuple(self.models)):
                 return value.model_dump_json(by_alias=True)  # type: ignore[union-attr]
             raise TypeError
     else:
@@ -41,7 +40,7 @@ class PydanticType(TypeDecorator):
         @override
         def process_bind_param(self, value: Any | None, dialect: Dialect) -> str:
             # 将 Pydantic 模型实例转换为 JSON
-            if isinstance(value, tuple(self.pydantic_models)):
+            if isinstance(value, tuple(self.models)):
                 return value.json(by_alias=True)  # type: ignore[union-attr]
             raise TypeError
 
@@ -49,18 +48,25 @@ class PydanticType(TypeDecorator):
     def process_result_value(self, value: Any | None, dialect: Dialect) -> BaseModel:
         # 将 JSON 转换回 Pydantic 模型实例
         if isinstance(value, str | bytes):
-            for i in self.pydantic_models:
+            for i in self.models:
                 try:
                     return type_validate_json(i, value)
                 except ValidationError:  # noqa: PERF203
                     ...
         raise ValueError
 
+    @property
+    def models(self) -> tuple[type[BaseModel], ...]:
+        models: set[type[BaseModel]] = set()
+        for i in self.get_model:
+            models.update(i())
+        models.update(self._models)
+        return tuple(models)
+
 
 class Bind(MappedAsDataclass, Model):
     id: Mapped[int] = mapped_column(init=False, primary_key=True)
-    chat_platform: Mapped[str] = mapped_column(String(32), index=True)
-    chat_account: Mapped[str] = mapped_column(index=True)
+    user_id: Mapped[int] = mapped_column(index=True)
     game_platform: Mapped[GameType] = mapped_column(String(32))
     game_account: Mapped[str]
 
@@ -70,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']] = 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/game_data_processor/__init__.py

@@ -1,44 +0,0 @@
-from typing import Any
-
-from nonebot.adapters import Bot
-from nonebot.exception import FinishedException
-from nonebot.matcher import Matcher
-from nonebot.message import run_postprocessor
-from nonebot_plugin_alconna import AlcMatches, AlconnaMatcher, At
-
-from ..utils.exception import MessageFormatError, NeedCatchError
-
-
-def add_default_handlers(matcher: type[AlconnaMatcher]) -> None:
-    @matcher.assign('query')
-    async def _(bot: Bot, matcher: Matcher, target: At):
-        if isinstance(target, At) and target.target == bot.self_id:
-            await matcher.finish('不能查询bot的信息')
-
-    @matcher.handle()
-    async def _(matcher: Matcher, account: MessageFormatError):
-        await matcher.finish(str(account))
-
-    @matcher.handle()
-    async def _(matcher: Matcher, matches: AlcMatches):
-        if matches.head_matched and matches.options != {} or matches.main_args == {}:
-            await matcher.finish(
-                (f'{matches.error_info!r}\n' if matches.error_info is not None else '')
-                + f'输入"{matches.header_result} --help"查看帮助'
-            )
-
-    @matcher.handle()
-    def _(other: Any):  # noqa: ANN401, ARG001
-        raise FinishedException
-
-
-from . import (  # noqa: F401, E402
-    io_data_processor,
-    top_data_processor,
-    tos_data_processor,
-)
-
-
-@run_postprocessor
-async def _(matcher: Matcher, exception: NeedCatchError):
-    await matcher.send(str(exception))

nonebot_plugin_tetris_stats/game_data_processor/io_data_processor/__init__.py

@@ -1,85 +0,0 @@
-from arclet.alconna import Alconna, AllParam, Arg, ArgFlag, Args, CommandMeta, Option
-from nonebot_plugin_alconna import At, on_alconna
-
-from ...utils.exception import MessageFormatError
-from ...utils.typing import Me
-from .. import add_default_handlers
-from ..constant import BIND_COMMAND, QUERY_COMMAND
-from .api import Player
-from .api.typing import Rank
-from .constant import USER_ID, USER_NAME
-
-
-def get_player(user_id_or_name: str) -> Player | MessageFormatError:
-    if USER_ID.match(user_id_or_name):
-        return Player(user_id=user_id_or_name, trust=True)
-    if USER_NAME.match(user_id_or_name):
-        return Player(user_name=user_id_or_name, trust=True)
-    return MessageFormatError('用户名/ID不合法')
-
-
-alc = on_alconna(
-    Alconna(
-        'io',
-        Option(
-            BIND_COMMAND[0],
-            Args(
-                Arg(
-                    'account',
-                    get_player,
-                    notice='IO 用户名 / ID',
-                    flags=[ArgFlag.HIDDEN],
-                )
-            ),
-            alias=BIND_COMMAND[1:],
-            compact=True,
-            dest='bind',
-            help_text='绑定 IO 账号',
-        ),
-        Option(
-            QUERY_COMMAND[0],
-            Args(
-                Arg(
-                    'target',
-                    At | Me,
-                    notice='@想要查询的人 | 自己',
-                    flags=[ArgFlag.HIDDEN, ArgFlag.OPTIONAL],
-                ),
-                Arg(
-                    'account',
-                    get_player,
-                    notice='IO 用户名 / ID',
-                    flags=[ArgFlag.HIDDEN, ArgFlag.OPTIONAL],
-                ),
-            ),
-            alias=QUERY_COMMAND[1:],
-            compact=True,
-            dest='query',
-            help_text='查询 IO 游戏信息',
-        ),
-        Option(
-            'rank',
-            Args(Arg('rank', Rank, notice='IO 段位')),
-            alias={'Rank', 'RANK', '段位'},
-            compact=True,
-            dest='rank',
-            help_text='查询 IO 段位信息',
-        ),
-        Arg('other', AllParam, flags=[ArgFlag.HIDDEN, ArgFlag.OPTIONAL]),
-        meta=CommandMeta(
-            description='查询 TETR.IO 的信息',
-            example='io绑定scdhh\nio查我\niorankx',
-            compact=True,
-            fuzzy_match=True,
-        ),
-    ),
-    skip_for_unmatch=False,
-    auto_send_output=True,
-    aliases={'IO'},
-)
-
-alc.shortcut('fkosk', {'command': 'io查', 'args': ['我'], 'fuzzy': False, 'humanized': 'An Easter egg!'})
-
-from . import bind, query, rank  # noqa: F401, E402
-
-add_default_handlers(alc)

nonebot_plugin_tetris_stats/game_data_processor/io_data_processor/query.py

@@ -1,296 +0,0 @@
-import contextlib
-from asyncio import gather
-from datetime import datetime, timedelta, timezone
-from hashlib import md5
-from math import ceil, floor
-from urllib.parse import urlunparse
-from zoneinfo import ZoneInfo
-
-from nonebot.adapters import Bot, Event
-from nonebot.matcher import Matcher
-from nonebot_plugin_alconna import At
-from nonebot_plugin_alconna.uniseg import UniMessage
-from nonebot_plugin_orm import get_session
-from nonebot_plugin_session import EventSession  # type: ignore[import-untyped]
-from nonebot_plugin_session_orm import get_session_persist_id  # type: ignore[import-untyped]
-from sqlalchemy import select
-
-from ...db import query_bind_info, trigger
-from ...utils.host import HostPage, get_self_netloc
-from ...utils.platform import get_platform
-from ...utils.render import TETRIOInfo, render
-from ...utils.render.schemas.base import Avatar
-from ...utils.render.schemas.tetrio_info import Data, Radar, Ranking, TetraLeague, TetraLeagueHistory
-from ...utils.render.schemas.tetrio_info import User as TemplateUser
-from ...utils.screenshot import screenshot
-from ...utils.typing import Me
-from ..constant import CANT_VERIFY_MESSAGE
-from . import alc
-from .api import Player, User, UserInfoSuccess
-from .api.models import TETRIOHistoricalData
-from .api.schemas.user_info import NeverPlayedLeague, NeverRatedLeague, RatedLeague
-from .api.schemas.user_records import SoloModeRecord, SoloRecord
-from .constant import GAME_TYPE, TR_MAX, TR_MIN
-
-UTC = timezone.utc
-
-
-@alc.assign('query')
-async def _(bot: Bot, event: Event, matcher: Matcher, target: At | Me, event_session: EventSession):
-    async with trigger(
-        session_persist_id=await get_session_persist_id(event_session),
-        game_platform=GAME_TYPE,
-        command_type='query',
-        command_args=[],
-    ):
-        async with get_session() as session:
-            bind = await query_bind_info(
-                session=session,
-                chat_platform=get_platform(bot),
-                chat_account=(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)
-        user, user_info, user_records = await gather(player.user, player.get_info(), player.get_records())
-        sprint = user_records.data.records.sprint
-        blitz = user_records.data.records.blitz
-        with contextlib.suppress(TypeError):
-            message += UniMessage.image(raw=await make_query_image(user, user_info, sprint.record, blitz.record))
-            await message.finish()
-        message += make_query_text(user_info, sprint, blitz)
-        await message.finish()
-
-
-@alc.assign('query')
-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='query',
-        command_args=[],
-    ):
-        user, user_info, user_records = await gather(account.user, account.get_info(), account.get_records())
-        sprint = user_records.data.records.sprint
-        blitz = user_records.data.records.blitz
-        with contextlib.suppress(TypeError):
-            await UniMessage.image(raw=await make_query_image(user, user_info, sprint.record, blitz.record)).finish()
-        await make_query_text(user_info, sprint, blitz).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)
-    if not historical_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)
-    ]
-
-    # 按照时间排序
-    histories = sorted(histories, key=lambda x: x.record_at)
-    for index, value in enumerate(histories):
-        # 在历史记录里找有没有今天0点后的数据
-        if value.record_at > today:
-            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
-
-
-async def make_query_image(
-    user: User, user_info: UserInfoSuccess, sprint: SoloRecord | None, blitz: SoloRecord | None
-) -> bytes:
-    league = user_info.data.user.league
-    if not isinstance(league, RatedLeague) or league.vs is None:
-        raise TypeError
-    user_name = user_info.data.user.username.upper()
-    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 is not None:
-        duration = timedelta(milliseconds=sprint.endcontext.final_time).total_seconds()
-        sprint_value = f'{duration:.1f}s' if duration < 60 else f'{duration // 60:.0f}m {duration % 60:.1f}s'  # noqa: PLR2004
-    else:
-        sprint_value = 'N/A'
-    blitz_value = f'{blitz.endcontext.score:,}' if blitz is not None else 'N/A'
-    async with HostPage(
-        await render(
-            'tetrio/info',
-            TETRIOInfo(
-                user=TemplateUser(
-                    avatar=f'https://tetr.io/user-content/avatars/{user_info.data.user.id}.jpg?rv={user_info.data.user.avatar_revision}'
-                    if user_info.data.user.avatar_revision is not None
-                    else Avatar(
-                        type='identicon',
-                        hash=md5(user_info.data.user.id.encode()).hexdigest(),  # noqa: S324
-                    ),
-                    name=user_name,
-                    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(urlunparse(('http', get_self_netloc(), f'/host/{page_hash}.html', '', '', '')))
-
-
-def make_query_text(user_info: UserInfoSuccess, sprint: SoloModeRecord, blitz: SoloModeRecord) -> UniMessage:
-    league = user_info.data.user.league
-    user_name = user_info.data.user.username.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)}, 最近十场的数据:'
-        lpm = league.pps * 24
-        message += f"\nL'PM: {round(lpm, 2)} ( {league.pps} pps )"
-        message += f'\nAPM: {league.apm} ( x{round(league.apm/lpm,2)} )'
-        if league.vs is not None:
-            adpm = league.vs * 0.6
-            message += f'\nADPM: {round(adpm,2)} ( x{round(adpm/lpm,2)} ) ( {league.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)

nonebot_plugin_tetris_stats/game_data_processor/tos_data_processor/query.py

@@ -1,170 +0,0 @@
-from asyncio import gather
-from dataclasses import dataclass
-from typing import Literal
-
-from nonebot.adapters import Bot, Event
-from nonebot.matcher import Matcher
-from nonebot_plugin_alconna import At
-from nonebot_plugin_alconna.uniseg import UniMessage
-from nonebot_plugin_orm import get_session
-from nonebot_plugin_session import EventSession  # type: ignore[import-untyped]
-from nonebot_plugin_session_orm import get_session_persist_id  # type: ignore[import-untyped]
-
-from ...db import query_bind_info, trigger
-from ...utils.metrics import TetrisMetricsProWithLPMADPM, get_metrics
-from ...utils.platform import get_platform
-from ...utils.typing import Me
-from ..constant import CANT_VERIFY_MESSAGE
-from . import alc
-from .api import Player
-from .constant import GAME_TYPE
-
-
-def add_special_handlers(
-    teaid_prefix: Literal['onebot-', 'kook-', 'discord-', 'qqguild-'], match_event: type[Event]
-) -> None:
-    @alc.assign('query')
-    async def _(event: Event, target: At | Me, event_session: EventSession):
-        if isinstance(event, match_event):
-            async with trigger(
-                session_persist_id=await get_session_persist_id(event_session),
-                game_platform=GAME_TYPE,
-                command_type='query',
-                command_args=[],
-            ):
-                await (
-                    await make_query_text(
-                        Player(
-                            teaid=f'{teaid_prefix}{target.target}'
-                            if isinstance(target, At)
-                            else f'{teaid_prefix}{event.get_user_id()}',
-                            trust=True,
-                        )
-                    )
-                ).finish()
-
-
-try:
-    from nonebot.adapters.onebot.v11 import MessageEvent as OB11MessageEvent
-
-    add_special_handlers('onebot-', OB11MessageEvent)
-except ImportError:
-    pass
-
-try:
-    from nonebot.adapters.qq.event import GuildMessageEvent as QQGuildMessageEvent
-    from nonebot.adapters.qq.event import QQMessageEvent
-
-    add_special_handlers('qqguild-', QQGuildMessageEvent)
-    add_special_handlers('onebot-', QQMessageEvent)
-except ImportError:
-    pass
-
-try:
-    from nonebot.adapters.kaiheila.event import MessageEvent as KookMessageEvent
-
-    add_special_handlers('kook-', KookMessageEvent)
-except ImportError:
-    pass
-
-try:
-    from nonebot.adapters.discord import MessageEvent as DiscordMessageEvent
-
-    add_special_handlers('discord-', DiscordMessageEvent)
-except ImportError:
-    pass
-
-
-@alc.assign('query')
-async def _(bot: Bot, event: Event, matcher: Matcher, target: At | Me, event_session: EventSession):
-    async with trigger(
-        session_persist_id=await get_session_persist_id(event_session),
-        game_platform=GAME_TYPE,
-        command_type='query',
-        command_args=[],
-    ):
-        async with get_session() as session:
-            bind = await query_bind_info(
-                session=session,
-                chat_platform=get_platform(bot),
-                chat_account=(target.target if isinstance(target, At) else event.get_user_id()),
-                game_platform=GAME_TYPE,
-            )
-        if bind is None:
-            await matcher.finish('未查询到绑定信息')
-        message = CANT_VERIFY_MESSAGE
-        await (message + await make_query_text(Player(teaid=bind.game_account, trust=True))).finish()
-
-
-@alc.assign('query')
-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='query',
-        command_args=[],
-    ):
-        await (await make_query_text(account)).finish()
-
-
-@dataclass
-class GameData:
-    game_num: int
-    metrics: TetrisMetricsProWithLPMADPM
-
-
-async def get_game_data(player: Player, query_num: int = 50) -> GameData | None:
-    """获取游戏数据"""
-    user_profile = await player.get_profile()
-    if user_profile.data == []:
-        return None
-    weighted_total_lpm = weighted_total_apm = weighted_total_adpm = total_time = 0.0
-    num = 0
-    for i in user_profile.data:
-        # 排除单人局和时间为0的游戏
-        # 茶: 不计算没挖掘的局, 即使apm和lpm也如此
-        if i.num_players == 1 or i.time == 0 or i.dig is None:
-            continue
-        # 加权计算
-        time = i.time / 1000
-        lpm = 24 * (i.pieces / time)
-        apm = (i.attack / time) * 60
-        adpm = ((i.attack + i.dig) / time) * 60
-        weighted_total_lpm += lpm * time
-        weighted_total_apm += apm * time
-        weighted_total_adpm += adpm * time
-        total_time += time
-        num += 1
-        if num >= query_num:
-            break
-    if num == 0:
-        return None
-    # TODO: 如果有效局数小于 {查询数} , 并且没有无dig信息的局, 且 user_profile.data 内有{请求数}个局, 则继续往前获取信息
-    metrics = get_metrics(
-        lpm=weighted_total_lpm / total_time, apm=weighted_total_apm / total_time, adpm=weighted_total_adpm / total_time
-    )
-    lpm = weighted_total_lpm / total_time
-    apm = weighted_total_apm / total_time
-    adpm = weighted_total_adpm / total_time
-    return GameData(game_num=num, metrics=metrics)
-
-
-async def make_query_text(player: Player) -> UniMessage:
-    user_info, game_data = await gather(player.get_info(), get_game_data(player))
-    user_data = user_info.data
-    message = f'用户 {user_data.name} ({user_data.teaid}) '
-    if user_data.ranked_games == '0':
-        message += '暂无段位统计数据'
-    else:
-        message += f', 段位分 {round(float(user_data.rating_now),2)}±{round(float(user_data.rd_now),2)} ({round(float(user_data.vol_now),2)}) '
-    if game_data is None:
-        message += ', 暂无游戏数据'
-    else:
-        message += f', 最近 {game_data.game_num} 局数据'
-        message += f"\nL'PM: {game_data.metrics.lpm} ( {game_data.metrics.pps} pps )"
-        message += f'\nAPM: {game_data.metrics.apm} ( x{game_data.metrics.apl} )'
-        message += f'\nADPM: {game_data.metrics.adpm} ( x{game_data.metrics.adpl} ) ( {game_data.metrics.vs}vs )'
-    message += f'\n40L: {float(user_data.pb_sprint)/1000:.2f}s' if user_data.pb_sprint != '2147483647' else ''
-    message += f'\nMarathon: {user_data.pb_marathon}' if user_data.pb_marathon != '0' else ''
-    message += f'\nChallenge: {user_data.pb_challenge}' if user_data.pb_challenge != '0' else ''
-    return UniMessage(message)

nonebot_plugin_tetris_stats/games/__init__.py

@@ -0,0 +1,55 @@
+from collections.abc import Callable
+
+from nonebot.adapters import Bot
+from nonebot.matcher import Matcher
+from nonebot.message import run_postprocessor
+from nonebot.typing import T_Handler
+from nonebot_plugin_alconna import AlcMatches, Alconna, At, CommandMeta, on_alconna
+
+from .. import ns
+from ..utils.exception import MessageFormatError, NeedCatchError
+
+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,
+)
+
+
+def add_block_handlers(handler: Callable[[T_Handler], T_Handler]) -> None:
+    @handler
+    async def _(bot: Bot, matcher: Matcher, target: At):
+        if isinstance(target, At) and target.target == bot.self_id:
+            await matcher.finish('不能查询bot的信息')
+
+
+from . import tetrio, top, tos  # noqa: F401, E402
+
+
+@alc.handle()
+async def _(matcher: Matcher, account: MessageFormatError):
+    await matcher.finish(str(account))
+
+
+@alc.handle()
+async def _(matcher: Matcher, matches: AlcMatches):
+    if matches.head_matched and matches.options != {} or matches.main_args == {}:
+        await matcher.finish(
+            (f'{matches.error_info!r}\n' if matches.error_info is not None else '')
+            + f'输入"{matches.header_result} --help"查看帮助'
+        )
+
+
+@run_postprocessor
+async def _(matcher: Matcher, exception: NeedCatchError):
+    await matcher.send(str(exception))

nonebot_plugin_tetris_stats/games/tetrio/__init__.py

@@ -0,0 +1,175 @@
+from arclet.alconna import Arg, ArgFlag, Args, Option, Subcommand
+from nonebot_plugin_alconna import At
+
+from ...utils.exception import MessageFormatError
+from ...utils.typing import Me
+from .. import add_block_handlers, alc, 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:
+    if USER_ID.match(user_id_or_name):
+        return Player(user_id=user_id_or_name, trust=True)
+    if USER_NAME.match(user_id_or_name):
+        return Player(user_name=user_id_or_name, trust=True)
+    return MessageFormatError('用户名/ID不合法')
+
+
+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(
+            'list',
+            Option('--max-tr', Arg('max_tr', float), help_text='TR的上限'),
+            Option('--min-tr', Arg('min_tr', float), help_text='TR的下限'),
+            Option('--limit', Arg('limit', int), help_text='查询数量'),
+            Option('--country', Arg('country', str), help_text='国家代码'),
+            help_text='查询 TETR.IO 段位排行榜',
+        ),
+        Subcommand(
+            'rank',
+            Args(Arg('rank', ValidRank, notice='TETR.IO 段位')),
+            help_text='查询 TETR.IO 段位信息',
+        ),
+        Subcommand(
+            'config',
+            Option(
+                '--default-template',
+                Arg('template', Template),
+                alias=['-DT', 'DefaultTemplate'],
+            ),
+        ),
+        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!',
+    },
+)
+
+add_block_handlers(alc.assign('TETRIO.query'))
+
+from . import bind, config, list, query, rank, record  # noqa: E402
+
+__all__ = [
+    'bind',
+    'config',
+    'list',
+    'query',
+    'rank',
+    'record',
+]

nonebot_plugin_tetris_stats/games/tetrio/api/player.py

@@ -11,7 +11,7 @@ from .models import TETRIOHistoricalData
 from .schemas.base import FailedModel
 from .schemas.user import User
 from .schemas.user_info import UserInfo, UserInfoSuccess
-from .schemas.user_records import UserRecords, UserRecordsSuccess
+from .schemas.user_records import SoloModeRecord, UserRecords, UserRecordsSuccess, Zen
 
 
 class Player:
@@ -43,7 +43,7 @@ class Player:
         if self.user_id is not None:
             return self.user_id
         if self.user_name is not None:
-            return self.user_name
+            return self.user_name.lower()
         msg = 'Invalid user'
         raise ValueError(msg)
 
@@ -100,3 +100,15 @@ class Player:
                 ),
             )
         return self._user_records
+
+    @property
+    async def sprint(self) -> SoloModeRecord:
+        return (await self.get_records()).data.records.sprint
+
+    @property
+    async def blitz(self) -> SoloModeRecord:
+        return (await self.get_records()).data.records.blitz
+
+    @property
+    async def zen(self) -> Zen:
+        return (await self.get_records()).data.zen

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_records.py

@@ -2,6 +2,7 @@ from datetime import datetime
 
 from pydantic import BaseModel, Field
 
+from .....utils.typing import Number
 from .base import FailedModel
 from .base import SuccessModel as BaseSuccessModel
 
@@ -35,7 +36,7 @@ class Garbage(BaseModel):
     sent: int
     received: int
     attack: int | None = None
-    cleared: int
+    cleared: int | None = None
 
 
 class Finesse(BaseModel):
@@ -45,7 +46,7 @@ class Finesse(BaseModel):
 
 
 class EndContext(BaseModel):
-    seed: int
+    seed: Number
     lines: int
     level_lines: int
     level_lines_needed: int
@@ -95,7 +96,7 @@ class MultiRecord(_Record):
 
 
 class SoloModeRecord(BaseModel):
-    record: SoloRecord
+    record: SoloRecord | None = None
     rank: int | None = None
 
 

nonebot_plugin_tetris_stats/games/tetrio/api/tetra_league.py

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

nonebot_plugin_tetris_stats/games/tetrio/api/typing.py

@@ -1,6 +1,6 @@
 from typing import Literal
 
-Rank = Literal[
+ValidRank = Literal[
     'x',
     'u',
     'ss',
@@ -18,5 +18,6 @@ Rank = Literal[
     'c-',
     'd+',
     'd',
-    'z',  # 未定级
 ]
+
+Rank = ValidRank | Literal['z']  # 未定级

nonebot_plugin_tetris_stats/games/tetrio/bind.py

@@ -1,17 +1,17 @@
+from asyncio import gather
 from hashlib import md5
-from urllib.parse import urlunparse
+from urllib.parse import urlencode
 
-from nonebot.adapters import Bot, Event
 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_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 ...db import BindStatus, create_or_update_bind, trigger
-from ...utils.avatar import get_avatar
 from ...utils.host import HostPage, get_self_netloc
-from ...utils.platform import get_platform
+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
@@ -20,34 +20,34 @@ from .api import Player
 from .constant import GAME_TYPE
 
 
-@alc.assign('bind')
-async def _(bot: Bot, event: Event, account: Player, event_session: EventSession, bot_info: UserInfo = BotUserInfo()):  # noqa: B008
+@alc.assign('TETRIO.bind')
+async def _(nb_user: User, account: Player, event_session: EventSession, bot_info: UserInfo = BotUserInfo()):  # noqa: B008
     async with trigger(
         session_persist_id=await get_session_persist_id(event_session),
         game_platform=GAME_TYPE,
         command_type='bind',
         command_args=[],
     ):
-        user = await account.user
+        user, user_info = await gather(account.user, account.get_info())
         async with get_session() as session:
             bind_status = await create_or_update_bind(
                 session=session,
-                chat_platform=get_platform(bot),
-                chat_account=event.get_user_id(),
+                user=nb_user,
                 game_platform=GAME_TYPE,
                 game_account=user.unique_identifier,
             )
-        user_info = await account.get_info()
         if bind_status in (BindStatus.SUCCESS, BindStatus.UPDATE):
+            netloc = get_self_netloc()
             async with HostPage(
                 await render(
-                    'binding',
+                    'v1/binding',
                     Bind(
                         platform='TETR.IO',
                         status='unknown',
                         user=People(
-                            avatar=f'https://tetr.io/user-content/avatars/{user_info.data.user.id}.jpg?rv={user_info.data.user.avatar_revision}'
+                            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(),
                         ),
@@ -59,6 +59,4 @@ async def _(bot: Bot, event: Event, account: Player, event_session: EventSession
                     ),
                 )
             ) as page_hash:
-                await UniMessage.image(
-                    raw=await screenshot(urlunparse(('http', get_self_netloc(), f'/host/{page_hash}.html', '', '', '')))
-                ).finish()
+                await UniMessage.image(raw=await screenshot(f'http://{netloc}/host/{page_hash}.html')).finish()

nonebot_plugin_tetris_stats/games/tetrio/config.py

@@ -0,0 +1,30 @@
+from nonebot_plugin_alconna.uniseg import UniMessage
+from nonebot_plugin_orm import async_scoped_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 nonebot_plugin_user import User  # type: ignore[import-untyped]
+from sqlalchemy import select
+
+from ...db import trigger
+from . import alc
+from .constant import GAME_TYPE
+from .models import TETRIOUserConfig
+from .typing import Template
+
+
+@alc.assign('TETRIO.config')
+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/list.py

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

nonebot_plugin_tetris_stats/games/tetrio/models.py

@@ -5,6 +5,7 @@ from sqlalchemy import JSON, DateTime, String
 from sqlalchemy.orm import Mapped, MappedAsDataclass, mapped_column
 
 from .api.typing import Rank
+from .typing import Template
 
 
 class IORank(MappedAsDataclass, Model):
@@ -26,3 +27,8 @@ class IORank(MappedAsDataclass, Model):
         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

@@ -0,0 +1,554 @@
+from asyncio import gather
+from collections import defaultdict
+from datetime import date, datetime, timedelta, timezone
+from hashlib import md5
+from math import ceil, floor
+from typing import ClassVar, TypeVar, overload
+from urllib.parse import urlencode
+from zoneinfo import ZoneInfo
+
+from aiofiles import open
+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.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_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 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.tetrio_info import Info as V1TemplateInfo
+from ...utils.render.schemas.tetrio.tetrio_info import Radar, TetraLeague, TetraLeagueHistory, TetraLeagueHistoryData
+from ...utils.render.schemas.tetrio.tetrio_info import User as V1TemplateUser
+from ...utils.render.schemas.tetrio.tetrio_user_info_v2 import Badge, Blitz, Sprint, Statistic, TetraLeagueStatistic
+from ...utils.render.schemas.tetrio.tetrio_user_info_v2 import Info as V2TemplateInfo
+from ...utils.render.schemas.tetrio.tetrio_user_info_v2 import TetraLeague as V2TemplateTetraLeague
+from ...utils.render.schemas.tetrio.tetrio_user_info_v2 import User as V2TemplateUser
+from ...utils.screenshot import screenshot
+from ...utils.typing import Me, Number
+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 .typing import Template
+
+UTC = timezone.utc
+
+driver = get_driver()
+
+
+@alc.assign('TETRIO.query')
+async def _(  # noqa: PLR0913
+    user: NBUser,
+    event: Event,
+    matcher: Matcher,
+    target: At | Me,
+    event_session: EventSession,
+    template: Template | None = None,
+):
+    async with trigger(
+        session_persist_id=await get_session_persist_id(event_session),
+        game_platform=GAME_TYPE,
+        command_type='query',
+        command_args=[f'--default-template {template}'] if template is not None else [],
+    ):
+        async with get_session() as session:
+            bind = await query_bind_info(
+                session=session,
+                user=await get_user(
+                    event_session.platform, target.target if isinstance(target, At) else event.get_user_id()
+                ),
+                game_platform=GAME_TYPE,
+            )
+            if template is None:
+                template = await session.scalar(
+                    select(TETRIOUserConfig.query_template).where(TETRIOUserConfig.id == user.id)
+                )
+        if bind is None:
+            await matcher.finish('未查询到绑定信息')
+        message = UniMessage(CANT_VERIFY_MESSAGE)
+        player = Player(user_id=bind.game_account, trust=True)
+        await (message + (await make_query_result(player, template or 'v1'))).finish()
+
+
+@alc.assign('TETRIO.query')
+async def _(user: NBUser, account: Player, event_session: EventSession, template: Template | None = None):
+    async with trigger(
+        session_persist_id=await get_session_persist_id(event_session),
+        game_platform=GAME_TYPE,
+        command_type='query',
+        command_args=[f'--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: TetraLeagueHistoryData,
+    behind_point: TetraLeagueHistoryData,
+    point_time: datetime,
+) -> TetraLeagueHistoryData:
+    """根据给出的 previous_point 和 behind_point, 推算 point_time 点处的数据
+
+    Args:
+        previous_point (Data): 前面的数据点
+        behind_point (Data): 后面的数据点
+        point_time (datetime): 要推算的点的位置
+
+    Returns:
+        Data: 要推算的点的数据
+    """
+    # 求两个点的斜率
+    slope = (behind_point.tr - previous_point.tr) / (
+        datetime.timestamp(behind_point.record_at) - datetime.timestamp(previous_point.record_at)
+    )
+    return TetraLeagueHistoryData(
+        record_at=point_time,
+        tr=previous_point.tr + slope * (datetime.timestamp(point_time) - datetime.timestamp(previous_point.record_at)),
+    )
+
+
+async def query_historical_data(user: User, user_info: UserInfoSuccess) -> list[TetraLeagueHistoryData]:
+    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 [
+            TetraLeagueHistoryData(record_at=today - forward, tr=user_info.data.user.league.rating),
+            TetraLeagueHistoryData(record_at=today.replace(microsecond=1000), tr=user_info.data.user.league.rating),
+        ]
+    histories = [
+        TetraLeagueHistoryData(
+            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],
+                TetraLeagueHistoryData(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, TetraLeagueHistoryData(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')
+
+
+N = TypeVar('N', int, float)
+
+
+def handling_special_value(value: N) -> N | None:
+    return value if value != -1 else None
+
+
+async def make_query_image_v2(player: Player) -> bytes:
+    user, user_info, sprint, blitz, zen = await gather(
+        player.user, player.get_info(), player.sprint, player.blitz, player.zen
+    )
+    league = get_league(user_info)
+    histories = await query_historical_data(user, user_info)
+
+    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'
+
+    play_time: str | None
+    if (game_time := handling_special_value(user_info.data.user.gametime)) is not None:
+        if game_time // 3600 > 0:
+            play_time = f'{game_time//3600:.0f}h {game_time % 3600 // 60:.0f}m {game_time % 60:.0f}s'
+        elif game_time // 60 > 0:
+            play_time = f'{game_time//60:.0f}m {game_time % 60:.0f}s'
+        else:
+            play_time = f'{game_time:.0f}s'
+    else:
+        play_time = game_time
+    netloc = get_self_netloc()
+    async with HostPage(
+        await render(
+            'v2/tetrio/user/info',
+            V2TemplateInfo(
+                user=V2TemplateUser(
+                    id=user.ID,
+                    name=user.name.upper(),
+                    bio=user_info.data.user.bio,
+                    banner=f'http://{netloc}/host/resource/tetrio/banners/{user.ID}?{urlencode({"revision": user_info.data.user.banner_revision})}'
+                    if user_info.data.user.banner_revision is not None and user_info.data.user.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
+                    else Avatar(
+                        type='identicon',
+                        hash=md5(user_info.data.user.id.encode()).hexdigest(),  # noqa: S324
+                    ),
+                    badges=[
+                        Badge(
+                            id=i.id,
+                            description=i.label,
+                            group=i.group,
+                            receive_at=i.ts if isinstance(i.ts, datetime) else None,
+                        )
+                        for i in user_info.data.user.badges
+                    ],
+                    country=user_info.data.user.country,
+                    role=user_info.data.user.role,
+                    xp=user_info.data.user.xp,
+                    friend_count=user_info.data.user.friend_count,
+                    supporter_tier=user_info.data.user.supporter_tier,
+                    bad_standing=user_info.data.user.badstanding or False,
+                    verified=user_info.data.user.verified,
+                    playtime=play_time,
+                    join_at=user_info.data.user.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,
+                    history=histories,
+                )
+                if isinstance(league, RatedLeague)
+                else None,
+                statistic=Statistic(
+                    total=handling_special_value(user_info.data.user.gamesplayed),
+                    wins=handling_special_value(user_info.data.user.gameswon),
+                ),
+                sprint=Sprint(
+                    time=sprint_value,
+                    global_rank=sprint.rank,
+                    play_at=sprint.record.ts,
+                )
+                if sprint.record is not None
+                else None,
+                blitz=Blitz(
+                    score=blitz.record.endcontext.score,
+                    global_rank=blitz.rank,
+                    play_at=blitz.record.ts,
+                )
+                if blitz.record is not None
+                else None,
+                zen=zen,
+            ),
+        ),
+    ) 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[TetraLeagueHistoryData]:
+        return [TetraLeagueHistoryData(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

@@ -30,14 +30,14 @@ 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 .model import IORank
+from .models import IORank
 
 UTC = timezone.utc
 
 driver = get_driver()
 
 
-@alc.assign('rank')
+@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),

nonebot_plugin_tetris_stats/games/tetrio/record/__init__.py

@@ -0,0 +1,6 @@
+from . import blitz, sprint
+
+__all__ = [
+    'blitz',
+    'sprint',
+]

nonebot_plugin_tetris_stats/games/tetrio/record/blitz.py

@@ -0,0 +1,137 @@
+from asyncio import gather
+from datetime import timedelta
+from hashlib import md5
+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.uniseg import UniMessage
+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 nonebot_plugin_user import get_user  # type: ignore[import-untyped]
+
+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.tetrio_record_base import Finesse, Max, Mini, Tspins, User
+from ....utils.render.schemas.tetrio.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
+
+
+@alc.assign('TETRIO.record.blitz')
+async def _(
+    event: Event,
+    matcher: Matcher,
+    target: At | Me,
+    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'],
+    ):
+        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, 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()} 的 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)
+    netloc = get_self_netloc()
+    async with HostPage(
+        page=await render(
+            'v2/tetrio/record/blitz',
+            Record(
+                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
+                    else Avatar(
+                        type='identicon',
+                        hash=md5(user.ID.encode()).hexdigest(),  # noqa: S324
+                    ),
+                ),
+                replay_id=blitz.record.replayid,
+                rank=blitz.rank,
+                statistic=Statistic(
+                    keys=endcontext.inputs,
+                    kpp=round(endcontext.inputs / endcontext.piecesplaced, 2),
+                    kps=round(endcontext.inputs / duration, 2),
+                    max=Max(
+                        combo=max((0, endcontext.topcombo - 1)),
+                        btb=max((0, endcontext.topbtb - 1)),
+                    ),
+                    pieces=endcontext.piecesplaced,
+                    pps=metrics.pps,
+                    lines=endcontext.lines,
+                    lpm=metrics.lpm,
+                    holds=endcontext.holds,
+                    score=endcontext.score,
+                    spp=round(endcontext.score / endcontext.piecesplaced, 2),
+                    single=clears.singles,
+                    double=clears.doubles,
+                    triple=clears.triples,
+                    quad=clears.quads,
+                    tspins=Tspins(
+                        total=clears.realtspins,
+                        single=clears.tspinsingles,
+                        double=clears.tspindoubles,
+                        triple=clears.triples,
+                        mini=Mini(
+                            total=clears.minitspins,
+                            single=clears.minitspinsingles,
+                            double=clears.minitspindoubles,
+                        ),
+                    ),
+                    all_clear=clears.allclear,
+                    finesse=Finesse(
+                        faults=endcontext.finesse.faults,
+                        accuracy=round(endcontext.finesse.perfectpieces / endcontext.piecesplaced * 100, 2),
+                    ),
+                    level=endcontext.level,
+                ),
+                play_at=blitz.record.ts,
+            ),
+        )
+    ) as page_hash:
+        return await screenshot(f'http://{netloc}/host/{page_hash}.html')

nonebot_plugin_tetris_stats/games/tetrio/record/sprint.py

@@ -0,0 +1,137 @@
+from asyncio import gather
+from datetime import timedelta
+from hashlib import md5
+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.uniseg import UniMessage
+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 nonebot_plugin_user import get_user  # type: ignore[import-untyped]
+
+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.tetrio_record_base import Finesse, Max, Mini, Tspins, User
+from ....utils.render.schemas.tetrio.tetrio_record_sprint 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
+
+
+@alc.assign('TETRIO.record.sprint')
+async def _(
+    event: Event,
+    matcher: Matcher,
+    target: At | Me,
+    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'],
+    ):
+        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, 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:
+        msg = f'未找到用户 {user.name.upper()} 的 40L 记录'
+        raise RecordNotFoundError(msg)
+    endcontext = sprint.record.endcontext
+    clears = endcontext.clears
+    duration = timedelta(milliseconds=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
+    metrics = get_metrics(pps=endcontext.piecesplaced / duration)
+    netloc = get_self_netloc()
+    async with HostPage(
+        page=await render(
+            'v2/tetrio/record/40l',
+            Record(
+                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
+                    else Avatar(
+                        type='identicon',
+                        hash=md5(user.ID.encode()).hexdigest(),  # noqa: S324
+                    ),
+                ),
+                time=sprint_value,
+                replay_id=sprint.record.replayid,
+                rank=sprint.rank,
+                statistic=Statistic(
+                    keys=endcontext.inputs,
+                    kpp=round(endcontext.inputs / endcontext.piecesplaced, 2),
+                    kps=round(endcontext.inputs / duration, 2),
+                    max=Max(
+                        combo=max((0, endcontext.topcombo - 1)),
+                        btb=max((0, endcontext.topbtb - 1)),
+                    ),
+                    pieces=endcontext.piecesplaced,
+                    pps=metrics.pps,
+                    lines=endcontext.lines,
+                    lpm=metrics.lpm,
+                    holds=endcontext.holds,
+                    score=endcontext.score,
+                    single=clears.singles,
+                    double=clears.doubles,
+                    triple=clears.triples,
+                    quad=clears.quads,
+                    tspins=Tspins(
+                        total=clears.realtspins,
+                        single=clears.tspinsingles,
+                        double=clears.tspindoubles,
+                        triple=clears.triples,
+                        mini=Mini(
+                            total=clears.minitspins,
+                            single=clears.minitspinsingles,
+                            double=clears.minitspindoubles,
+                        ),
+                    ),
+                    all_clear=clears.allclear,
+                    finesse=Finesse(
+                        faults=endcontext.finesse.faults,
+                        accuracy=round(endcontext.finesse.perfectpieces / endcontext.piecesplaced * 100, 2),
+                    ),
+                ),
+                play_at=sprint.record.ts,
+            ),
+        )
+    ) as page_hash:
+        return await screenshot(f'http://{netloc}/host/{page_hash}.html')

nonebot_plugin_tetris_stats/games/tetrio/typing.py

@@ -0,0 +1,3 @@
+from typing import Literal
+
+Template = Literal['v1', 'v2']

nonebot_plugin_tetris_stats/games/top/__init__.py

@@ -1,10 +1,9 @@
-from arclet.alconna import Alconna, AllParam, Arg, ArgFlag, Args, CommandMeta, Option
-from nonebot_plugin_alconna import At, on_alconna
+from arclet.alconna import Arg, ArgFlag, Args, Subcommand
+from nonebot_plugin_alconna import At
 
 from ...utils.exception import MessageFormatError
 from ...utils.typing import Me
-from .. import add_default_handlers
-from ..constant import BIND_COMMAND, QUERY_COMMAND
+from .. import add_block_handlers, alc, command
 from .api import Player
 from .constant import USER_NAME
 
@@ -15,31 +14,28 @@ def get_player(name: str) -> Player | MessageFormatError:
     return MessageFormatError('用户名/ID不合法')
 
 
-alc = on_alconna(
-    Alconna(
-        'top',
-        Option(
-            BIND_COMMAND[0],
+command.add(
+    Subcommand(
+        'TOP',
+        Subcommand(
+            'bind',
             Args(
                 Arg(
                     'account',
                     get_player,
-                    notice='TOP 用户名',
+                    notice='TOP 用户名 / ID',
                     flags=[ArgFlag.HIDDEN],
                 )
             ),
-            alias=BIND_COMMAND[1:],
-            compact=True,
-            dest='bind',
             help_text='绑定 TOP 账号',
         ),
-        Option(
-            QUERY_COMMAND[0],
+        Subcommand(
+            'query',
             Args(
                 Arg(
                     'target',
                     At | Me,
-                    notice='@想要查询的人 | 自己',
+                    notice='@想要查询的人 / 自己',
                     flags=[ArgFlag.HIDDEN, ArgFlag.OPTIONAL],
                 ),
                 Arg(
@@ -49,24 +45,15 @@ alc = on_alconna(
                     flags=[ArgFlag.HIDDEN, ArgFlag.OPTIONAL],
                 ),
             ),
-            alias=QUERY_COMMAND[1:],
-            compact=True,
-            dest='query',
             help_text='查询 TOP 游戏信息',
         ),
-        Arg('other', AllParam, flags=[ArgFlag.HIDDEN, ArgFlag.OPTIONAL]),
-        meta=CommandMeta(
-            description='查询 TetrisOnline波兰服 的信息',
-            example='top绑定scdhh\ntop查我',
-            compact=True,
-            fuzzy_match=True,
-        ),
-    ),
-    skip_for_unmatch=False,
-    auto_send_output=True,
-    aliases={'TOP'},
+        help_text='TOP 游戏相关指令',
+    )
 )
 
-from . import bind, query  # noqa: E402, F401
+alc.shortcut('(?i:top)(?i:绑定|绑|bind)', {'command': 'tstats TOP bind', 'humanized': 'top绑定'})
+alc.shortcut('(?i:top)(?i:查询|查|query|stats)', {'command': 'tstats TOP query', 'humanized': 'top查'})
 
-add_default_handlers(alc)
+add_block_handlers(alc.assign('TOP.query'))
+
+from . import bind, query  # noqa: E402, F401

nonebot_plugin_tetris_stats/games/top/bind.py

@@ -1,16 +1,13 @@
-from urllib.parse import urlunparse
-
-from nonebot.adapters import Bot
 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_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 ...db import BindStatus, create_or_update_bind, trigger
-from ...utils.avatar import get_avatar
 from ...utils.host import HostPage, get_self_netloc
-from ...utils.platform import get_platform
+from ...utils.image import get_avatar
 from ...utils.render import Bind, render
 from ...utils.render.schemas.base import People
 from ...utils.screenshot import screenshot
@@ -19,13 +16,13 @@ from .api import Player
 from .constant import GAME_TYPE
 
 
-@alc.assign('bind')
+@alc.assign('TOP.bind')
 async def _(
-    bot: Bot,
+    nb_user: User,
     account: Player,
     event_session: EventSession,
-    bot_info: UserInfo = BotUserInfo(),  # noqa: B008
     event_user_info: UserInfo = EventUserInfo(),  # noqa: B008
+    bot_info: UserInfo = BotUserInfo(),  # noqa: B008
 ):
     async with trigger(
         session_persist_id=await get_session_persist_id(event_session),
@@ -37,15 +34,14 @@ async def _(
         async with get_session() as session:
             bind_status = await create_or_update_bind(
                 session=session,
-                chat_platform=get_platform(bot),
-                chat_account=event_user_info.user_id,
+                user=nb_user,
                 game_platform=GAME_TYPE,
                 game_account=user.unique_identifier,
             )
         if bind_status in (BindStatus.SUCCESS, BindStatus.UPDATE):
             async with HostPage(
                 await render(
-                    'binding',
+                    'v1/binding',
                     Bind(
                         platform=GAME_TYPE,
                         status='unknown',
@@ -62,5 +58,5 @@ async def _(
                 )
             ) as page_hash:
                 await UniMessage.image(
-                    raw=await screenshot(urlunparse(('http', get_self_netloc(), f'/host/{page_hash}.html', '', '', '')))
+                    raw=await screenshot(f'http://{get_self_netloc()}/host/{page_hash}.html')
                 ).finish()

nonebot_plugin_tetris_stats/games/top/query.py

@@ -1,14 +1,14 @@
-from nonebot.adapters import Bot, Event
+from nonebot.adapters import Event
 from nonebot.matcher import Matcher
 from nonebot_plugin_alconna import At
 from nonebot_plugin_alconna.uniseg import UniMessage
 from nonebot_plugin_orm import get_session
 from nonebot_plugin_session import EventSession  # type: ignore[import-untyped]
 from nonebot_plugin_session_orm import get_session_persist_id  # type: ignore[import-untyped]
+from nonebot_plugin_user import get_user  # type: ignore[import-untyped]
 
 from ...db import query_bind_info, trigger
 from ...utils.metrics import get_metrics
-from ...utils.platform import get_platform
 from ...utils.typing import Me
 from ..constant import CANT_VERIFY_MESSAGE
 from . import alc
@@ -17,8 +17,8 @@ from .api.schemas.user_profile import UserProfile
 from .constant import GAME_TYPE
 
 
-@alc.assign('query')
-async def _(bot: Bot, event: Event, matcher: Matcher, target: At | Me, event_session: EventSession):
+@alc.assign('TOP.query')
+async def _(event: Event, matcher: Matcher, target: At | Me, event_session: EventSession):
     async with trigger(
         session_persist_id=await get_session_persist_id(event_session),
         game_platform=GAME_TYPE,
@@ -28,8 +28,9 @@ async def _(bot: Bot, event: Event, matcher: Matcher, target: At | Me, event_ses
         async with get_session() as session:
             bind = await query_bind_info(
                 session=session,
-                chat_platform=get_platform(bot),
-                chat_account=(target.target if isinstance(target, At) else event.get_user_id()),
+                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:
@@ -38,7 +39,7 @@ async def _(bot: Bot, event: Event, matcher: Matcher, target: At | Me, event_ses
         await (message + make_query_text(await Player(user_name=bind.game_account, trust=True).get_profile())).finish()
 
 
-@alc.assign('query')
+@alc.assign('TOP.query')
 async def _(account: Player, event_session: EventSession):
     async with trigger(
         session_persist_id=await get_session_persist_id(event_session),

nonebot_plugin_tetris_stats/games/tos/__init__.py

@@ -1,10 +1,9 @@
-from arclet.alconna import Alconna, AllParam, Arg, ArgFlag, Args, CommandMeta, Option
-from nonebot_plugin_alconna import At, on_alconna
+from arclet.alconna import Arg, ArgFlag, Args, Subcommand
+from nonebot_plugin_alconna import At
 
 from ...utils.exception import MessageFormatError
 from ...utils.typing import Me
-from .. import add_default_handlers
-from ..constant import BIND_COMMAND, QUERY_COMMAND
+from .. import add_block_handlers, alc, command
 from .api import Player
 from .constant import USER_NAME
 
@@ -20,31 +19,28 @@ def get_player(teaid_or_name: str) -> Player | MessageFormatError:
     return MessageFormatError('用户名/ID不合法')
 
 
-alc = on_alconna(
-    Alconna(
-        '茶服',
-        Option(
-            BIND_COMMAND[0],
+command.add(
+    Subcommand(
+        'TOS',
+        Subcommand(
+            'bind',
             Args(
                 Arg(
                     'account',
                     get_player,
-                    notice='茶服 用户名 / TeaID',
+                    notice='茶服 用户名 / ID',
                     flags=[ArgFlag.HIDDEN],
                 )
             ),
-            alias=BIND_COMMAND[1:],
-            compact=True,
-            dest='bind',
             help_text='绑定 茶服 账号',
         ),
-        Option(
-            QUERY_COMMAND[0],
+        Subcommand(
+            'query',
             Args(
                 Arg(
                     'target',
                     At | Me,
-                    notice='@想要查询的人 | 自己',
+                    notice='@想要查询的人 / 自己',
                     flags=[ArgFlag.HIDDEN, ArgFlag.OPTIONAL],
                 ),
                 Arg(
@@ -53,27 +49,16 @@ alc = on_alconna(
                     notice='茶服 用户名 / TeaID',
                     flags=[ArgFlag.HIDDEN, ArgFlag.OPTIONAL],
                 ),
-                # 如果放在一个 Union Args 里, 验证顺序不能保证, 可能出错
             ),
-            alias=QUERY_COMMAND[1:],
-            compact=True,
-            dest='query',
             help_text='查询 茶服 游戏信息',
         ),
-        Arg('other', AllParam, flags=[ArgFlag.HIDDEN, ArgFlag.OPTIONAL]),
-        meta=CommandMeta(
-            description='查询 TetrisOnline茶服 的信息',
-            example='茶服查我',
-            compact=True,
-            fuzzy_match=True,
-        ),
-    ),
-    skip_for_unmatch=False,
-    auto_send_output=True,
-    aliases={'tos', 'TOS'},
+        help_text='茶服 游戏相关指令',
+    )
 )
 
+alc.shortcut('(?i:tos|茶服)(?i:绑定|绑|bind)', {'command': 'tstats TOS bind', 'humanized': '茶服绑定'})
+alc.shortcut('(?i:tos|茶服)(?i:查询|查|query|stats)', {'command': 'tstats TOS query', 'humanized': '茶服查'})
+
+add_block_handlers(alc.assign('TOS.query'))
 
 from . import bind, query  # noqa: E402, F401
-
-add_default_handlers(alc)

nonebot_plugin_tetris_stats/games/tos/bind.py

@@ -1,16 +1,13 @@
-from urllib.parse import urlunparse
-
-from nonebot.adapters import Bot
 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_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 ...db import BindStatus, create_or_update_bind, trigger
-from ...utils.avatar import get_avatar
 from ...utils.host import HostPage, get_self_netloc
-from ...utils.platform import get_platform
+from ...utils.image import get_avatar
 from ...utils.render import Bind, render
 from ...utils.render.schemas.base import People
 from ...utils.screenshot import screenshot
@@ -19,13 +16,13 @@ from .api import Player
 from .constant import GAME_TYPE
 
 
-@alc.assign('bind')
+@alc.assign('TOS.bind')
 async def _(
-    bot: Bot,
+    nb_user: User,
     account: Player,
     event_session: EventSession,
-    bot_info: UserInfo = BotUserInfo(),  # noqa: B008
     event_user_info: UserInfo = EventUserInfo(),  # noqa: B008
+    bot_info: UserInfo = BotUserInfo(),  # noqa: B008
 ):
     async with trigger(
         session_persist_id=await get_session_persist_id(event_session),
@@ -37,8 +34,7 @@ async def _(
         async with get_session() as session:
             bind_status = await create_or_update_bind(
                 session=session,
-                chat_platform=get_platform(bot),
-                chat_account=event_user_info.user_id,
+                user=nb_user,
                 game_platform=GAME_TYPE,
                 game_account=user.unique_identifier,
             )
@@ -46,7 +42,7 @@ async def _(
         if bind_status in (BindStatus.SUCCESS, BindStatus.UPDATE):
             async with HostPage(
                 await render(
-                    'binding',
+                    'v1/binding',
                     Bind(
                         platform=GAME_TYPE,
                         status='unknown',
@@ -62,5 +58,5 @@ async def _(
                 )
             ) as page_hash:
                 await UniMessage.image(
-                    raw=await screenshot(urlunparse(('http', get_self_netloc(), f'/host/{page_hash}.html', '', '', '')))
+                    raw=await screenshot(f'http://{get_self_netloc()}/host/{page_hash}.html')
                 ).finish()

nonebot_plugin_tetris_stats/games/tos/query.py

@@ -0,0 +1,252 @@
+from asyncio import gather
+from datetime import timedelta
+from http import HTTPStatus
+from typing import Literal, NamedTuple
+
+from nonebot.adapters import Event
+from nonebot.matcher import Matcher
+from nonebot_plugin_alconna import At
+from nonebot_plugin_alconna.uniseg import UniMessage
+from nonebot_plugin_orm import get_session
+from nonebot_plugin_session import EventSession  # type: ignore[import-untyped]
+from nonebot_plugin_session_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 ...db import query_bind_info, trigger
+from ...utils.exception import RequestError
+from ...utils.host import HostPage, get_self_netloc
+from ...utils.image import get_avatar
+from ...utils.metrics import TetrisMetricsProWithLPMADPM, get_metrics
+from ...utils.render import render
+from ...utils.render.schemas.base import People, Ranking
+from ...utils.render.schemas.tos_info import Info, Multiplayer, Radar
+from ...utils.screenshot import screenshot
+from ...utils.typing import Me, Number
+from ..constant import CANT_VERIFY_MESSAGE
+from . import alc
+from .api import Player
+from .api.schemas.user_info import UserInfoSuccess
+from .constant import GAME_TYPE
+
+
+def add_special_handlers(
+    teaid_prefix: Literal['onebot-', 'kook-', 'discord-', 'qqguild-'], match_event: type[Event]
+) -> None:
+    @alc.assign('TOS.query')
+    async def _(
+        event: Event,
+        target: At | Me,
+        event_session: EventSession,
+        event_user_info: UserInfo = EventUserInfo(),  # noqa: B008
+    ):
+        if isinstance(event, match_event):
+            async with trigger(
+                session_persist_id=await get_session_persist_id(event_session),
+                game_platform=GAME_TYPE,
+                command_type='query',
+                command_args=[],
+            ):
+                player = Player(
+                    teaid=f'{teaid_prefix}{target.target}'
+                    if isinstance(target, At)
+                    else f'{teaid_prefix}{event.get_user_id()}',
+                    trust=True,
+                )
+                try:
+                    user_info, game_data = await gather(player.get_info(), get_game_data(player))
+                    if game_data is not None:
+                        await UniMessage.image(
+                            raw=await make_query_image(user_info, game_data, event_user_info)
+                        ).finish()
+                    await make_query_text(user_info, game_data).finish()
+                except RequestError as e:
+                    if e.status_code == HTTPStatus.BAD_REQUEST and '未找到此用户' in e.message:
+                        return
+                    raise
+
+
+try:
+    from nonebot.adapters.onebot.v11 import MessageEvent as OB11MessageEvent
+
+    add_special_handlers('onebot-', OB11MessageEvent)
+except ImportError:
+    pass
+
+try:
+    from nonebot.adapters.qq.event import GuildMessageEvent as QQGuildMessageEvent
+    from nonebot.adapters.qq.event import QQMessageEvent
+
+    add_special_handlers('qqguild-', QQGuildMessageEvent)
+    add_special_handlers('onebot-', QQMessageEvent)
+except ImportError:
+    pass
+
+try:
+    from nonebot.adapters.kaiheila.event import MessageEvent as KookMessageEvent
+
+    add_special_handlers('kook-', KookMessageEvent)
+except ImportError:
+    pass
+
+try:
+    from nonebot.adapters.discord import MessageEvent as DiscordMessageEvent
+
+    add_special_handlers('discord-', DiscordMessageEvent)
+except ImportError:
+    pass
+
+
+@alc.assign('TOS.query')
+async def _(
+    event: Event,
+    matcher: Matcher,
+    target: At | Me,
+    event_session: EventSession,
+    event_user_info: UserInfo = EventUserInfo(),  # noqa: B008
+):
+    async with trigger(
+        session_persist_id=await get_session_persist_id(event_session),
+        game_platform=GAME_TYPE,
+        command_type='query',
+        command_args=[],
+    ):
+        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 = CANT_VERIFY_MESSAGE
+        player = Player(teaid=bind.game_account, trust=True)
+        user_info, game_data = await gather(player.get_info(), get_game_data(player))
+        if game_data is not None:
+            await (
+                message + UniMessage.image(raw=await make_query_image(user_info, game_data, event_user_info))
+            ).finish()
+        await (message + make_query_text(user_info, game_data)).finish()
+
+
+@alc.assign('TOS.query')
+async def _(account: Player, event_session: EventSession, event_user_info: UserInfo = EventUserInfo()):  # noqa: B008
+    async with trigger(
+        session_persist_id=await get_session_persist_id(event_session),
+        game_platform=GAME_TYPE,
+        command_type='query',
+        command_args=[],
+    ):
+        user_info, game_data = await gather(account.get_info(), get_game_data(account))
+        if game_data is not None:
+            await UniMessage.image(raw=await make_query_image(user_info, game_data, event_user_info)).finish()
+        await make_query_text(user_info, game_data).finish()
+
+
+class GameData(NamedTuple):
+    game_num: int
+    metrics: TetrisMetricsProWithLPMADPM
+    OR: Number
+    dspp: Number
+    ge: Number
+
+
+async def get_game_data(player: Player, query_num: int = 50) -> GameData | None:
+    """获取游戏数据"""
+    user_profile = await player.get_profile()
+    if user_profile.data == []:
+        return None
+    weighted_total_lpm = weighted_total_apm = weighted_total_adpm = total_time = 0.0
+    total_attack = total_dig = total_offset = total_pieses = total_receive = num = 0
+    for i in user_profile.data:
+        # 排除单人局和时间为0的游戏
+        # 茶: 不计算没挖掘的局, 即使apm和lpm也如此
+        if i.num_players == 1 or i.time == 0 or i.dig is None:
+            continue
+        # 加权计算
+        time = i.time / 1000
+        lpm = 24 * (i.pieces / time)
+        apm = (i.attack / time) * 60
+        adpm = ((i.attack + i.dig) / time) * 60
+        weighted_total_lpm += lpm * time
+        weighted_total_apm += apm * time
+        weighted_total_adpm += adpm * time
+        total_attack += i.attack
+        total_dig += i.dig
+        total_offset += i.offset
+        total_pieses += i.pieces
+        total_receive += i.receive
+        total_time += time
+        num += 1
+        if num >= query_num:
+            break
+    if num == 0:
+        return None
+    # TODO: 如果有效局数小于 {查询数} , 并且没有无dig信息的局, 且 user_profile.data 内有{请求数}个局, 则继续往前获取信息
+    metrics = get_metrics(
+        lpm=weighted_total_lpm / total_time, apm=weighted_total_apm / total_time, adpm=weighted_total_adpm / total_time
+    )
+    return GameData(
+        game_num=num,
+        metrics=metrics,
+        OR=total_offset / total_receive * 100,
+        dspp=total_dig / total_pieses,
+        ge=2 * ((total_attack * total_dig) / total_pieses**2),
+    )
+
+
+async def make_query_image(user_info: UserInfoSuccess, game_data: GameData, event_user_info: UserInfo) -> bytes:
+    metrics = game_data.metrics
+    duration = timedelta(milliseconds=float(user_info.data.pb_sprint)).total_seconds()
+    sprint_value = f'{duration:.3f}s' if duration < 60 else f'{duration // 60:.0f}m {duration % 60:.3f}s'  # noqa: PLR2004
+    async with HostPage(
+        await render(
+            'v1/tos/info',
+            Info(
+                user=People(avatar=await get_avatar(event_user_info, 'Data URI', None), name=user_info.data.name),
+                ranking=Ranking(rating=float(user_info.data.ranking), rd=round(float(user_info.data.rd_now), 2)),
+                multiplayer=Multiplayer(
+                    pps=metrics.pps,
+                    lpm=metrics.lpm,
+                    apm=metrics.apm,
+                    apl=metrics.apl,
+                    vs=metrics.vs,
+                    adpm=metrics.adpm,
+                    adpl=metrics.adpl,
+                ),
+                radar=Radar(
+                    app=(app := (metrics.apm / (60 * metrics.pps))),
+                    OR=game_data.OR,
+                    dspp=game_data.dspp,
+                    ci=150 * game_data.dspp - 125 * app + 50 * (metrics.vs / metrics.apm) - 25,
+                    ge=game_data.ge,
+                ),
+                sprint=sprint_value,
+                challenge=f'{int(user_info.data.pb_challenge):,}' if user_info.data.pb_challenge != '0' else 'N/A',
+                marathon=f'{int(user_info.data.pb_marathon):,}' if user_info.data.pb_marathon != '0' else 'N/A',
+            ),
+        )
+    ) as page_hash:
+        return await screenshot(f'http://{get_self_netloc()}/host/{page_hash}.html')
+
+
+def make_query_text(user_info: UserInfoSuccess, game_data: GameData | None) -> UniMessage:
+    user_data = user_info.data
+    message = f'用户 {user_data.name} ({user_data.teaid}) '
+    if user_data.ranked_games == '0':
+        message += '暂无段位统计数据'
+    else:
+        message += f', 段位分 {round(float(user_data.rating_now),2)}±{round(float(user_data.rd_now),2)} ({round(float(user_data.vol_now),2)}) '
+    if game_data is None:
+        message += ', 暂无游戏数据'
+    else:
+        message += f', 最近 {game_data.game_num} 局数据'
+        message += f"\nL'PM: {game_data.metrics.lpm} ( {game_data.metrics.pps} pps )"
+        message += f'\nAPM: {game_data.metrics.apm} ( x{game_data.metrics.apl} )'
+        message += f'\nADPM: {game_data.metrics.adpm} ( x{game_data.metrics.adpl} ) ( {game_data.metrics.vs}vs )'
+    message += f'\n40L: {float(user_data.pb_sprint)/1000:.2f}s' if user_data.pb_sprint != '2147483647' else ''
+    message += f'\nMarathon: {user_data.pb_marathon}' if user_data.pb_marathon != '0' else ''
+    message += f'\nChallenge: {user_data.pb_challenge}' if user_data.pb_challenge != '0' else ''
+    return UniMessage(message)

nonebot_plugin_tetris_stats/utils/exception.py

@@ -27,6 +27,14 @@ class MessageFormatError(NeedCatchError):
     """用户发送的消息格式不正确"""
 
 
+class RecordNotFoundError(NeedCatchError):
+    """找不到用户的某种记录"""
+
+
+class FallbackError(NeedCatchError):
+    """需要回滚至更通用的方法"""
+
+
 class DoNotCatchError(TetrisStatsError):
     """不应该被捕获的异常基类"""
 
@@ -35,5 +43,5 @@ class WhatTheFuckError(DoNotCatchError):
     """用于表示不应该出现的情况 ("""
 
 
-class HandleNotFinishedError(DoNotCatchError):
-    """任务没有正常完成处理的错误"""
+class NoFallbackError(DoNotCatchError):  # 暂时没用 但是先写了
+    """没有可用的回退方法"""

nonebot_plugin_tetris_stats/utils/host.py

@@ -1,26 +1,27 @@
 from hashlib import sha256
 from ipaddress import IPv4Address, IPv6Address
-from typing import TYPE_CHECKING, ClassVar
+from typing import TYPE_CHECKING, ClassVar, Literal
 
-from fastapi import FastAPI, status
-from fastapi.responses import HTMLResponse
+from fastapi import FastAPI, Path, status
+from fastapi.responses import FileResponse, HTMLResponse, Response
 from fastapi.staticfiles import StaticFiles
 from nonebot import get_app, get_driver
 from nonebot.log import logger
-from nonebot_plugin_localstore import get_cache_dir  # type: ignore[import-untyped]
 
+from ..config.config import CACHE_PATH
+from .image import img_to_png
+from .request import Request
 from .templates import templates_dir
 
 if TYPE_CHECKING:
     from pydantic import IPvAnyAddress
 
-app = get_app()
+app: FastAPI = get_app()
 
 driver = get_driver()
 
 global_config = driver.config
 
-cache_dir = get_cache_dir('nonebot_plugin_tetris_stats')
 
 if not isinstance(app, FastAPI):
     msg = '本插件需要 FastAPI 驱动器才能运行'
@@ -60,6 +61,22 @@ async def _(page_hash: str) -> HTMLResponse:
     return NOT_FOUND
 
 
+@app.get('/host/resource/tetrio/{resource_type}/{user_id}', status_code=status.HTTP_200_OK)
+async def _(
+    resource_type: Literal['avatars', 'banners'], revision: int, user_id: str = Path(regex=r'^[a-f0-9]{24}$')
+) -> Response:
+    if not (path := CACHE_PATH / 'tetrio' / resource_type / f'{user_id}_{revision}.png').exists():
+        path.parent.mkdir(parents=True, exist_ok=True)
+        path.write_bytes(
+            img_to_png(
+                await Request.request(
+                    f'https://tetr.io/user-content/{resource_type}/{user_id}.jpg?rv={revision}', is_json=False
+                )
+            )
+        )
+    return FileResponse(path)
+
+
 def get_self_netloc() -> str:
     host: IPv4Address | IPv6Address | IPvAnyAddress = global_config.host
     if isinstance(host, IPv4Address):

nonebot_plugin_tetris_stats/utils/image.py

@@ -52,3 +52,11 @@ async def get_avatar(user: UserInfo, scheme: Literal['Data URI', 'bytes'], defau
             raise TypeError(msg)
         return f'data:{Image.MIME[avatar_format]};base64,{b64encode(bot_avatar).decode()}'
     return bot_avatar
+
+
+def img_to_png(image: bytes) -> bytes:
+    """将图片转换为 PNG 格式"""
+    result = BytesIO()
+    with Image.open(BytesIO(image)) as img:
+        img.save(result, 'PNG')
+    return result.getvalue()

nonebot_plugin_tetris_stats/utils/platform.py

@@ -1,19 +0,0 @@
-from nonebot.adapters import Bot
-
-
-def get_platform(bot: Bot) -> str:
-    try:
-        from nonebot.adapters.onebot.v12 import Bot as OB12Bot
-
-        if isinstance(bot, OB12Bot):
-            return bot.platform
-    except ImportError:
-        pass
-    try:
-        from nonebot.adapters.satori import Bot as SaBot
-
-        if isinstance(bot, SaBot):
-            return bot.platform
-    except ImportError:
-        pass
-    return bot.type

nonebot_plugin_tetris_stats/utils/render/__init__.py

@@ -5,7 +5,13 @@ from nonebot.compat import PYDANTIC_V2
 
 from ..templates import templates_dir
 from .schemas.bind import Bind
-from .schemas.tetrio_info import TETRIOInfo
+from .schemas.tetrio.tetrio_info import Info as TETRIOInfo
+from .schemas.tetrio.tetrio_record_blitz import Record as TETRIORecordBlitz
+from .schemas.tetrio.tetrio_record_sprint import Record as TETRIORecordSprint
+from .schemas.tetrio.tetrio_user_info_v2 import Info as TETRIOUserInfoV2
+from .schemas.tetrio.tetrio_user_list_v2 import List as TETRIOUserListV2
+from .schemas.top_info import Info as TOPInfo
+from .schemas.tos_info import Info as TOSInfo
 
 env = Environment(
     loader=FileSystemLoader(templates_dir), autoescape=True, trim_blocks=True, lstrip_blocks=True, enable_async=True
@@ -13,17 +19,62 @@ env = Environment(
 
 
 @overload
-async def render(render_type: Literal['binding'], data: Bind) -> str: ...
+async def render(render_type: Literal['v1/binding'], data: Bind) -> str: ...
 
 
 @overload
-async def render(render_type: Literal['tetrio/info'], data: TETRIOInfo) -> str: ...
+async def render(render_type: Literal['v1/tetrio/info'], data: TETRIOInfo) -> str: ...
 
 
-async def render(render_type: Literal['binding', 'tetrio/info'], data: Bind | TETRIOInfo) -> 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/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: ...
+
+
+async def render(
+    render_type: Literal[
+        'v1/binding',
+        'v1/tetrio/info',
+        'v1/top/info',
+        'v1/tos/info',
+        'v2/tetrio/user/info',
+        'v2/tetrio/user/list',
+        'v2/tetrio/record/40l',
+        'v2/tetrio/record/blitz',
+    ],
+    data: Bind
+    | TETRIOInfo
+    | TOPInfo
+    | TOSInfo
+    | TETRIOUserInfoV2
+    | TETRIOUserListV2
+    | TETRIORecordSprint
+    | TETRIORecordBlitz,
+) -> str:
     if PYDANTIC_V2:
-        return await env.get_template('index.html').render_async(path=render_type, data=data.model_dump_json())
-    return await env.get_template('index.html').render_async(path=render_type, data=data.json())
+        return await env.get_template('index.html').render_async(
+            path=render_type, data=data.model_dump_json(by_alias=True)
+        )
+    return await env.get_template('index.html').render_async(path=render_type, data=data.json(by_alias=True))
 
 
-__all__ = ['render', 'Bind', 'TETRIOInfo']
+__all__ = ['render']

nonebot_plugin_tetris_stats/utils/render/schemas/base.py

@@ -2,6 +2,8 @@ from typing import Literal
 
 from pydantic import BaseModel
 
+from ...typing import Number
+
 
 class Avatar(BaseModel):
     type: Literal['identicon']
@@ -11,3 +13,8 @@ class Avatar(BaseModel):
 class People(BaseModel):
     avatar: str | Avatar
     name: str
+
+
+class Ranking(BaseModel):
+    rating: Number
+    rd: Number

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

@@ -0,0 +1,58 @@
+from pydantic import BaseModel
+
+from ..base import People
+
+
+class User(People):
+    id: str
+
+
+class Max(BaseModel):
+    combo: int
+    btb: int
+
+
+class Mini(BaseModel):
+    total: int
+    single: int
+    double: int
+
+
+class Tspins(BaseModel):
+    total: int
+    single: int
+    double: int
+    triple: int
+
+    mini: Mini
+
+
+class Finesse(BaseModel):
+    faults: int
+    accuracy: float
+
+
+class RecordStatistic(BaseModel):
+    keys: int
+    kpp: float
+    kps: float
+
+    max: Max
+
+    pieces: int
+    pps: float
+    lines: int
+    lpm: float
+    holds: int | None
+    score: int
+
+    single: int
+    double: int
+    triple: int
+    quad: int
+
+    tspins: Tspins
+
+    all_clear: int
+
+    finesse: Finesse

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

@@ -0,0 +1,22 @@
+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/tetrio_record_sprint.py

@@ -0,0 +1,18 @@
+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/render/schemas/tetrio/tetrio_user_info_v2.py

@@ -0,0 +1,100 @@
+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 .base import TetraLeagueHistoryData
+
+
+class Badge(BaseModel):
+    id: str
+    description: str
+    group: str | None
+    receive_at: datetime | None
+
+
+class User(BaseModel):
+    id: str
+    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 | None
+    supporter_tier: int
+
+    verified: bool
+    bad_standing: bool
+
+    badges: list[Badge]
+    xp: Number
+
+    playtime: str | None
+    join_at: datetime | None
+
+
+class Statistic(BaseModel):
+    total: int | None
+    wins: int | None
+
+
+class TetraLeagueStatistic(BaseModel):
+    total: int
+    wins: int
+
+
+class TetraLeague(BaseModel):
+    rank: Rank
+    highest_rank: Rank
+
+    tr: Number
+
+    glicko: Number
+    rd: Number
+
+    global_rank: int | None
+    country_rank: int | None
+
+    pps: Number
+
+    apm: Number
+    apl: Number
+
+    vs: Number | None
+    adpl: Number | None
+
+    statistic: TetraLeagueStatistic
+
+    decaying: bool
+
+    history: list[TetraLeagueHistoryData]
+
+
+class Sprint(BaseModel):
+    time: str
+    global_rank: int | None
+    play_at: datetime
+
+
+class Blitz(BaseModel):
+    score: int
+    global_rank: int | None
+    play_at: datetime
+
+
+class Info(BaseModel):
+    user: User
+    tetra_league: TetraLeague | None
+    statistic: Statistic | None
+    sprint: Sprint | None
+    blitz: Blitz | None
+    zen: Zen

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

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

nonebot_plugin_tetris_stats/utils/render/schemas/tetrio_info.py

@@ -1,77 +0,0 @@
-from datetime import datetime
-from typing import Annotated, ClassVar
-
-from nonebot.compat import PYDANTIC_V2
-from pydantic import BaseModel
-
-from ....game_data_processor.io_data_processor.api.typing import Rank
-from ...typing import Number
-from .base import People
-
-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 Ranking(BaseModel):
-    rating: Number
-    rd: Number
-
-
-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 TETRIOInfo(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/top_info.py

@@ -0,0 +1,17 @@
+from pydantic import BaseModel
+
+from ...typing import Number
+from .base import People
+
+
+class Data(BaseModel):
+    pps: Number
+    lpm: Number
+    apm: Number
+    apl: Number
+
+
+class Info(BaseModel):
+    user: People
+    today: Data
+    history: Data

nonebot_plugin_tetris_stats/utils/render/schemas/tos_info.py

@@ -0,0 +1,32 @@
+from pydantic import BaseModel, Field
+
+from ...typing import Number
+from .base import People, Ranking
+
+
+class Multiplayer(BaseModel):
+    pps: Number
+    lpm: Number
+    apm: Number
+    apl: Number
+    vs: Number
+    adpm: Number
+    adpl: Number
+
+
+class Radar(BaseModel):
+    app: Number
+    OR: Number = Field(serialization_alias='or')
+    dspp: Number
+    ci: Number
+    ge: Number
+
+
+class Info(BaseModel):
+    user: People
+    ranking: Ranking
+    multiplayer: Multiplayer
+    radar: Radar
+    sprint: str
+    challenge: str
+    marathon: str

nonebot_plugin_tetris_stats/utils/retry.py

@@ -1,38 +1,44 @@
 from asyncio import sleep
-from collections.abc import Awaitable, Callable
+from collections.abc import Callable, Coroutine
+from contextlib import suppress
 from datetime import timedelta
 from functools import wraps
-from typing import TypeVar, cast
+from typing import Any, ParamSpec, TypeVar
 
 from nonebot.log import logger
+from nonebot_plugin_alconna.uniseg import SerializeFailed, UniMessage
 
 T = TypeVar('T')
+P = ParamSpec('P')
 
 
 def retry(
     max_attempts: int = 3,
     exception_type: type[BaseException] | tuple[type[BaseException], ...] = Exception,
     delay: timedelta | None = None,
-) -> Callable[[Callable[..., Awaitable[T]]], Callable[..., Awaitable[T]]]:
-    def decorator(func: Callable[..., Awaitable[T]]) -> Callable[..., Awaitable[T]]:
+    reply: str | UniMessage | None = None,
+) -> Callable[[Callable[P, Coroutine[Any, Any, T]]], Callable[P, Coroutine[Any, Any, T]]]:
+    def decorator(func: Callable[P, Coroutine[Any, Any, T]]) -> Callable[P, Coroutine[Any, Any, T]]:
         @wraps(func)
-        async def wrapper(*args, **kwargs) -> T:  # noqa: ANN002, ANN003
-            attempts = 0
-            while attempts < max_attempts + 1:
+        async def wrapper(*args: P.args, **kwargs: P.kwargs) -> T:
+            for i in range(max_attempts + 1):
+                if i > 0:
+                    message = f'Retrying: {func.__name__} ({i}/{max_attempts})'
+                    logger.debug(message)
+                    with suppress(SerializeFailed):
+                        await UniMessage(reply or message).send()
+                if i == max_attempts:
+                    break
                 try:
                     return await func(*args, **kwargs)
-                except exception_type as e:  # noqa: PERF203
+                except exception_type as e:
+                    if i == max_attempts:
+                        raise
                     logger.exception(e)
-                    attempts += 1
-                    if attempts <= max_attempts:
-                        if delay is not None:
-                            await sleep(delay.total_seconds())
-                        logger.debug(f'Retrying: {func.__name__} ({attempts}/{max_attempts})')
-                        continue
-                    raise
-            msg = 'Unexpectedly reached the end of the retry loop'
-            raise RuntimeError(msg)
+                    if delay is not None:
+                        await sleep(delay.total_seconds())
+            return await func(*args, **kwargs)
 
-        return cast(Callable[..., Awaitable[T]], wrapper)
+        return wrapper
 
     return decorator

nonebot_plugin_tetris_stats/utils/screenshot.py

@@ -1,11 +1,29 @@
+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
+
+config = get_plugin_config(Config)
 
 
+@retry(exception_type=TimeoutError, reply='截图失败, 重试中')
 async def screenshot(url: str) -> bytes:
     browser = await BrowserManager.get_browser()
     async with (
-        await browser.new_page(no_viewport=True, viewport={'width': 0, 'height': 0}) 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')
-        return await page.screenshot(full_page=True, type='png')
+        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/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.2.0'
+version = '1.3.5'
 description = '一款基于 NoneBot2 的用于查询 Tetris 相关游戏数据的插件'
 authors = ['scdhh <wallfjjd@gmail.com>']
 readme = 'README.md'
@@ -11,15 +11,16 @@ 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.4"
 nonebot-plugin-userinfo = "^0.2.4"
 aiocache = "^0.12.2"
-aiofiles = "^23.2.1"
+aiofiles = ">=23.2.1,<25.0.0"
 httpx = "^0.27.0"
 jinja2 = "^3.1.3"
 lxml = '^5.1.0'
@@ -33,19 +34,21 @@ zstandard = "^0.22.0"
 [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'
 pandas-stubs = '>=1.5.2,<3.0.0'
-nonebot-plugin-orm = { extras = ["default"], version = ">=0.3,<0.8" }
+nonebot2 = { extras = ["all"], version = "^2.3.0" }
 nonebot-adapter-discord = "^0.1.3"
 nonebot-adapter-kaiheila = "^0.3.4"
 nonebot-adapter-onebot = "^2.4.1"
 nonebot-adapter-qq = "^1.4.4"
-nonebot-adapter-satori = "^0.11.4"
+nonebot-adapter-satori = ">=0.11.4,<0.13.0"
+nonebot-plugin-orm = { extras = ["default"], version = ">=0.3,<0.8" }
 
 [tool.poetry.group.debug.dependencies]
+memory-profiler = "^0.61.0"
 objprint = '^0.2.2'
 viztracer = "^0.16.2"