🔖 1.1.0

* 🔥 删除现有模板

* ✨ 自动克隆模板仓库

* 🔥 删除 identicon 相关代码

* 🚚 修改静态文件路径

* ✨ 使用新模板进行渲染

* ✨ 每次渲染都获取一次模板, 以应对实时更新

* ✨ TETR.IO 绑定图使用新模板

* 🚚 修改网络路径

* ✨ TOP 绑定图使用新模板

* ✨ TOS 绑定图使用新模板

* 🐛 防止截图超时

* 🐛 Pydantic V1 会把 float 转换成 int

* ✏️ 模板字段名写错了

* ✨ 兼容 Pydantic V1

* ✨ TETR.IO 查询图使用新模板

* 🐛 在查询的用户没有历史记录时不去查询更多记录

.github/dependabot.yml

@@ -7,6 +7,6 @@ version: 2
 updates:
   - package-ecosystem: "pip" # See documentation for possible values
     directory: "/" # Location of package manifests
-    target-branch: "dev"
+    target-branch: "main"
     schedule:
       interval: "daily"

.github/workflows/Release.yml

@@ -8,18 +8,42 @@ on:
 jobs:
   release:
     runs-on: ubuntu-latest
+    permissions:
+      id-token: write
+      contents: write
     steps:
       - uses: actions/checkout@v3
+
+      - name: Install poetry
+        run: pipx install poetry
+        shell: bash
+
       - uses: actions/setup-python@v4
         with:
           python-version: '3.11'
-      - name: Install Poetry
+          cache: "poetry"
+
+      - run: poetry install
         shell: bash
+
+      - name: Get Version
+        id: version
         run: |
-          pip install poetry
-      - name: Build
-        shell: bash
-        run: |
-          poetry install
-          poetry env use python
-          poetry publish --build -u ${{ secrets.USERNAME }} -p ${{ secrets.PASSWORD }} -n
+          echo "VERSION=$(poetry version -s)" >> $GITHUB_OUTPUT
+          echo "TAG_VERSION=${GITHUB_REF#refs/tags/}" >> $GITHUB_OUTPUT
+          echo "TAG_NAME=${GITHUB_REF#refs/tags/}" >> $GITHUB_OUTPUT
+
+      - name: Check Version
+        if: steps.version.outputs.VERSION != steps.version.outputs.TAG_VERSION
+        run: exit 1
+
+      - name: Build Package
+        run: poetry build
+
+      - name: Publish Package to PyPI
+        uses: pypa/gh-action-pypi-publish@release/v1
+
+      - name: Publish Package to GitHub Release
+        run: gh release create ${{ steps.version.outputs.TAG_NAME }} dist/*.tar.gz dist/*.whl -t "🔖 ${{ steps.version.outputs.TAG_NAME }}" --generate-notes
+        env:
+          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}

.gitignore

@@ -5,6 +5,17 @@ Untitled*
 *copy*
 .vscode
 *dev*
-*cache*
+*_cache*
 *backup*
 *.pyc
+node_modules
+.prettier*
+package.json
+pnpm-lock.yaml
+*.drawio.svg
+package-lock.json
+*Zone.Identifier
+.env*
+bot.py
+TODO
+*.fish

nonebot_plugin_tetris_stats/__init__.py

@@ -13,10 +13,11 @@ __plugin_meta__ = PluginMetadata(
     description='一个用于查询 Tetris 相关游戏玩家数据的插件',
     usage='发送 {游戏名} --help 查询使用方法',
     type='application',
-    homepage='https://github.com/shoucandanghehe/nonebot-plugin-tetris-stats',
+    homepage='https://github.com/A-minos/nonebot-plugin-tetris-stats',
     extra={
         'orm_version_location': migrations,
     },
 )
 
 from . import game_data_processor  # noqa: F401, E402
+from .utils import host  # noqa: F401, E402

nonebot_plugin_tetris_stats/config/config.py

@@ -11,4 +11,4 @@ CACHE_PATH: Path = get_cache_dir('nonebot_plugin_tetris_stats')
 class Config(BaseModel):
     """配置类"""
 
-    db_url: str = 'sqlite://data/nonebot_plugin_tetris_stats/data.db'
+    tetris_req_timeout: float = 30.0

nonebot_plugin_tetris_stats/config/migrations/09d4bb60160d_rename_field.py

@@ -0,0 +1,51 @@
+"""Rename field
+
+迁移 ID: 09d4bb60160d
+父迁移: b9d65badc713
+创建时间: 2024-04-23 23:42:04.541672
+
+"""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = '09d4bb60160d'
+down_revision: str | Sequence[str] | None = 'b9d65badc713'
+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_iorank', schema=None) as batch_op:
+        batch_op.alter_column('create_time', new_column_name='update_time', existing_type=sa.DateTime())
+        batch_op.drop_index('ix_nonebot_plugin_tetris_stats_iorank_create_time')
+    op.create_index(
+        batch_op.f('ix_nonebot_plugin_tetris_stats_iorank_update_time'),
+        'nonebot_plugin_tetris_stats_iorank',
+        ['update_time'],
+        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_iorank', schema=None) as batch_op:
+        batch_op.alter_column('update_time', new_column_name='create_time')
+        batch_op.drop_index(batch_op.f('ix_nonebot_plugin_tetris_stats_iorank_update_time'))
+    op.create_index(
+        'ix_nonebot_plugin_tetris_stats_iorank_create_time',
+        'nonebot_plugin_tetris_stats_iorank',
+        ['create_time'],
+        unique=False,
+    )
+    # ### end Alembic commands ###

nonebot_plugin_tetris_stats/config/migrations/0d50142b780f_add_field.py

@@ -0,0 +1,43 @@
+"""add field
+
+迁移 ID: 0d50142b780f
+父迁移: 09d4bb60160d
+创建时间: 2024-04-24 14:55:08.064098
+
+"""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = '0d50142b780f'
+down_revision: str | Sequence[str] | None = '09d4bb60160d'
+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_iorank', schema=None) as batch_op:
+        batch_op.add_column(sa.Column('file_hash', sa.String(length=128), nullable=True))
+        batch_op.create_index(
+            batch_op.f('ix_nonebot_plugin_tetris_stats_iorank_file_hash'), ['file_hash'], 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_iorank', schema=None) as batch_op:
+        batch_op.drop_index(batch_op.f('ix_nonebot_plugin_tetris_stats_iorank_file_hash'))
+        batch_op.drop_column('file_hash')
+
+    # ### end Alembic commands ###

nonebot_plugin_tetris_stats/config/migrations/6c3206f90cc3_add_field.py

@@ -0,0 +1,65 @@
+"""Add redundant platform field
+
+迁移 ID: 6c3206f90cc3
+父迁移: 9f6582279ce2
+创建时间: 2023-11-26 20:15:56.033892
+
+"""
+from __future__ import annotations
+
+from collections.abc import Sequence
+
+from alembic import op
+from sqlalchemy.ext.automap import automap_base
+from sqlalchemy.orm import Session
+from ujson import dumps, loads
+
+revision: str = '6c3206f90cc3'
+down_revision: str | Sequence[str] | None = '9f6582279ce2'
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade(name: str = '') -> None:
+    if name:
+        return
+
+    Base = automap_base()  # noqa: N806
+    connection = op.get_bind()
+    Base.prepare(autoload_with=connection)
+
+    HistoricalData = Base.classes.nonebot_plugin_tetris_stats_historicaldata  # noqa: N806
+
+    with Session(connection) as session:
+        for row in session.query(HistoricalData):
+            platform = row.game_platform
+            game_user = loads(row.game_user)
+            processed_data = loads(row.processed_data)
+            game_user['platform'] = platform
+            processed_data['platform'] = platform
+            row.game_user = dumps(game_user)
+            row.processed_data = dumps(processed_data)
+            session.add(row)
+        session.commit()
+
+
+def downgrade(name: str = '') -> None:
+    if name:
+        return
+
+    Base = automap_base()  # noqa: N806
+    connection = op.get_bind()
+    Base.prepare(autoload_with=connection)
+
+    HistoricalData = Base.classes.nonebot_plugin_tetris_stats_historicaldata  # noqa: N806
+
+    with Session(connection) as session:
+        for row in session.query(HistoricalData):
+            game_user = loads(row.game_user)
+            processed_data = loads(row.processed_data)
+            game_user.pop('platform', None)
+            processed_data.pop('platform', None)
+            row.game_user = dumps(game_user)
+            row.processed_data = dumps(processed_data)
+            session.add(row)
+        session.commit()

nonebot_plugin_tetris_stats/config/migrations/8a91210ce14d_correct_the_data_in_historicaldata.py

@@ -0,0 +1,92 @@
+"""Correct the data in HistoricalData
+
+迁移 ID: 8a91210ce14d
+父迁移: 0d50142b780f
+创建时间: 2024-05-06 08:16:38.487214
+
+"""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+
+from alembic import op
+from nonebot.log import logger
+from sqlalchemy import select
+from sqlalchemy.ext.automap import automap_base
+from sqlalchemy.orm import Session
+
+revision: str = '8a91210ce14d'
+down_revision: str | Sequence[str] | None = '0d50142b780f'
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade(name: str = '') -> None:
+    if name:
+        return
+    from nonebot_plugin_tetris_stats.version import __version__
+
+    if __version__ != '1.0.3':
+        logger.critical('本迁移需要1.0.3版本, 请先锁定版本至1.0.3版本再执行本迁移')
+        raise RuntimeError('本迁移需要1.0.3版本, 请先锁定版本至1.0.3版本再执行本迁移')
+
+    from nonebot.compat import PYDANTIC_V2, type_validate_json
+    from pydantic import BaseModel, ValidationError
+    from rich.progress import (
+        BarColumn,
+        MofNCompleteColumn,
+        Progress,
+        TaskProgressColumn,
+        TextColumn,
+        TimeRemainingColumn,
+    )
+
+    from nonebot_plugin_tetris_stats.game_data_processor.schemas import BaseProcessedData
+
+    Base = automap_base()  # noqa: N806
+    Base.prepare(autoload_with=op.get_bind())
+    HistoricalData = Base.classes.nonebot_plugin_tetris_stats_historicaldata  # noqa: N806
+    if PYDANTIC_V2:
+
+        def model_to_json(value: BaseModel) -> str:
+            return value.model_dump_json(by_alias=True)
+    else:
+
+        def model_to_json(value: BaseModel) -> str:
+            return value.json(by_alias=True)
+
+    models = BaseProcessedData.__subclasses__()
+
+    def json_to_model(value: str) -> BaseModel:
+        for i in models:
+            try:
+                return type_validate_json(i, value)
+            except ValidationError:  # noqa: PERF203
+                ...
+        raise ValueError
+
+    with Session(op.get_bind()) as session:
+        count = session.query(HistoricalData).count()
+        with Progress(
+            TextColumn('[progress.description]{task.description}'),
+            BarColumn(),
+            MofNCompleteColumn(),
+            TaskProgressColumn(),
+            TimeRemainingColumn(),
+        ) as progress:
+            task_id = progress.add_task('[cyan]Updateing:', total=count)
+            for i in range(0, count, 100):
+                for j in session.scalars(
+                    select(HistoricalData).where(HistoricalData.id > i).order_by(HistoricalData.id).limit(100)
+                ):
+                    model = json_to_model(j.processed_data)
+                    j.processed_data = model_to_json(model)
+                    progress.update(task_id, advance=1)
+                session.commit()
+    logger.success('Corrected HistoricalData')
+
+
+def downgrade(name: str = '') -> None:
+    if name:
+        return

nonebot_plugin_tetris_stats/config/migrations/9cd1647db502_merge_old_db.py

@@ -64,7 +64,7 @@ def upgrade(name: str = '') -> None:
     if name:
         return
     try:
-        db_path = Path(config.db_path)
+        db_path = Path(config.db_url)
     except AttributeError:
         db_path = Path('data/nonebot_plugin_tetris_stats/data.db')
     if db_path.exists() is False:
@@ -84,7 +84,7 @@ def upgrade(name: str = '') -> None:
             raise RuntimeError('nonebot_plugin_tetris_stats: 请先安装 0.4.4 版本完成迁移之后再升级')
         logger.info('nonebot_plugin_tetris_stats: 发现来自老版本的数据, 正在迁移...')
         migrate_old_data(connection)
-        db_path.unlink()
+    db_path.unlink()
 
 
 def downgrade(name: str = '') -> None:

nonebot_plugin_tetris_stats/config/migrations/9f6582279ce2_recreate_historical_data.py

@@ -0,0 +1,112 @@
+"""Recreate HistoricalData
+
+迁移 ID: 9f6582279ce2
+父迁移: 9cd1647db502
+创建时间: 2023-11-21 08:35:50.393246
+
+"""
+from __future__ import annotations
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+from sqlalchemy.dialects import sqlite
+
+from nonebot_plugin_tetris_stats.db.models import PydanticType
+
+revision: str = '9f6582279ce2'
+down_revision: str | Sequence[str] | None = '9cd1647db502'
+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_historicaldata', schema=None) as batch_op:
+        batch_op.drop_index('ix_nonebot_plugin_tetris_stats_historicaldata_command_type')
+        batch_op.drop_index('ix_nonebot_plugin_tetris_stats_historicaldata_game_platform')
+        batch_op.drop_index('ix_nonebot_plugin_tetris_stats_historicaldata_source_account')
+        batch_op.drop_index('ix_nonebot_plugin_tetris_stats_historicaldata_source_type')
+
+    op.drop_table('nonebot_plugin_tetris_stats_historicaldata')
+
+    op.create_table(
+        'nonebot_plugin_tetris_stats_historicaldata',
+        sa.Column('id', sa.Integer(), nullable=False),
+        sa.Column('trigger_time', sa.DateTime(), nullable=False),
+        sa.Column('bot_platform', sa.String(length=32), nullable=True),
+        sa.Column('bot_account', sa.String(), nullable=True),
+        sa.Column('source_type', sa.String(length=32), nullable=True),
+        sa.Column('source_account', sa.String(), nullable=True),
+        sa.Column('message', sa.PickleType(), nullable=True),
+        sa.Column('game_platform', sa.String(length=32), nullable=False),
+        sa.Column('command_type', sa.String(length=16), nullable=False),
+        sa.Column('command_args', sa.JSON(), nullable=False),
+        sa.Column('game_user', PydanticType(list), nullable=False),
+        sa.Column('processed_data', PydanticType(list), nullable=False),
+        sa.Column('finish_time', sa.DateTime(), nullable=False),
+        sa.PrimaryKeyConstraint('id', name=op.f('pk_nonebot_plugin_tetris_stats_historicaldata')),
+    )
+    with op.batch_alter_table('nonebot_plugin_tetris_stats_historicaldata', schema=None) as batch_op:
+        batch_op.create_index(
+            batch_op.f('ix_nonebot_plugin_tetris_stats_historicaldata_command_type'), ['command_type'], unique=False
+        )
+        batch_op.create_index(
+            batch_op.f('ix_nonebot_plugin_tetris_stats_historicaldata_game_platform'), ['game_platform'], unique=False
+        )
+        batch_op.create_index(
+            batch_op.f('ix_nonebot_plugin_tetris_stats_historicaldata_source_account'), ['source_account'], unique=False
+        )
+        batch_op.create_index(
+            batch_op.f('ix_nonebot_plugin_tetris_stats_historicaldata_source_type'), ['source_type'], 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_historicaldata', schema=None) as batch_op:
+        batch_op.drop_index(batch_op.f('ix_nonebot_plugin_tetris_stats_historicaldata_source_type'))
+        batch_op.drop_index(batch_op.f('ix_nonebot_plugin_tetris_stats_historicaldata_source_account'))
+        batch_op.drop_index(batch_op.f('ix_nonebot_plugin_tetris_stats_historicaldata_game_platform'))
+        batch_op.drop_index(batch_op.f('ix_nonebot_plugin_tetris_stats_historicaldata_command_type'))
+
+    op.drop_table('nonebot_plugin_tetris_stats_historicaldata')
+
+    op.create_table(
+        'nonebot_plugin_tetris_stats_historicaldata',
+        sa.Column('id', sa.INTEGER(), nullable=False),
+        sa.Column('trigger_time', sa.DATETIME(), nullable=False),
+        sa.Column('bot_platform', sa.VARCHAR(length=32), nullable=True),
+        sa.Column('bot_account', sa.VARCHAR(), nullable=True),
+        sa.Column('source_type', sa.VARCHAR(length=32), nullable=True),
+        sa.Column('source_account', sa.VARCHAR(), nullable=True),
+        sa.Column('message', sa.BLOB(), nullable=True),
+        sa.Column('game_platform', sa.VARCHAR(length=32), nullable=False),
+        sa.Column('command_type', sa.VARCHAR(length=16), nullable=False),
+        sa.Column('command_args', sqlite.JSON(), nullable=False),
+        sa.Column('game_user', sa.BLOB(), nullable=False),
+        sa.Column('processed_data', sa.BLOB(), nullable=False),
+        sa.Column('finish_time', sa.DATETIME(), nullable=False),
+        sa.PrimaryKeyConstraint('id', name='pk_nonebot_plugin_tetris_stats_historicaldata'),
+    )
+    with op.batch_alter_table('nonebot_plugin_tetris_stats_historicaldata', schema=None) as batch_op:
+        batch_op.create_index(
+            'ix_nonebot_plugin_tetris_stats_historicaldata_source_type', ['source_type'], unique=False
+        )
+        batch_op.create_index(
+            'ix_nonebot_plugin_tetris_stats_historicaldata_source_account', ['source_account'], unique=False
+        )
+        batch_op.create_index(
+            'ix_nonebot_plugin_tetris_stats_historicaldata_game_platform', ['game_platform'], unique=False
+        )
+        batch_op.create_index(
+            'ix_nonebot_plugin_tetris_stats_historicaldata_command_type', ['command_type'], unique=False
+        )
+
+    # ### end Alembic commands ###

nonebot_plugin_tetris_stats/config/migrations/b7fbdafc339a_add_user_unique_identifier_field_to_.py

@@ -0,0 +1,103 @@
+"""Add user_unique_identifier field to HistoricalData
+
+迁移 ID: b7fbdafc339a
+父迁移: 8a91210ce14d
+创建时间: 2024-05-07 16:55:29.527215
+
+"""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+from nonebot.log import logger
+
+revision: str = 'b7fbdafc339a'
+down_revision: str | Sequence[str] | None = '8a91210ce14d'
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade(name: str = '') -> None:
+    if name:
+        return
+    from nonebot_plugin_tetris_stats.version import __version__
+
+    if __version__ != '1.0.4':
+        logger.critical('本迁移需要1.0.4版本, 请先锁定版本至1.0.4版本再执行本迁移')
+        raise RuntimeError('本迁移需要1.0.4版本, 请先锁定版本至1.0.4版本再执行本迁移')
+    from nonebot.compat import type_validate_json
+    from pydantic import ValidationError
+    from rich.progress import (
+        BarColumn,
+        MofNCompleteColumn,
+        Progress,
+        TaskProgressColumn,
+        TextColumn,
+        TimeRemainingColumn,
+    )
+    from sqlalchemy import select
+    from sqlalchemy.ext.automap import automap_base
+    from sqlalchemy.orm import Session
+
+    from nonebot_plugin_tetris_stats.game_data_processor.schemas import BaseUser
+
+    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))
+        batch_op.create_index(
+            batch_op.f('ix_nonebot_plugin_tetris_stats_historicaldata_user_unique_identifier'),
+            ['user_unique_identifier'],
+            unique=False,
+        )
+    Base = automap_base()  # noqa: N806
+    connection = op.get_bind()
+    Base.prepare(autoload_with=connection)
+    HistoricalData = Base.classes.nonebot_plugin_tetris_stats_historicaldata  # noqa: N806
+
+    models: list[type[BaseUser]] = BaseUser.__subclasses__()
+
+    def json_to_model(value: str) -> BaseUser:
+        for i in models:
+            try:
+                return type_validate_json(i, value)
+            except ValidationError:  # noqa: PERF203
+                ...
+        raise ValueError
+
+    with Session(op.get_bind()) as session:
+        count = session.query(HistoricalData).count()
+        with Progress(
+            TextColumn('[progress.description]{task.description}'),
+            BarColumn(),
+            MofNCompleteColumn(),
+            TaskProgressColumn(),
+            TimeRemainingColumn(),
+        ) as progress:
+            task_id = progress.add_task('[cyan]Updateing:', total=count)
+            for i in range(0, count, 100):
+                for j in session.scalars(
+                    select(HistoricalData).where(HistoricalData.id > i).order_by(HistoricalData.id).limit(100)
+                ):
+                    model = json_to_model(j.game_user)
+                    try:
+                        j.user_unique_identifier = model.unique_identifier
+                    except ValueError:
+                        session.delete(j)
+                    progress.update(task_id, advance=1)
+                session.commit()
+    with op.batch_alter_table('nonebot_plugin_tetris_stats_historicaldata', schema=None) as batch_op:
+        batch_op.alter_column('user_unique_identifier', existing_type=sa.VARCHAR(length=32), nullable=False)
+    logger.success('database upgrade success')
+
+
+def downgrade(name: str = '') -> None:
+    if name:
+        return
+    # ### commands auto generated by Alembic - please adjust! ###
+    with op.batch_alter_table('nonebot_plugin_tetris_stats_historicaldata', schema=None) as batch_op:
+        batch_op.drop_index(batch_op.f('ix_nonebot_plugin_tetris_stats_historicaldata_user_unique_identifier'))
+        batch_op.drop_column('user_unique_identifier')
+
+    # ### end Alembic commands ###

nonebot_plugin_tetris_stats/config/migrations/b9d65badc713_del_old_tos_bind_data.py

@@ -0,0 +1,38 @@
+"""Del old TOS bind data
+
+迁移 ID: b9d65badc713
+父迁移: 6c3206f90cc3
+创建时间: 2023-12-30 00:27:40.991704
+
+"""
+from __future__ import annotations
+
+from collections.abc import Sequence
+
+from alembic import op
+from sqlalchemy.ext.automap import automap_base
+from sqlalchemy.orm import Session
+
+revision: str = 'b9d65badc713'
+down_revision: str | Sequence[str] | None = '6c3206f90cc3'
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade(name: str = '') -> None:
+    if name:
+        return
+
+    Base = automap_base()  # noqa: N806
+    connection = op.get_bind()
+    Base.prepare(autoload_with=connection)
+
+    Bind = Base.classes.nonebot_plugin_tetris_stats_bind  # noqa: N806
+    with Session(connection) as session:
+        session.query(Bind).filter(Bind.game_platform == 'TOS').delete()
+        session.commit()
+
+
+def downgrade(name: str = '') -> None:
+    if name:
+        return

nonebot_plugin_tetris_stats/db/__init__.py

@@ -1,3 +1,5 @@
+from enum import Enum, auto
+
 from nonebot_plugin_orm import AsyncSession
 from sqlalchemy import select
 
@@ -5,6 +7,11 @@ from ..utils.typing import GameType
 from .models import Bind
 
 
+class BindStatus(Enum):
+    SUCCESS = auto()
+    UPDATE = auto()
+
+
 async def query_bind_info(
     session: AsyncSession,
     chat_platform: str,
@@ -27,7 +34,7 @@ async def create_or_update_bind(
     chat_account: str,
     game_platform: GameType,
     game_account: str,
-) -> str:
+) -> BindStatus:
     bind = await query_bind_info(
         session=session,
         chat_platform=chat_platform,
@@ -42,9 +49,9 @@ async def create_or_update_bind(
             game_account=game_account,
         )
         session.add(bind)
-        message = '绑定成功'
+        message = BindStatus.SUCCESS
     else:
         bind.game_account = game_account
-        message = '更新绑定成功'
+        message = BindStatus.UPDATE
     await session.commit()
     return message

nonebot_plugin_tetris_stats/db/models.py

@@ -1,14 +1,56 @@
+from collections.abc import Callable, Sequence
 from datetime import datetime
+from typing import Any
 
 from nonebot.adapters import Message
+from nonebot.compat import PYDANTIC_V2, type_validate_json
 from nonebot_plugin_orm import Model
-from sqlalchemy import JSON, DateTime, PickleType, String
+from pydantic import BaseModel, ValidationError
+from sqlalchemy import JSON, DateTime, Dialect, PickleType, String, TypeDecorator
 from sqlalchemy.orm import Mapped, MappedAsDataclass, mapped_column
+from typing_extensions import override
 
-from ..game_data_processor import ProcessedData, User
+from ..game_data_processor.schemas import BaseProcessedData, BaseUser
 from ..utils.typing import CommandType, GameType
 
 
+class PydanticType(TypeDecorator):
+    impl = JSON
+
+    @override
+    def __init__(self, get_model: Callable[[], Sequence[type[BaseModel]]], *args: Any, **kwargs: Any):
+        self.get_model = get_model
+        super().__init__(*args, **kwargs)
+
+    if PYDANTIC_V2:
+
+        @override
+        def process_bind_param(self, value: Any | None, dialect: Dialect) -> str:
+            # 将 Pydantic 模型实例转换为 JSON
+            if isinstance(value, tuple(self.get_model())):
+                return value.model_dump_json(by_alias=True)  # type: ignore[union-attr]
+            raise TypeError
+    else:
+
+        @override
+        def process_bind_param(self, value: Any | None, dialect: Dialect) -> str:
+            # 将 Pydantic 模型实例转换为 JSON
+            if isinstance(value, tuple(self.get_model())):
+                return value.json(by_alias=True)  # type: ignore[union-attr]
+            raise TypeError
+
+    @override
+    def process_result_value(self, value: Any | None, dialect: Dialect) -> BaseModel:
+        # 将 JSON 转换回 Pydantic 模型实例
+        if isinstance(value, str | bytes):
+            for i in self.get_model():
+                try:
+                    return type_validate_json(i, value)
+                except ValidationError:  # noqa: PERF203
+                    ...
+        raise ValueError
+
+
 class Bind(MappedAsDataclass, Model):
     id: Mapped[int] = mapped_column(init=False, primary_key=True)
     chat_platform: Mapped[str] = mapped_column(String(32), index=True)
@@ -28,6 +70,9 @@ class HistoricalData(MappedAsDataclass, Model):
     game_platform: Mapped[GameType] = mapped_column(String(32), index=True, init=False)
     command_type: Mapped[CommandType] = mapped_column(String(16), index=True, init=False)
     command_args: Mapped[list[str]] = mapped_column(JSON, init=False)
-    game_user: Mapped[User] = mapped_column(PickleType, init=False)
-    processed_data: Mapped[ProcessedData] = mapped_column(PickleType, init=False)
+    user_unique_identifier: Mapped[str] = mapped_column(String(32), index=True, init=False)
+    game_user: Mapped[BaseUser] = mapped_column(PydanticType(get_model=BaseUser.__subclasses__), init=False)
+    processed_data: Mapped[BaseProcessedData] = mapped_column(
+        PydanticType(get_model=BaseProcessedData.__subclasses__), init=False
+    )
     finish_time: Mapped[datetime] = mapped_column(DateTime, init=False)

nonebot_plugin_tetris_stats/game_data_processor/__init__.py

@@ -1,26 +1,20 @@
 from abc import ABC, abstractmethod
-from dataclasses import dataclass
-from datetime import UTC, datetime
+from datetime import datetime, timezone
+from typing import Any
 
+from nonebot.matcher import Matcher
+from nonebot_plugin_alconna import AlcMatches, AlconnaMatcher
+from nonebot_plugin_alconna.uniseg import UniMessage
+from nonebot_plugin_userinfo import UserInfo  # type: ignore[import-untyped]
+
+from ..utils.exception import MessageFormatError
+from ..utils.recorder import Recorder
 from ..utils.typing import CommandType, GameType
+from .schemas import BaseProcessedData as ProcessedData
+from .schemas import BaseRawResponse as RawResponse
+from .schemas import BaseUser as User
 
-
-@dataclass
-class User:
-    """游戏用户"""
-
-
-@dataclass
-class RawResponse:
-    """原始请求数据"""
-
-
-@dataclass
-class ProcessedData:
-    """处理/验证后的数据"""
-
-
-from ..utils.recorder import Recorder  # noqa: E402 避免循环导入
+UTC = timezone.utc
 
 
 class Processor(ABC):
@@ -49,32 +43,56 @@ class Processor(ABC):
         raise NotImplementedError
 
     @abstractmethod
-    async def handle_bind(self, platform: str, account: str) -> str:
+    async def handle_bind(
+        self,
+        platform: str,
+        account: str,
+        bot_info: UserInfo,
+        *args: Any,  # noqa: ANN401
+        **kwargs: Any,  # noqa: ANN401
+    ) -> UniMessage:
         """处理绑定消息"""
         raise NotImplementedError
 
     @abstractmethod
-    async def handle_query(self) -> str:
+    async def handle_query(self) -> UniMessage:
         """处理查询消息"""
         raise NotImplementedError
 
-    @abstractmethod
-    async def generate_message(self) -> str:
-        """生成消息"""
-        raise NotImplementedError
-
     def __del__(self) -> None:
         finish_time = datetime.now(tz=UTC)
+        if Recorder.is_error_event(self.event_id):
+            Recorder.del_error_event(self.event_id)
+            return
         historical_data = Recorder.get_historical_data(self.event_id)
         historical_data.game_platform = self.game_platform
         historical_data.command_type = self.command_type
         historical_data.command_args = self.command_args
+        historical_data.user_unique_identifier = self.user.unique_identifier
         historical_data.game_user = self.user
         historical_data.processed_data = self.processed_data
         historical_data.finish_time = finish_time
         Recorder.update_historical_data(self.event_id, historical_data)
 
 
+def add_default_handlers(matcher: type[AlconnaMatcher]) -> None:
+    @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()
+    async def _(matcher: Matcher, other: Any):  # noqa: ANN401
+        await matcher.finish()
+
+
 from . import (  # noqa: F401, E402
     io_data_processor,
     top_data_processor,

nonebot_plugin_tetris_stats/game_data_processor/io_data_processor/__init__.py

@@ -1,23 +1,29 @@
-from datetime import timedelta
+from datetime import datetime, timedelta, timezone
+from zoneinfo import ZoneInfo
 
-from arclet.alconna import Alconna, Arg, ArgFlag, Args, CommandMeta, Option
+from arclet.alconna import Alconna, AllParam, Arg, ArgFlag, Args, CommandMeta, Option
 from nonebot.adapters import Bot, Event
 from nonebot.matcher import Matcher
-from nonebot_plugin_alconna import AlcMatches, At, on_alconna
+from nonebot_plugin_alconna import At, on_alconna
+from nonebot_plugin_alconna.uniseg import UniMessage
 from nonebot_plugin_orm import get_session
-from sqlalchemy import select
+from nonebot_plugin_userinfo import BotUserInfo, UserInfo  # type: ignore[import-untyped]
+from sqlalchemy import func, select
 
 from ...db import query_bind_info
-from ...utils.exception import MessageFormatError, NeedCatchError
+from ...utils.exception import HandleNotFinishedError, NeedCatchError
 from ...utils.metrics import get_metrics
 from ...utils.platform import get_platform
 from ...utils.typing import Me
+from .. import add_default_handlers
 from ..constant import BIND_COMMAND, QUERY_COMMAND
 from .constant import GAME_TYPE
 from .model import IORank
-from .processor import Processor, User, check_rank_data, identify_user_info
+from .processor import Processor, User, identify_user_info
 from .typing import Rank
 
+UTC = timezone.utc
+
 alc = on_alconna(
     Alconna(
         'io',
@@ -65,6 +71,7 @@ alc = on_alconna(
             dest='rank',
             help_text='查询 IO 段位信息',
         ),
+        Arg('other', AllParam, flags=[ArgFlag.HIDDEN, ArgFlag.OPTIONAL]),
         meta=CommandMeta(
             description='查询 TETR.IO 的信息',
             example='io绑定scdhh\nio查我\niorankx',
@@ -81,16 +88,15 @@ alc.shortcut('fkosk', {'command': 'io查', 'args': ['我']})
 
 
 @alc.assign('bind')
-async def _(bot: Bot, event: Event, matcher: Matcher, account: User):
-    proc = Processor(
-        event_id=id(event),
-        user=account,
-        command_args=[],
-    )
+async def _(bot: Bot, event: Event, matcher: Matcher, account: User, bot_info: UserInfo = BotUserInfo()):  # noqa: B008
+    proc = Processor(event_id=id(event), user=account, command_args=[])
     try:
-        await matcher.finish(await proc.handle_bind(platform=get_platform(bot), account=event.get_user_id()))
+        await (
+            await proc.handle_bind(platform=get_platform(bot), account=event.get_user_id(), bot_info=bot_info)
+        ).finish()
     except NeedCatchError as e:
-        await matcher.finish(str(e))
+        await matcher.send(str(e))
+        raise HandleNotFinishedError from e
 
 
 @alc.assign('query')
@@ -111,9 +117,10 @@ async def _(bot: Bot, event: Event, matcher: Matcher, target: At | Me):
         command_args=[],
     )
     try:
-        await matcher.finish(message + await proc.handle_query())
+        await (UniMessage(message) + await proc.handle_query()).finish()
     except NeedCatchError as e:
-        await matcher.finish(str(e))
+        await matcher.send(str(e))
+        raise HandleNotFinishedError from e
 
 
 @alc.assign('query')
@@ -124,27 +131,39 @@ async def _(event: Event, matcher: Matcher, account: User):
         command_args=[],
     )
     try:
-        await matcher.finish(await proc.handle_query())
+        await (await proc.handle_query()).finish()
     except NeedCatchError as e:
-        await matcher.finish(str(e))
+        await matcher.send(str(e))
+        raise HandleNotFinishedError from e
 
 
 @alc.assign('rank')
-async def _(event: Event, matcher: Matcher, rank: Rank):
+async def _(matcher: Matcher, rank: Rank):
     if rank == 'z':
         await matcher.finish('暂不支持查询未知段位')
-    try:
-        await check_rank_data()
-    except NeedCatchError as e:
-        await matcher.finish(str(f'段位信息获取失败\n{e}'))
     async with get_session() as session:
-        data = (
-            await session.scalars(select(IORank).where(IORank.rank == rank).order_by(IORank.id.desc()).limit(5))
-        ).all()
-    latest_data = data[0]
-    message = f'{rank.upper()} 段 分数线 {latest_data.tr_line:.2f} TR, {latest_data.player_count} 名玩家\n'
-    if len(data) > 1:
-        message += f'对比 {(latest_data.create_time-data[-1].create_time).total_seconds()/3600:.2f} 小时前趋势: {f"↑{difference:.2f}" if (difference:=latest_data.tr_line-data[-1].tr_line) > 0 else f"↓{-difference:.2f}" if difference < 0 else "→"}'
+        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)
@@ -169,19 +188,9 @@ async def _(event: Event, matcher: Matcher, rank: Rank):
         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.create_time+timedelta(hours=8)).strftime("%Y-%m-%d %H:%M:%S")}'
+        f'数据更新时间: {latest_data.update_time.replace(tzinfo=UTC).astimezone(ZoneInfo("Asia/Shanghai")).strftime("%Y-%m-%d %H:%M:%S")}'
     )
     await matcher.finish(message)
 
 
-@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:
-        await matcher.finish(
-            f'{matches.error_info!r}\n' if matches.error_info is not None else '' + '输入"io --help"查看帮助'
-        )
+add_default_handlers(alc)

nonebot_plugin_tetris_stats/game_data_processor/io_data_processor/cache.py

@@ -0,0 +1,30 @@
+from datetime import datetime, timezone
+
+from aiocache import Cache as ACache  # type: ignore[import-untyped]
+from nonebot.compat import type_validate_json
+from nonebot.log import logger
+
+from ...utils.request import Request
+from .schemas.base import FailedModel, SuccessModel
+
+UTC = timezone.utc
+
+
+class Cache:
+    cache = ACache(ACache.MEMORY)
+
+    @classmethod
+    async def get(cls, url: str) -> bytes:
+        cached_data = await cls.cache.get(url)
+        if cached_data is None:
+            response_data = await Request.request(url)
+            parsed_data: SuccessModel | FailedModel = type_validate_json(SuccessModel | FailedModel, response_data)  # type: ignore[arg-type]
+            if isinstance(parsed_data, SuccessModel):
+                await cls.cache.add(
+                    url,
+                    response_data,
+                    (parsed_data.cache.cached_until - datetime.now(UTC)).total_seconds(),
+                )
+            return response_data
+        logger.debug(f'{url}: Cache hit!')
+        return cached_data

nonebot_plugin_tetris_stats/game_data_processor/io_data_processor/constant.py

@@ -1,7 +1,8 @@
-from ...utils.typing import GameType
+from typing import Literal
+
 from .typing import Rank
 
-GAME_TYPE: GameType = 'IO'
+GAME_TYPE: Literal['IO'] = 'IO'
 BASE_URL = 'https://ch.tetr.io/api/'
 RANK_PERCENTILE: dict[Rank, float] = {
     'x': 1,
@@ -22,3 +23,5 @@ RANK_PERCENTILE: dict[Rank, float] = {
     'd+': 97.5,
     'd': 100,
 }
+TR_MIN = 0
+TR_MAX = 25000

nonebot_plugin_tetris_stats/game_data_processor/io_data_processor/model.py

@@ -1,4 +1,4 @@
-from datetime import UTC, datetime
+from datetime import datetime, timezone
 
 from nonebot_plugin_orm import Model
 from sqlalchemy import JSON, DateTime, String
@@ -6,6 +6,8 @@ from sqlalchemy.orm import Mapped, MappedAsDataclass, mapped_column
 
 from .typing import Rank
 
+UTC = timezone.utc
+
 
 class IORank(MappedAsDataclass, Model):
     id: Mapped[int] = mapped_column(init=False, primary_key=True)
@@ -21,9 +23,8 @@ class IORank(MappedAsDataclass, Model):
     high_pps: Mapped[tuple[dict[str, str], float]] = mapped_column(JSON)
     high_apm: Mapped[tuple[dict[str, str], float]] = mapped_column(JSON)
     high_vs: Mapped[tuple[dict[str, str], float]] = mapped_column(JSON)
-    create_time: Mapped[datetime] = mapped_column(
+    update_time: Mapped[datetime] = mapped_column(
         DateTime,
-        default=lambda: datetime.now(tz=UTC),
         index=True,
-        init=False,
     )
+    file_hash: Mapped[str | None] = mapped_column(String(128), index=True)

nonebot_plugin_tetris_stats/game_data_processor/io_data_processor/processor.py

@@ -1,63 +1,57 @@
+from asyncio import gather
 from collections import defaultdict
 from collections.abc import Callable
-from dataclasses import asdict, dataclass
-from datetime import UTC, datetime, timedelta
-from math import floor
+from datetime import datetime, timedelta, timezone
+from hashlib import md5, sha512
+from math import ceil, floor
 from re import match
 from statistics import mean
+from typing import Literal
+from urllib.parse import urlunparse
+from zoneinfo import ZoneInfo
 
+from aiofiles import open
 from nonebot import get_driver
+from nonebot.compat import type_validate_json
+from nonebot.utils import run_sync
+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 pydantic import parse_raw_as
+from nonebot_plugin_userinfo import UserInfo as NBUserInfo  # type: ignore[import-untyped]
 from sqlalchemy import select
+from typing_extensions import override
+from zstandard import ZstdCompressor
 
-from ...db import create_or_update_bind
+from ...db import BindStatus, create_or_update_bind
+from ...db.models import HistoricalData
+from ...utils.avatar import get_avatar
 from ...utils.exception import MessageFormatError, RequestError, WhatTheFuckError
-from ...utils.request import Request, splice_url
-from ...utils.typing import GameType
-from .. import ProcessedData as ProcessedDataMeta
+from ...utils.host import HostPage, get_self_netloc
+from ...utils.render import Bind, TETRIOInfo, render
+from ...utils.request import splice_url
+from ...utils.retry import retry
+from ...utils.screenshot import screenshot
 from .. import Processor as ProcessorMeta
-from .. import RawResponse as RawResponseMeta
-from .. import User as UserMeta
-from .constant import BASE_URL, GAME_TYPE, RANK_PERCENTILE
+from .cache import Cache
+from .constant import BASE_URL, GAME_TYPE, RANK_PERCENTILE, TR_MAX, TR_MIN
 from .model import IORank
-from .schemas.league_all import FailedModel as LeagueAllFailed
+from .schemas.base import FailedModel
 from .schemas.league_all import LeagueAll
 from .schemas.league_all import ValidUser as LeagueAllUser
-from .schemas.user_info import FailedModel as InfoFailed
-from .schemas.user_info import (
-    NeverPlayedLeague,
-    NeverRatedLeague,
-    UserInfo,
-)
+from .schemas.response import ProcessedData, RawResponse
+from .schemas.user import User
+from .schemas.user_info import NeverPlayedLeague, NeverRatedLeague, RatedLeague, UserInfo
 from .schemas.user_info import SuccessModel as InfoSuccess
-from .schemas.user_records import FailedModel as RecordsFailed
-from .schemas.user_records import SoloRecord, UserRecords
+from .schemas.user_records import MultiRecord, SoloRecord, UserRecords
 from .schemas.user_records import SuccessModel as RecordsSuccess
 from .typing import Rank
 
+UTC = timezone.utc
+
 driver = get_driver()
 
 
-@dataclass
-class User(UserMeta):
-    ID: str | None = None
-    name: str | None = None
-
-
-@dataclass
-class RawResponse(RawResponseMeta):
-    user_info: bytes | None = None
-    user_records: bytes | None = None
-
-
-@dataclass
-class ProcessedData(ProcessedDataMeta):
-    user_info: InfoSuccess | None = None
-    user_records: RecordsSuccess | None = None
-
-
 def identify_user_info(info: str) -> User | MessageFormatError:
     if match(r'^[a-f0-9]{24}$', info):
         return User(ID=info)
@@ -66,40 +60,292 @@ def identify_user_info(info: str) -> User | MessageFormatError:
     return MessageFormatError('用户名/ID不合法')
 
 
+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: TETRIOInfo.TetraLeagueHistory.Data,
+    behind_point: TETRIOInfo.TetraLeagueHistory.Data,
+    point_time: datetime,
+) -> TETRIOInfo.TetraLeagueHistory.Data:
+    """根据给出的 previous_point 和 behind_point, 推算 point_time 点处的数据
+
+    Args:
+        previous_point (TETRIOInfo.TetraLeagueHistory.Data): 前面的数据点
+        behind_point (TETRIOInfo.TetraLeagueHistory.Data): 后面的数据点
+        point_time (datetime): 要推算的点的位置
+
+    Returns:
+        TETRIOInfo.TetraLeagueHistory.Data: 要推算的点的数据
+    """
+    # 求两个点的斜率
+    slope = (behind_point.tr - previous_point.tr) / (
+        datetime.timestamp(behind_point.record_at) - datetime.timestamp(previous_point.record_at)
+    )
+    return TETRIOInfo.TetraLeagueHistory.Data(
+        record_at=point_time,
+        tr=previous_point.tr + slope * (datetime.timestamp(point_time) - datetime.timestamp(previous_point.record_at)),
+    )
+
+
 class Processor(ProcessorMeta):
     user: User
     raw_response: RawResponse
     processed_data: ProcessedData
 
+    @override
     def __init__(self, event_id: int, user: User, command_args: list[str]) -> None:
         super().__init__(event_id, user, command_args)
         self.raw_response = RawResponse()
         self.processed_data = ProcessedData()
 
     @property
-    def game_platform(self) -> GameType:
+    @override
+    def game_platform(self) -> Literal['IO']:
         return GAME_TYPE
 
-    async def handle_bind(self, platform: str, account: str) -> str:
+    @override
+    async def handle_bind(self, platform: str, account: str, bot_info: NBUserInfo) -> UniMessage:
         """处理绑定消息"""
         self.command_type = 'bind'
         await self.get_user()
         if self.user.ID is None:
             raise  # FIXME: 不知道怎么才能把这类型给变过来了
         async with get_session() as session:
-            return await create_or_update_bind(
+            bind_status = await create_or_update_bind(
                 session=session,
                 chat_platform=platform,
                 chat_account=account,
                 game_platform=GAME_TYPE,
                 game_account=self.user.ID,
             )
+        bot_avatar = await get_avatar(bot_info, 'Data URI', '../../static/logo/logo.svg')
+        user_info = await self.get_user_info()
+        if bind_status in (BindStatus.SUCCESS, BindStatus.UPDATE):
+            async with HostPage(
+                await render(
+                    'binding',
+                    Bind(
+                        platform='TETR.IO',
+                        status='unknown',
+                        user=Bind.People(
+                            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 f'{{"type":"identicon","hash":"{md5(user_info.data.user.id.encode()).hexdigest()}"}}',  # noqa: S324
+                            name=user_info.data.user.username.upper(),
+                        ),
+                        bot=Bind.People(
+                            avatar=bot_avatar,
+                            name=bot_info.user_name,
+                        ),
+                        command='io查我',
+                    ),
+                )
+            ) as page_hash:
+                message = UniMessage.image(
+                    raw=await screenshot(urlunparse(('http', get_self_netloc(), f'/host/{page_hash}.html', '', '', '')))
+                )
+        return message
 
-    async def handle_query(self) -> str:
+    @override
+    async def handle_query(self) -> UniMessage:
         """处理查询消息"""
         self.command_type = 'query'
         await self.get_user()
-        return await self.generate_message()
+        user_info, user_records = await gather(self.get_user_info(), self.get_user_records())
+        sprint = user_records.data.records.sprint
+        blitz = user_records.data.records.blitz
+        if isinstance(sprint.record, MultiRecord) or isinstance(blitz.record, MultiRecord):
+            raise WhatTheFuckError('单人游戏记录是多人游戏记录')
+        try:
+            return UniMessage.image(raw=await self.make_query_image(self.user, user_info, sprint.record, blitz.record))
+        except TypeError:
+            ...
+        # fallback
+        league = user_info.data.user.league
+        user_name = user_info.data.user.username.upper()
+        ret_message = ''
+        if isinstance(league, NeverPlayedLeague):
+            ret_message += f'用户 {user_name} 没有排位统计数据'
+        else:
+            if isinstance(league, NeverRatedLeague):
+                ret_message += f'用户 {user_name} 暂未完成定级赛, 最近十场的数据:'
+            else:
+                if league.rank == 'z':
+                    ret_message += f'用户 {user_name} 暂无段位, {round(league.rating,2)} TR'
+                else:
+                    ret_message += (
+                        f'{league.rank.upper()} 段用户 {user_name} {round(league.rating,2)} TR (#{league.standing})'
+                    )
+                ret_message += f', 段位分 {round(league.glicko,2)}±{round(league.rd,2)}, 最近十场的数据:'
+            lpm = league.pps * 24
+            ret_message += f"\nL'PM: {round(lpm, 2)} ( {league.pps} pps )"
+            ret_message += f'\nAPM: {league.apm} ( x{round(league.apm/lpm,2)} )'
+            if league.vs is not None:
+                adpm = league.vs * 0.6
+                ret_message += f'\nADPM: {round(adpm,2)} ( x{round(adpm/lpm,2)} ) ( {league.vs}vs )'
+        if sprint.record is not None:
+            ret_message += f'\n40L: {round(sprint.record.endcontext.final_time/1000,2)}s'
+            ret_message += f' ( #{sprint.rank} )' if sprint.rank is not None else ''
+        if blitz.record is not None:
+            ret_message += f'\nBlitz: {blitz.record.endcontext.score}'
+            ret_message += f' ( #{blitz.rank} )' if blitz.rank is not None else ''
+        return UniMessage(ret_message)
+
+    @staticmethod
+    async def query_historical_data(user: User, user_info: InfoSuccess) -> list[TETRIOInfo.TetraLeagueHistory.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(HistoricalData)
+                    .where(HistoricalData.trigger_time >= start_time)
+                    .where(HistoricalData.game_platform == GAME_TYPE)
+                    .where(HistoricalData.user_unique_identifier == user.unique_identifier)
+                )
+            ).all()
+            if historical_data:
+                extra = (
+                    await session.scalars(
+                        select(HistoricalData)
+                        .where(HistoricalData.game_platform == GAME_TYPE)
+                        .where(HistoricalData.user_unique_identifier == user.unique_identifier)
+                        .order_by(HistoricalData.id.desc())
+                        .where(HistoricalData.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)
+
+        histories = [
+            TETRIOInfo.TetraLeagueHistory.Data(
+                record_at=i.processed_data.user_info.cache.cached_at.astimezone(ZoneInfo('Asia/Shanghai')),
+                tr=i.processed_data.user_info.data.user.league.rating,
+            )
+            for i in historical_data
+            if isinstance(i.processed_data, ProcessedData)
+            and i.processed_data.user_info is not None
+            and isinstance(i.processed_data.user_info.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],
+                    TETRIOInfo.TetraLeagueHistory.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, TETRIOInfo.TetraLeagueHistory.Data(record_at=today - forward, tr=histories[0].tr))
+        return histories
+
+    @staticmethod
+    async def make_query_image(
+        user: User, user_info: InfoSuccess, 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 Processor.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=TETRIOInfo.User(
+                        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 f'{{"type":"identicon","hash":"{md5(user_info.data.user.id.encode()).hexdigest()}"}}',  # noqa: S324
+                        name=user_name,
+                        bio=user_info.data.user.bio,
+                    ),
+                    ranking=TETRIOInfo.Ranking(
+                        rating=round(league.glicko, 2),
+                        rd=round(league.rd, 2),
+                    ),
+                    tetra_league=TETRIOInfo.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=TETRIOInfo.TetraLeagueHistory(
+                        data=histories,
+                        split_interval=split_value,
+                        min_tr=value_min,
+                        max_tr=value_max,
+                        offset=offset,
+                    ),
+                    radar=TETRIOInfo.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', '', '', '')))
 
     async def get_user(self) -> None:
         """
@@ -113,11 +359,11 @@ class Processor(ProcessorMeta):
     async def get_user_info(self) -> InfoSuccess:
         """获取用户数据"""
         if self.processed_data.user_info is None:
-            self.raw_response.user_info = await Request.request(
+            self.raw_response.user_info = await Cache.get(
                 splice_url([BASE_URL, 'users/', f'{self.user.ID or self.user.name}'])
             )
-            user_info: UserInfo = parse_raw_as(UserInfo, self.raw_response.user_info)  # type: ignore[arg-type]
-            if isinstance(user_info, InfoFailed):
+            user_info: UserInfo = type_validate_json(UserInfo, self.raw_response.user_info)  # type: ignore[arg-type]
+            if isinstance(user_info, FailedModel):
                 raise RequestError(f'用户信息请求错误:\n{user_info.error}')
             self.processed_data.user_info = user_info
         return self.processed_data.user_info
@@ -125,73 +371,25 @@ class Processor(ProcessorMeta):
     async def get_user_records(self) -> RecordsSuccess:
         """获取Solo数据"""
         if self.processed_data.user_records is None:
-            self.raw_response.user_records = await Request.request(
-                splice_url(
-                    [
-                        BASE_URL,
-                        'users/',
-                        f'{self.user.ID or self.user.name}/',
-                        'records',
-                    ]
-                )
+            self.raw_response.user_records = await Cache.get(
+                splice_url([BASE_URL, 'users/', f'{self.user.ID or self.user.name}/', 'records'])
             )
-            user_records: UserRecords = parse_raw_as(
-                UserRecords,  # type: ignore[arg-type]
-                self.raw_response.user_records,
-            )
-            if isinstance(user_records, RecordsFailed):
+            user_records: UserRecords = type_validate_json(UserRecords, self.raw_response.user_records)  # type: ignore[arg-type]
+            if isinstance(user_records, FailedModel):
                 raise RequestError(f'用户Solo数据请求错误:\n{user_records.error}')
             self.processed_data.user_records = user_records
         return self.processed_data.user_records
 
-    async def generate_message(self) -> str:
-        """生成消息"""
-        user_info = await self.get_user_info()
-        user_name = user_info.data.user.username.upper()
-        league = user_info.data.user.league
-        ret_message = ''
-        if isinstance(league, NeverPlayedLeague):
-            ret_message += f'用户 {user_name} 没有排位统计数据'
-        else:
-            if isinstance(league, NeverRatedLeague):
-                ret_message += f'用户 {user_name} 暂未完成定级赛, 最近十场的数据:'
-            elif league.rank == 'z':
-                ret_message += f'用户 {user_name} 暂无段位, {round(league.rating,2)} TR'
-            else:
-                ret_message += (
-                    f'{league.rank.upper()} 段用户 {user_name} {round(league.rating,2)} TR (#{league.standing})'
-                )
-                ret_message += f', 段位分 {round(league.glicko,2)}±{round(league.rd,2)}, 最近十场的数据:'
-            lpm = league.pps * 24
-            ret_message += f"\nL'PM: {round(lpm, 2)} ( {league.pps} pps )"
-            ret_message += f'\nAPM: {league.apm} ( x{round(league.apm/(league.pps*24),2)} )'
-            if league.vs is not None:
-                adpm = league.vs * 0.6
-                ret_message += f'\nADPM: {round(adpm,2)} ( x{round(adpm/lpm,2)} ) ( {league.vs}vs )'
-        user_records = await self.get_user_records()
-        sprint = user_records.data.records.sprint
-        if sprint.record is not None:
-            if not isinstance(sprint.record, SoloRecord):
-                raise WhatTheFuckError('40L记录不是单人记录')
-            ret_message += f'\n40L: {round(sprint.record.endcontext.final_time/1000,2)}s'
-            ret_message += f' ( #{sprint.rank} )' if sprint.rank is not None else ''
-        blitz = user_records.data.records.blitz
-        if blitz.record is not None:
-            if not isinstance(blitz.record, SoloRecord):
-                raise WhatTheFuckError('Blitz记录不是单人记录')
-            ret_message += f'\nBlitz: {blitz.record.endcontext.score}'
-            ret_message += f' ( #{blitz.rank} )' if blitz.rank is not None else ''
-        return ret_message
-
 
 @scheduler.scheduled_job('cron', hour='0,6,12,18', minute=0)
+@retry(exception_type=RequestError, delay=timedelta(minutes=15))
 async def get_io_rank_data() -> None:
-    league_all: LeagueAll = parse_raw_as(
+    league_all: LeagueAll = type_validate_json(
         LeagueAll,  # type: ignore[arg-type]
-        await Request.request(splice_url([BASE_URL, 'users/lists/league/all'])),
+        (data := await Cache.get(splice_url([BASE_URL, 'users/lists/league/all']))),
     )
-    if isinstance(league_all, LeagueAllFailed):
-        raise RequestError(f'用户Solo数据请求错误:\n{league_all.error}')
+    if isinstance(league_all, FailedModel):
+        raise RequestError(f'排行榜数据请求错误:\n{league_all.error}')
 
     def pps(user: LeagueAllUser) -> float:
         return user.league.pps
@@ -214,7 +412,11 @@ async def get_io_rank_data() -> None:
         sort: Callable[[list[LeagueAllUser], Callable[[LeagueAllUser], float]], LeagueAllUser],
     ) -> tuple[dict[str, str], float]:
         user = sort(users, field)
-        return asdict(User(ID=user.id, name=user.username)), field(user)
+        return User(ID=user.id, name=user.username).dict(), field(user)
+
+    data_hash: str | None = await run_sync((await run_sync(sha512)(data)).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)(data))
 
     users = [i for i in league_all.data.users if isinstance(i, LeagueAllUser)]
     rank_to_users: defaultdict[Rank, list[LeagueAllUser]] = defaultdict(list)
@@ -239,6 +441,8 @@ async def get_io_rank_data() -> None:
                 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_all.cache.cached_at,
+                file_hash=data_hash,
             )
         )
     async with get_session() as session:
@@ -247,8 +451,8 @@ async def get_io_rank_data() -> None:
 
 
 @driver.on_startup
-async def check_rank_data() -> None:
+async def _() -> None:
     async with get_session() as session:
-        latest_time = await session.scalar(select(IORank.create_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_io_rank_data()
+        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_io_rank_data()

nonebot_plugin_tetris_stats/game_data_processor/io_data_processor/schemas/league_all.py

@@ -5,9 +5,19 @@ from .base import FailedModel
 from .base import SuccessModel as BaseSuccessModel
 
 
+class _User(BaseModel):
+    id: str = Field(..., alias='_id')
+    username: str
+    role: str
+    xp: float
+    supporter: bool
+    verified: bool
+    country: str | None = None
+
+
 class SuccessModel(BaseSuccessModel):
     class Data(BaseModel):
-        class ValidUser(BaseModel):
+        class ValidUser(_User):
             class League(BaseModel):
                 gamesplayed: int
                 gameswon: int
@@ -21,37 +31,23 @@ class SuccessModel(BaseSuccessModel):
                 vs: float
                 decaying: bool
 
-            id: str = Field(..., alias='_id')
-            username: str
-            role: str
-            xp: float
             league: League
-            supporter: bool
-            verified: bool
-            country: str | None
 
-        class InvalidUser(BaseModel):
+        class InvalidUser(_User):
             class League(BaseModel):
                 gamesplayed: int
                 gameswon: int
                 rating: float
-                glicko: float | None
-                rd: float | None
+                glicko: float | None = None
+                rd: float | None = None
                 rank: Rank
                 bestrank: Rank
-                apm: float | None
-                pps: float | None
-                vs: float | None
+                apm: float | None = None
+                pps: float | None = None
+                vs: float | None = None
                 decaying: bool
 
-            id: str = Field(..., alias='_id')
-            username: str
-            role: str
-            xp: float
             league: League
-            supporter: bool
-            verified: bool
-            country: str | None
 
         users: list[ValidUser | InvalidUser]
 

nonebot_plugin_tetris_stats/game_data_processor/io_data_processor/schemas/response.py

@@ -0,0 +1,21 @@
+from typing import Literal
+
+from ... import ProcessedData as ProcessedDataMeta
+from ... import RawResponse as RawResponseMeta
+from ..constant import GAME_TYPE
+from .user_info import SuccessModel as InfoSuccess
+from .user_records import SuccessModel as RecordsSuccess
+
+
+class RawResponse(RawResponseMeta):
+    platform: Literal['IO'] = GAME_TYPE
+
+    user_info: bytes | None = None
+    user_records: bytes | None = None
+
+
+class ProcessedData(ProcessedDataMeta):
+    platform: Literal['IO'] = GAME_TYPE
+
+    user_info: InfoSuccess | None = None
+    user_records: RecordsSuccess | None = None

nonebot_plugin_tetris_stats/game_data_processor/io_data_processor/schemas/user.py

@@ -0,0 +1,17 @@
+from typing import Literal
+
+from ...schemas import BaseUser
+from ..constant import GAME_TYPE
+
+
+class User(BaseUser):
+    platform: Literal['IO'] = GAME_TYPE
+
+    ID: str | None = None
+    name: str | None = None
+
+    @property
+    def unique_identifier(self) -> str:
+        if self.ID is None:
+            raise ValueError('不完整的User!')
+        return self.ID

nonebot_plugin_tetris_stats/game_data_processor/io_data_processor/schemas/user_info.py

@@ -14,7 +14,8 @@ class SuccessModel(BaseSuccessModel):
             class Badge(BaseModel):
                 id: str
                 label: str
-                ts: datetime | None
+                group: str | None = None
+                ts: datetime | Literal[False] | None = None
 
             class NeverPlayedLeague(BaseModel):
                 gamesplayed: Literal[0]
@@ -29,9 +30,9 @@ class SuccessModel(BaseSuccessModel):
                 prev_at: Literal[-1]
                 percentile: Literal[-1]
                 percentile_rank: Literal['z']
-                apm: None
-                pps: None
-                vs: None
+                apm: None = Field(None)
+                pps: None = Field(None)
+                vs: None = Field(None)
                 decaying: bool
 
             class NeverRatedLeague(BaseModel):
@@ -60,8 +61,8 @@ class SuccessModel(BaseSuccessModel):
                 bestrank: Rank
                 standing: int
                 standing_local: int
-                next_rank: Rank | None
-                prev_rank: Rank | None
+                next_rank: Rank | None = None
+                prev_rank: Rank | None = None
                 next_at: int
                 prev_at: int
                 percentile: float
@@ -70,7 +71,7 @@ class SuccessModel(BaseSuccessModel):
                 rd: float
                 apm: float
                 pps: float
-                vs: float | None
+                vs: float | None = None
                 decaying: bool
 
             class Connections(BaseModel):
@@ -78,41 +79,41 @@ class SuccessModel(BaseSuccessModel):
                     id: str
                     username: str
 
-                discord: Discord | None
+                discord: Discord | None = None
 
             class Distinguishment(BaseModel):
-                type: str  # noqa: A003
+                type: str
 
             id: str = Field(..., alias='_id')
             username: str
             role: Literal['anon', 'user', 'bot', 'halfmod', 'mod', 'admin', 'sysop', 'banned']
-            ts: datetime | None
-            botmaster: str | None
+            ts: datetime | None = None
+            botmaster: str | None = None
             badges: list[Badge]
             xp: float
             gamesplayed: int
             gameswon: int
             gametime: float
-            country: str | None
-            badstanding: bool | None
-            supporter: bool | None  # osk说是必有, 但实际上不是 fk osk
+            country: str | None = None
+            badstanding: bool | None = None
+            supporter: bool | None = None  # osk说是必有, 但实际上不是 fk osk
             supporter_tier: int
             verified: bool
             league: NeverPlayedLeague | NeverRatedLeague | RatedLeague
-            avatar_revision: int | None
+            avatar_revision: int | None = None
             """This user's avatar ID. Get their avatar at
 
             https://tetr.io/user-content/avatars/{ USERID }.jpg?rv={ AVATAR_REVISION }"""
-            banner_revision: int | None
+            banner_revision: int | None = None
             """This user's banner ID. Get their banner at
 
             https://tetr.io/user-content/banners/{ USERID }.jpg?rv={ BANNER_REVISION }
 
             Ignore this field if the user is not a supporter."""
-            bio: str | None
+            bio: str | None = None
             connections: Connections
-            friend_count: int
-            distinguishment: Distinguishment | None
+            friend_count: int | None = None
+            distinguishment: Distinguishment | None = None
 
         user: User
 

nonebot_plugin_tetris_stats/game_data_processor/io_data_processor/schemas/user_records.py

@@ -12,13 +12,14 @@ class EndContext(BaseModel):
         zero: bool
         locked: bool
         prev: int
-        frameoffset: int
+        frameoffset: int | None = None
 
     class Clears(BaseModel):
         singles: int
         doubles: int
         triples: int
         quads: int
+        pentas: int | None = None
         realtspins: int
         minitspins: int
         minitspinsingles: int
@@ -32,7 +33,7 @@ class EndContext(BaseModel):
     class Garbage(BaseModel):
         sent: int
         received: int
-        attack: int
+        attack: int | None = None
         cleared: int
 
     class Finesse(BaseModel):
@@ -45,18 +46,18 @@ class EndContext(BaseModel):
     level_lines: int
     level_lines_needed: int
     inputs: int
-    holds: int
+    holds: int | None = None
     time: Time
     score: int
-    zenlevel: int
-    zenprogress: int
+    zenlevel: int | None = None
+    zenprogress: int | None = None
     level: int
     combo: int
-    currentcombopower: int | None  # WTF
+    currentcombopower: int | None = None  # WTF
     topcombo: int
     btb: int
     topbtb: int
-    currentbtbchainpower: int | None  # WTF * 2
+    currentbtbchainpower: int | None = None  # WTF * 2
     tspins: int
     piecesplaced: int
     clears: Clears
@@ -67,45 +68,37 @@ class EndContext(BaseModel):
     gametype: str
 
 
-class BaseModeRecord(BaseModel):
-    class SoloRecord(BaseModel):
-        class User(BaseModel):
-            id: str = Field(..., alias='_id')
-            username: str
+class _User(BaseModel):
+    id: str = Field(..., alias='_id')
+    username: str
 
-        id: str = Field(..., alias='_id')
-        stream: str
-        replayid: str
-        user: User
-        ts: datetime
-        ismulti: bool | None
+
+class _Record(BaseModel):
+    id: str = Field(..., alias='_id')
+    stream: str
+    replayid: str
+    user: _User
+    ts: datetime
+    ismulti: bool | None = None
+
+
+class BaseModeRecord(BaseModel):
+    class SoloRecord(_Record):
         endcontext: EndContext
 
-    class MultiRecord(BaseModel):
-        class User(BaseModel):
-            id: str = Field(..., alias='_id')
-            username: str
-
-        id: str = Field(..., alias='_id')
-        stream: str
-        replayid: str
-        user: User
-        ts: datetime
-        ismulti: bool | None
+    class MultiRecord(_Record):
         endcontext: list[EndContext]
 
-    record: SoloRecord | MultiRecord | None
-    rank: int | None
+    record: SoloRecord | MultiRecord | None = None
+    rank: int | None = None
 
 
 class SuccessModel(BaseSuccessModel):
     class Data(BaseModel):
         class Records(BaseModel):
-            class Sprint(BaseModeRecord):
-                ...
+            class Sprint(BaseModeRecord): ...
 
-            class Blitz(BaseModeRecord):
-                ...
+            class Blitz(BaseModeRecord): ...
 
             sprint: Sprint = Field(..., alias='40l')
             blitz: Blitz

nonebot_plugin_tetris_stats/game_data_processor/schemas.py

@@ -0,0 +1,31 @@
+from abc import ABC, abstractmethod
+
+from pydantic import BaseModel
+
+from ..utils.typing import GameType
+
+
+class Base(BaseModel):
+    platform: GameType
+
+
+class BaseUser(ABC, Base):
+    """游戏用户"""
+
+    def __eq__(self, __value: object) -> bool:
+        if isinstance(__value, BaseUser):
+            return self.unique_identifier == __value.unique_identifier
+        return False
+
+    @property
+    @abstractmethod
+    def unique_identifier(self) -> str:
+        raise NotImplementedError
+
+
+class BaseRawResponse(Base):
+    """原始请求数据"""
+
+
+class BaseProcessedData(Base):
+    """处理/验证后的数据"""

nonebot_plugin_tetris_stats/game_data_processor/top_data_processor/__init__.py

@@ -1,13 +1,16 @@
-from arclet.alconna import Alconna, Arg, ArgFlag, Args, CommandMeta, Option
+from arclet.alconna import Alconna, AllParam, Arg, ArgFlag, Args, CommandMeta, Option
 from nonebot.adapters import Bot, Event
 from nonebot.matcher import Matcher
-from nonebot_plugin_alconna import AlcMatches, At, on_alconna
+from nonebot_plugin_alconna import At, on_alconna
+from nonebot_plugin_alconna.uniseg import UniMessage
 from nonebot_plugin_orm import get_session
+from nonebot_plugin_userinfo import BotUserInfo, EventUserInfo, UserInfo  # type: ignore[import-untyped]
 
 from ...db import query_bind_info
-from ...utils.exception import MessageFormatError, NeedCatchError
+from ...utils.exception import HandleNotFinishedError, NeedCatchError
 from ...utils.platform import get_platform
 from ...utils.typing import Me
+from .. import add_default_handlers
 from ..constant import BIND_COMMAND, QUERY_COMMAND
 from .constant import GAME_TYPE
 from .processor import Processor, User, identify_user_info
@@ -51,6 +54,7 @@ alc = on_alconna(
             dest='query',
             help_text='查询 TOP 游戏信息',
         ),
+        Arg('other', AllParam, flags=[ArgFlag.HIDDEN, ArgFlag.OPTIONAL]),
         meta=CommandMeta(
             description='查询 TetrisOnline波兰服 的信息',
             example='top绑定scdhh\ntop查我',
@@ -65,16 +69,28 @@ alc = on_alconna(
 
 
 @alc.assign('bind')
-async def _(bot: Bot, event: Event, matcher: Matcher, account: User):
+async def _(  # noqa: PLR0913
+    bot: Bot,
+    event: Event,
+    matcher: Matcher,
+    account: User,
+    bot_info: UserInfo = BotUserInfo(),  # noqa: B008
+    user_info: UserInfo = EventUserInfo(),  # noqa: B008
+):
     proc = Processor(
         event_id=id(event),
         user=account,
         command_args=[],
     )
     try:
-        await matcher.finish(await proc.handle_bind(platform=get_platform(bot), account=event.get_user_id()))
+        await (
+            await proc.handle_bind(
+                platform=get_platform(bot), account=event.get_user_id(), bot_info=bot_info, user_info=user_info
+            )
+        ).finish()
     except NeedCatchError as e:
-        await matcher.finish(str(e))
+        await matcher.send(str(e))
+        raise HandleNotFinishedError from e
 
 
 @alc.assign('query')
@@ -95,9 +111,10 @@ async def _(bot: Bot, event: Event, matcher: Matcher, target: At | Me):
         command_args=[],
     )
     try:
-        await matcher.finish(message + await proc.handle_query())
+        await (UniMessage(message) + await proc.handle_query()).finish()
     except NeedCatchError as e:
-        await matcher.finish(str(e))
+        await matcher.send(str(e))
+        raise HandleNotFinishedError from e
 
 
 @alc.assign('query')
@@ -108,19 +125,10 @@ async def _(event: Event, matcher: Matcher, account: User):
         command_args=[],
     )
     try:
-        await matcher.finish(await proc.handle_query())
+        await (await proc.handle_query()).finish()
     except NeedCatchError as e:
-        await matcher.finish(str(e))
+        await matcher.send(str(e))
+        raise HandleNotFinishedError from e
 
 
-@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:
-        await matcher.finish(
-            f'{matches.error_info!r}\n' if matches.error_info is not None else '' + '输入"top --help"查看帮助'
-        )
+add_default_handlers(alc)

nonebot_plugin_tetris_stats/game_data_processor/top_data_processor/constant.py

@@ -1,4 +1,4 @@
-from ...utils.typing import GameType
+from typing import Literal
 
-GAME_TYPE: GameType = 'TOP'
+GAME_TYPE: Literal['TOP'] = 'TOP'
 BASE_URL = 'http://tetrisonline.pl/top/'

nonebot_plugin_tetris_stats/game_data_processor/top_data_processor/processor.py

@@ -2,37 +2,38 @@ from contextlib import suppress
 from dataclasses import dataclass
 from io import StringIO
 from re import match
-from typing import NoReturn
-from urllib.parse import urlencode
+from typing import Literal
+from urllib.parse import urlencode, urlunparse
 
 from lxml import etree
+from nonebot_plugin_alconna.uniseg import UniMessage
 from nonebot_plugin_orm import get_session
+from nonebot_plugin_userinfo import UserInfo  # type: ignore[import-untyped]
 from pandas import read_html
+from typing_extensions import override
 
-from ...db import create_or_update_bind
+from ...db import BindStatus, create_or_update_bind
+from ...utils.avatar import get_avatar
 from ...utils.exception import MessageFormatError, RequestError
+from ...utils.host import HostPage, get_self_netloc
+from ...utils.render import Bind, render
 from ...utils.request import Request, splice_url
-from ...utils.typing import GameType
-from .. import ProcessedData as ProcessedDataMeta
+from ...utils.screenshot import screenshot
 from .. import Processor as ProcessorMeta
-from .. import RawResponse as RawResponseMeta
-from .. import User as UserMeta
+from ..schemas import BaseUser
 from .constant import BASE_URL, GAME_TYPE
+from .schemas.response import ProcessedData, RawResponse
 
 
-@dataclass
-class User(UserMeta):
+class User(BaseUser):
+    platform: Literal['TOP'] = GAME_TYPE
+
     name: str
 
-
-@dataclass
-class RawResponse(RawResponseMeta):
-    user_profile: bytes | None = None
-
-
-@dataclass
-class ProcessedData(ProcessedDataMeta):
-    user_profile: str | None = None
+    @property
+    @override
+    def unique_identifier(self) -> str:
+        return self.name
 
 
 @dataclass
@@ -58,33 +59,74 @@ class Processor(ProcessorMeta):
     raw_response: RawResponse
     processed_data: ProcessedData
 
+    @override
     def __init__(self, event_id: int, user: User, command_args: list[str]) -> None:
         super().__init__(event_id, user, command_args)
         self.raw_response = RawResponse()
         self.processed_data = ProcessedData()
 
     @property
-    def game_platform(self) -> GameType:
+    @override
+    def game_platform(self) -> Literal['TOP']:
         return GAME_TYPE
 
-    async def handle_bind(self, platform: str, account: str) -> str:
+    @override
+    async def handle_bind(self, platform: str, account: str, bot_info: UserInfo, user_info: UserInfo) -> UniMessage:
         """处理绑定消息"""
         self.command_type = 'bind'
         await self.check_user()
         async with get_session() as session:
-            return await create_or_update_bind(
+            bind_status = await create_or_update_bind(
                 session=session,
                 chat_platform=platform,
                 chat_account=account,
                 game_platform=GAME_TYPE,
                 game_account=self.user.name,
             )
+        if bind_status in (BindStatus.SUCCESS, BindStatus.UPDATE):
+            async with HostPage(
+                await render(
+                    'binding',
+                    Bind(
+                        platform=self.game_platform,
+                        status='unknown',
+                        user=Bind.People(
+                            avatar=await get_avatar(user_info, 'Data URI', None),
+                            name=(await self.get_user_name()).upper(),
+                        ),
+                        bot=Bind.People(
+                            avatar=await get_avatar(bot_info, 'Data URI', '../../static/logo/logo.svg'),
+                            name=bot_info.user_name,
+                        ),
+                        command='top查我',
+                    ),
+                )
+            ) as page_hash:
+                message = UniMessage.image(
+                    raw=await screenshot(urlunparse(('http', get_self_netloc(), f'/host/{page_hash}.html', '', '', '')))
+                )
+        return message
 
-    async def handle_query(self) -> str:
+    @override
+    async def handle_query(self) -> UniMessage:
         """处理查询消息"""
         self.command_type = 'query'
         await self.check_user()
-        return await self.generate_message()
+        game_data = await self.get_game_data()
+        message = ''
+        if game_data.day is not None:
+            message += f'用户 {self.user.name} 24小时内统计数据为: '
+            message += f"\nL'PM: {round(game_data.day.lpm,2)} ( {round(game_data.day.lpm/24,2)} pps )"
+            message += f'\nAPM: {round(game_data.day.apm,2)} ( x{round(game_data.day.apm/game_data.day.lpm,2)} )'
+        else:
+            message += f'用户 {self.user.name} 暂无24小时内统计数据'
+        if game_data.total is not None:
+            message += '\n历史统计数据为: '
+            message += f"\nL'PM: {round(game_data.total.lpm,2)} ( {round(game_data.total.lpm/24,2)} pps )"
+            message += f'\nAPM: {round(game_data.total.apm,2)} ( x{round(game_data.total.apm/game_data.total.lpm,2)} )'
+        else:
+            message += '\n暂无历史统计数据'
+        return UniMessage(message)
 
     async def get_user_profile(self) -> str:
         """获取用户信息"""
@@ -94,10 +136,9 @@ class Processor(ProcessorMeta):
             self.processed_data.user_profile = self.raw_response.user_profile.decode()
         return self.processed_data.user_profile
 
-    async def check_user(self) -> None | NoReturn:
+    async def check_user(self) -> None:
         if 'user not found!' in await self.get_user_profile():
             raise RequestError('用户不存在!')
-        return None
 
     async def get_user_name(self) -> str:
         """获取用户名"""
@@ -122,21 +163,3 @@ class Processor(ProcessorMeta):
         dataframe = read_html(table, encoding='utf-8', header=0)[0]
         total = Data(lpm=dataframe['lpm'].mean(), apm=dataframe['apm'].mean()) if len(dataframe) != 0 else None
         return GameData(day=day, total=total)
-
-    async def generate_message(self) -> str:
-        """生成消息"""
-        game_data = await self.get_game_data()
-        message = ''
-        if game_data.day is not None:
-            message += f'用户 {self.user.name} 24小时内统计数据为: '
-            message += f"\nL'PM: {round(game_data.day.lpm,2)} ( {round(game_data.day.lpm/24,2)} pps )"
-            message += f'\nAPM: {round(game_data.day.apm,2)} ( x{round(game_data.day.apm/game_data.day.lpm,2)} )'
-        else:
-            message += f'用户 {self.user.name} 暂无24小时内统计数据'
-        if game_data.total is not None:
-            message += '\n历史统计数据为: '
-            message += f"\nL'PM: {round(game_data.total.lpm,2)} ( {round(game_data.total.lpm/24,2)} pps )"
-            message += f'\nAPM: {round(game_data.total.apm,2)} ( x{round(game_data.total.apm/game_data.total.lpm,2)} )'
-        else:
-            message += '\n暂无历史统计数据'
-        return message

nonebot_plugin_tetris_stats/game_data_processor/top_data_processor/schemas/response.py

@@ -0,0 +1,16 @@
+from typing import Literal
+
+from ...schemas import BaseProcessedData, BaseRawResponse
+from ..constant import GAME_TYPE
+
+
+class RawResponse(BaseRawResponse):
+    platform: Literal['TOP'] = GAME_TYPE
+
+    user_profile: bytes | None = None
+
+
+class ProcessedData(BaseProcessedData):
+    platform: Literal['TOP'] = GAME_TYPE
+
+    user_profile: str | None = None

nonebot_plugin_tetris_stats/game_data_processor/tos_data_processor/__init__.py

@@ -1,13 +1,18 @@
-from arclet.alconna import Alconna, Arg, ArgFlag, Args, CommandMeta, Option
+from typing import NoReturn
+
+from arclet.alconna import Alconna, AllParam, Arg, ArgFlag, Args, CommandMeta, Option
 from nonebot.adapters import Bot, Event
 from nonebot.matcher import Matcher
-from nonebot_plugin_alconna import AlcMatches, At, on_alconna
+from nonebot_plugin_alconna import At, on_alconna
+from nonebot_plugin_alconna.uniseg import UniMessage
 from nonebot_plugin_orm import get_session
+from nonebot_plugin_userinfo import BotUserInfo, EventUserInfo, UserInfo  # type: ignore[import-untyped]
 
 from ...db import query_bind_info
-from ...utils.exception import MessageFormatError, NeedCatchError
+from ...utils.exception import HandleNotFinishedError, NeedCatchError, RequestError
 from ...utils.platform import get_platform
 from ...utils.typing import Me
+from .. import add_default_handlers
 from ..constant import BIND_COMMAND, QUERY_COMMAND
 from .constant import GAME_TYPE
 from .processor import Processor, User, identify_user_info
@@ -52,6 +57,7 @@ alc = on_alconna(
             dest='query',
             help_text='查询 茶服 游戏信息',
         ),
+        Arg('other', AllParam, flags=[ArgFlag.HIDDEN, ArgFlag.OPTIONAL]),
         meta=CommandMeta(
             description='查询 TetrisOnline茶服 的信息',
             example='茶服查我',
@@ -64,42 +70,90 @@ alc = on_alconna(
     aliases={'tos', 'TOS'},
 )
 
-try:
-    from nonebot.adapters.onebot.v11 import GROUP, MessageEvent
-    from nonebot.adapters.onebot.v11 import Bot as OB11Bot
 
-    @alc.assign('bind')
-    async def _(event: MessageEvent, matcher: Matcher):
-        await matcher.finish('QQ 平台无需绑定')
+async def finish_special_query(matcher: Matcher, proc: Processor) -> NoReturn:
+    try:
+        await (await proc.handle_query()).finish()
+    except NeedCatchError as e:
+        if isinstance(e, RequestError) and '未找到此用户' in e.message:
+            matcher.skip()
+        await matcher.send(str(e))
+        raise HandleNotFinishedError from e
+
+
+try:
+    from nonebot.adapters.onebot.v11 import GROUP as OB11GROUP
+    from nonebot.adapters.onebot.v11 import Bot as OB11Bot
+    from nonebot.adapters.onebot.v11 import MessageEvent as OB11MessageEvent
 
     @alc.assign('query')
-    async def _(bot: OB11Bot, event: MessageEvent, matcher: Matcher, target: At | Me):
-        if event.is_tome() and await GROUP(bot, event):
+    async def _(bot: OB11Bot, event: OB11MessageEvent, matcher: Matcher, target: At | Me):
+        if event.is_tome() and await OB11GROUP(bot, event):
             await matcher.finish('不能查询bot的信息')
         proc = Processor(
             event_id=id(event),
-            user=User(teaid=target.target if isinstance(target, At) else event.get_user_id()),
+            user=User(teaid=f'onebot-{target.target}' if isinstance(target, At) else f'onebot-{event.get_user_id()}'),
             command_args=[],
         )
-        try:
-            await matcher.finish(await proc.handle_query())
-        except NeedCatchError as e:
-            await matcher.finish(str(e))
+        await finish_special_query(matcher, proc)
+
+except ImportError:
+    pass
+
+try:
+    from nonebot.adapters.kaiheila.event import MessageEvent as KookMessageEvent
+
+    @alc.assign('query')
+    async def _(event: KookMessageEvent, matcher: Matcher, target: At | Me):
+        proc = Processor(
+            event_id=id(event),
+            user=User(teaid=f'kook-{target.target}' if isinstance(target, At) else f'kook-{event.get_user_id()}'),
+            command_args=[],
+        )
+        await finish_special_query(matcher, proc)
+
+except ImportError:
+    pass
+
+try:
+    from nonebot.adapters.discord import MessageEvent as DiscordMessageEvent
+
+    @alc.assign('query')
+    async def _(event: DiscordMessageEvent, matcher: Matcher, target: At | Me):
+        proc = Processor(
+            event_id=id(event),
+            user=User(teaid=f'discord-{target.target}' if isinstance(target, At) else f'discord-{event.get_user_id()}'),
+            command_args=[],
+        )
+        await finish_special_query(matcher, proc)
+
 except ImportError:
     pass
 
 
 @alc.assign('bind')
-async def _(bot: Bot, event: Event, matcher: Matcher, account: User):
+async def _(  # noqa: PLR0913
+    bot: Bot,
+    event: Event,
+    matcher: Matcher,
+    account: User,
+    bot_info: UserInfo = BotUserInfo(),  # noqa: B008
+    user_info: UserInfo = EventUserInfo(),  # noqa: B008
+):
     proc = Processor(
         event_id=id(event),
         user=account,
         command_args=[],
     )
     try:
-        await matcher.finish(await proc.handle_bind(platform=get_platform(bot), account=event.get_user_id()))
+        await (
+            await proc.handle_bind(
+                platform=get_platform(bot), account=event.get_user_id(), bot_info=bot_info, nb_user_info=user_info
+            )
+        ).finish()
     except NeedCatchError as e:
-        await matcher.finish(str(e))
+        await matcher.send(str(e))
+        raise HandleNotFinishedError from e
 
 
 @alc.assign('query')
@@ -116,13 +170,14 @@ async def _(bot: Bot, event: Event, matcher: Matcher, target: At | Me):
     message = '* 由于无法验证绑定信息, 不能保证查询到的用户为本人\n'
     proc = Processor(
         event_id=id(event),
-        user=User(name=bind.game_account),
+        user=User(teaid=bind.game_account),
         command_args=[],
     )
     try:
-        await matcher.finish(message + await proc.handle_query())
+        await (UniMessage(message) + await proc.handle_query()).finish()
     except NeedCatchError as e:
-        await matcher.finish(str(e))
+        await matcher.send(str(e))
+        raise HandleNotFinishedError from e
 
 
 @alc.assign('query')
@@ -133,19 +188,10 @@ async def _(event: Event, matcher: Matcher, account: User):
         command_args=[],
     )
     try:
-        await matcher.finish(await proc.handle_query())
+        await (await proc.handle_query()).finish()
     except NeedCatchError as e:
-        await matcher.finish(str(e))
+        await matcher.send(str(e))
+        raise HandleNotFinishedError from e
 
 
-@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:
-        await matcher.finish(
-            f'{matches.error_info!r}\n' if matches.error_info is not None else '' + '输入"茶服 --help"查看帮助'
-        )
+add_default_handlers(alc)

nonebot_plugin_tetris_stats/game_data_processor/tos_data_processor/constant.py

@@ -1,4 +1,10 @@
-from ...utils.typing import GameType
+from typing import Literal
 
-GAME_TYPE: GameType = 'TOS'
-BASE_URL = 'https://teatube.cn:8888/'
+GAME_TYPE: Literal['TOS'] = 'TOS'
+BASE_URL = {
+    'https://teatube.cn:8888/',
+    'http://cafuuchino1.studio26f.org:19970',
+    'http://cafuuchino2.studio26f.org:19970',
+    'http://cafuuchino3.studio26f.org:19970',
+    'http://cafuuchino4.studio26f.org:19970',
+}

nonebot_plugin_tetris_stats/game_data_processor/tos_data_processor/processor.py

@@ -1,41 +1,43 @@
 from dataclasses import dataclass
 from re import match
-from typing import Any
-from urllib.parse import urlencode
+from typing import Literal
+from urllib.parse import urlencode, urlunparse
 
+from httpx import TimeoutException
+from nonebot.compat import type_validate_json
+from nonebot_plugin_alconna.uniseg import UniMessage
 from nonebot_plugin_orm import get_session
-from pydantic import parse_raw_as
+from nonebot_plugin_userinfo import UserInfo as NBUserInfo  # type: ignore[import-untyped]
+from typing_extensions import override
 
-from ...db import create_or_update_bind
+from ...db import BindStatus, create_or_update_bind
+from ...utils.avatar import get_avatar
 from ...utils.exception import MessageFormatError, RequestError
+from ...utils.host import HostPage, get_self_netloc
+from ...utils.render import Bind, render
 from ...utils.request import Request, splice_url
-from ...utils.typing import GameType
-from .. import ProcessedData as ProcessedDataMeta
+from ...utils.screenshot import screenshot
 from .. import Processor as ProcessorMeta
-from .. import RawResponse as RawResponseMeta
-from .. import User as UserMeta
+from ..schemas import BaseUser
 from .constant import BASE_URL, GAME_TYPE
+from .schemas.response import ProcessedData, RawResponse
 from .schemas.user_info import SuccessModel as InfoSuccess
 from .schemas.user_info import UserInfo
 from .schemas.user_profile import UserProfile
 
 
-@dataclass
-class User(UserMeta):
+class User(BaseUser):
+    platform: Literal['TOS'] = GAME_TYPE
+
     teaid: str | None = None
     name: str | None = None
 
-
-@dataclass
-class RawResponse(RawResponseMeta):
-    user_profile: dict[frozenset[tuple[str, Any]], bytes]
-    user_info: bytes | None = None
-
-
-@dataclass
-class ProcessedData(ProcessedDataMeta):
-    user_profile: dict[frozenset[tuple[str, Any]], UserProfile]
-    user_info: InfoSuccess | None = None
+    @property
+    @override
+    def unique_identifier(self) -> str:
+        if self.teaid is None:
+            raise ValueError('不完整的User!')
+        return self.teaid
 
 
 @dataclass
@@ -60,7 +62,7 @@ def identify_user_info(info: str) -> User | MessageFormatError:
         and 2 <= len(info) <= 18  # noqa: PLR2004
     ):
         return User(name=info)
-    if info.isdigit():
+    if info.startswith(('onebot-', 'qqguild-', 'kook-', 'discord-')) and info.split('-', maxsplit=1)[1].isdigit():
         return User(teaid=info)
     return MessageFormatError('用户名/QQ号不合法')
 
@@ -70,35 +72,79 @@ class Processor(ProcessorMeta):
     raw_response: RawResponse
     processed_data: ProcessedData
 
+    @override
     def __init__(self, event_id: int, user: User, command_args: list[str]) -> None:
         super().__init__(event_id, user, command_args)
         self.raw_response = RawResponse(user_profile={})
         self.processed_data = ProcessedData(user_profile={})
 
     @property
-    def game_platform(self) -> GameType:
+    @override
+    def game_platform(self) -> Literal['TOS']:
         return GAME_TYPE
 
-    async def handle_bind(self, platform: str, account: str) -> str:
+    @override
+    async def handle_bind(
+        self, platform: str, account: str, bot_info: NBUserInfo, nb_user_info: NBUserInfo
+    ) -> UniMessage:
         """处理绑定消息"""
         self.command_type = 'bind'
         await self.get_user()
-        if self.user.name is None:
-            raise  # FIXME: 不知道怎么才能把这类型给变过来了
         async with get_session() as session:
-            return await create_or_update_bind(
+            bind_status = await create_or_update_bind(
                 session=session,
                 chat_platform=platform,
                 chat_account=account,
                 game_platform=GAME_TYPE,
-                game_account=self.user.name,
+                game_account=self.user.unique_identifier,
             )
+        user_info = await self.get_user_info()
+        if bind_status in (BindStatus.SUCCESS, BindStatus.UPDATE):
+            async with HostPage(
+                await render(
+                    'binding',
+                    Bind(
+                        platform=self.game_platform,
+                        status='unknown',
+                        user=Bind.People(
+                            avatar=await get_avatar(nb_user_info, 'Data URI', None), name=user_info.data.name
+                        ),
+                        bot=Bind.People(
+                            avatar=await get_avatar(bot_info, 'Data URI', '../../static/logo/logo.svg'),
+                            name=bot_info.user_name,
+                        ),
+                        command='茶服查我',
+                    ),
+                )
+            ) as page_hash:
+                message = UniMessage.image(
+                    raw=await screenshot(urlunparse(('http', get_self_netloc(), f'/host/{page_hash}.html', '', '', '')))
+                )
+        return message
 
-    async def handle_query(self) -> str:
+    @override
+    async def handle_query(self) -> UniMessage:
         """处理查询消息"""
         self.command_type = 'query'
         await self.get_user()
-        return await self.generate_message()
+        user_info = (await self.get_user_info()).data
+        message = f'用户 {user_info.name} ({user_info.teaid}) '
+        if user_info.ranked_games == '0':
+            message += '暂无段位统计数据'
+        else:
+            message += f', 段位分 {round(float(user_info.rating_now),2)}±{round(float(user_info.rd_now),2)} ({round(float(user_info.vol_now),2)}) '
+        game_data = await self.get_game_data()
+        if game_data is None:
+            message += ', 暂无游戏数据'
+        else:
+            message += f', 最近 {game_data.num} 局数据'
+            message += f"\nL'PM: {game_data.lpm} ( {game_data.pps} pps )"
+            message += f'\nAPM: {game_data.apm} ( x{game_data.apl} )'
+            message += f'\nADPM: {game_data.adpm} ( x{game_data.adpl} ) ( {game_data.vs}vs )'
+        message += f'\n40L: {float(user_info.pb_sprint)/1000:.2f}s' if user_info.pb_sprint != '2147483647' else ''
+        message += f'\nMarathon: {user_info.pb_marathon}' if user_info.pb_marathon != '0' else ''
+        message += f'\nChallenge: {user_info.pb_challenge}' if user_info.pb_challenge != '0' else ''
+        return UniMessage(message)
 
     async def get_user(self) -> None:
         """
@@ -113,45 +159,60 @@ class Processor(ProcessorMeta):
         """获取用户信息"""
         if self.processed_data.user_info is None:
             if self.user.teaid is not None:
-                url = splice_url(
-                    [
-                        BASE_URL,
-                        'getTeaIdInfo',
-                        f'?{urlencode({"teaId":self.user.teaid})}',
-                    ]
-                )
+                url = [
+                    splice_url(
+                        [
+                            i,
+                            'getTeaIdInfo',
+                            f'?{urlencode({"teaId":self.user.teaid})}',
+                        ]
+                    )
+                    for i in BASE_URL
+                ]
             else:
-                url = splice_url(
-                    [
-                        BASE_URL,
-                        'getUsernameInfo',
-                        f'?{urlencode({"username":self.user.name})}',
-                    ]
-                )
-            self.raw_response.user_info = await Request.request(url)
-            user_info: UserInfo = parse_raw_as(UserInfo, self.raw_response.user_info)  # type: ignore[arg-type]
+                url = [
+                    splice_url(
+                        [
+                            i,
+                            'getUsernameInfo',
+                            f'?{urlencode({"username":self.user.name})}',
+                        ]
+                    )
+                    for i in BASE_URL
+                ]
+            self.raw_response.user_info = await Request.failover_request(
+                url, failover_code=[502], failover_exc=(TimeoutException,)
+            )
+            user_info: UserInfo = type_validate_json(UserInfo, self.raw_response.user_info)  # type: ignore[arg-type]
             if not isinstance(user_info, InfoSuccess):
                 raise RequestError(f'用户信息请求错误:\n{user_info.error}')
             self.processed_data.user_info = user_info
         return self.processed_data.user_info
 
-    async def get_user_profile(self, other_parameter: dict[str, Any] | None = None) -> UserProfile:
+    async def get_user_profile(self, other_parameter: dict[str, str | bytes] | None = None) -> UserProfile:
         """获取用户数据"""
         if other_parameter is None:
             other_parameter = {}
-        fset = frozenset(other_parameter.items())
-        if self.processed_data.user_profile.get(fset) is None:
-            self.raw_response.user_profile[fset] = await Request.request(
-                splice_url(
-                    [
-                        BASE_URL,
-                        'getProfile',
-                        f'?{urlencode({"id":self.user.teaid or self.user.name},**other_parameter)}',
-                    ]
-                )
+        params = urlencode(dict(sorted(other_parameter.items())))
+        if self.processed_data.user_profile.get(params) is None:
+            self.raw_response.user_profile[params] = await Request.failover_request(
+                [
+                    splice_url(
+                        [
+                            i,
+                            'getProfile',
+                            f'?{urlencode({"id":self.user.teaid or self.user.name,**other_parameter})}',
+                        ]
+                    )
+                    for i in BASE_URL
+                ],
+                failover_code=[502],
+                failover_exc=(TimeoutException,),
             )
-            self.processed_data.user_profile[fset] = UserProfile.parse_raw(self.raw_response.user_profile[fset])
-        return self.processed_data.user_profile[fset]
+            self.processed_data.user_profile[params] = type_validate_json(
+                UserProfile, self.raw_response.user_profile[params]
+            )
+        return self.processed_data.user_profile[params]
 
     async def get_game_data(self) -> GameData | None:
         """获取游戏数据"""
@@ -194,28 +255,3 @@ class Processor(ProcessorMeta):
             adpl=round((adpm / lpm), 2),
             vs=round((adpm / 60 * 100), 2),
         )
-
-    async def generate_message(self) -> str:
-        """生成消息"""
-        user_info = (await self.get_user_info()).data
-        message = f'用户 {user_info.name} ({user_info.teaid}) '
-        if user_info.ranked_games == '0':
-            message += '暂无段位统计数据'
-        else:
-            message += f', 段位分 {round(float(user_info.rating_now),2)}±{round(float(user_info.rd_now),2)} ({round(float(user_info.vol_now),2)}) '
-        game_data = await self.get_game_data()
-        if game_data is None:
-            message += ', 暂无游戏数据'
-        else:
-            message += f', 最近 {game_data.num} 局数据'
-            message += f"\nL'PM: {game_data.lpm} ( {game_data.pps} pps )"
-            message += f'\nAPM: {game_data.apm} ( x{game_data.apl} )'
-            message += f'\nADPM: {game_data.adpm} ( x{game_data.adpl} ) ( {game_data.vs}vs )'
-        message += (
-            f'\n40L: {float(user_info.pb_sprint)/1000:.2f}s'
-            if user_info.pb_sprint != 2147483647  # noqa: PLR2004
-            else ''
-        )
-        message += f'\nMarathon: {user_info.pb_marathon}' if user_info.pb_marathon != 0 else ''
-        message += f'\nChallenge: {user_info.pb_challenge}' if user_info.pb_challenge != 0 else ''
-        return message

nonebot_plugin_tetris_stats/game_data_processor/tos_data_processor/schemas/response.py

@@ -0,0 +1,20 @@
+from typing import Literal
+
+from ...schemas import BaseProcessedData, BaseRawResponse
+from ..constant import GAME_TYPE
+from .user_info import SuccessModel as InfoSuccess
+from .user_profile import UserProfile
+
+
+class RawResponse(BaseRawResponse):
+    platform: Literal['TOS'] = GAME_TYPE
+
+    user_profile: dict[str, bytes]
+    user_info: bytes | None = None
+
+
+class ProcessedData(BaseProcessedData):
+    platform: Literal['TOS'] = GAME_TYPE
+
+    user_profile: dict[str, UserProfile]
+    user_info: InfoSuccess | None = None

nonebot_plugin_tetris_stats/utils/avatar.py

@@ -0,0 +1,52 @@
+from base64 import b64encode
+from io import BytesIO
+from typing import Literal, overload
+
+from nonebot_plugin_userinfo import UserInfo  # type: ignore[import-untyped]
+from PIL import Image
+
+
+@overload
+async def get_avatar(user: UserInfo, scheme: Literal['Data URI'], default: str | None) -> str:
+    """获取用户头像的指定格式
+
+    Args:
+        user (UserInfo): 要获取的用户
+        scheme (Literal['Data URI']): 格式
+        default (str | None): 获取不到时的默认值
+
+    Raises:
+        TypeError: Can't get avatar: 当获取不到头像并且没有设置默认值时抛出
+        TypeError: Can't get avatar format: 当获取到的头像无法识别格式时抛出
+
+    Returns:
+        str: Data URI 格式的头像
+    """
+
+
+@overload
+async def get_avatar(user: UserInfo, scheme: Literal['bytes'], default: str | None) -> bytes:
+    """获取用户头像的指定格式
+
+    Args:
+        user (UserInfo): 要获取的用户
+        scheme (Literal['bytes']): 格式
+        default (str | None): 获取不到时的默认值
+
+    Returns:
+        bytes: bytes 格式的头像
+    """
+
+
+async def get_avatar(user: UserInfo, scheme: Literal['Data URI', 'bytes'], default: str | None) -> str | bytes:
+    if user.user_avatar is None:
+        if default is None:
+            raise TypeError("Can't get avatar")
+        return default
+    bot_avatar = await user.user_avatar.get_image()
+    if scheme == 'Data URI':
+        avatar_format = Image.open(BytesIO(bot_avatar)).format
+        if avatar_format is None:
+            raise TypeError("Can't get avatar format")
+        return f'data:{Image.MIME[avatar_format]};base64,{b64encode(bot_avatar).decode()}'
+    return bot_avatar

nonebot_plugin_tetris_stats/utils/browser.py

@@ -1,6 +1,7 @@
 import sys
 from os import environ
 from platform import system
+from re import sub
 
 from nonebot import get_driver
 from nonebot.log import logger
@@ -33,18 +34,18 @@ class BrowserManager:
             raise ImportError('加载失败, Windows 必须设置 FASTAPI_RELOAD=false 才能正常运行 playwright')
         logger.info('开始 安装/更新 playwright 浏览器')
         environ['PLAYWRIGHT_DOWNLOAD_HOST'] = 'https://npmmirror.com/mirrors/playwright/'
-        if cls._handle_error(cls._call_playwright(['', 'install', 'firefox'])):
+        if cls._call_playwright(['', 'install', 'firefox']):
             logger.success('安装/更新 playwright 浏览器成功')
         else:
             logger.warning('playwright 浏览器 安装/更新 失败, 尝试使用原始仓库下载')
             del environ['PLAYWRIGHT_DOWNLOAD_HOST']
-            if cls._handle_error(cls._call_playwright(['', 'install', 'firefox'])):
+            if cls._call_playwright(['', 'install', 'firefox']):
                 logger.success('安装/更新 playwright 浏览器成功')
             else:
                 logger.error('安装/更新 playwright 浏览器失败')
         try:
             await cls._start_browser()
-        except BaseException as e:  # noqa: BLE001 不知道会有什么异常, 交给用户解决
+        except BaseException as e:  # 不知道会有什么异常, 交给用户解决
             raise ImportError(
                 'playwright 启动失败, 请尝试在命令行运行 playwright install-deps firefox, 如果仍然启动失败, 请参考上面的报错👆'
             ) from e
@@ -52,26 +53,20 @@ class BrowserManager:
             logger.success('playwright 启动成功')
 
     @classmethod
-    def _call_playwright(cls, argv: list[str]) -> BaseException:
+    def _call_playwright(cls, argv: list[str]) -> bool:
         """等价于调用 playwright 的命令行程序"""
         argv_backup = sys.argv.copy()
-        from re import sub
-
         sys.argv[0] = sub(r'(-script\.pyw|\.exe)?$', '', sys.argv[0])
         sys.argv = argv
         try:
             main()
-        except BaseException as e:  # noqa: BLE001 不在这里处理 playwright 的异常
-            return e
+        except SystemExit as e:
+            return e.code == 0
+        except BaseException:  # noqa: BLE001
+            return False
         finally:
             sys.argv = argv_backup
-        return SystemExit(0)
-
-    @classmethod
-    def _handle_error(cls, error: BaseException) -> bool:
-        if isinstance(error, SystemExit) and error.code == 0:
-            return True
-        return False
+        return True
 
     @classmethod
     async def _start_browser(cls) -> Browser:

nonebot_plugin_tetris_stats/utils/exception.py

@@ -15,21 +15,25 @@ class NeedCatchError(TetrisStatsError):
     """需要被捕获的异常基类"""
 
 
-class DoNotCatchError(TetrisStatsError):
-    """不应该被捕获的异常基类"""
-
-
 class RequestError(NeedCatchError):
     """请求错误"""
 
+    def __init__(self, message: str = '', *, status_code: int | None = None):
+        super().__init__(message)
+        self.status_code = status_code
+
 
 class MessageFormatError(NeedCatchError):
     """用户发送的消息格式不正确"""
 
 
-class DatabaseVersionError(DoNotCatchError):
-    """数据库版本错误"""
+class DoNotCatchError(TetrisStatsError):
+    """不应该被捕获的异常基类"""
 
 
 class WhatTheFuckError(DoNotCatchError):
     """用于表示不应该出现的情况 ("""
+
+
+class HandleNotFinishedError(DoNotCatchError):
+    """任务没有正常完成处理的错误"""

nonebot_plugin_tetris_stats/utils/host.py

@@ -0,0 +1,66 @@
+from hashlib import sha256
+from ipaddress import IPv4Address, IPv6Address
+from typing import ClassVar
+
+from fastapi import FastAPI, status
+from fastapi.responses import HTMLResponse
+from fastapi.staticfiles import StaticFiles
+from nonebot import get_app, get_driver
+from nonebot_plugin_localstore import get_cache_dir  # type: ignore[import-untyped]
+from pydantic import IPvAnyAddress
+
+from .templates import templates_dir
+
+app = get_app()
+
+driver = get_driver()
+
+global_config = driver.config
+
+cache_dir = get_cache_dir('nonebot_plugin_tetris_stats')
+
+if not isinstance(app, FastAPI):
+    raise RuntimeError('本插件需要 FastAPI 驱动器才能运行')  # noqa: TRY004
+
+NOT_FOUND = HTMLResponse('404 Not Found', status_code=status.HTTP_404_NOT_FOUND)
+
+
+class HostPage:
+    pages: ClassVar[dict[str, str]] = {}
+
+    def __init__(self, page: str) -> None:
+        self.page_hash = sha256(page.encode()).hexdigest()
+        self.pages[self.page_hash] = page
+
+    async def __aenter__(self) -> str:
+        return self.page_hash
+
+    async def __aexit__(self, exc_type, exc, tb) -> None:  # noqa: ANN001
+        self.pages.pop(self.page_hash, None)
+
+
+app.mount(
+    '/host/assets',
+    StaticFiles(directory=templates_dir / 'assets'),
+    name='assets',
+)
+
+
+@app.get('/host/{page_hash}.html', status_code=status.HTTP_200_OK)
+async def _(page_hash: str) -> HTMLResponse:
+    if page_hash in HostPage.pages:
+        return HTMLResponse(HostPage.pages[page_hash])
+    return NOT_FOUND
+
+
+def get_self_netloc() -> str:
+    host: IPv4Address | IPv6Address | IPvAnyAddress = global_config.host
+    if isinstance(host, IPv4Address):
+        if host == IPv4Address('0.0.0.0'):  # noqa: S104
+            host = IPv4Address('127.0.0.1')
+        netloc = f'{host}:{global_config.port}'
+    else:
+        if host == IPv6Address('::'):
+            host = IPv6Address('::1')
+        netloc = f'[{host}]:{global_config.port}'
+    return netloc

nonebot_plugin_tetris_stats/utils/metrics.py

@@ -144,7 +144,7 @@ class TetrisMetricsProWithLPMADPM(TetrisMetricsBasicWithLPM, TetrisMetricsBaseWi
 
 
 @overload
-def get_metrics(  # noqa: PLR0913
+def get_metrics(
     *,
     pps: Number,
     lpm: None = None,
@@ -157,7 +157,7 @@ def get_metrics(  # noqa: PLR0913
 
 
 @overload
-def get_metrics(  # noqa: PLR0913
+def get_metrics(
     *,
     pps: None = None,
     lpm: Number,
@@ -170,7 +170,7 @@ def get_metrics(  # noqa: PLR0913
 
 
 @overload
-def get_metrics(  # noqa: PLR0913
+def get_metrics(
     *,
     pps: None = None,
     lpm: None = None,
@@ -183,7 +183,7 @@ def get_metrics(  # noqa: PLR0913
 
 
 @overload
-def get_metrics(  # noqa: PLR0913
+def get_metrics(
     *,
     pps: None = None,
     lpm: None = None,
@@ -196,7 +196,7 @@ def get_metrics(  # noqa: PLR0913
 
 
 @overload
-def get_metrics(  # noqa: PLR0913
+def get_metrics(
     *,
     pps: Number,
     lpm: None = None,
@@ -209,7 +209,7 @@ def get_metrics(  # noqa: PLR0913
 
 
 @overload
-def get_metrics(  # noqa: PLR0913
+def get_metrics(
     *,
     pps: None = None,
     lpm: Number,
@@ -222,7 +222,7 @@ def get_metrics(  # noqa: PLR0913
 
 
 @overload
-def get_metrics(  # noqa: PLR0913
+def get_metrics(
     *,
     pps: Number,
     lpm: None = None,
@@ -235,7 +235,7 @@ def get_metrics(  # noqa: PLR0913
 
 
 @overload
-def get_metrics(  # noqa: PLR0913
+def get_metrics(
     *,
     pps: Number,
     lpm: None = None,
@@ -248,7 +248,7 @@ def get_metrics(  # noqa: PLR0913
 
 
 @overload
-def get_metrics(  # noqa: PLR0913
+def get_metrics(
     *,
     pps: None = None,
     lpm: Number,
@@ -261,7 +261,7 @@ def get_metrics(  # noqa: PLR0913
 
 
 @overload
-def get_metrics(  # noqa: PLR0913
+def get_metrics(
     *,
     pps: None = None,
     lpm: Number,

nonebot_plugin_tetris_stats/utils/recorder.py

@@ -1,4 +1,4 @@
-from datetime import UTC, datetime
+from datetime import datetime, timezone
 from typing import ClassVar
 
 from nonebot import get_driver, get_plugin
@@ -9,12 +9,15 @@ from nonebot_plugin_orm import get_session
 
 from ..db.models import HistoricalData
 
+UTC = timezone.utc
+
 driver = get_driver()
 
 
 class Recorder:
     matchers: ClassVar[set[type[Matcher]]] = set()
     historical_data: ClassVar[dict[int, tuple[HistoricalData, bool]]] = {}
+    error_event: ClassVar[set[int]] = set()
 
     @classmethod
     def create_historical_data(cls, event_id: int, historical_data: HistoricalData) -> None:
@@ -32,17 +35,27 @@ class Recorder:
 
     @classmethod
     async def save_historical_data(cls, event_id: int) -> None:
-        if event_id not in cls.historical_data:
-            raise KeyError
-        historical_data, completed = cls.historical_data.pop(event_id)
+        historical_data, completed = cls.del_historical_data(event_id)
         if completed:
             async with get_session() as session:
                 session.add(historical_data)
                 await session.commit()
 
     @classmethod
-    def del_historical_data(cls, event_id: int) -> None:
-        cls.historical_data.pop(event_id)
+    def del_historical_data(cls, event_id: int) -> tuple[HistoricalData, bool]:
+        return cls.historical_data.pop(event_id)
+
+    @classmethod
+    def add_error_event(cls, event_id: int) -> None:
+        cls.error_event.add(event_id)
+
+    @classmethod
+    def del_error_event(cls, event_id: int) -> None:
+        cls.error_event.remove(event_id)
+
+    @classmethod
+    def is_error_event(cls, event_id: int) -> bool:
+        return event_id in cls.error_event
 
 
 @driver.on_startup
@@ -73,7 +86,9 @@ def _(bot: Bot, event: Event, matcher: Matcher):
 @run_postprocessor
 async def _(event: Event, matcher: Matcher, exception: Exception | None):
     if isinstance(matcher, tuple(Recorder.matchers)):
+        event_id = id(event)
         if exception is not None:
-            Recorder.del_historical_data(id(event))
+            Recorder.add_error_event(event_id)
+            Recorder.del_historical_data(event_id)
         else:
-            await Recorder.save_historical_data(id(event))
+            await Recorder.save_historical_data(event_id)

nonebot_plugin_tetris_stats/utils/render.py

@@ -0,0 +1,104 @@
+from datetime import datetime
+from typing import Annotated, ClassVar, Literal, overload
+
+from jinja2 import Environment, FileSystemLoader
+from nonebot.compat import PYDANTIC_V2
+from pydantic import BaseModel
+
+from ..game_data_processor.io_data_processor.typing import Rank
+from .templates import templates_dir
+from .typing import Number
+
+if PYDANTIC_V2:
+    from pydantic import PlainSerializer
+
+env = Environment(
+    loader=FileSystemLoader(templates_dir), autoescape=True, trim_blocks=True, lstrip_blocks=True, enable_async=True
+)
+
+
+def format_datetime_to_timestamp(dt: datetime) -> int:
+    return int(dt.timestamp() * 1000)
+
+
+class Bind(BaseModel):
+    class People(BaseModel):
+        avatar: str
+        name: str
+
+    platform: Literal['TETR.IO', 'TOP', 'TOS']
+    status: Literal['error', 'success', 'unknown', 'unlink', 'unverified']
+    user: People
+    bot: People
+    command: str
+
+
+class TETRIOInfo(BaseModel):
+    class User(BaseModel):
+        avatar: str
+        name: str
+        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 TetraLeagueHistory(BaseModel):
+        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
+
+        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
+
+    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}
+
+
+@overload
+async def render(render_type: Literal['binding'], data: Bind) -> str: ...
+
+
+@overload
+async def render(render_type: Literal['tetrio/info'], data: TETRIOInfo) -> str: ...
+
+
+async def render(render_type: Literal['binding', 'tetrio/info'], data: Bind | TETRIOInfo) -> 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())

nonebot_plugin_tetris_stats/utils/request.py

@@ -1,17 +1,20 @@
+from collections.abc import Sequence
+from http import HTTPStatus
 from urllib.parse import urljoin, urlparse
 
 from aiofiles import open
 from httpx import AsyncClient, HTTPError
-from nonebot import get_driver
+from nonebot import get_driver, get_plugin_config
 from nonebot.log import logger
 from playwright.async_api import Response
 from ujson import JSONDecodeError, dumps, loads
 
-from ..config.config import CACHE_PATH
+from ..config.config import CACHE_PATH, Config
 from .browser import BrowserManager
 from .exception import RequestError
 
 driver = get_driver()
+config = get_plugin_config(Config)
 
 
 @driver.on_startup
@@ -37,7 +40,7 @@ def splice_url(url_list: list[str]) -> str:
 class Request:
     """网络请求相关类"""
 
-    _CACHE_FILE = CACHE_PATH.joinpath('cloudflare_cache.json')
+    _CACHE_FILE = CACHE_PATH / 'cloudflare_cache.json'
     _headers: dict | None = None
     _cookies: dict | None = None
 
@@ -45,36 +48,35 @@ class Request:
     async def _anti_cloudflare(cls, url: str) -> bytes:
         """用firefox硬穿五秒盾"""
         browser = await BrowserManager.get_browser()
-        context = await browser.new_context()
-        page = await context.new_page()
-        response = await page.goto(url)
-        attempts = 0
-        while attempts < 60:  # noqa: PLR2004
-            attempts += 1
-            text = await page.locator('body').text_content()
-            if text is None:
-                await page.wait_for_timeout(1000)
-                continue
-            if await page.title() == 'Please Wait... | Cloudflare':
-                logger.warning('疑似触发了 Cloudflare 的验证码')
-                break
-            try:
-                loads(text)
-            except JSONDecodeError:
-                await page.wait_for_timeout(1000)
-            else:
-                if not isinstance(response, Response):
-                    raise RequestError('api请求失败')
-                cls._headers = await response.request.all_headers()
+        async with await browser.new_context() as context, await context.new_page() as page:
+            response = await page.goto(url)
+            attempts = 0
+            while attempts < 60:  # noqa: PLR2004
+                attempts += 1
+                text = await page.locator('body').text_content()
+                if text is None:
+                    await page.wait_for_timeout(1000)
+                    continue
+                if await page.title() == 'Please Wait... | Cloudflare':
+                    logger.warning('疑似触发了 Cloudflare 的验证码')
+                    break
                 try:
-                    cls._cookies = {i['name']: i['value'] for i in await context.cookies()}
-                except KeyError:
-                    cls._cookies = None
-                await page.close()
-                await context.close()
-                return await response.body()
-        await page.close()
-        await context.close()
+                    loads(text)
+                except JSONDecodeError:
+                    await page.wait_for_timeout(1000)
+                else:
+                    if not isinstance(response, Response):
+                        raise RequestError('api请求失败')
+                    cls._headers = await response.request.all_headers()
+                    try:
+                        cls._cookies = {
+                            name: value
+                            for i in await context.cookies()
+                            if (name := i.get('name')) is not None and (value := i.get('value')) is not None
+                        }
+                    except KeyError:
+                        cls._cookies = None
+                    return await response.body()
         raise RequestError('绕过五秒盾失败')
 
     @classmethod
@@ -115,14 +117,49 @@ class Request:
     async def request(cls, url: str, *, is_json: bool = True) -> bytes:
         """请求api"""
         try:
-            async with AsyncClient(cookies=cls._cookies) as session:
+            async with AsyncClient(cookies=cls._cookies, timeout=config.tetris_req_timeout) as session:
                 response = await session.get(url, headers=cls._headers)
+                if response.status_code != HTTPStatus.OK:
+                    raise RequestError(
+                        f'请求错误 code: {response.status_code} {HTTPStatus(response.status_code).phrase}\n{response.text}',
+                        status_code=response.status_code,
+                    )
                 if is_json:
                     loads(response.content)
                 return response.content
         except HTTPError as e:
-            raise RequestError(f'请求错误\n{e!r}') from e
+            raise RequestError(f'请求错误 \n{e!r}') from e
         except JSONDecodeError:
             if urlparse(url).netloc.lower().endswith('tetr.io'):
                 return await cls._anti_cloudflare(url)
             raise
+
+    @classmethod
+    async def failover_request(
+        cls,
+        urls: Sequence[str],
+        *,
+        failover_code: Sequence[int],
+        failover_exc: tuple[type[BaseException], ...],
+        is_json: bool = True,
+    ) -> bytes:
+        error_list: list[RequestError] = []
+        for i in urls:
+            logger.debug(f'尝试请求 {i}')
+            try:
+                return await cls.request(i, is_json=is_json)
+            except RequestError as e:
+                if e.status_code in failover_code:  # 如果状态码在 failover_code 中, 则继续尝试下一个URL
+                    error_list.append(e)
+                    continue
+                # 如果状态码不在故障转移列表中, 则查找异常栈, 如果异常栈内有 failover_exc 内的异常类型, 则继续尝试下一个URL
+                tb = e.__traceback__
+                while tb is not None:
+                    if isinstance(tb.tb_frame.f_locals.get('exc_value'), failover_exc):
+                        error_list.append(e)
+                        break
+                    tb = tb.tb_next
+                else:
+                    raise
+                continue
+        raise RequestError(f'所有地址皆不可用\n{error_list!r}')

nonebot_plugin_tetris_stats/utils/retry.py

@@ -0,0 +1,37 @@
+from asyncio import sleep
+from collections.abc import Awaitable, Callable
+from datetime import timedelta
+from functools import wraps
+from typing import TypeVar, cast
+
+from nonebot.log import logger
+
+T = TypeVar('T')
+
+
+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]]:
+        @wraps(func)
+        async def wrapper(*args, **kwargs) -> T:  # noqa: ANN002, ANN003
+            attempts = 0
+            while attempts < max_attempts + 1:
+                try:
+                    return await func(*args, **kwargs)
+                except exception_type as e:  # noqa: PERF203
+                    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
+            raise RuntimeError('Unexpectedly reached the end of the retry loop')
+
+        return cast(Callable[..., Awaitable[T]], wrapper)
+
+    return decorator

nonebot_plugin_tetris_stats/utils/screenshot.py

@@ -0,0 +1,11 @@
+from .browser import BrowserManager
+
+
+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 page.goto(url)
+        await page.wait_for_load_state('networkidle')
+        return await page.screenshot(full_page=True, type='png')

nonebot_plugin_tetris_stats/utils/templates.py

@@ -0,0 +1,60 @@
+from asyncio.subprocess import PIPE, create_subprocess_exec
+from shutil import rmtree
+
+from nonebot import get_driver
+from nonebot.log import logger
+from nonebot_plugin_localstore import get_data_dir  # type: ignore[import-untyped]
+
+driver = get_driver()
+
+templates_dir = get_data_dir('nonebot_plugin_tetris_stats') / 'templates'
+
+
+@driver.on_startup
+async def init_templates() -> None:
+    try:
+        await create_subprocess_exec('git', '--version', stdout=PIPE)
+    except FileNotFoundError as e:
+        raise RuntimeError(
+            '未找到 git, 请确保 git 已安装并在环境变量中\n安装步骤请参阅: https://git-scm.com/book/zh/v2/%E8%B5%B7%E6%AD%A5-%E5%AE%89%E8%A3%85-Git'
+        ) from e
+    if not templates_dir.exists():
+        logger.info('模板仓库不存在, 正在尝试初始化...')
+        proc = await create_subprocess_exec(
+            'git',
+            'clone',
+            '-b',
+            'gh-pages',
+            'https://github.com/A-Minos/tetris-stats-templates',
+            templates_dir,
+            '--depth=1',
+            stdout=PIPE,
+            stderr=PIPE,
+        )
+        stdout, stderr = await proc.communicate()
+        if proc.returncode != 0:
+            for i in stderr.decode().splitlines():
+                logger.error(i)
+            raise RuntimeError('初始化模板仓库失败')
+        logger.success('模板仓库初始化成功')
+        return
+    proc = await create_subprocess_exec(
+        'git', 'rev-parse', '--is-inside-work-tree', stdout=PIPE, stderr=PIPE, cwd=templates_dir
+    )
+    stdout, stderr = await proc.communicate()
+    if proc.returncode != 0:
+        for i in stderr.decode().splitlines():
+            logger.error(i)
+        logger.warning('模板仓库状态异常, 尝试重新初始化')
+        rmtree(templates_dir)
+        await init_templates()
+        return
+    logger.info('正在更新模板仓库...')
+    proc = await create_subprocess_exec('git', 'pull', stdout=PIPE, stderr=PIPE, cwd=templates_dir)
+    stdout, stderr = await proc.communicate()
+    logger.info(stdout.decode().strip())
+    if proc.returncode != 0:
+        for i in stderr.decode().splitlines():
+            logger.error(i)
+        raise RuntimeError('更新模板仓库失败')
+    logger.success('模板仓库更新成功')

nonebot_plugin_tetris_stats/utils/typing.py

@@ -1,7 +1,7 @@
 from collections.abc import Awaitable, Callable
 from typing import Any, Literal
 
-Number = int | float
+Number = float | int
 GameType = Literal['IO', 'TOP', 'TOS']
 CommandType = Literal['bind', 'query']
 AsyncCallable = Callable[..., Awaitable[Any]]

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = 'nonebot-plugin-tetris-stats'
-version = '1.0.0.a2'
+version = '1.1.0'
 description = '一款基于 NoneBot2 的用于查询 Tetris 相关游戏数据的插件'
 authors = ['scdhh <wallfjjd@gmail.com>']
 readme = 'README.md'
@@ -10,39 +10,51 @@ license = 'AGPL-3.0'
 
 [tool.poetry.dependencies]
 python = '^3.10'
-nonebot2 = '^2.0.0-beta.3'
-lxml = '^4.9.1'
+nonebot2 = { extras = ["fastapi"], version = "^2.3.0" }
+lxml = '^5.1.0'
 pandas = '>=1.4.3,<3.0.0'
-playwright = '^1.24.1'
-ujson = '^5.4.0'
+playwright = '^1.41.2'
+ujson = '^5.9.0'
 aiofiles = "^23.2.1"
-nonebot-plugin-orm = ">=0.1.1,<0.6.0"
-nonebot-plugin-localstore = "^0.5.1"
-httpx = "^0.25.0"
-nonebot-plugin-alconna = ">=0.30,<0.34"
-nonebot-plugin-apscheduler = "^0.3.0"
+nonebot-plugin-orm = ">=0.1.1,<0.8.0"
+nonebot-plugin-localstore = "^0.6.0"
+httpx = "^0.27.0"
+nonebot-plugin-alconna = ">=0.40"
+nonebot-plugin-apscheduler = "^0.4.0"
+aiocache = "^0.12.2"
+zstandard = "^0.22.0"
+jinja2 = "^3.1.3"
+nonebot-plugin-userinfo = "^0.2.4"
+pillow = "^10.3.0"
+rich = "^13.7.1"
 
 [tool.poetry.group.dev.dependencies]
-mypy = '>=0.991,<1.8'
-types-ujson = '^5.7.0'
+mypy = '>=1.9'
+types-ujson = '^5.9.0'
 pandas-stubs = '>=1.5.2,<3.0.0'
-ruff = '>=0.0.239,<0.1.6'
-types-aiofiles = "^23.2.0.0"
-nonebot2 = { extras = ["fastapi"], version = "^2.1.1" }
-types-lxml = "^2023.3.28"
-nonebot-plugin-orm = { extras = ["default"], version = ">=0.3,<0.6" }
-nonebot-adapter-onebot = "^2.3.1"
-nonebot-adapter-satori = "^0.7.0"
+ruff = '>=0.3.0'
+types-aiofiles = "^23.2.0.20240106"
+types-lxml = "^2024.2.9"
+nonebot-plugin-orm = { extras = ["default"], version = ">=0.3,<0.8" }
+nonebot-adapter-onebot = "^2.4.1"
+nonebot-adapter-satori = "^0.11.4"
+nonebot-adapter-kaiheila = "^0.3.4"
+nonebot-adapter-discord = "^0.1.3"
+types-pillow = "^10.2.0.20240423"
 
 [tool.poetry.group.debug.dependencies]
 objprint = '^0.2.2'
-viztracer = "^0.16.0"
+viztracer = "^0.16.2"
 
 [build-system]
 requires = ['poetry-core>=1.0.0']
 build-backend = 'poetry.core.masonry.api'
 
 [tool.ruff]
+line-length = 120
+target-version = "py310"
+
+[tool.ruff.lint]
 select = [
     'F',     # pyflakes
     'E',     # pycodestyle errors
@@ -83,14 +95,12 @@ ignore = [
     'ANN202', # 向 NoneBot 注册的函数
     'TRY003',
 ]
-line-length = 120
-target-version = "py310"
 flake8-quotes = { inline-quotes = 'single', multiline-quotes = 'double' }
 
-[tool.ruff.flake8-annotations]
+[tool.ruff.lint.flake8-annotations]
 mypy-init-return = true
 
-[tool.ruff.flake8-builtins]
+[tool.ruff.lint.flake8-builtins]
 builtins-ignorelist = ["id"]
 
 [tool.ruff.format]