mirror of
https://github.com/A-Minos/nonebot-plugin-tetris-stats.git
synced 2026-03-05 05:36:54 +08:00
134 lines
4.6 KiB
Python
134 lines
4.6 KiB
Python
from nonebot.adapters import Event
|
|
from nonebot.matcher import Matcher
|
|
from nonebot_plugin_alconna import At
|
|
from nonebot_plugin_alconna.uniseg import Image, UniMessage
|
|
from nonebot_plugin_orm import get_session
|
|
from nonebot_plugin_uninfo import Uninfo
|
|
from nonebot_plugin_uninfo.orm import get_session_persist_id
|
|
from nonebot_plugin_user import get_user
|
|
|
|
from ...db import query_bind_info, trigger
|
|
from ...i18n import Lang
|
|
from ...utils.exception import FallbackError
|
|
from ...utils.lang import get_lang
|
|
from ...utils.metrics import TetrisMetricsBasicWithLPM, get_metrics
|
|
from ...utils.render import render_image
|
|
from ...utils.render.avatar import get_avatar
|
|
from ...utils.render.schemas.base import People, Trending
|
|
from ...utils.render.schemas.v1.top.info import Data as InfoData
|
|
from ...utils.render.schemas.v1.top.info import Info
|
|
from ...utils.typedefs import Me
|
|
from . import alc
|
|
from .api import Player
|
|
from .api.schemas.user_profile import Data, UserProfile
|
|
from .constant import GAME_TYPE
|
|
|
|
|
|
@alc.assign('TOP.query')
|
|
async def _(event: Event, matcher: Matcher, target: At | Me, event_session: Uninfo):
|
|
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,
|
|
user=await get_user(
|
|
event_session.scope, target.target if isinstance(target, At) else event.get_user_id()
|
|
),
|
|
game_platform=GAME_TYPE,
|
|
)
|
|
if bind is None:
|
|
await matcher.finish(Lang.bind.not_found())
|
|
await (
|
|
UniMessage.i18n(Lang.interaction.warning.unverified)
|
|
+ (
|
|
UniMessage('\n')
|
|
if not (
|
|
result := await make_query_result(
|
|
await Player(user_name=bind.game_account, trust=True).get_profile()
|
|
)
|
|
).has(Image)
|
|
else UniMessage()
|
|
)
|
|
+ result
|
|
).finish()
|
|
|
|
|
|
@alc.assign('TOP.query')
|
|
async def _(account: Player, event_session: Uninfo):
|
|
async with trigger(
|
|
session_persist_id=await get_session_persist_id(event_session),
|
|
game_platform=GAME_TYPE,
|
|
command_type='query',
|
|
command_args=[],
|
|
):
|
|
await (await make_query_result(await account.get_profile())).finish()
|
|
|
|
|
|
def get_avg_metrics(data: list[Data]) -> TetrisMetricsBasicWithLPM:
|
|
total_lpm = total_apm = 0.0
|
|
for value in data:
|
|
total_lpm += value.lpm
|
|
total_apm += value.apm
|
|
num = len(data)
|
|
return get_metrics(lpm=total_lpm / num, apm=total_apm / num)
|
|
|
|
|
|
async def make_query_image(profile: UserProfile) -> bytes:
|
|
if profile.today is None or profile.total is None:
|
|
raise FallbackError
|
|
today = get_metrics(lpm=profile.today.lpm, apm=profile.today.apm)
|
|
history = get_avg_metrics(profile.total)
|
|
return await render_image(
|
|
Info(
|
|
user=People(avatar=get_avatar(profile.user_name), name=profile.user_name),
|
|
today=InfoData(
|
|
pps=today.pps,
|
|
lpm=today.lpm,
|
|
lpm_trending=Trending.KEEP,
|
|
apm=today.apm,
|
|
apl=today.apl,
|
|
apm_trending=Trending.KEEP,
|
|
),
|
|
historical=InfoData(
|
|
pps=history.pps,
|
|
lpm=history.lpm,
|
|
lpm_trending=Trending.KEEP,
|
|
apm=history.apm,
|
|
apl=history.apl,
|
|
apm_trending=Trending.KEEP,
|
|
),
|
|
lang=get_lang(),
|
|
),
|
|
)
|
|
|
|
|
|
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 += Lang.stats.daily_stats(name=profile.user_name)
|
|
message += Lang.stats.lpm(lpm=today.lpm, pps=today.pps)
|
|
message += Lang.stats.apm(apm=today.apm, apl=today.apl)
|
|
else:
|
|
message += Lang.stats.no_daily(name=profile.user_name)
|
|
if profile.total is not None:
|
|
total = get_avg_metrics(profile.total)
|
|
message += Lang.stats.history_stats()
|
|
message += Lang.stats.lpm(lpm=total.lpm, pps=total.pps)
|
|
message += Lang.stats.apm(apm=total.apm, apl=total.apl)
|
|
else:
|
|
message += Lang.stats.no_history()
|
|
return UniMessage(message)
|
|
|
|
|
|
async def make_query_result(profile: UserProfile) -> UniMessage:
|
|
try:
|
|
return UniMessage.image(raw=await make_query_image(profile))
|
|
except FallbackError:
|
|
...
|
|
return make_query_text(profile)
|