diff --git a/nonebot_plugin_tetris_stats/games/tetrio/__init__.py b/nonebot_plugin_tetris_stats/games/tetrio/__init__.py index 3f3c5d2..eadd47a 100644 --- a/nonebot_plugin_tetris_stats/games/tetrio/__init__.py +++ b/nonebot_plugin_tetris_stats/games/tetrio/__init__.py @@ -92,7 +92,15 @@ command.add( ), Subcommand( 'rank', - Args(Arg('rank', ValidRank, notice='TETR.IO 段位')), + Option( + '--all', + dest='all', + ), + Option( + '--detail', + Arg('rank', ValidRank), + alias=['-D'], + ), help_text='查询 TETR.IO 段位信息', ), Subcommand( @@ -108,57 +116,54 @@ command.add( ) ) + +def rank_wrapper(slot: int | str, content: str | None): + if slot == 'rank' and not content: + return '--all' + if content is not None: + return f'--detail {content.lower()}' + return content + + alc.shortcut( '(?i:io)(?i:绑定|绑|bind)', - { - 'command': 'tstats TETR.IO bind', - 'humanized': 'io绑定', - }, + command='tstats TETR.IO bind', + humanized='io绑定', ) alc.shortcut( '(?i:io)(?i:查询|查|query|stats)', - { - 'command': 'tstats TETR.IO query', - 'humanized': 'io查', - }, + command='tstats TETR.IO query', + humanized='io查', ) alc.shortcut( '(?i:io)(?i:记录|record)(?i:40l)', - { - 'command': 'tstats TETR.IO record --40l', - 'humanized': 'io记录40l', - }, + command='tstats TETR.IO record --40l', + humanized='io记录40l', ) alc.shortcut( '(?i:io)(?i:记录|record)(?i:blitz)', - { - 'command': 'tstats TETR.IO record --blitz', - 'humanized': 'io记录blitz', - }, + command='tstats TETR.IO record --blitz', + humanized='io记录blitz', ) alc.shortcut( - '(?i:io)(?i:段位|段|rank)', - { - 'command': 'tstats TETR.IO rank', - 'humanized': 'iorank', - }, + r'(?i:io)(?i:段位|段|rank)\s*(?P[a-zA-Z+-]{0,2})', + command='tstats TETR.IO rank {rank}', + humanized='iorank', + fuzzy=False, + wrapper=rank_wrapper, ) alc.shortcut( '(?i:io)(?i:配置|配|config)', - { - 'command': 'tstats TETR.IO config', - 'humanized': 'io配置', - }, + command='tstats TETR.IO config', + humanized='io配置', ) alc.shortcut( 'fkosk', - { - 'command': 'tstats TETR.IO query', - 'args': ['我'], - 'fuzzy': False, - 'humanized': 'An Easter egg!', - }, + command='tstats TETR.IO query', + arguments=['我'], + fuzzy=False, + humanized='An Easter egg!', ) add_block_handlers(alc.assign('TETRIO.query')) diff --git a/nonebot_plugin_tetris_stats/games/tetrio/constant.py b/nonebot_plugin_tetris_stats/games/tetrio/constant.py index 9bb0a94..1ba147b 100644 --- a/nonebot_plugin_tetris_stats/games/tetrio/constant.py +++ b/nonebot_plugin_tetris_stats/games/tetrio/constant.py @@ -1,13 +1,13 @@ from re import compile from typing import Literal -from .api.typing import Rank +from .api.typing import ValidRank GAME_TYPE: Literal['IO'] = 'IO' BASE_URL = 'https://ch.tetr.io/api/' -RANK_PERCENTILE: dict[Rank, float] = { +RANK_PERCENTILE: dict[ValidRank, float] = { 'x': 1, 'u': 5, 'ss': 11, diff --git a/nonebot_plugin_tetris_stats/games/tetrio/models.py b/nonebot_plugin_tetris_stats/games/tetrio/models.py index c2f927b..b36c410 100644 --- a/nonebot_plugin_tetris_stats/games/tetrio/models.py +++ b/nonebot_plugin_tetris_stats/games/tetrio/models.py @@ -4,13 +4,13 @@ from nonebot_plugin_orm import Model from sqlalchemy import JSON, DateTime, String from sqlalchemy.orm import Mapped, MappedAsDataclass, mapped_column -from .api.typing import Rank +from .api.typing import ValidRank from .typing import Template class IORank(MappedAsDataclass, Model): id: Mapped[int] = mapped_column(init=False, primary_key=True) - rank: Mapped[Rank] = mapped_column(String(2), index=True) + rank: Mapped[ValidRank] = mapped_column(String(2), index=True) tr_line: Mapped[float] player_count: Mapped[int] low_pps: Mapped[tuple[dict[str, str], float]] = mapped_column(JSON) diff --git a/nonebot_plugin_tetris_stats/games/tetrio/rank.py b/nonebot_plugin_tetris_stats/games/tetrio/rank.py deleted file mode 100644 index 7c0b047..0000000 --- a/nonebot_plugin_tetris_stats/games/tetrio/rank.py +++ /dev/null @@ -1,174 +0,0 @@ -from collections import defaultdict -from collections.abc import Callable -from datetime import datetime, timedelta, timezone -from hashlib import sha512 -from math import floor -from statistics import mean -from zoneinfo import ZoneInfo - -from aiofiles import open -from nonebot import get_driver -from nonebot.compat import model_dump -from nonebot.matcher import Matcher -from nonebot.utils import run_sync -from nonebot_plugin_apscheduler import scheduler # type: ignore[import-untyped] -from nonebot_plugin_localstore import get_data_file # type: ignore[import-untyped] -from nonebot_plugin_orm import get_session -from nonebot_plugin_session import EventSession # type: ignore[import-untyped] -from nonebot_plugin_session_orm import get_session_persist_id # type: ignore[import-untyped] -from sqlalchemy import func, select -from zstandard import ZstdCompressor - -from ...db import trigger -from ...utils.exception import RequestError -from ...utils.metrics import get_metrics -from ...utils.retry import retry -from . import alc -from .api.schemas.base import FailedModel -from .api.schemas.tetra_league import ValidUser -from .api.schemas.user import User -from .api.tetra_league import full_export -from .api.typing import Rank -from .constant import GAME_TYPE, RANK_PERCENTILE -from .models import IORank - -UTC = timezone.utc - -driver = get_driver() - - -@alc.assign('TETRIO.rank') -async def _(matcher: Matcher, rank: Rank, event_session: EventSession): - async with trigger( - session_persist_id=await get_session_persist_id(event_session), - game_platform=GAME_TYPE, - command_type='rank', - command_args=[], - ): - if rank == 'z': - await matcher.finish('暂不支持查询未知段位') - async with get_session() as session: - latest_data = ( - await session.scalars(select(IORank).where(IORank.rank == rank).order_by(IORank.id.desc()).limit(1)) - ).one() - compare_data = ( - await session.scalars( - select(IORank) - .where(IORank.rank == rank) - .order_by( - func.abs( - func.julianday(IORank.update_time) - - func.julianday(latest_data.update_time - timedelta(hours=24)) - ) - ) - .limit(1) - ) - ).one() - message = '' - if (datetime.now(UTC) - latest_data.update_time.replace(tzinfo=UTC)) > timedelta(hours=7): - message += 'Warning: 数据超过7小时未更新, 请联系Bot主人查看后台\n' - message += f'{rank.upper()} 段 分数线 {latest_data.tr_line:.2f} TR, {latest_data.player_count} 名玩家\n' - if compare_data.id != latest_data.id: - message += f'对比 {(latest_data.update_time-compare_data.update_time).total_seconds()/3600:.2f} 小时前趋势: {f"↑{difference:.2f}" if (difference:=latest_data.tr_line-compare_data.tr_line) > 0 else f"↓{-difference:.2f}" if difference < 0 else "→"}' - else: - message += '暂无对比数据' - avg = get_metrics(pps=latest_data.avg_pps, apm=latest_data.avg_apm, vs=latest_data.avg_vs) - low_pps = get_metrics(pps=latest_data.low_pps[1]) - low_vs = get_metrics(vs=latest_data.low_vs[1]) - max_pps = get_metrics(pps=latest_data.high_pps[1]) - max_vs = get_metrics(vs=latest_data.high_vs[1]) - message += ( - '\n' - '平均数据:\n' - f"L'PM: {avg.lpm} ( {avg.pps} pps )\n" - f'APM: {avg.apm} ( x{avg.apl} )\n' - f'ADPM: {avg.adpm} ( x{avg.adpl} ) ( {avg.vs}vs )\n' - '\n' - '最低数据:\n' - f"L'PM: {low_pps.lpm} ( {low_pps.pps} pps ) By: {latest_data.low_pps[0]['name'].upper()}\n" - f'APM: {latest_data.low_apm[1]} By: {latest_data.low_apm[0]["name"].upper()}\n' - f'ADPM: {low_vs.adpm} ( {low_vs.vs}vs ) By: {latest_data.low_vs[0]["name"].upper()}\n' - '\n' - '最高数据:\n' - f"L'PM: {max_pps.lpm} ( {max_pps.pps} pps ) By: {latest_data.high_pps[0]['name'].upper()}\n" - f'APM: {latest_data.high_apm[1]} By: {latest_data.high_apm[0]["name"].upper()}\n' - f'ADPM: {max_vs.adpm} ( {max_vs.vs}vs ) By: {latest_data.high_vs[0]["name"].upper()}\n' - '\n' - f'数据更新时间: {latest_data.update_time.replace(tzinfo=UTC).astimezone(ZoneInfo("Asia/Shanghai")).strftime("%Y-%m-%d %H:%M:%S")}' - ) - await matcher.finish(message) - - -@scheduler.scheduled_job('cron', hour='0,6,12,18', minute=0) -@retry(exception_type=RequestError, delay=timedelta(minutes=15)) -async def get_tetra_league_data() -> None: - league, original = await full_export(with_original=True) - if isinstance(league, FailedModel): - msg = f'排行榜数据请求错误:\n{league.error}' - raise RequestError(msg) - - def pps(user: ValidUser) -> float: - return user.league.pps - - def apm(user: ValidUser) -> float: - return user.league.apm - - def vs(user: ValidUser) -> float: - return user.league.vs - - def _min(users: list[ValidUser], field: Callable[[ValidUser], float]) -> ValidUser: - return min(users, key=field) - - def _max(users: list[ValidUser], field: Callable[[ValidUser], float]) -> ValidUser: - return max(users, key=field) - - def build_extremes_data( - users: list[ValidUser], - field: Callable[[ValidUser], float], - sort: Callable[[list[ValidUser], Callable[[ValidUser], float]], ValidUser], - ) -> tuple[dict[str, str], float]: - user = sort(users, field) - return model_dump(User(ID=user.id, name=user.username)), field(user) - - data_hash: str | None = await run_sync((await run_sync(sha512)(original)).hexdigest)() - async with open(get_data_file('nonebot_plugin_tetris_stats', f'{data_hash}.json.zst'), mode='wb') as file: - await file.write(await run_sync(ZstdCompressor(level=12, threads=-1).compress)(original)) - - users = [i for i in league.data.users if isinstance(i, ValidUser)] - rank_to_users: defaultdict[Rank, list[ValidUser]] = defaultdict(list) - for i in users: - rank_to_users[i.league.rank].append(i) - rank_info: list[IORank] = [] - for rank, percentile in RANK_PERCENTILE.items(): - offset = floor((percentile / 100) * len(users)) - 1 - tr_line = users[offset].league.rating - rank_users = rank_to_users[rank] - rank_info.append( - IORank( - rank=rank, - tr_line=tr_line, - player_count=len(rank_users), - low_pps=(build_extremes_data(rank_users, pps, _min)), - low_apm=(build_extremes_data(rank_users, apm, _min)), - low_vs=(build_extremes_data(rank_users, vs, _min)), - avg_pps=mean({i.league.pps for i in rank_users}), - avg_apm=mean({i.league.apm for i in rank_users}), - avg_vs=mean({i.league.vs for i in rank_users}), - high_pps=(build_extremes_data(rank_users, pps, _max)), - high_apm=(build_extremes_data(rank_users, apm, _max)), - high_vs=(build_extremes_data(rank_users, vs, _max)), - update_time=league.cache.cached_at, - file_hash=data_hash, - ) - ) - async with get_session() as session: - session.add_all(rank_info) - await session.commit() - - -@driver.on_startup -async def _() -> None: - async with get_session() as session: - latest_time = await session.scalar(select(IORank.update_time).order_by(IORank.id.desc()).limit(1)) - if latest_time is None or datetime.now(tz=UTC) - latest_time.replace(tzinfo=UTC) > timedelta(hours=6): - await get_tetra_league_data() diff --git a/nonebot_plugin_tetris_stats/games/tetrio/rank/__init__.py b/nonebot_plugin_tetris_stats/games/tetrio/rank/__init__.py new file mode 100644 index 0000000..623ded0 --- /dev/null +++ b/nonebot_plugin_tetris_stats/games/tetrio/rank/__init__.py @@ -0,0 +1,113 @@ +from collections import defaultdict +from collections.abc import Callable +from datetime import datetime, timedelta, timezone +from hashlib import sha512 +from math import floor +from statistics import mean +from typing import TYPE_CHECKING + +from aiofiles import open +from nonebot import get_driver +from nonebot.compat import model_dump +from nonebot.utils import run_sync +from nonebot_plugin_apscheduler import scheduler +from nonebot_plugin_localstore import get_data_file +from nonebot_plugin_orm import get_session +from sqlalchemy import select +from zstandard import ZstdCompressor + +from ....utils.exception import RequestError +from ....utils.retry import retry +from ..api.schemas.base import FailedModel +from ..api.schemas.tetra_league import ValidUser +from ..api.schemas.user import User +from ..api.tetra_league import full_export +from ..constant import RANK_PERCENTILE +from ..models import IORank + +if TYPE_CHECKING: + from ..api.typing import Rank + +UTC = timezone.utc + +driver = get_driver() + + +@scheduler.scheduled_job('cron', hour='0,6,12,18', minute=0) +@retry(exception_type=RequestError, delay=timedelta(minutes=15)) +async def get_tetra_league_data() -> None: + league, original = await full_export(with_original=True) + if isinstance(league, FailedModel): + msg = f'排行榜数据请求错误:\n{league.error}' + raise RequestError(msg) + + def pps(user: ValidUser) -> float: + return user.league.pps + + def apm(user: ValidUser) -> float: + return user.league.apm + + def vs(user: ValidUser) -> float: + return user.league.vs + + def _min(users: list[ValidUser], field: Callable[[ValidUser], float]) -> ValidUser: + return min(users, key=field) + + def _max(users: list[ValidUser], field: Callable[[ValidUser], float]) -> ValidUser: + return max(users, key=field) + + def build_extremes_data( + users: list[ValidUser], + field: Callable[[ValidUser], float], + sort: Callable[[list[ValidUser], Callable[[ValidUser], float]], ValidUser], + ) -> tuple[dict[str, str], float]: + user = sort(users, field) + return model_dump(User(ID=user.id, name=user.username)), field(user) + + data_hash: str | None = await run_sync((await run_sync(sha512)(original)).hexdigest)() + async with open(get_data_file('nonebot_plugin_tetris_stats', f'{data_hash}.json.zst'), mode='wb') as file: + await file.write(await run_sync(ZstdCompressor(level=12, threads=-1).compress)(original)) + + users = [i for i in league.data.users if isinstance(i, ValidUser)] + rank_to_users: defaultdict[Rank, list[ValidUser]] = defaultdict(list) + for i in users: + rank_to_users[i.league.rank].append(i) + rank_info: list[IORank] = [] + for rank, percentile in RANK_PERCENTILE.items(): + offset = floor((percentile / 100) * len(users)) - 1 + tr_line = users[offset].league.rating + rank_users = rank_to_users[rank] + rank_info.append( + IORank( + rank=rank, + tr_line=tr_line, + player_count=len(rank_users), + low_pps=(build_extremes_data(rank_users, pps, _min)), + low_apm=(build_extremes_data(rank_users, apm, _min)), + low_vs=(build_extremes_data(rank_users, vs, _min)), + avg_pps=mean({i.league.pps for i in rank_users}), + avg_apm=mean({i.league.apm for i in rank_users}), + avg_vs=mean({i.league.vs for i in rank_users}), + high_pps=(build_extremes_data(rank_users, pps, _max)), + high_apm=(build_extremes_data(rank_users, apm, _max)), + high_vs=(build_extremes_data(rank_users, vs, _max)), + update_time=league.cache.cached_at, + file_hash=data_hash, + ) + ) + async with get_session() as session: + session.add_all(rank_info) + await session.commit() + + +@driver.on_startup +async def _() -> None: + async with get_session() as session: + latest_time = await session.scalar(select(IORank.update_time).order_by(IORank.id.desc()).limit(1)) + if latest_time is None or datetime.now(tz=UTC) - latest_time.replace(tzinfo=UTC) > timedelta(hours=6): + await get_tetra_league_data() + + +from . import all, detail # noqa: E402 + +__all__ = ['all', 'detail'] diff --git a/nonebot_plugin_tetris_stats/games/tetrio/rank/all.py b/nonebot_plugin_tetris_stats/games/tetrio/rank/all.py new file mode 100644 index 0000000..cafa80a --- /dev/null +++ b/nonebot_plugin_tetris_stats/games/tetrio/rank/all.py @@ -0,0 +1,78 @@ +from datetime import timedelta + +from nonebot_plugin_alconna import UniMessage +from nonebot_plugin_orm import get_session +from nonebot_plugin_session import EventSession # type: ignore[import-untyped] +from nonebot_plugin_session_orm import get_session_persist_id # type: ignore[import-untyped] +from sqlalchemy import func, select + +from ....db import trigger +from ....utils.host import HostPage, get_self_netloc +from ....utils.metrics import get_metrics +from ....utils.render import render +from ....utils.render.schemas.tetrio.tetrio_rank import AverageData, Data, ItemData +from ....utils.screenshot import screenshot +from .. import alc +from ..constant import GAME_TYPE +from ..models import IORank + + +@alc.assign('TETRIO.rank.all') +async def _(event_session: EventSession): + async with trigger( + session_persist_id=await get_session_persist_id(event_session), + game_platform=GAME_TYPE, + command_type='rank', + command_args=['--all'], + ): + async with get_session() as session: + latest_update_time = ( + await session.scalars(select(IORank.update_time).order_by(IORank.id.desc()).limit(1)) + ).one() + compare_time = ( + await session.scalars( + select(IORank.update_time) + .order_by( + func.abs( + func.julianday(IORank.update_time) + - func.julianday(latest_update_time - timedelta(hours=24)) + ) + ) + .limit(1) + ) + ).one() + latest_data = ( + await session.scalars( + select(IORank).where(IORank.update_time == latest_update_time).order_by(IORank.tr_line.desc()) + ) + ).all() + compare_data = ( + await session.scalars( + select(IORank).where(IORank.update_time == compare_time).order_by(IORank.tr_line.desc()) + ) + ).all() + async with HostPage( + await render( + 'v2/tetrio/rank', + Data( + items={ + i[0].rank: ItemData( + require_tr=round(i[0].tr_line, 2), + trending=round(i[0].tr_line - i[1].tr_line, 2), + average_data=AverageData( + pps=(metrics := get_metrics(pps=i[0].avg_pps, apm=i[0].avg_apm, vs=i[0].avg_vs)).pps, + apm=metrics.apm, + apl=metrics.apl, + vs=metrics.vs, + adpl=metrics.adpl, + ), + players=i[0].player_count, + ) + for i in zip(latest_data, compare_data, strict=True) + }, + updated_at=latest_update_time, + ), + ) + ) as page_hash: + await screenshot(f'http://{get_self_netloc()}/host/{page_hash}.html') + await UniMessage.image(raw=await screenshot(f'http://{get_self_netloc()}/host/{page_hash}.html')).finish() diff --git a/nonebot_plugin_tetris_stats/games/tetrio/rank/detail.py b/nonebot_plugin_tetris_stats/games/tetrio/rank/detail.py new file mode 100644 index 0000000..7d27ff4 --- /dev/null +++ b/nonebot_plugin_tetris_stats/games/tetrio/rank/detail.py @@ -0,0 +1,138 @@ +from datetime import datetime, timedelta, timezone +from zoneinfo import ZoneInfo + +from nonebot import get_driver +from nonebot_plugin_alconna import UniMessage +from nonebot_plugin_orm import get_session +from nonebot_plugin_session import EventSession # type: ignore[import-untyped] +from nonebot_plugin_session_orm import get_session_persist_id # type: ignore[import-untyped] +from sqlalchemy import func, select + +from ....db import trigger +from ....utils.host import HostPage, get_self_netloc +from ....utils.metrics import get_metrics +from ....utils.render import render +from ....utils.render.schemas.tetrio.tetrio_rank_detail import Data, SpecialData +from ....utils.screenshot import screenshot +from .. import alc +from ..api.typing import ValidRank +from ..constant import GAME_TYPE +from ..models import IORank + +UTC = timezone.utc + +driver = get_driver() + + +@alc.assign('TETRIO.rank') +async def _(rank: ValidRank, event_session: EventSession): + async with trigger( + session_persist_id=await get_session_persist_id(event_session), + game_platform=GAME_TYPE, + command_type='rank', + command_args=[f'--detail {rank}'], + ): + async with get_session() as session: + latest_data = ( + await session.scalars(select(IORank).where(IORank.rank == rank).order_by(IORank.id.desc()).limit(1)) + ).one() + compare_data = ( + await session.scalars( + select(IORank) + .where(IORank.rank == rank) + .order_by( + func.abs( + func.julianday(IORank.update_time) + - func.julianday(latest_data.update_time - timedelta(hours=24)) + ) + ) + .limit(1) + ) + ).one() + await UniMessage.image(raw=await make_image(latest_data, compare_data)).finish() + + +async def make_image(latest_data: IORank, compare_data: IORank) -> bytes: + avg = get_metrics(pps=latest_data.avg_pps, apm=latest_data.avg_apm, vs=latest_data.avg_vs) + low_pps = get_metrics(pps=latest_data.low_pps[1]) + low_vs = get_metrics(vs=latest_data.low_vs[1]) + max_pps = get_metrics(pps=latest_data.high_pps[1]) + max_vs = get_metrics(vs=latest_data.high_vs[1]) + async with HostPage( + await render( + 'v2/tetrio/rank/detail', + Data( + name=latest_data.rank, + trending=round(latest_data.tr_line - compare_data.tr_line, 2), + require_tr=round(latest_data.tr_line, 2), + players=latest_data.player_count, + minimum_data=SpecialData( + apm=latest_data.low_apm[1], + pps=low_pps.pps, + lpm=low_pps.lpm, + vs=low_vs.vs, + adpm=low_vs.adpm, + apm_holder=latest_data.low_apm[0]['name'].upper(), + pps_holder=latest_data.low_pps[0]['name'].upper(), + vs_holder=latest_data.low_vs[0]['name'].upper(), + ), + average_data=SpecialData( + apm=avg.apm, + pps=avg.pps, + lpm=avg.lpm, + vs=avg.vs, + adpm=avg.adpm, + apl=avg.apl, + adpl=avg.adpl, + ), + maximum_data=SpecialData( + apm=latest_data.high_apm[1], + pps=max_pps.pps, + lpm=max_pps.lpm, + vs=max_vs.vs, + adpm=max_vs.adpm, + apm_holder=latest_data.high_apm[0]['name'].upper(), + pps_holder=latest_data.high_pps[0]['name'].upper(), + vs_holder=latest_data.high_vs[0]['name'].upper(), + ), + updated_at=latest_data.update_time.replace(tzinfo=UTC).astimezone(ZoneInfo('Asia/Shanghai')), + ), + ) + ) as page_hash: + return await screenshot(f'http://{get_self_netloc()}/host/{page_hash}.html') + + +async def make_text(latest_data: IORank, compare_data: IORank) -> str: + message = '' + if (datetime.now(UTC) - latest_data.update_time.replace(tzinfo=UTC)) > timedelta(hours=7): + message += 'Warning: 数据超过7小时未更新, 请联系Bot主人查看后台\n' + message += f'{latest_data.rank.upper()} 段 分数线 {latest_data.tr_line:.2f} TR, {latest_data.player_count} 名玩家\n' + if compare_data.id != latest_data.id: + message += f'对比 {(latest_data.update_time-compare_data.update_time).total_seconds()/3600:.2f} 小时前趋势: {f"↑{difference:.2f}" if (difference:=latest_data.tr_line-compare_data.tr_line) > 0 else f"↓{-difference:.2f}" if difference < 0 else "→"}' + else: + message += '暂无对比数据' + avg = get_metrics(pps=latest_data.avg_pps, apm=latest_data.avg_apm, vs=latest_data.avg_vs) + low_pps = get_metrics(pps=latest_data.low_pps[1]) + low_vs = get_metrics(vs=latest_data.low_vs[1]) + max_pps = get_metrics(pps=latest_data.high_pps[1]) + max_vs = get_metrics(vs=latest_data.high_vs[1]) + message += ( + '\n' + '平均数据:\n' + f"L'PM: {avg.lpm} ( {avg.pps} pps )\n" + f'APM: {avg.apm} ( x{avg.apl} )\n' + f'ADPM: {avg.adpm} ( x{avg.adpl} ) ( {avg.vs}vs )\n' + '\n' + '最低数据:\n' + f"L'PM: {low_pps.lpm} ( {low_pps.pps} pps ) By: {latest_data.low_pps[0]['name'].upper()}\n" + f'APM: {latest_data.low_apm[1]} By: {latest_data.low_apm[0]["name"].upper()}\n' + f'ADPM: {low_vs.adpm} ( {low_vs.vs}vs ) By: {latest_data.low_vs[0]["name"].upper()}\n' + '\n' + '最高数据:\n' + f"L'PM: {max_pps.lpm} ( {max_pps.pps} pps ) By: {latest_data.high_pps[0]['name'].upper()}\n" + f'APM: {latest_data.high_apm[1]} By: {latest_data.high_apm[0]["name"].upper()}\n' + f'ADPM: {max_vs.adpm} ( {max_vs.vs}vs ) By: {latest_data.high_vs[0]["name"].upper()}\n' + '\n' + f'数据更新时间: {latest_data.update_time.replace(tzinfo=UTC).astimezone(ZoneInfo("Asia/Shanghai")).strftime("%Y-%m-%d %H:%M:%S")}' + ) + return message diff --git a/nonebot_plugin_tetris_stats/utils/render/__init__.py b/nonebot_plugin_tetris_stats/utils/render/__init__.py index 722d1e3..21ab05e 100644 --- a/nonebot_plugin_tetris_stats/utils/render/__init__.py +++ b/nonebot_plugin_tetris_stats/utils/render/__init__.py @@ -6,6 +6,8 @@ from nonebot.compat import PYDANTIC_V2 from ..templates import templates_dir from .schemas.bind import Bind from .schemas.tetrio.tetrio_info import Info as TETRIOInfo +from .schemas.tetrio.tetrio_rank import Data as TETRIORankData +from .schemas.tetrio.tetrio_rank_detail import Data as TETRIORankDetailData from .schemas.tetrio.tetrio_record_blitz import Record as TETRIORecordBlitz from .schemas.tetrio.tetrio_record_sprint import Record as TETRIORecordSprint from .schemas.tetrio.tetrio_user_info_v2 import Info as TETRIOUserInfoV2 @@ -20,34 +22,24 @@ env = Environment( @overload async def render(render_type: Literal['v1/binding'], data: Bind) -> str: ... - - @overload async def render(render_type: Literal['v1/tetrio/info'], data: TETRIOInfo) -> str: ... - - @overload async def render(render_type: Literal['v1/top/info'], data: TOPInfo) -> str: ... - - @overload async def render(render_type: Literal['v1/tos/info'], data: TOSInfo) -> str: ... - - @overload async def render(render_type: Literal['v2/tetrio/user/info'], data: TETRIOUserInfoV2) -> str: ... - - @overload async def render(render_type: Literal['v2/tetrio/user/list'], data: TETRIOUserListV2) -> str: ... - - @overload async def render(render_type: Literal['v2/tetrio/record/40l'], data: TETRIORecordSprint) -> str: ... - - @overload async def render(render_type: Literal['v2/tetrio/record/blitz'], data: TETRIORecordBlitz) -> str: ... +@overload +async def render(render_type: Literal['v2/tetrio/rank'], data: TETRIORankData) -> str: ... +@overload +async def render(render_type: Literal['v2/tetrio/rank/detail'], data: TETRIORankDetailData) -> str: ... async def render( @@ -60,6 +52,8 @@ async def render( 'v2/tetrio/user/list', 'v2/tetrio/record/40l', 'v2/tetrio/record/blitz', + 'v2/tetrio/rank', + 'v2/tetrio/rank/detail', ], data: Bind | TETRIOInfo @@ -68,7 +62,9 @@ async def render( | TETRIOUserInfoV2 | TETRIOUserListV2 | TETRIORecordSprint - | TETRIORecordBlitz, + | TETRIORecordBlitz + | TETRIORankData + | TETRIORankDetailData, ) -> str: if PYDANTIC_V2: return await env.get_template('index.html').render_async( diff --git a/nonebot_plugin_tetris_stats/utils/render/schemas/tetrio/tetrio_rank.py b/nonebot_plugin_tetris_stats/utils/render/schemas/tetrio/tetrio_rank.py new file mode 100644 index 0000000..0d6a4ae --- /dev/null +++ b/nonebot_plugin_tetris_stats/utils/render/schemas/tetrio/tetrio_rank.py @@ -0,0 +1,25 @@ +from datetime import datetime + +from pydantic import BaseModel + +from .....games.tetrio.api.typing import ValidRank + + +class AverageData(BaseModel): + pps: float + apm: float + apl: float + vs: float + adpl: float + + +class ItemData(BaseModel): + require_tr: float + trending: float + average_data: AverageData + players: int + + +class Data(BaseModel): + items: dict[ValidRank, ItemData] + updated_at: datetime diff --git a/nonebot_plugin_tetris_stats/utils/render/schemas/tetrio/tetrio_rank_detail.py b/nonebot_plugin_tetris_stats/utils/render/schemas/tetrio/tetrio_rank_detail.py new file mode 100644 index 0000000..a208e52 --- /dev/null +++ b/nonebot_plugin_tetris_stats/utils/render/schemas/tetrio/tetrio_rank_detail.py @@ -0,0 +1,31 @@ +from datetime import datetime + +from pydantic import BaseModel + +from .....games.tetrio.api.typing import ValidRank + + +class SpecialData(BaseModel): + apm: float + pps: float + lpm: float + vs: float + adpm: float + apl: float | None = None + adpl: float | None = None + apm_holder: str | None = None + pps_holder: str | None = None + vs_holder: str | None = None + + +class Data(BaseModel): + name: ValidRank + trending: float + require_tr: float + players: int + + minimum_data: SpecialData + average_data: SpecialData + maximum_data: SpecialData + + updated_at: datetime