🔖 1.4.0

Bumps [ruff](https://github.com/astral-sh/ruff) from 0.5.1 to 0.5.2.
- [Release notes](https://github.com/astral-sh/ruff/releases)
- [Changelog](https://github.com/astral-sh/ruff/blob/main/CHANGELOG.md)
- [Commits](https://github.com/astral-sh/ruff/compare/0.5.1...0.5.2)

---
updated-dependencies:
- dependency-name: ruff
  dependency-type: direct:development
  update-type: version-update:semver-patch
...

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

.github/workflows/TypeCheck.yml

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

.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

@@ -8,13 +8,14 @@ require('nonebot_plugin_orm')
 require('nonebot_plugin_session_orm')
 require('nonebot_plugin_session')
 require('nonebot_plugin_user')
+require('nonebot_plugin_userinfo')
 
 from nonebot_plugin_alconna import namespace  # noqa: E402
 
 with namespace('tetris_stats') as ns:
     ns.enable_message_cache = False
 
-from .config.config import migrations  # noqa: E402
+from .config import migrations  # noqa: E402
 
 __plugin_meta__ = PluginMetadata(
     name='Tetris Stats',

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/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/db/__init__.py

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

nonebot_plugin_tetris_stats/db/models.py

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

nonebot_plugin_tetris_stats/games/__init__.py

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

nonebot_plugin_tetris_stats/games/tetrio/__init__.py

@@ -3,7 +3,7 @@ from nonebot_plugin_alconna import At
 
 from ...utils.exception import MessageFormatError
 from ...utils.typing import Me
-from .. import add_block_handlers, alc
+from .. import add_block_handlers, alc, command
 from .api import Player
 from .api.typing import ValidRank
 from .constant import USER_ID, USER_NAME
@@ -18,7 +18,7 @@ def get_player(user_id_or_name: str) -> Player | MessageFormatError:
     return MessageFormatError('用户名/ID不合法')
 
 
-alc.command.add(
+command.add(
     Subcommand(
         'TETR.IO',
         Subcommand(
@@ -57,23 +57,124 @@ alc.command.add(
             ),
             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 段位')),
+            Option(
+                '--all',
+                dest='all',
+            ),
+            Option(
+                '--detail',
+                Arg('rank', ValidRank),
+                alias=['-D'],
+            ),
             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查'})
+
+def rank_wrapper(slot: int | str, content: str | None):
+    if slot == 'rank' and not content:
+        return '--all'
+    if content is not None:
+        return f'--detail {content.lower()}'
+    return content
+
 
 alc.shortcut(
-    'fkosk', {'command': 'tstats TETR.IO query', 'args': ['我'], 'fuzzy': False, 'humanized': 'An Easter egg!'}
+    '(?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(
+    r'(?i:io)(?i:段位|段|rank)\s*(?P<rank>[a-zA-Z+-]{0,2})',
+    command='tstats TETR.IO rank {rank}',
+    humanized='iorank',
+    fuzzy=False,
+    wrapper=rank_wrapper,
+)
+alc.shortcut(
+    '(?i:io)(?i:配置|配|config)',
+    command='tstats TETR.IO config',
+    humanized='io配置',
+)
+
+alc.shortcut(
+    'fkosk',
+    command='tstats TETR.IO query',
+    arguments=['我'],
+    fuzzy=False,
+    humanized='An Easter egg!',
 )
 
 add_block_handlers(alc.assign('TETRIO.query'))
 
-from . import bind, query, rank  # noqa: F401, E402
+from . import bind, config, list, query, rank, record  # noqa: E402
+
+__all__ = [
+    'bind',
+    'config',
+    'list',
+    'query',
+    'rank',
+    'record',
+]

nonebot_plugin_tetris_stats/games/tetrio/api/models.py

@@ -15,8 +15,3 @@ class TETRIOHistoricalData(MappedAsDataclass, Model):
     api_type: Mapped[Literal['User Info', 'User Records']] = mapped_column(String(16), index=True)
     data: Mapped[SuccessModel] = mapped_column(PydanticType(get_model=[SuccessModel.__subclasses__], models=set()))
     update_time: Mapped[datetime] = mapped_column(DateTime, index=True)
-
-
-class TETRIOUserConfig(MappedAsDataclass, Model):
-    id: Mapped[int] = mapped_column(init=False, primary_key=True)
-    query_template: Mapped[Literal['v1', 'v2']] = mapped_column(String(2))

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:
@@ -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/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/bind.py

@@ -47,6 +47,7 @@ async def _(nb_user: User, account: Player, event_session: EventSession, bot_inf
                         user=People(
                             avatar=f'http://{netloc}/host/resource/tetrio/avatars/{user.ID}?{urlencode({"revision": user_info.data.user.avatar_revision})}'
                             if user_info.data.user.avatar_revision is not None
+                            and user_info.data.user.avatar_revision != 0
                             else Avatar(type='identicon', hash=md5(user_info.data.user.id.encode()).hexdigest()),  # noqa: S324
                             name=user_info.data.user.username.upper(),
                         ),

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

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

nonebot_plugin_tetris_stats/games/tetrio/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

@@ -4,12 +4,13 @@ from nonebot_plugin_orm import Model
 from sqlalchemy import JSON, DateTime, String
 from sqlalchemy.orm import Mapped, MappedAsDataclass, mapped_column
 
-from .api.typing import Rank
+from .api.typing import ValidRank
+from .typing import Template
 
 
 class IORank(MappedAsDataclass, Model):
     id: Mapped[int] = mapped_column(init=False, primary_key=True)
-    rank: Mapped[Rank] = mapped_column(String(2), index=True)
+    rank: Mapped[ValidRank] = mapped_column(String(2), index=True)
     tr_line: Mapped[float]
     player_count: Mapped[int]
     low_pps: Mapped[tuple[dict[str, str], float]] = mapped_column(JSON)
@@ -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

@@ -10,7 +10,7 @@ from zoneinfo import ZoneInfo
 from aiofiles import open
 from nonebot import get_driver
 from nonebot.adapters import Event
-from nonebot.compat import model_dump, type_validate_json
+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
@@ -19,6 +19,7 @@ from nonebot_plugin_localstore import get_data_file  # type: ignore[import-untyp
 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
@@ -29,13 +30,13 @@ from ...utils.host import HostPage, get_self_netloc
 from ...utils.metrics import TetrisMetricsProWithPPSVS, get_metrics
 from ...utils.render import render
 from ...utils.render.schemas.base import Avatar, Ranking
-from ...utils.render.schemas.tetrio_info import Data, Radar, TetraLeague, TetraLeagueHistory
-from ...utils.render.schemas.tetrio_info import Info as V1TemplateInfo
-from ...utils.render.schemas.tetrio_info import User as V1TemplateUser
-from ...utils.render.schemas.tetrio_info_v2 import Badge, Blitz, Sprint, Statistic, TetraLeagueStatistic, Zen
-from ...utils.render.schemas.tetrio_info_v2 import Info as V2TemplateInfo
-from ...utils.render.schemas.tetrio_info_v2 import TetraLeague as V2TemplateTetraLeague
-from ...utils.render.schemas.tetrio_info_v2 import User as V2TemplateUser
+from ...utils.render.schemas.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
@@ -44,9 +45,8 @@ 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 .api.schemas.user_records import SoloModeRecord, UserRecordsSuccess
 from .constant import GAME_TYPE, TR_MAX, TR_MIN
-from .model import IORank
+from .models import IORank, TETRIOUserConfig
 from .typing import Template
 
 UTC = timezone.utc
@@ -55,7 +55,8 @@ driver = get_driver()
 
 
 @alc.assign('TETRIO.query')
-async def _(
+async def _(  # noqa: PLR0913
+    user: NBUser,
     event: Event,
     matcher: Matcher,
     target: At | Me,
@@ -66,7 +67,7 @@ async def _(
         session_persist_id=await get_session_persist_id(event_session),
         game_platform=GAME_TYPE,
         command_type='query',
-        command_args=[],
+        command_args=[f'--default-template {template}'] if template is not None else [],
     ):
         async with get_session() as session:
             bind = await query_bind_info(
@@ -76,6 +77,10 @@ async def _(
                 ),
                 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)
@@ -84,13 +89,18 @@ async def _(
 
 
 @alc.assign('TETRIO.query')
-async def _(account: Player, event_session: EventSession, template: Template | None = None):
+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=[],
+        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()
 
 
@@ -119,10 +129,10 @@ def get_split(value_max: int, value_min: int) -> tuple[int, int]:
 
 
 def get_specified_point(
-    previous_point: Data,
-    behind_point: Data,
+    previous_point: TetraLeagueHistoryData,
+    behind_point: TetraLeagueHistoryData,
     point_time: datetime,
-) -> Data:
+) -> TetraLeagueHistoryData:
     """根据给出的 previous_point 和 behind_point, 推算 point_time 点处的数据
 
     Args:
@@ -137,13 +147,13 @@ def get_specified_point(
     slope = (behind_point.tr - previous_point.tr) / (
         datetime.timestamp(behind_point.record_at) - datetime.timestamp(previous_point.record_at)
     )
-    return Data(
+    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[Data]:
+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)
@@ -173,11 +183,11 @@ async def query_historical_data(user: User, user_info: UserInfoSuccess) -> list[
     full_export_data = FullExport.get_data(user.unique_identifier)
     if not historical_data and not full_export_data:
         return [
-            Data(record_at=today - forward, tr=user_info.data.user.league.rating),
-            Data(record_at=today.replace(microsecond=1000), tr=user_info.data.user.league.rating),
+            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 = [
-        Data(
+        TetraLeagueHistoryData(
             record_at=i.update_time.astimezone(ZoneInfo('Asia/Shanghai')),
             tr=i.data.data.user.league.rating,
         )
@@ -198,7 +208,7 @@ async def query_historical_data(user: User, user_info: UserInfoSuccess) -> list[
         histories.append(
             get_specified_point(
                 histories[-1],
-                Data(record_at=user_info.cache.cached_at, tr=user_info.data.user.league.rating),
+                TetraLeagueHistoryData(record_at=user_info.cache.cached_at, tr=user_info.data.user.league.rating),
                 today.replace(microsecond=1000),
             )
         )
@@ -209,7 +219,7 @@ async def query_historical_data(user: User, user_info: UserInfoSuccess) -> list[
             today - forward,
         )
     else:
-        histories.insert(0, Data(record_at=today - forward, tr=histories[0].tr))
+        histories.insert(0, TetraLeagueHistoryData(record_at=today - forward, tr=histories[0].tr))
     return histories
 
 
@@ -233,29 +243,20 @@ def get_league(
     raise FallbackError
 
 
-def get_sprint(user_records: UserRecordsSuccess) -> SoloModeRecord:
-    return user_records.data.records.sprint
-
-
-def get_blitz(user_records: UserRecordsSuccess) -> SoloModeRecord:
-    return user_records.data.records.blitz
-
-
 async def make_query_image_v1(player: Player) -> bytes:
-    user, user_info, user_records = await gather(player.user, player.get_info(), player.get_records())
+    user, user_info, sprint, blitz = await gather(player.user, player.get_info(), player.sprint, player.blitz)
     league = get_league(user_info, RatedLeague)
-    sprint, blitz = get_sprint(user_records).record, get_blitz(user_records).record
     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 is not None:
-        duration = timedelta(milliseconds=sprint.endcontext.final_time).total_seconds()
+    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.endcontext.score:,}' if blitz is not None else '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(
@@ -263,7 +264,7 @@ async def make_query_image_v1(player: Player) -> bytes:
             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
+                    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
@@ -317,9 +318,11 @@ def handling_special_value(value: N) -> N | None:
 
 
 async def make_query_image_v2(player: Player) -> bytes:
-    user, user_info, user_records = await gather(player.user, player.get_info(), player.get_records())
+    user, user_info, sprint, blitz, zen = await gather(
+        player.user, player.get_info(), player.sprint, player.blitz, player.zen
+    )
     league = get_league(user_info)
-    sprint, blitz = get_sprint(user_records), get_blitz(user_records)
+    histories = await query_historical_data(user, user_info)
 
     if sprint.record is not None:
         duration = timedelta(milliseconds=sprint.record.endcontext.final_time).total_seconds()
@@ -340,17 +343,17 @@ async def make_query_image_v2(player: Player) -> bytes:
     netloc = get_self_netloc()
     async with HostPage(
         await render(
-            'v2/tetrio/info',
+            'v2/tetrio/user/info',
             V2TemplateInfo(
                 user=V2TemplateUser(
                     id=user.ID,
                     name=user.name.upper(),
                     bio=user_info.data.user.bio,
-                    banner=f'http://{netloc}/host/resource/tetrio/banners/{user.ID}?{urlencode({"revision": user_info.data.user.avatar_revision})}'
+                    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
+                    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
@@ -365,8 +368,9 @@ async def make_query_image_v2(player: Player) -> bytes:
                         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 or 0,
+                    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,
@@ -395,6 +399,7 @@ async def make_query_image_v2(player: Player) -> bytes:
                         wins=league.gameswon,
                     ),
                     decaying=league.decaying,
+                    history=histories,
                 )
                 if isinstance(league, RatedLeague)
                 else None,
@@ -416,7 +421,7 @@ async def make_query_image_v2(player: Player) -> bytes:
                 )
                 if blitz.record is not None
                 else None,
-                zen=Zen.model_validate(model_dump(user_records.data.zen)),
+                zen=zen,
             ),
         ),
     ) as page_hash:
@@ -424,9 +429,8 @@ async def make_query_image_v2(player: Player) -> bytes:
 
 
 async def make_query_text(player: Player) -> UniMessage:
-    user, user_info, user_records = await gather(player.user, player.get_info(), player.get_records())
+    user, user_info, sprint, blitz = await gather(player.user, player.get_info(), player.sprint, player.blitz)
     league = get_league(user_info)
-    sprint, blitz = get_sprint(user_records), get_blitz(user_records)
 
     user_name = user.name.upper()
 
@@ -509,8 +513,8 @@ class FullExport:
         cls.latest_update = datetime.now(tz=ZoneInfo('Asia/Shanghai')).date()
 
     @classmethod
-    def get_data(cls, unique_identifier: str) -> list[Data]:
-        return [Data(record_at=i[0], tr=i[1]) for i in cls.cache[unique_identifier]]
+    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:

nonebot_plugin_tetris_stats/games/tetrio/rank.py

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

nonebot_plugin_tetris_stats/games/tetrio/rank/__init__.py

@@ -0,0 +1,113 @@
+from collections import defaultdict
+from collections.abc import Callable
+from datetime import datetime, timedelta, timezone
+from hashlib import sha512
+from math import floor
+from statistics import mean
+from typing import TYPE_CHECKING
+
+from aiofiles import open
+from nonebot import get_driver
+from nonebot.compat import model_dump
+from nonebot.utils import run_sync
+from nonebot_plugin_apscheduler import scheduler
+from nonebot_plugin_localstore import get_data_file
+from nonebot_plugin_orm import get_session
+from sqlalchemy import select
+from zstandard import ZstdCompressor
+
+from ....utils.exception import RequestError
+from ....utils.retry import retry
+from ..api.schemas.base import FailedModel
+from ..api.schemas.tetra_league import ValidUser
+from ..api.schemas.user import User
+from ..api.tetra_league import full_export
+from ..constant import RANK_PERCENTILE
+from ..models import IORank
+
+if TYPE_CHECKING:
+    from ..api.typing import Rank
+
+UTC = timezone.utc
+
+driver = get_driver()
+
+
+@scheduler.scheduled_job('cron', hour='0,6,12,18', minute=0)
+@retry(exception_type=RequestError, delay=timedelta(minutes=15))
+async def get_tetra_league_data() -> None:
+    league, original = await full_export(with_original=True)
+    if isinstance(league, FailedModel):
+        msg = f'排行榜数据请求错误:\n{league.error}'
+        raise RequestError(msg)
+
+    def pps(user: ValidUser) -> float:
+        return user.league.pps
+
+    def apm(user: ValidUser) -> float:
+        return user.league.apm
+
+    def vs(user: ValidUser) -> float:
+        return user.league.vs
+
+    def _min(users: list[ValidUser], field: Callable[[ValidUser], float]) -> ValidUser:
+        return min(users, key=field)
+
+    def _max(users: list[ValidUser], field: Callable[[ValidUser], float]) -> ValidUser:
+        return max(users, key=field)
+
+    def build_extremes_data(
+        users: list[ValidUser],
+        field: Callable[[ValidUser], float],
+        sort: Callable[[list[ValidUser], Callable[[ValidUser], float]], ValidUser],
+    ) -> tuple[dict[str, str], float]:
+        user = sort(users, field)
+        return model_dump(User(ID=user.id, name=user.username)), field(user)
+
+    data_hash: str | None = await run_sync((await run_sync(sha512)(original)).hexdigest)()
+    async with open(get_data_file('nonebot_plugin_tetris_stats', f'{data_hash}.json.zst'), mode='wb') as file:
+        await file.write(await run_sync(ZstdCompressor(level=12, threads=-1).compress)(original))
+
+    users = [i for i in league.data.users if isinstance(i, ValidUser)]
+    rank_to_users: defaultdict[Rank, list[ValidUser]] = defaultdict(list)
+    for i in users:
+        rank_to_users[i.league.rank].append(i)
+    rank_info: list[IORank] = []
+    for rank, percentile in RANK_PERCENTILE.items():
+        offset = floor((percentile / 100) * len(users)) - 1
+        tr_line = users[offset].league.rating
+        rank_users = rank_to_users[rank]
+        rank_info.append(
+            IORank(
+                rank=rank,
+                tr_line=tr_line,
+                player_count=len(rank_users),
+                low_pps=(build_extremes_data(rank_users, pps, _min)),
+                low_apm=(build_extremes_data(rank_users, apm, _min)),
+                low_vs=(build_extremes_data(rank_users, vs, _min)),
+                avg_pps=mean({i.league.pps for i in rank_users}),
+                avg_apm=mean({i.league.apm for i in rank_users}),
+                avg_vs=mean({i.league.vs for i in rank_users}),
+                high_pps=(build_extremes_data(rank_users, pps, _max)),
+                high_apm=(build_extremes_data(rank_users, apm, _max)),
+                high_vs=(build_extremes_data(rank_users, vs, _max)),
+                update_time=league.cache.cached_at,
+                file_hash=data_hash,
+            )
+        )
+    async with get_session() as session:
+        session.add_all(rank_info)
+        await session.commit()
+
+
+@driver.on_startup
+async def _() -> None:
+    async with get_session() as session:
+        latest_time = await session.scalar(select(IORank.update_time).order_by(IORank.id.desc()).limit(1))
+    if latest_time is None or datetime.now(tz=UTC) - latest_time.replace(tzinfo=UTC) > timedelta(hours=6):
+        await get_tetra_league_data()
+
+
+from . import all, detail  # noqa: E402
+
+__all__ = ['all', 'detail']

nonebot_plugin_tetris_stats/games/tetrio/rank/all.py

@@ -0,0 +1,78 @@
+from datetime import timedelta
+
+from nonebot_plugin_alconna 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 func, select
+
+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_rank import AverageData, Data, ItemData
+from ....utils.screenshot import screenshot
+from .. import alc
+from ..constant import GAME_TYPE
+from ..models import IORank
+
+
+@alc.assign('TETRIO.rank.all')
+async def _(event_session: EventSession):
+    async with trigger(
+        session_persist_id=await get_session_persist_id(event_session),
+        game_platform=GAME_TYPE,
+        command_type='rank',
+        command_args=['--all'],
+    ):
+        async with get_session() as session:
+            latest_update_time = (
+                await session.scalars(select(IORank.update_time).order_by(IORank.id.desc()).limit(1))
+            ).one()
+            compare_time = (
+                await session.scalars(
+                    select(IORank.update_time)
+                    .order_by(
+                        func.abs(
+                            func.julianday(IORank.update_time)
+                            - func.julianday(latest_update_time - timedelta(hours=24))
+                        )
+                    )
+                    .limit(1)
+                )
+            ).one()
+            latest_data = (
+                await session.scalars(
+                    select(IORank).where(IORank.update_time == latest_update_time).order_by(IORank.tr_line.desc())
+                )
+            ).all()
+            compare_data = (
+                await session.scalars(
+                    select(IORank).where(IORank.update_time == compare_time).order_by(IORank.tr_line.desc())
+                )
+            ).all()
+        async with HostPage(
+            await render(
+                'v2/tetrio/rank',
+                Data(
+                    items={
+                        i[0].rank: ItemData(
+                            require_tr=round(i[0].tr_line, 2),
+                            trending=round(i[0].tr_line - i[1].tr_line, 2),
+                            average_data=AverageData(
+                                pps=(metrics := get_metrics(pps=i[0].avg_pps, apm=i[0].avg_apm, vs=i[0].avg_vs)).pps,
+                                apm=metrics.apm,
+                                apl=metrics.apl,
+                                vs=metrics.vs,
+                                adpl=metrics.adpl,
+                            ),
+                            players=i[0].player_count,
+                        )
+                        for i in zip(latest_data, compare_data, strict=True)
+                    },
+                    updated_at=latest_update_time,
+                ),
+            )
+        ) as page_hash:
+            await screenshot(f'http://{get_self_netloc()}/host/{page_hash}.html')
+            await UniMessage.image(raw=await screenshot(f'http://{get_self_netloc()}/host/{page_hash}.html')).finish()

nonebot_plugin_tetris_stats/games/tetrio/rank/detail.py

@@ -0,0 +1,138 @@
+from datetime import datetime, timedelta, timezone
+from zoneinfo import ZoneInfo
+
+from nonebot import get_driver
+from nonebot_plugin_alconna 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 func, select
+
+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_rank_detail import Data, SpecialData
+from ....utils.screenshot import screenshot
+from .. import alc
+from ..api.typing import ValidRank
+from ..constant import GAME_TYPE
+from ..models import IORank
+
+UTC = timezone.utc
+
+driver = get_driver()
+
+
+@alc.assign('TETRIO.rank')
+async def _(rank: ValidRank, event_session: EventSession):
+    async with trigger(
+        session_persist_id=await get_session_persist_id(event_session),
+        game_platform=GAME_TYPE,
+        command_type='rank',
+        command_args=[f'--detail {rank}'],
+    ):
+        async with get_session() as session:
+            latest_data = (
+                await session.scalars(select(IORank).where(IORank.rank == rank).order_by(IORank.id.desc()).limit(1))
+            ).one()
+            compare_data = (
+                await session.scalars(
+                    select(IORank)
+                    .where(IORank.rank == rank)
+                    .order_by(
+                        func.abs(
+                            func.julianday(IORank.update_time)
+                            - func.julianday(latest_data.update_time - timedelta(hours=24))
+                        )
+                    )
+                    .limit(1)
+                )
+            ).one()
+        await UniMessage.image(raw=await make_image(latest_data, compare_data)).finish()
+
+
+async def make_image(latest_data: IORank, compare_data: IORank) -> bytes:
+    avg = get_metrics(pps=latest_data.avg_pps, apm=latest_data.avg_apm, vs=latest_data.avg_vs)
+    low_pps = get_metrics(pps=latest_data.low_pps[1])
+    low_vs = get_metrics(vs=latest_data.low_vs[1])
+    max_pps = get_metrics(pps=latest_data.high_pps[1])
+    max_vs = get_metrics(vs=latest_data.high_vs[1])
+    async with HostPage(
+        await render(
+            'v2/tetrio/rank/detail',
+            Data(
+                name=latest_data.rank,
+                trending=round(latest_data.tr_line - compare_data.tr_line, 2),
+                require_tr=round(latest_data.tr_line, 2),
+                players=latest_data.player_count,
+                minimum_data=SpecialData(
+                    apm=latest_data.low_apm[1],
+                    pps=low_pps.pps,
+                    lpm=low_pps.lpm,
+                    vs=low_vs.vs,
+                    adpm=low_vs.adpm,
+                    apm_holder=latest_data.low_apm[0]['name'].upper(),
+                    pps_holder=latest_data.low_pps[0]['name'].upper(),
+                    vs_holder=latest_data.low_vs[0]['name'].upper(),
+                ),
+                average_data=SpecialData(
+                    apm=avg.apm,
+                    pps=avg.pps,
+                    lpm=avg.lpm,
+                    vs=avg.vs,
+                    adpm=avg.adpm,
+                    apl=avg.apl,
+                    adpl=avg.adpl,
+                ),
+                maximum_data=SpecialData(
+                    apm=latest_data.high_apm[1],
+                    pps=max_pps.pps,
+                    lpm=max_pps.lpm,
+                    vs=max_vs.vs,
+                    adpm=max_vs.adpm,
+                    apm_holder=latest_data.high_apm[0]['name'].upper(),
+                    pps_holder=latest_data.high_pps[0]['name'].upper(),
+                    vs_holder=latest_data.high_vs[0]['name'].upper(),
+                ),
+                updated_at=latest_data.update_time.replace(tzinfo=UTC).astimezone(ZoneInfo('Asia/Shanghai')),
+            ),
+        )
+    ) as page_hash:
+        return await screenshot(f'http://{get_self_netloc()}/host/{page_hash}.html')
+
+
+async def make_text(latest_data: IORank, compare_data: IORank) -> str:
+    message = ''
+    if (datetime.now(UTC) - latest_data.update_time.replace(tzinfo=UTC)) > timedelta(hours=7):
+        message += 'Warning: 数据超过7小时未更新, 请联系Bot主人查看后台\n'
+    message += f'{latest_data.rank.upper()} 段 分数线 {latest_data.tr_line:.2f} TR, {latest_data.player_count} 名玩家\n'
+    if compare_data.id != latest_data.id:
+        message += f'对比 {(latest_data.update_time-compare_data.update_time).total_seconds()/3600:.2f} 小时前趋势: {f"↑{difference:.2f}" if (difference:=latest_data.tr_line-compare_data.tr_line) > 0 else f"↓{-difference:.2f}" if difference < 0 else "→"}'
+    else:
+        message += '暂无对比数据'
+    avg = get_metrics(pps=latest_data.avg_pps, apm=latest_data.avg_apm, vs=latest_data.avg_vs)
+    low_pps = get_metrics(pps=latest_data.low_pps[1])
+    low_vs = get_metrics(vs=latest_data.low_vs[1])
+    max_pps = get_metrics(pps=latest_data.high_pps[1])
+    max_vs = get_metrics(vs=latest_data.high_vs[1])
+    message += (
+        '\n'
+        '平均数据:\n'
+        f"L'PM: {avg.lpm} ( {avg.pps} pps )\n"
+        f'APM: {avg.apm} ( x{avg.apl} )\n'
+        f'ADPM: {avg.adpm} ( x{avg.adpl} ) ( {avg.vs}vs )\n'
+        '\n'
+        '最低数据:\n'
+        f"L'PM: {low_pps.lpm} ( {low_pps.pps} pps ) By: {latest_data.low_pps[0]['name'].upper()}\n"
+        f'APM: {latest_data.low_apm[1]} By: {latest_data.low_apm[0]["name"].upper()}\n'
+        f'ADPM: {low_vs.adpm} ( {low_vs.vs}vs ) By: {latest_data.low_vs[0]["name"].upper()}\n'
+        '\n'
+        '最高数据:\n'
+        f"L'PM: {max_pps.lpm} ( {max_pps.pps} pps ) By: {latest_data.high_pps[0]['name'].upper()}\n"
+        f'APM: {latest_data.high_apm[1]} By: {latest_data.high_apm[0]["name"].upper()}\n'
+        f'ADPM: {max_vs.adpm} ( {max_vs.vs}vs ) By: {latest_data.high_vs[0]["name"].upper()}\n'
+        '\n'
+        f'数据更新时间: {latest_data.update_time.replace(tzinfo=UTC).astimezone(ZoneInfo("Asia/Shanghai")).strftime("%Y-%m-%d %H:%M:%S")}'
+    )
+    return message

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.tspintriples,
+                        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.tspintriples,
+                        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/top/__init__.py

@@ -3,7 +3,7 @@ from nonebot_plugin_alconna import At
 
 from ...utils.exception import MessageFormatError
 from ...utils.typing import Me
-from .. import add_block_handlers, alc
+from .. import add_block_handlers, alc, command
 from .api import Player
 from .constant import USER_NAME
 
@@ -14,7 +14,7 @@ def get_player(name: str) -> Player | MessageFormatError:
     return MessageFormatError('用户名/ID不合法')
 
 
-alc.command.add(
+command.add(
     Subcommand(
         'TOP',
         Subcommand(
@@ -51,8 +51,8 @@ alc.command.add(
     )
 )
 
-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查'})
+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_block_handlers(alc.assign('TOP.query'))
 

nonebot_plugin_tetris_stats/games/tos/__init__.py

@@ -3,7 +3,7 @@ from nonebot_plugin_alconna import At
 
 from ...utils.exception import MessageFormatError
 from ...utils.typing import Me
-from .. import add_block_handlers, alc
+from .. import add_block_handlers, alc, command
 from .api import Player
 from .constant import USER_NAME
 
@@ -19,7 +19,7 @@ def get_player(teaid_or_name: str) -> Player | MessageFormatError:
     return MessageFormatError('用户名/ID不合法')
 
 
-alc.command.add(
+command.add(
     Subcommand(
         'TOS',
         Subcommand(
@@ -56,8 +56,8 @@ alc.command.add(
     )
 )
 
-alc.shortcut('(?i:tos|茶服)(?i:绑|绑定|bind)', {'command': 'tstats TOS bind', 'humanized': '茶服绑定'})
-alc.shortcut('(?i:tos|茶服)(?i:查|查询|query|stats)', {'command': 'tstats TOS query', 'humanized': '茶服查'})
+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'))
 

nonebot_plugin_tetris_stats/utils/exception.py

@@ -27,6 +27,10 @@ class MessageFormatError(NeedCatchError):
     """用户发送的消息格式不正确"""
 
 
+class RecordNotFoundError(NeedCatchError):
+    """找不到用户的某种记录"""
+
+
 class FallbackError(NeedCatchError):
     """需要回滚至更通用的方法"""
 

nonebot_plugin_tetris_stats/utils/render/__init__.py

@@ -5,8 +5,13 @@ from nonebot.compat import PYDANTIC_V2
 
 from ..templates import templates_dir
 from .schemas.bind import Bind
-from .schemas.tetrio_info import Info as TETRIOInfo
-from .schemas.tetrio_info_v2 import Info as TETRIOInfoV2
+from .schemas.tetrio.tetrio_info import Info as TETRIOInfo
+from .schemas.tetrio.tetrio_rank import Data as TETRIORankData
+from .schemas.tetrio.tetrio_rank_detail import Data as TETRIORankDetailData
+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
 
@@ -17,27 +22,49 @@ env = Environment(
 
 @overload
 async def render(render_type: Literal['v1/binding'], data: Bind) -> str: ...
-
-
 @overload
 async def render(render_type: Literal['v1/tetrio/info'], data: TETRIOInfo) -> str: ...
-
-
-@overload
-async def render(render_type: Literal['v2/tetrio/info'], data: TETRIOInfoV2) -> 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: ...
+@overload
+async def render(render_type: Literal['v2/tetrio/rank'], data: TETRIORankData) -> str: ...
+@overload
+async def render(render_type: Literal['v2/tetrio/rank/detail'], data: TETRIORankDetailData) -> str: ...
 
 
 async def render(
-    render_type: Literal['v1/binding', 'v1/tetrio/info', 'v2/tetrio/info', 'v1/top/info', 'v1/tos/info'],
-    data: Bind | TETRIOInfo | TETRIOInfoV2 | TOPInfo | TOSInfo,
+    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',
+        'v2/tetrio/rank',
+        'v2/tetrio/rank/detail',
+    ],
+    data: Bind
+    | TETRIOInfo
+    | TOPInfo
+    | TOSInfo
+    | TETRIOUserInfoV2
+    | TETRIOUserListV2
+    | TETRIORecordSprint
+    | TETRIORecordBlitz
+    | TETRIORankData
+    | TETRIORankDetailData,
 ) -> str:
     if PYDANTIC_V2:
         return await env.get_template('index.html').render_async(
@@ -46,4 +73,4 @@ async def render(
     return await env.get_template('index.html').render_async(path=render_type, data=data.json(by_alias=True))
 
 
-__all__ = ['render', 'Bind']
+__all__ = ['render']

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

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

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

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

nonebot_plugin_tetris_stats/utils/render/schemas/tetrio/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

@@ -1,10 +1,13 @@
 from datetime import datetime
+from typing import Literal
 
 from pydantic import BaseModel
 
-from ....games.tetrio.api.typing import Rank
-from ...typing import Number
-from .base import Avatar
+from .....games.tetrio.api.schemas.user_records import Zen
+from .....games.tetrio.api.typing import Rank
+from ....typing import Number
+from ..base import Avatar
+from .base import TetraLeagueHistoryData
 
 
 class Badge(BaseModel):
@@ -19,12 +22,14 @@ class User(BaseModel):
     name: str
     country: str | None
 
+    role: Literal['anon', 'user', 'bot', 'halfmod', 'mod', 'admin', 'sysop', 'banned']
+
     avatar: str | Avatar
     banner: str | None
 
     bio: str | None
 
-    friend_count: int
+    friend_count: int | None
     supporter_tier: int
 
     verified: bool
@@ -71,6 +76,8 @@ class TetraLeague(BaseModel):
 
     decaying: bool
 
+    history: list[TetraLeagueHistoryData]
+
 
 class Sprint(BaseModel):
     time: str
@@ -84,11 +91,6 @@ class Blitz(BaseModel):
     play_at: datetime
 
 
-class Zen(BaseModel):
-    score: int
-    level: int
-
-
 class Info(BaseModel):
     user: User
     tetra_league: TetraLeague | None

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

nonebot_plugin_tetris_stats/utils/screenshot.py

@@ -1,15 +1,29 @@
-from playwright.async_api import TimeoutError
+from nonebot import get_plugin_config
+from playwright.async_api import TimeoutError, ViewportSize
 
+from ..config.config import Config
 from .browser import BrowserManager
 from .retry import retry
 
+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(viewport={'width': 3000, 'height': 3000}) as page,
+        await browser.new_page(device_scale_factor=config.tetris_screenshot_quality) as page,
     ):
         await page.goto(url)
         await page.wait_for_load_state('networkidle')
+        size: ViewportSize = await page.evaluate("""
+            () => {
+                const element = document.querySelector('#content');
+                return {
+                    width: element.offsetWidth,
+                    height: element.offsetHeight,
+                };
+            };
+        """)
+        await page.set_viewport_size(size)
         return await page.locator('id=content').screenshot(timeout=5000, type='png')

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