🎨 重命名一些模块

nonebot_plugin_tetris_stats/games/top/__init__.py

@@ -0,0 +1,72 @@
+from arclet.alconna import Alconna, AllParam, Arg, ArgFlag, Args, CommandMeta, Option
+from nonebot_plugin_alconna import At, on_alconna
+
+from ...utils.exception import MessageFormatError
+from ...utils.typing import Me
+from .. import add_default_handlers
+from ..constant import BIND_COMMAND, QUERY_COMMAND
+from .api import Player
+from .constant import USER_NAME
+
+
+def get_player(name: str) -> Player | MessageFormatError:
+    if USER_NAME.match(name):
+        return Player(user_name=name, trust=True)
+    return MessageFormatError('用户名/ID不合法')
+
+
+alc = on_alconna(
+    Alconna(
+        'top',
+        Option(
+            BIND_COMMAND[0],
+            Args(
+                Arg(
+                    'account',
+                    get_player,
+                    notice='TOP 用户名',
+                    flags=[ArgFlag.HIDDEN],
+                )
+            ),
+            alias=BIND_COMMAND[1:],
+            compact=True,
+            dest='bind',
+            help_text='绑定 TOP 账号',
+        ),
+        Option(
+            QUERY_COMMAND[0],
+            Args(
+                Arg(
+                    'target',
+                    At | Me,
+                    notice='@想要查询的人 | 自己',
+                    flags=[ArgFlag.HIDDEN, ArgFlag.OPTIONAL],
+                ),
+                Arg(
+                    'account',
+                    get_player,
+                    notice='TOP 用户名',
+                    flags=[ArgFlag.HIDDEN, ArgFlag.OPTIONAL],
+                ),
+            ),
+            alias=QUERY_COMMAND[1:],
+            compact=True,
+            dest='query',
+            help_text='查询 TOP 游戏信息',
+        ),
+        Arg('other', AllParam, flags=[ArgFlag.HIDDEN, ArgFlag.OPTIONAL]),
+        meta=CommandMeta(
+            description='查询 TetrisOnline波兰服 的信息',
+            example='top绑定scdhh\ntop查我',
+            compact=True,
+            fuzzy_match=True,
+        ),
+    ),
+    skip_for_unmatch=False,
+    auto_send_output=True,
+    aliases={'TOP'},
+)
+
+from . import bind, query  # noqa: E402, F401
+
+add_default_handlers(alc)

nonebot_plugin_tetris_stats/games/top/api/__init__.py

@@ -0,0 +1,3 @@
+from .player import Player
+
+__all__ = ['Player']

nonebot_plugin_tetris_stats/games/top/api/models.py

@@ -0,0 +1,17 @@
+from datetime import datetime
+from typing import Literal
+
+from nonebot_plugin_orm import Model
+from sqlalchemy import DateTime, String
+from sqlalchemy.orm import Mapped, MappedAsDataclass, mapped_column
+
+from ....db.models import PydanticType
+from .schemas.user_profile import UserProfile
+
+
+class TOPHistoricalData(MappedAsDataclass, Model):
+    id: Mapped[int] = mapped_column(init=False, primary_key=True)
+    user_unique_identifier: Mapped[str] = mapped_column(String(24), index=True)
+    api_type: Mapped[Literal['User Profile']] = mapped_column(String(16), index=True)
+    data: Mapped[UserProfile] = mapped_column(PydanticType(get_model=[], models={UserProfile}))
+    update_time: Mapped[datetime] = mapped_column(DateTime, index=True)

nonebot_plugin_tetris_stats/games/top/api/player.py

@@ -0,0 +1,71 @@
+from contextlib import suppress
+from datetime import datetime, timezone
+from io import StringIO
+from urllib.parse import urlencode
+
+from lxml import etree
+from pandas import read_html
+
+from ....db import anti_duplicate_add
+from ....utils.request import Request, splice_url
+from ..constant import BASE_URL, USER_NAME
+from .models import TOPHistoricalData
+from .schemas.user import User
+from .schemas.user_profile import Data, UserProfile
+
+UTC = timezone.utc
+
+
+class Player:
+    def __init__(self, *, user_name: str, trust: bool = False) -> None:
+        self.user_name = user_name
+        if not trust and not USER_NAME.match(self.user_name):
+            msg = 'Invalid user name'
+            raise ValueError(msg)
+        self.__user: User | None = None
+        self._user_profile: UserProfile | None = None
+
+    @property
+    async def user(self) -> User:
+        if self.__user is None:
+            profile = await self.get_profile()
+            self.__user = User(user_name=profile.user_name)
+        return self.__user
+
+    async def get_profile(self) -> UserProfile:
+        """获取用户信息"""
+        if self._user_profile is None:
+            url = splice_url([BASE_URL, 'profile.php', f'?{urlencode({"user":self.user_name})}'])
+            raw_user_profile = await Request.request(url, is_json=False)
+            self._user_profile = self._parse_profile(raw_user_profile)
+            await anti_duplicate_add(
+                TOPHistoricalData,
+                TOPHistoricalData(
+                    user_unique_identifier=(await self.user).unique_identifier,
+                    api_type='User Profile',
+                    data=self._user_profile,
+                    update_time=datetime.now(tz=UTC),
+                ),
+            )
+        return self._user_profile
+
+    def _parse_profile(self, original_user_profile: bytes) -> UserProfile:
+        html = etree.HTML(original_user_profile)
+        user_name = html.xpath('//div[@class="mycontent"]/h1/text()')[0].replace("'s profile", '')
+        today = None
+        with suppress(ValueError):
+            today = Data(
+                lpm=float(str(html.xpath('//div[@class="mycontent"]/text()[3]')[0]).replace('lpm:', '').strip()),
+                apm=float(str(html.xpath('//div[@class="mycontent"]/text()[4]')[0]).replace('apm:', '').strip()),
+            )
+        table = StringIO(
+            etree.tostring(
+                html.xpath('//div[@class="mycontent"]/table[@class="mytable"]')[0],
+                encoding='utf-8',
+            ).decode()
+        )
+        dataframe = read_html(table, encoding='utf-8', header=0)[0]
+        total: list[Data] = []
+        for _, value in dataframe.iterrows():
+            total.append(Data(lpm=value['lpm'], apm=value['apm']))
+        return UserProfile(user_name=user_name, today=today, total=total)

nonebot_plugin_tetris_stats/games/top/api/schemas/user.py

@@ -0,0 +1,17 @@
+from typing import Literal
+
+from typing_extensions import override
+
+from ....schemas import BaseUser
+from ...constant import GAME_TYPE
+
+
+class User(BaseUser):
+    platform: Literal['TOP'] = GAME_TYPE
+
+    user_name: str
+
+    @property
+    @override
+    def unique_identifier(self) -> str:
+        return self.user_name

nonebot_plugin_tetris_stats/games/top/api/schemas/user_profile.py

@@ -0,0 +1,12 @@
+from pydantic import BaseModel
+
+
+class Data(BaseModel):
+    lpm: float
+    apm: float
+
+
+class UserProfile(BaseModel):
+    user_name: str
+    today: Data | None
+    total: list[Data] | None

nonebot_plugin_tetris_stats/games/top/bind.py

@@ -0,0 +1,66 @@
+from urllib.parse import urlunparse
+
+from nonebot.adapters import Bot
+from nonebot_plugin_alconna.uniseg import UniMessage
+from nonebot_plugin_orm import get_session
+from nonebot_plugin_session import EventSession  # type: ignore[import-untyped]
+from nonebot_plugin_session_orm import get_session_persist_id  # type: ignore[import-untyped]
+from nonebot_plugin_userinfo import BotUserInfo, EventUserInfo, UserInfo  # type: ignore[import-untyped]
+
+from ...db import BindStatus, create_or_update_bind, trigger
+from ...utils.avatar import get_avatar
+from ...utils.host import HostPage, get_self_netloc
+from ...utils.platform import get_platform
+from ...utils.render import Bind, render
+from ...utils.render.schemas.base import People
+from ...utils.screenshot import screenshot
+from . import alc
+from .api import Player
+from .constant import GAME_TYPE
+
+
+@alc.assign('bind')
+async def _(
+    bot: Bot,
+    account: Player,
+    event_session: EventSession,
+    bot_info: UserInfo = BotUserInfo(),  # noqa: B008
+    event_user_info: UserInfo = EventUserInfo(),  # noqa: B008
+):
+    async with trigger(
+        session_persist_id=await get_session_persist_id(event_session),
+        game_platform=GAME_TYPE,
+        command_type='bind',
+        command_args=[],
+    ):
+        user = await account.user
+        async with get_session() as session:
+            bind_status = await create_or_update_bind(
+                session=session,
+                chat_platform=get_platform(bot),
+                chat_account=event_user_info.user_id,
+                game_platform=GAME_TYPE,
+                game_account=user.unique_identifier,
+            )
+        if bind_status in (BindStatus.SUCCESS, BindStatus.UPDATE):
+            async with HostPage(
+                await render(
+                    'binding',
+                    Bind(
+                        platform=GAME_TYPE,
+                        status='unknown',
+                        user=People(
+                            avatar=await get_avatar(event_user_info, 'Data URI', None),
+                            name=user.user_name,
+                        ),
+                        bot=People(
+                            avatar=await get_avatar(bot_info, 'Data URI', '../../static/logo/logo.svg'),
+                            name=bot_info.user_name,
+                        ),
+                        command='top查我',
+                    ),
+                )
+            ) as page_hash:
+                await UniMessage.image(
+                    raw=await screenshot(urlunparse(('http', get_self_netloc(), f'/host/{page_hash}.html', '', '', '')))
+                ).finish()

nonebot_plugin_tetris_stats/games/top/constant.py

@@ -0,0 +1,8 @@
+from re import compile
+from typing import Literal
+
+GAME_TYPE: Literal['TOP'] = 'TOP'
+
+BASE_URL = 'http://tetrisonline.pl/top/'
+
+USER_NAME = compile(r'^[a-zA-Z0-9_]{1,16}$')

nonebot_plugin_tetris_stats/games/top/query.py

@@ -0,0 +1,73 @@
+from nonebot.adapters import Bot, Event
+from nonebot.matcher import Matcher
+from nonebot_plugin_alconna import At
+from nonebot_plugin_alconna.uniseg import UniMessage
+from nonebot_plugin_orm import get_session
+from nonebot_plugin_session import EventSession  # type: ignore[import-untyped]
+from nonebot_plugin_session_orm import get_session_persist_id  # type: ignore[import-untyped]
+
+from ...db import query_bind_info, trigger
+from ...utils.metrics import get_metrics
+from ...utils.platform import get_platform
+from ...utils.typing import Me
+from ..constant import CANT_VERIFY_MESSAGE
+from . import alc
+from .api import Player
+from .api.schemas.user_profile import UserProfile
+from .constant import GAME_TYPE
+
+
+@alc.assign('query')
+async def _(bot: Bot, event: Event, matcher: Matcher, target: At | Me, event_session: EventSession):
+    async with trigger(
+        session_persist_id=await get_session_persist_id(event_session),
+        game_platform=GAME_TYPE,
+        command_type='query',
+        command_args=[],
+    ):
+        async with get_session() as session:
+            bind = await query_bind_info(
+                session=session,
+                chat_platform=get_platform(bot),
+                chat_account=(target.target if isinstance(target, At) else event.get_user_id()),
+                game_platform=GAME_TYPE,
+            )
+        if bind is None:
+            await matcher.finish('未查询到绑定信息')
+        message = CANT_VERIFY_MESSAGE
+        await (message + make_query_text(await Player(user_name=bind.game_account, trust=True).get_profile())).finish()
+
+
+@alc.assign('query')
+async def _(account: Player, event_session: EventSession):
+    async with trigger(
+        session_persist_id=await get_session_persist_id(event_session),
+        game_platform=GAME_TYPE,
+        command_type='query',
+        command_args=[],
+    ):
+        await (make_query_text(await account.get_profile())).finish()
+
+
+def make_query_text(profile: UserProfile) -> UniMessage:
+    message = ''
+    if profile.today is not None:
+        today = get_metrics(lpm=profile.today.lpm, apm=profile.today.apm)
+        message += f'用户 {profile.user_name} 24小时内统计数据为: '
+        message += f"\nL'PM: {today.lpm} ( {today.pps} pps )"
+        message += f'\nAPM: {today.apm} ( x{today.apl} )'
+    else:
+        message += f'用户 {profile.user_name} 暂无24小时内统计数据'
+    if profile.total is not None:
+        total_lpm = total_apm = 0.0
+        for value in profile.total:
+            total_lpm += value.lpm
+            total_apm += value.apm
+        num = len(profile.total)
+        total = get_metrics(lpm=total_lpm / num, apm=total_apm / num)
+        message += '\n历史统计数据为: '
+        message += f"\nL'PM: {total.lpm} ( {total.pps} pps )"
+        message += f'\nAPM: {total.apm} ( x{total.apl} )'
+    else:
+        message += '\n暂无历史统计数据'
+    return UniMessage(message)