🔖 1.1.4

* 🔥 删除现有模板

* ✨ 自动克隆模板仓库

* 🔥 删除 identicon 相关代码

* 🚚 修改静态文件路径

* ✨ 使用新模板进行渲染

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

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

* 🚚 修改网络路径

* ✨ TOP 绑定图使用新模板

* ✨ TOS 绑定图使用新模板

* 🐛 防止截图超时

* 🐛 Pydantic V1 会把 float 转换成 int

* ✏️ 模板字段名写错了

* ✨ 兼容 Pydantic V1

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

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

.gitignore

@@ -8,3 +8,14 @@ Untitled*
 *_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/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/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/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/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

@@ -3,10 +3,12 @@ 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 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.schemas import BaseProcessedData, BaseUser
 from ..utils.typing import CommandType, GameType
@@ -15,25 +17,38 @@ from ..utils.typing import CommandType, GameType
 class PydanticType(TypeDecorator):
     impl = JSON
 
-    def __init__(self, get_model: Callable[[], Sequence[type[BaseModel]]], *args: Any, **kwargs: Any):  # noqa: ANN401
+    @override
+    def __init__(self, get_model: Callable[[], Sequence[type[BaseModel]]], *args: Any, **kwargs: Any):
         self.get_model = get_model
         super().__init__(*args, **kwargs)
 
-    def process_bind_param(self, value: Any | None, dialect: Dialect) -> str:  # noqa: ANN401
-        # 将 Pydantic 模型实例转换为 JSON
-        if isinstance(value, tuple(self.get_model())):
-            return value.json()  # type: ignore[union-attr]
-        raise TypeError
+    if PYDANTIC_V2:
 
-    def process_result_value(self, value: Any | None, dialect: Dialect) -> BaseModel:  # noqa: ANN401
+        @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 i.parse_raw(value)
+                    return type_validate_json(i, value)
                 except ValidationError:  # noqa: PERF203
                     ...
-        raise TypeError
+        raise ValueError
 
 
 class Bind(MappedAsDataclass, Model):
@@ -55,6 +70,7 @@ 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)
+    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

nonebot_plugin_tetris_stats/game_data_processor/__init__.py

@@ -1,9 +1,11 @@
 from abc import ABC, abstractmethod
-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
@@ -12,6 +14,8 @@ from .schemas import BaseProcessedData as ProcessedData
 from .schemas import BaseRawResponse as RawResponse
 from .schemas import BaseUser as User
 
+UTC = timezone.utc
+
 
 class Processor(ABC):
     event_id: int
@@ -39,20 +43,22 @@ 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):
@@ -62,6 +68,7 @@ class Processor(ABC):
         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

nonebot_plugin_tetris_stats/game_data_processor/io_data_processor/__init__.py

@@ -1,11 +1,13 @@
-from datetime import UTC, datetime, timedelta
+from datetime import datetime, timedelta, timezone
 from zoneinfo import ZoneInfo
 
 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 At, on_alconna
+from nonebot_plugin_alconna.uniseg import UniMessage
 from nonebot_plugin_orm import get_session
+from nonebot_plugin_userinfo import BotUserInfo, UserInfo  # type: ignore[import-untyped]
 from sqlalchemy import func, select
 
 from ...db import query_bind_info
@@ -20,6 +22,8 @@ from .model import IORank
 from .processor import Processor, User, identify_user_info
 from .typing import Rank
 
+UTC = timezone.utc
+
 alc = on_alconna(
     Alconna(
         'io',
@@ -84,14 +88,12 @@ 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.send(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.send(str(e))
         raise HandleNotFinishedError from e
@@ -115,7 +117,7 @@ async def _(bot: Bot, event: Event, matcher: Matcher, target: At | Me):
         command_args=[],
     )
     try:
-        await matcher.send(message + await proc.handle_query())
+        await (UniMessage(message) + await proc.handle_query()).finish()
     except NeedCatchError as e:
         await matcher.send(str(e))
         raise HandleNotFinishedError from e
@@ -129,7 +131,7 @@ async def _(event: Event, matcher: Matcher, account: User):
         command_args=[],
     )
     try:
-        await matcher.send(await proc.handle_query())
+        await (await proc.handle_query()).finish()
     except NeedCatchError as e:
         await matcher.send(str(e))
         raise HandleNotFinishedError from e
@@ -149,19 +151,19 @@ async def _(matcher: Matcher, rank: Rank):
                 .where(IORank.rank == rank)
                 .order_by(
                     func.abs(
-                        func.julianday(IORank.create_time)
-                        - func.julianday(latest_data.create_time - timedelta(hours=24))
+                        func.julianday(IORank.update_time)
+                        - func.julianday(latest_data.update_time - timedelta(hours=24))
                     )
                 )
                 .limit(1)
             )
         ).one()
     message = ''
-    if (datetime.now(UTC) - latest_data.create_time.replace(tzinfo=UTC)) > timedelta(hours=7):
+    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.create_time-compare_data.create_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 "→"}'
+        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)
@@ -186,7 +188,7 @@ async def _(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.replace(tzinfo=UTC).astimezone(ZoneInfo("Asia/Shanghai")).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)
 

nonebot_plugin_tetris_stats/game_data_processor/io_data_processor/cache.py

@@ -1,12 +1,14 @@
-from datetime import UTC, datetime
+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 pydantic import parse_raw_as
 
 from ...utils.request import Request
 from .schemas.base import FailedModel, SuccessModel
 
+UTC = timezone.utc
+
 
 class Cache:
     cache = ACache(ACache.MEMORY)
@@ -16,7 +18,7 @@ class Cache:
         cached_data = await cls.cache.get(url)
         if cached_data is None:
             response_data = await Request.request(url)
-            parsed_data: SuccessModel | FailedModel = parse_raw_as(SuccessModel | FailedModel, response_data)  # type: ignore[arg-type]
+            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,

nonebot_plugin_tetris_stats/game_data_processor/io_data_processor/constant.py

@@ -23,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,38 +1,54 @@
+from asyncio import gather
 from collections import defaultdict
 from collections.abc import Callable
-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.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 .cache import Cache
-from .constant import BASE_URL, GAME_TYPE, RANK_PERCENTILE
+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.response import ProcessedData, RawResponse
 from .schemas.user import User
-from .schemas.user_info import FailedModel as InfoFailed
-from .schemas.user_info import NeverPlayedLeague, NeverRatedLeague, UserInfo
+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()
 
 
@@ -44,79 +60,131 @@ 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
+    @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()
-
-    async def get_user(self) -> None:
-        """
-        用于获取 UserName 和 UserID 的函数
-        """
-        if self.user.name is None:
-            self.user.name = (await self.get_user_info()).data.user.username
-        if self.user.ID is None:
-            self.user.ID = (await self.get_user_info()).data.user.id
-
-    async def get_user_info(self) -> InfoSuccess:
-        """获取用户数据"""
-        if self.processed_data.user_info is None:
-            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):
-                raise RequestError(f'用户信息请求错误:\n{user_info.error}')
-            self.processed_data.user_info = user_info
-        return self.processed_data.user_info
-
-    async def get_user_records(self) -> RecordsSuccess:
-        """获取Solo数据"""
-        if self.processed_data.user_records is None:
-            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, self.raw_response.user_records)  # type: ignore[arg-type]
-            if isinstance(user_records, RecordsFailed):
-                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()
+        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} 没有排位统计数据'
@@ -133,31 +201,203 @@ class Processor(ProcessorMeta):
                 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)} )'
+            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 )'
-        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
+        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)
+        if not historical_data:
+            return [
+                TETRIOInfo.TetraLeagueHistory.Data(record_at=today - forward, tr=user_info.data.user.league.rating),
+                TETRIOInfo.TetraLeagueHistory.Data(
+                    record_at=today.replace(microsecond=1000), tr=user_info.data.user.league.rating
+                ),
+            ]
+        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 TETRIOInfo.User.Avatar(
+                            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:
+        """
+        用于获取 UserName 和 UserID 的函数
+        """
+        if self.user.name is None:
+            self.user.name = (await self.get_user_info()).data.user.username
+        if self.user.ID is None:
+            self.user.ID = (await self.get_user_info()).data.user.id
+
+    async def get_user_info(self) -> InfoSuccess:
+        """获取用户数据"""
+        if self.processed_data.user_info is None:
+            self.raw_response.user_info = await Cache.get(
+                splice_url([BASE_URL, 'users/', f'{self.user.ID or self.user.name}'])
+            )
+            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
+
+    async def get_user_records(self) -> RecordsSuccess:
+        """获取Solo数据"""
+        if self.processed_data.user_records is None:
+            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 = 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
 
 
 @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(LeagueAll, await Cache.get(splice_url([BASE_URL, 'users/lists/league/all'])))  # type: ignore[arg-type]
-    if isinstance(league_all, LeagueAllFailed):
+    league_all: LeagueAll = type_validate_json(
+        LeagueAll,  # type: ignore[arg-type]
+        (data := await Cache.get(splice_url([BASE_URL, 'users/lists/league/all']))),
+    )
+    if isinstance(league_all, FailedModel):
         raise RequestError(f'排行榜数据请求错误:\n{league_all.error}')
 
     def pps(user: LeagueAllUser) -> float:
@@ -183,6 +423,10 @@ async def get_io_rank_data() -> None:
         user = sort(users, field)
         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)
     for i in users:
@@ -206,6 +450,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:
@@ -216,6 +462,6 @@ async def get_io_rank_data() -> None:
 @driver.on_startup
 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))
+        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/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]
@@ -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 | None
-            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,14 +12,14 @@ class EndContext(BaseModel):
         zero: bool
         locked: bool
         prev: int
-        frameoffset: int | None
+        frameoffset: int | None = None
 
     class Clears(BaseModel):
         singles: int
         doubles: int
         triples: int
         quads: int
-        pentas: int | None
+        pentas: int | None = None
         realtspins: int
         minitspins: int
         minitspinsingles: int
@@ -33,7 +33,7 @@ class EndContext(BaseModel):
     class Garbage(BaseModel):
         sent: int
         received: int
-        attack: int | None
+        attack: int | None = None
         cleared: int
 
     class Finesse(BaseModel):
@@ -46,18 +46,18 @@ class EndContext(BaseModel):
     level_lines: int
     level_lines_needed: int
     inputs: int
-    holds: int | None
+    holds: int | None = None
     time: Time
     score: int
-    zenlevel: int | None
-    zenprogress: int | None
+    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
@@ -68,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/top_data_processor/__init__.py

@@ -2,7 +2,9 @@ from arclet.alconna import Alconna, AllParam, Arg, ArgFlag, Args, CommandMeta, O
 from nonebot.adapters import Bot, Event
 from nonebot.matcher import Matcher
 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 HandleNotFinishedError, NeedCatchError
@@ -67,14 +69,25 @@ 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.send(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.send(str(e))
         raise HandleNotFinishedError from e
@@ -98,7 +111,7 @@ async def _(bot: Bot, event: Event, matcher: Matcher, target: At | Me):
         command_args=[],
     )
     try:
-        await matcher.send(message + await proc.handle_query())
+        await (UniMessage(message) + await proc.handle_query()).finish()
     except NeedCatchError as e:
         await matcher.send(str(e))
         raise HandleNotFinishedError from e
@@ -112,7 +125,7 @@ async def _(event: Event, matcher: Matcher, account: User):
         command_args=[],
     )
     try:
-        await matcher.send(await proc.handle_query())
+        await (await proc.handle_query()).finish()
     except NeedCatchError as e:
         await matcher.send(str(e))
         raise HandleNotFinishedError from e

nonebot_plugin_tetris_stats/game_data_processor/top_data_processor/processor.py

@@ -2,16 +2,23 @@ from contextlib import suppress
 from dataclasses import dataclass
 from io import StringIO
 from re import match
-from typing import Literal, 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.screenshot import screenshot
 from .. import Processor as ProcessorMeta
 from ..schemas import BaseUser
 from .constant import BASE_URL, GAME_TYPE
@@ -24,6 +31,7 @@ class User(BaseUser):
     name: str
 
     @property
+    @override
     def unique_identifier(self) -> str:
         return self.name
 
@@ -51,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
+    @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:
         """获取用户信息"""
@@ -87,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:
         """获取用户名"""
@@ -115,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/tos_data_processor/__init__.py

@@ -4,7 +4,9 @@ from arclet.alconna import Alconna, AllParam, Arg, ArgFlag, Args, CommandMeta, O
 from nonebot.adapters import Bot, Event
 from nonebot.matcher import Matcher
 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 HandleNotFinishedError, NeedCatchError, RequestError
@@ -71,7 +73,7 @@ alc = on_alconna(
 
 async def finish_special_query(matcher: Matcher, proc: Processor) -> NoReturn:
     try:
-        await matcher.finish(await proc.handle_query())
+        await (await proc.handle_query()).finish()
     except NeedCatchError as e:
         if isinstance(e, RequestError) and '未找到此用户' in e.message:
             matcher.skip()
@@ -94,6 +96,7 @@ try:
             command_args=[],
         )
         await finish_special_query(matcher, proc)
+
 except ImportError:
     pass
 
@@ -108,6 +111,7 @@ try:
             command_args=[],
         )
         await finish_special_query(matcher, proc)
+
 except ImportError:
     pass
 
@@ -122,19 +126,31 @@ try:
             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.send(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.send(str(e))
         raise HandleNotFinishedError from e
@@ -158,7 +174,7 @@ async def _(bot: Bot, event: Event, matcher: Matcher, target: At | Me):
         command_args=[],
     )
     try:
-        await matcher.send(message + await proc.handle_query())
+        await (UniMessage(message) + await proc.handle_query()).finish()
     except NeedCatchError as e:
         await matcher.send(str(e))
         raise HandleNotFinishedError from e
@@ -172,7 +188,7 @@ async def _(event: Event, matcher: Matcher, account: User):
         command_args=[],
     )
     try:
-        await matcher.send(await proc.handle_query())
+        await (await proc.handle_query()).finish()
     except NeedCatchError as e:
         await matcher.send(str(e))
         raise HandleNotFinishedError from e

nonebot_plugin_tetris_stats/game_data_processor/tos_data_processor/constant.py

@@ -1,4 +1,10 @@
 from typing import Literal
 
 GAME_TYPE: Literal['TOS'] = 'TOS'
-BASE_URL = 'https://teatube.cn:8888/'
+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,14 +1,22 @@
 from dataclasses import dataclass
 from re import match
 from typing import Literal
-from urllib.parse import urlencode
+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.screenshot import screenshot
 from .. import Processor as ProcessorMeta
 from ..schemas import BaseUser
 from .constant import BASE_URL, GAME_TYPE
@@ -25,6 +33,7 @@ class User(BaseUser):
     name: str | None = None
 
     @property
+    @override
     def unique_identifier(self) -> str:
         if self.teaid is None:
             raise ValueError('不完整的User!')
@@ -63,33 +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
+    @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()
         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.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:
         """
@@ -104,23 +159,31 @@ 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
@@ -132,16 +195,23 @@ class Processor(ProcessorMeta):
             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.request(
-                splice_url(
-                    [
-                        BASE_URL,
-                        'getProfile',
-                        f'?{urlencode({"id":self.user.teaid or self.user.name,**other_parameter})}',
-                    ]
-                )
+            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[params] = type_validate_json(
+                UserProfile, self.raw_response.user_profile[params]
             )
-            self.processed_data.user_profile[params] = UserProfile.parse_raw(self.raw_response.user_profile[params])
         return self.processed_data.user_profile[params]
 
     async def get_game_data(self) -> GameData | None:
@@ -185,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/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

@@ -45,7 +45,7 @@ class BrowserManager:
                 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

nonebot_plugin_tetris_stats/utils/exception.py

@@ -18,6 +18,10 @@ class NeedCatchError(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):
     """用户发送的消息格式不正确"""

nonebot_plugin_tetris_stats/utils/host.py

@@ -0,0 +1,70 @@
+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.log import logger
+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)
+
+
+@driver.on_startup
+def _():
+    app.mount(
+        '/host/assets',
+        StaticFiles(directory=templates_dir / 'assets'),
+        name='assets',
+    )
+    logger.success('assets mounted')
+
+
+@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,6 +9,8 @@ from nonebot_plugin_orm import get_session
 
 from ..db.models import HistoricalData
 
+UTC = timezone.utc
+
 driver = get_driver()
 
 

nonebot_plugin_tetris_stats/utils/render.py

@@ -0,0 +1,108 @@
+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):
+        class Avatar(BaseModel):
+            type: Literal['identicon']
+            hash: str
+
+        avatar: str | Avatar
+        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,9 +1,10 @@
+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
@@ -13,7 +14,7 @@ from .browser import BrowserManager
 from .exception import RequestError
 
 driver = get_driver()
-config = Config.parse_obj(driver.config)
+config = get_plugin_config(Config)
 
 
 @driver.on_startup
@@ -68,7 +69,11 @@ class Request:
                         raise RequestError('api请求失败')
                     cls._headers = await response.request.all_headers()
                     try:
-                        cls._cookies = {i['name']: i['value'] for i in await context.cookies()}
+                        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()
@@ -116,7 +121,8 @@ class Request:
                 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}'
+                        f'请求错误 code: {response.status_code} {HTTPStatus(response.status_code).phrase}\n{response.text}',
+                        status_code=response.status_code,
                     )
                 if is_json:
                     loads(response.content)
@@ -127,3 +133,33 @@ class Request:
             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/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,70 @@
+from asyncio.subprocess import PIPE, create_subprocess_exec
+from shutil import rmtree
+
+from nonebot import get_driver
+from nonebot.log import logger
+from nonebot.permission import SUPERUSER
+from nonebot_plugin_alconna import on_alconna
+from nonebot_plugin_localstore import get_data_dir  # type: ignore[import-untyped]
+
+driver = get_driver()
+
+templates_dir = get_data_dir('nonebot_plugin_tetris_stats') / 'templates'
+
+alc = on_alconna('更新模板', permission=SUPERUSER)
+
+
+@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('模板仓库更新成功')
+
+
+@alc.handle()
+async def _():
+    await init_templates()
+    await alc.finish('模板仓库更新成功')

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.a10'
+version = '1.1.4'
 description = '一款基于 NoneBot2 的用于查询 Tetris 相关游戏数据的插件'
 authors = ['scdhh <wallfjjd@gmail.com>']
 readme = 'README.md'
@@ -10,42 +10,51 @@ license = 'AGPL-3.0'
 
 [tool.poetry.dependencies]
 python = '^3.10'
-nonebot2 = '^2.1.3'
-lxml = '^5.0.0'
+nonebot2 = { extras = ["fastapi"], version = "^2.3.0" }
+lxml = '^5.1.0'
 pandas = '>=1.4.3,<3.0.0'
-playwright = '^1.40.0'
+playwright = '^1.41.2'
 ujson = '^5.9.0'
 aiofiles = "^23.2.1"
-nonebot-plugin-orm = ">=0.1.1,<0.7.0"
-nonebot-plugin-localstore = "^0.5.1"
-httpx = "^0.26.0"
-nonebot-plugin-alconna = ">=0.30,<0.36"
-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.9'
+mypy = '>=1.9'
 types-ujson = '^5.9.0'
 pandas-stubs = '>=1.5.2,<3.0.0'
-ruff = '>=0.0.239,<0.1.10'
-types-aiofiles = "^23.2.0.0"
-nonebot2 = { extras = ["fastapi"], version = "^2.1.3" }
-types-lxml = "^2023.3.28"
-nonebot-plugin-orm = { extras = ["default"], version = ">=0.3,<0.7" }
-nonebot-adapter-onebot = "^2.3.1"
-nonebot-adapter-satori = "^0.8.1"
-nonebot-adapter-kaiheila = "^0.3.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.1"
+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
@@ -86,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]