mirror of
https://github.com/A-Minos/nonebot-plugin-tetris-stats.git
synced 2026-03-05 05:36:54 +08:00
Compare commits
8 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| a57811b0d3 | |||
|
|
7a5170936b | ||
|
|
068c508f57 | ||
|
|
0648ca021b | ||
|
|
65e7fed32b | ||
|
|
fdbb2f3f6e | ||
| 144c223fe9 | |||
|
|
52a6d95434 |
2
.github/workflows/Test.yml
vendored
2
.github/workflows/Test.yml
vendored
@@ -53,6 +53,6 @@ jobs:
|
||||
runs-on: ubuntu-latest
|
||||
steps:
|
||||
- name: Decide whether the needed jobs succeeded or failed
|
||||
uses: re-actors/alls-green@223e4bb7a751b91f43eda76992bcfbf23b8b0302
|
||||
uses: re-actors/alls-green@2765efec08f0fd63e83ad900f5fd75646be69ff6
|
||||
with:
|
||||
jobs: ${{ toJSON(needs) }}
|
||||
|
||||
@@ -1,3 +1,5 @@
|
||||
from pathlib import Path
|
||||
|
||||
from nonebot import get_driver, get_plugin_config
|
||||
from nonebot_plugin_localstore import get_plugin_cache_dir, get_plugin_data_dir
|
||||
from pydantic import BaseModel, Field
|
||||
@@ -14,11 +16,17 @@ class Proxy(BaseModel):
|
||||
top: str | None = None
|
||||
|
||||
|
||||
class Dev(BaseModel):
|
||||
enabled: bool = False
|
||||
template_path: Path | None = None
|
||||
enable_template_check: bool = True
|
||||
|
||||
|
||||
class ScopedConfig(BaseModel):
|
||||
request_timeout: float = 30.0
|
||||
screenshot_quality: float = 2
|
||||
proxy: Proxy = Field(default_factory=Proxy)
|
||||
development: bool = False
|
||||
dev: Dev = Field(default_factory=Dev)
|
||||
|
||||
|
||||
class Config(BaseModel):
|
||||
|
||||
@@ -0,0 +1,42 @@
|
||||
"""add verify field
|
||||
|
||||
迁移 ID: 2ff388a8c486
|
||||
父迁移: 3588702dd3a4
|
||||
创建时间: 2025-07-22 18:09:09.734164
|
||||
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
import sqlalchemy as sa
|
||||
from alembic import op
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from collections.abc import Sequence
|
||||
|
||||
revision: str = '2ff388a8c486'
|
||||
down_revision: str | Sequence[str] | None = '3588702dd3a4'
|
||||
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('nb_t_bind', schema=None) as batch_op:
|
||||
batch_op.add_column(sa.Column('verify', sa.Boolean(), nullable=False, server_default='false'))
|
||||
|
||||
# ### end Alembic commands ###
|
||||
|
||||
|
||||
def downgrade(name: str = '') -> None:
|
||||
if name:
|
||||
return
|
||||
# ### commands auto generated by Alembic - please adjust! ###
|
||||
with op.batch_alter_table('nb_t_bind', schema=None) as batch_op:
|
||||
batch_op.drop_column('verify')
|
||||
|
||||
# ### end Alembic commands ###
|
||||
@@ -42,6 +42,8 @@ async def create_or_update_bind(
|
||||
user: User,
|
||||
game_platform: GameType,
|
||||
game_account: str,
|
||||
*,
|
||||
verify: bool = False,
|
||||
) -> BindStatus:
|
||||
bind = await query_bind_info(
|
||||
session=session,
|
||||
@@ -53,11 +55,13 @@ async def create_or_update_bind(
|
||||
user_id=user.id,
|
||||
game_platform=game_platform,
|
||||
game_account=game_account,
|
||||
verify=verify,
|
||||
)
|
||||
session.add(bind)
|
||||
status = BindStatus.SUCCESS
|
||||
else:
|
||||
bind.game_account = game_account
|
||||
bind.verify = verify
|
||||
status = BindStatus.UPDATE
|
||||
await session.commit()
|
||||
return status
|
||||
|
||||
@@ -71,6 +71,7 @@ class Bind(MappedAsDataclass, Model):
|
||||
user_id: Mapped[int] = mapped_column(index=True)
|
||||
game_platform: Mapped[GameType] = mapped_column(String(32))
|
||||
game_account: Mapped[str]
|
||||
verify: Mapped[bool]
|
||||
|
||||
|
||||
class TriggerHistoricalDataV2(MappedAsDataclass, Model):
|
||||
|
||||
@@ -23,7 +23,7 @@ command = Subcommand(
|
||||
)
|
||||
|
||||
|
||||
from . import bind, config, list, query, rank, record, unbind # noqa: A004, E402
|
||||
from . import bind, config, list, query, rank, record, unbind, verify # noqa: A004, E402
|
||||
|
||||
main_command.add(command)
|
||||
|
||||
@@ -36,4 +36,5 @@ __all__ = [
|
||||
'rank',
|
||||
'record',
|
||||
'unbind',
|
||||
'verify',
|
||||
]
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
from asyncio import gather
|
||||
from hashlib import md5
|
||||
from secrets import choice
|
||||
|
||||
@@ -13,12 +14,12 @@ from yarl import URL
|
||||
|
||||
from ...config.config import global_config
|
||||
from ...db import BindStatus, create_or_update_bind, trigger
|
||||
from ...utils.host import HostPage, get_self_netloc
|
||||
from ...utils.host import get_self_netloc
|
||||
from ...utils.image import get_avatar
|
||||
from ...utils.lang import get_lang
|
||||
from ...utils.render import Bind, render
|
||||
from ...utils.render import render_image
|
||||
from ...utils.render.schemas.base import Avatar, People
|
||||
from ...utils.screenshot import screenshot
|
||||
from ...utils.render.schemas.bind import Bind
|
||||
from . import alc, command, get_player
|
||||
from .api import Player
|
||||
from .constant import GAME_TYPE
|
||||
@@ -44,6 +45,42 @@ alc.shortcut(
|
||||
humanized='io绑定',
|
||||
)
|
||||
|
||||
try:
|
||||
from nonebot.adapters.discord import MessageCreateEvent
|
||||
|
||||
@alc.assign('TETRIO.bind')
|
||||
async def _(_: MessageCreateEvent, nb_user: User, account: Player, event_session: Uninfo, interface: QryItrface):
|
||||
async with trigger(
|
||||
session_persist_id=await get_session_persist_id(event_session),
|
||||
game_platform=GAME_TYPE,
|
||||
command_type='bind',
|
||||
command_args=[],
|
||||
):
|
||||
user, user_info = await gather(account.user, account.get_info())
|
||||
verify = (
|
||||
user_info.data.connections.discord is not None
|
||||
and user_info.data.connections.discord.id == event_session.user.id
|
||||
)
|
||||
async with get_session() as session:
|
||||
bind_status = await create_or_update_bind(
|
||||
session=session,
|
||||
user=nb_user,
|
||||
game_platform=GAME_TYPE,
|
||||
game_account=user.unique_identifier,
|
||||
verify=verify,
|
||||
)
|
||||
if bind_status in (BindStatus.SUCCESS, BindStatus.UPDATE):
|
||||
await UniMessage.image(
|
||||
raw=await make_bind_image(
|
||||
player=account,
|
||||
event_session=event_session,
|
||||
interface=interface,
|
||||
verify=verify,
|
||||
)
|
||||
).finish()
|
||||
except ImportError:
|
||||
pass
|
||||
|
||||
|
||||
@alc.assign('TETRIO.bind')
|
||||
async def _(nb_user: User, account: Player, event_session: Uninfo, interface: QryItrface):
|
||||
@@ -62,36 +99,45 @@ async def _(nb_user: User, account: Player, event_session: Uninfo, interface: Qr
|
||||
game_account=user.unique_identifier,
|
||||
)
|
||||
if bind_status in (BindStatus.SUCCESS, BindStatus.UPDATE):
|
||||
netloc = get_self_netloc()
|
||||
async with HostPage(
|
||||
await render(
|
||||
'v1/binding',
|
||||
Bind(
|
||||
platform='TETR.IO',
|
||||
type='unknown',
|
||||
user=People(
|
||||
avatar=str(
|
||||
URL(f'http://{netloc}/host/resource/tetrio/avatars/{user.ID}')
|
||||
% {'revision': avatar_revision}
|
||||
)
|
||||
if (avatar_revision := (await account.avatar_revision)) is not None and avatar_revision != 0
|
||||
else Avatar(type='identicon', hash=md5(user.ID.encode()).hexdigest()), # noqa: S324
|
||||
name=user.name.upper(),
|
||||
),
|
||||
bot=People(
|
||||
avatar=await get_avatar(
|
||||
(
|
||||
bot_user := await interface.get_user(event_session.self_id)
|
||||
or UninfoUser(id=event_session.self_id)
|
||||
),
|
||||
'Data URI',
|
||||
'../../static/logo/logo.svg',
|
||||
),
|
||||
name=bot_user.nick or bot_user.name or choice(list(global_config.nickname) or ['bot']),
|
||||
),
|
||||
prompt='io查我',
|
||||
lang=get_lang(),
|
||||
),
|
||||
await UniMessage.image(
|
||||
raw=await make_bind_image(
|
||||
player=account,
|
||||
event_session=event_session,
|
||||
interface=interface,
|
||||
verify=None,
|
||||
)
|
||||
) as page_hash:
|
||||
await UniMessage.image(raw=await screenshot(f'http://{netloc}/host/{page_hash}.html')).finish()
|
||||
).finish()
|
||||
|
||||
|
||||
async def make_bind_image(
|
||||
player: Player, event_session: Uninfo, interface: QryItrface, *, verify: bool | None = None
|
||||
) -> bytes:
|
||||
(user, avatar_revision) = await gather(player.user, player.avatar_revision)
|
||||
return await render_image(
|
||||
Bind(
|
||||
platform='TETR.IO',
|
||||
type='unknown' if verify is None else 'success' if verify else 'unverified',
|
||||
user=People(
|
||||
avatar=str(
|
||||
URL(f'http://{get_self_netloc()}/host/resource/tetrio/avatars/{user.ID}')
|
||||
% {'revision': avatar_revision}
|
||||
)
|
||||
if avatar_revision is not None and avatar_revision != 0
|
||||
else Avatar(type='identicon', hash=md5(user.ID.encode()).hexdigest()), # noqa: S324
|
||||
name=user.name.upper(),
|
||||
),
|
||||
bot=People(
|
||||
avatar=await get_avatar(
|
||||
(
|
||||
bot_user := await interface.get_user(event_session.self_id)
|
||||
or UninfoUser(id=event_session.self_id)
|
||||
),
|
||||
'Data URI',
|
||||
'../../static/logo/logo.svg',
|
||||
),
|
||||
name=bot_user.nick or bot_user.name or choice(list(global_config.nickname) or ['bot']),
|
||||
),
|
||||
prompt='io查我',
|
||||
lang=get_lang(),
|
||||
),
|
||||
)
|
||||
|
||||
@@ -4,12 +4,10 @@ from nonebot_plugin_uninfo import Uninfo
|
||||
from nonebot_plugin_uninfo.orm import get_session_persist_id
|
||||
|
||||
from ...db import trigger
|
||||
from ...utils.host import HostPage, get_self_netloc
|
||||
from ...utils.lang import get_lang
|
||||
from ...utils.metrics import get_metrics
|
||||
from ...utils.render import render
|
||||
from ...utils.render import render_image
|
||||
from ...utils.render.schemas.v2.tetrio.user.list import Data, List, TetraLeague, User
|
||||
from ...utils.screenshot import screenshot
|
||||
from .. import alc
|
||||
from . import command
|
||||
from .api.leaderboards import by
|
||||
@@ -59,9 +57,8 @@ async def _(
|
||||
country=country,
|
||||
)
|
||||
league = await by('league', parameter)
|
||||
async with HostPage(
|
||||
await render(
|
||||
'v2/tetrio/user/list',
|
||||
await UniMessage.image(
|
||||
raw=await render_image(
|
||||
List(
|
||||
show_index=True,
|
||||
data=[
|
||||
@@ -92,5 +89,4 @@ async def _(
|
||||
lang=get_lang(),
|
||||
),
|
||||
)
|
||||
) as page_hash:
|
||||
await UniMessage.image(raw=await screenshot(f'http://{get_self_netloc()}/host/{page_hash}.html')).finish()
|
||||
).finish()
|
||||
|
||||
@@ -6,14 +6,13 @@ from yarl import URL
|
||||
|
||||
from ....utils.chart import get_split, get_value_bounds, handle_history_data
|
||||
from ....utils.exception import FallbackError
|
||||
from ....utils.host import HostPage, get_self_netloc
|
||||
from ....utils.host import get_self_netloc
|
||||
from ....utils.lang import get_lang
|
||||
from ....utils.metrics import get_metrics
|
||||
from ....utils.render import render
|
||||
from ....utils.render import render_image
|
||||
from ....utils.render.schemas.base import Avatar, Trending
|
||||
from ....utils.render.schemas.v1.base import History
|
||||
from ....utils.render.schemas.v1.tetrio.user.info import Info, Multiplayer, Singleplayer, User
|
||||
from ....utils.screenshot import screenshot
|
||||
from ....utils.render.schemas.v1.tetrio.info import Info, Multiplayer, Singleplayer, User
|
||||
from ..api import Player
|
||||
from ..api.schemas.summaries.league import RatedData
|
||||
from ..constant import TR_MAX, TR_MIN
|
||||
@@ -40,61 +39,57 @@ async def make_query_image_v1(player: Player) -> bytes:
|
||||
else:
|
||||
sprint_value = 'N/A'
|
||||
blitz_value = f'{blitz.data.record.results.stats.score:,}' if blitz.data.record is not None else 'N/A'
|
||||
netloc = get_self_netloc()
|
||||
dsps: float
|
||||
dspp: float
|
||||
# make mypy happy
|
||||
async with HostPage(
|
||||
page=await render(
|
||||
'v1/tetrio/info',
|
||||
Info(
|
||||
user=User(
|
||||
avatar=str(
|
||||
URL(f'http://{netloc}/host/resource/tetrio/avatars/{user.ID}') % {'revision': avatar_revision}
|
||||
)
|
||||
if avatar_revision is not None and avatar_revision != 0
|
||||
else Avatar(
|
||||
type='identicon',
|
||||
hash=md5(user.ID.encode()).hexdigest(), # noqa: S324
|
||||
),
|
||||
name=user.name.upper(),
|
||||
bio=user_info.data.bio,
|
||||
return await render_image(
|
||||
Info(
|
||||
user=User(
|
||||
avatar=str(
|
||||
URL(f'http://{get_self_netloc()}/host/resource/tetrio/avatars/{user.ID}')
|
||||
% {'revision': avatar_revision}
|
||||
)
|
||||
if avatar_revision is not None and avatar_revision != 0
|
||||
else Avatar(
|
||||
type='identicon',
|
||||
hash=md5(user.ID.encode()).hexdigest(), # noqa: S324
|
||||
),
|
||||
multiplayer=Multiplayer(
|
||||
glicko=f'{round(league_data.glicko, 2):,}',
|
||||
rd=round(league_data.rd, 2),
|
||||
rank=league_data.rank,
|
||||
tr=f'{round(league_data.tr, 2):,}',
|
||||
global_rank=league_data.standing,
|
||||
history=History(
|
||||
data=histories,
|
||||
split_interval=split_value,
|
||||
min_value=values.value_min,
|
||||
max_value=values.value_max,
|
||||
offset=offset,
|
||||
),
|
||||
lpm=(metrics := get_metrics(pps=league_data.pps, apm=league_data.apm, vs=league_data.vs)).lpm,
|
||||
pps=metrics.pps,
|
||||
lpm_trending=Trending.KEEP,
|
||||
apm=metrics.apm,
|
||||
apl=metrics.apl,
|
||||
apm_trending=Trending.KEEP,
|
||||
adpm=metrics.adpm,
|
||||
vs=metrics.vs,
|
||||
adpl=metrics.adpl,
|
||||
adpm_trending=Trending.KEEP,
|
||||
app=(app := (league_data.apm / (60 * league_data.pps))),
|
||||
dsps=(dsps := ((league_data.vs / 100) - (league_data.apm / 60))),
|
||||
dspp=(dspp := (dsps / league_data.pps)),
|
||||
ci=150 * dspp - 125 * app + 50 * (league_data.vs / league_data.apm) - 25,
|
||||
ge=2 * ((app * dsps) / league_data.pps),
|
||||
),
|
||||
singleplayer=Singleplayer(
|
||||
sprint=sprint_value,
|
||||
blitz=blitz_value,
|
||||
),
|
||||
lang=get_lang(),
|
||||
name=user.name.upper(),
|
||||
bio=user_info.data.bio,
|
||||
),
|
||||
)
|
||||
) as page_hash:
|
||||
return await screenshot(f'http://{netloc}/host/{page_hash}.html')
|
||||
multiplayer=Multiplayer(
|
||||
glicko=f'{round(league_data.glicko, 2):,}',
|
||||
rd=round(league_data.rd, 2),
|
||||
rank=league_data.rank,
|
||||
tr=f'{round(league_data.tr, 2):,}',
|
||||
global_rank=league_data.standing,
|
||||
history=History(
|
||||
data=histories,
|
||||
split_interval=split_value,
|
||||
min_value=values.value_min,
|
||||
max_value=values.value_max,
|
||||
offset=offset,
|
||||
),
|
||||
lpm=(metrics := get_metrics(pps=league_data.pps, apm=league_data.apm, vs=league_data.vs)).lpm,
|
||||
pps=metrics.pps,
|
||||
lpm_trending=Trending.KEEP,
|
||||
apm=metrics.apm,
|
||||
apl=metrics.apl,
|
||||
apm_trending=Trending.KEEP,
|
||||
adpm=metrics.adpm,
|
||||
vs=metrics.vs,
|
||||
adpl=metrics.adpl,
|
||||
adpm_trending=Trending.KEEP,
|
||||
app=(app := (league_data.apm / (60 * league_data.pps))),
|
||||
dsps=(dsps := ((league_data.vs / 100) - (league_data.apm / 60))),
|
||||
dspp=(dspp := (dsps / league_data.pps)),
|
||||
ci=150 * dspp - 125 * app + 50 * (league_data.vs / league_data.apm) - 25,
|
||||
ge=2 * ((app * dsps) / league_data.pps),
|
||||
),
|
||||
singleplayer=Singleplayer(
|
||||
sprint=sprint_value,
|
||||
blitz=blitz_value,
|
||||
),
|
||||
lang=get_lang(),
|
||||
),
|
||||
)
|
||||
|
||||
@@ -5,10 +5,10 @@ from hashlib import md5
|
||||
from yarl import URL
|
||||
|
||||
from ....utils.exception import FallbackError
|
||||
from ....utils.host import HostPage, get_self_netloc
|
||||
from ....utils.host import get_self_netloc
|
||||
from ....utils.lang import get_lang
|
||||
from ....utils.metrics import get_metrics
|
||||
from ....utils.render import render
|
||||
from ....utils.render import render_image
|
||||
from ....utils.render.schemas.base import Avatar
|
||||
from ....utils.render.schemas.v2.tetrio.user.info import (
|
||||
Achievement,
|
||||
@@ -25,7 +25,6 @@ from ....utils.render.schemas.v2.tetrio.user.info import (
|
||||
Zen,
|
||||
Zenith,
|
||||
)
|
||||
from ....utils.screenshot import screenshot
|
||||
from ..api import Player
|
||||
from ..api.schemas.summaries.league import InvalidData, NeverPlayedData, NeverRatedData
|
||||
from .tools import flow_to_history, handling_special_value
|
||||
@@ -74,137 +73,133 @@ async def make_query_image_v2(player: Player) -> bytes:
|
||||
except FallbackError:
|
||||
history = None
|
||||
netloc = get_self_netloc()
|
||||
async with HostPage(
|
||||
await render(
|
||||
'v2/tetrio/user/info',
|
||||
Info(
|
||||
user=User(
|
||||
id=user.ID,
|
||||
name=user.name.upper(),
|
||||
country=user_info.data.country,
|
||||
role=user_info.data.role,
|
||||
botmaster=user_info.data.botmaster,
|
||||
avatar=str(
|
||||
URL(f'http://{netloc}/host/resource/tetrio/avatars/{user.ID}') % {'revision': avatar_revision}
|
||||
)
|
||||
if avatar_revision is not None and avatar_revision != 0
|
||||
else Avatar(
|
||||
type='identicon',
|
||||
hash=md5(user.ID.encode()).hexdigest(), # noqa: S324
|
||||
),
|
||||
banner=str(
|
||||
URL(f'http://{netloc}/host/resource/tetrio/banners/{user.ID}') % {'revision': banner_revision}
|
||||
)
|
||||
if banner_revision is not None and banner_revision != 0
|
||||
else None,
|
||||
bio=user_info.data.bio,
|
||||
friend_count=user_info.data.friend_count,
|
||||
supporter_tier=user_info.data.supporter_tier,
|
||||
bad_standing=user_info.data.badstanding or False,
|
||||
badges=[
|
||||
Badge(
|
||||
id=i.id,
|
||||
description=i.label,
|
||||
group=i.group,
|
||||
receive_at=i.ts if isinstance(i.ts, datetime) else None,
|
||||
)
|
||||
for i in user_info.data.badges
|
||||
],
|
||||
xp=user_info.data.xp,
|
||||
ar=user_info.data.ar,
|
||||
achievements=[
|
||||
Achievement(
|
||||
key=i.achievement_id,
|
||||
rank_type=i.rank_type,
|
||||
ar_type=i.ar_type,
|
||||
stub=i.stub,
|
||||
rank=i.rank,
|
||||
achieved_score=i.achieved_score,
|
||||
pos=i.pos,
|
||||
progress=i.progress,
|
||||
total=i.total,
|
||||
)
|
||||
for i in achievements.data
|
||||
],
|
||||
playtime=play_time,
|
||||
join_at=user_info.data.ts,
|
||||
),
|
||||
tetra_league=TetraLeague(
|
||||
rank=league.data.rank,
|
||||
highest_rank='z' if isinstance(league.data, NeverRatedData) else league.data.bestrank,
|
||||
tr=round(league.data.tr, 2),
|
||||
glicko=round(league.data.glicko, 2),
|
||||
rd=round(league.data.rd, 2),
|
||||
global_rank=league.data.standing,
|
||||
country_rank=league.data.standing_local,
|
||||
pps=(metrics := get_metrics(pps=league.data.pps, apm=league.data.apm, vs=league.data.vs)).pps,
|
||||
apm=metrics.apm,
|
||||
apl=metrics.apl,
|
||||
vs=metrics.vs,
|
||||
adpl=metrics.adpl,
|
||||
statistic=TetraLeagueStatistic(total=league.data.gamesplayed, wins=league.data.gameswon),
|
||||
decaying=league.data.decaying,
|
||||
history=history,
|
||||
return await render_image(
|
||||
Info(
|
||||
user=User(
|
||||
id=user.ID,
|
||||
name=user.name.upper(),
|
||||
country=user_info.data.country,
|
||||
role=user_info.data.role,
|
||||
botmaster=user_info.data.botmaster,
|
||||
avatar=str(
|
||||
URL(f'http://{netloc}/host/resource/tetrio/avatars/{user.ID}') % {'revision': avatar_revision}
|
||||
)
|
||||
if not isinstance(league.data, NeverPlayedData | InvalidData)
|
||||
else None,
|
||||
zenith=Zenith(
|
||||
week=Week(
|
||||
altitude=zenith.data.record.results.stats.zenith.altitude,
|
||||
global_rank=zenith.data.rank,
|
||||
country_rank=zenith.data.rank_local,
|
||||
play_at=zenith.data.record.ts,
|
||||
)
|
||||
if zenith.data.record is not None
|
||||
else None,
|
||||
best=Best(
|
||||
altitude=zenith.data.best.record.results.stats.zenith.altitude,
|
||||
global_rank=zenith.data.best.rank,
|
||||
play_at=zenith.data.best.record.ts,
|
||||
)
|
||||
if zenith.data.best.record is not None
|
||||
else None,
|
||||
if avatar_revision is not None and avatar_revision != 0
|
||||
else Avatar(
|
||||
type='identicon',
|
||||
hash=md5(user.ID.encode()).hexdigest(), # noqa: S324
|
||||
),
|
||||
zenithex=Zenith(
|
||||
week=Week(
|
||||
altitude=zenithex.data.record.results.stats.zenith.altitude,
|
||||
global_rank=zenithex.data.rank,
|
||||
country_rank=zenithex.data.rank_local,
|
||||
play_at=zenithex.data.record.ts,
|
||||
)
|
||||
if zenithex.data.record is not None
|
||||
else None,
|
||||
best=Best(
|
||||
altitude=zenithex.data.best.record.results.stats.zenith.altitude,
|
||||
global_rank=zenithex.data.best.rank,
|
||||
play_at=zenithex.data.best.record.ts,
|
||||
)
|
||||
if zenithex.data.best.record is not None
|
||||
else None,
|
||||
),
|
||||
statistic=Statistic(
|
||||
total=handling_special_value(user_info.data.gamesplayed),
|
||||
wins=handling_special_value(user_info.data.gameswon),
|
||||
),
|
||||
sprint=Sprint(
|
||||
time=sprint_value,
|
||||
global_rank=sprint.data.rank,
|
||||
country_rank=sprint.data.rank_local,
|
||||
play_at=sprint.data.record.ts,
|
||||
banner=str(
|
||||
URL(f'http://{netloc}/host/resource/tetrio/banners/{user.ID}') % {'revision': banner_revision}
|
||||
)
|
||||
if sprint.data.record is not None
|
||||
if banner_revision is not None and banner_revision != 0
|
||||
else None,
|
||||
blitz=Blitz(
|
||||
score=blitz.data.record.results.stats.score,
|
||||
global_rank=blitz.data.rank,
|
||||
country_rank=blitz.data.rank_local,
|
||||
play_at=blitz.data.record.ts,
|
||||
)
|
||||
if blitz.data.record is not None
|
||||
else None,
|
||||
zen=Zen(level=zen.data.level, score=zen.data.score),
|
||||
lang=get_lang(),
|
||||
bio=user_info.data.bio,
|
||||
friend_count=user_info.data.friend_count,
|
||||
supporter_tier=user_info.data.supporter_tier,
|
||||
bad_standing=user_info.data.badstanding or False,
|
||||
badges=[
|
||||
Badge(
|
||||
id=i.id,
|
||||
description=i.label,
|
||||
group=i.group,
|
||||
receive_at=i.ts if isinstance(i.ts, datetime) else None,
|
||||
)
|
||||
for i in user_info.data.badges
|
||||
],
|
||||
xp=user_info.data.xp,
|
||||
ar=user_info.data.ar,
|
||||
achievements=[
|
||||
Achievement(
|
||||
key=i.achievement_id,
|
||||
rank_type=i.rank_type,
|
||||
ar_type=i.ar_type,
|
||||
stub=i.stub,
|
||||
rank=i.rank,
|
||||
achieved_score=i.achieved_score,
|
||||
pos=i.pos,
|
||||
progress=i.progress,
|
||||
total=i.total,
|
||||
)
|
||||
for i in achievements.data
|
||||
],
|
||||
playtime=play_time,
|
||||
join_at=user_info.data.ts,
|
||||
),
|
||||
tetra_league=TetraLeague(
|
||||
rank=league.data.rank,
|
||||
highest_rank='z' if isinstance(league.data, NeverRatedData) else league.data.bestrank,
|
||||
tr=round(league.data.tr, 2),
|
||||
glicko=round(league.data.glicko, 2),
|
||||
rd=round(league.data.rd, 2),
|
||||
global_rank=league.data.standing,
|
||||
country_rank=league.data.standing_local,
|
||||
pps=(metrics := get_metrics(pps=league.data.pps, apm=league.data.apm, vs=league.data.vs)).pps,
|
||||
apm=metrics.apm,
|
||||
apl=metrics.apl,
|
||||
vs=metrics.vs,
|
||||
adpl=metrics.adpl,
|
||||
statistic=TetraLeagueStatistic(total=league.data.gamesplayed, wins=league.data.gameswon),
|
||||
decaying=league.data.decaying,
|
||||
history=history,
|
||||
)
|
||||
if not isinstance(league.data, NeverPlayedData | InvalidData)
|
||||
else None,
|
||||
zenith=Zenith(
|
||||
week=Week(
|
||||
altitude=zenith.data.record.results.stats.zenith.altitude,
|
||||
global_rank=zenith.data.rank,
|
||||
country_rank=zenith.data.rank_local,
|
||||
play_at=zenith.data.record.ts,
|
||||
)
|
||||
if zenith.data.record is not None
|
||||
else None,
|
||||
best=Best(
|
||||
altitude=zenith.data.best.record.results.stats.zenith.altitude,
|
||||
global_rank=zenith.data.best.rank,
|
||||
play_at=zenith.data.best.record.ts,
|
||||
)
|
||||
if zenith.data.best.record is not None
|
||||
else None,
|
||||
),
|
||||
zenithex=Zenith(
|
||||
week=Week(
|
||||
altitude=zenithex.data.record.results.stats.zenith.altitude,
|
||||
global_rank=zenithex.data.rank,
|
||||
country_rank=zenithex.data.rank_local,
|
||||
play_at=zenithex.data.record.ts,
|
||||
)
|
||||
if zenithex.data.record is not None
|
||||
else None,
|
||||
best=Best(
|
||||
altitude=zenithex.data.best.record.results.stats.zenith.altitude,
|
||||
global_rank=zenithex.data.best.rank,
|
||||
play_at=zenithex.data.best.record.ts,
|
||||
)
|
||||
if zenithex.data.best.record is not None
|
||||
else None,
|
||||
),
|
||||
statistic=Statistic(
|
||||
total=handling_special_value(user_info.data.gamesplayed),
|
||||
wins=handling_special_value(user_info.data.gameswon),
|
||||
),
|
||||
sprint=Sprint(
|
||||
time=sprint_value,
|
||||
global_rank=sprint.data.rank,
|
||||
country_rank=sprint.data.rank_local,
|
||||
play_at=sprint.data.record.ts,
|
||||
)
|
||||
if sprint.data.record is not None
|
||||
else None,
|
||||
blitz=Blitz(
|
||||
score=blitz.data.record.results.stats.score,
|
||||
global_rank=blitz.data.rank,
|
||||
country_rank=blitz.data.rank_local,
|
||||
play_at=blitz.data.record.ts,
|
||||
)
|
||||
if blitz.data.record is not None
|
||||
else None,
|
||||
zen=Zen(level=zen.data.level, score=zen.data.score),
|
||||
lang=get_lang(),
|
||||
),
|
||||
) as page_hash:
|
||||
return await screenshot(f'http://{netloc}/host/{page_hash}.html')
|
||||
)
|
||||
|
||||
@@ -137,7 +137,7 @@ async def get_tetra_league_data() -> None:
|
||||
await session.commit()
|
||||
|
||||
|
||||
if not config.tetris.development:
|
||||
if not config.tetris.dev.enabled:
|
||||
|
||||
@driver.on_startup
|
||||
async def _() -> None:
|
||||
|
||||
@@ -5,20 +5,18 @@ from nonebot_plugin_alconna import Option, Subcommand, 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 sqlalchemy import func, select
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.orm import selectinload
|
||||
|
||||
from ....db import trigger
|
||||
from ....utils.host import HostPage, get_self_netloc
|
||||
from ....utils.lang import get_lang
|
||||
from ....utils.metrics import get_metrics
|
||||
from ....utils.render import render
|
||||
from ....utils.render import render_image
|
||||
from ....utils.render.schemas.v1.tetrio.rank import Data as DataV1
|
||||
from ....utils.render.schemas.v1.tetrio.rank import ItemData as ItemDataV1
|
||||
from ....utils.render.schemas.v2.tetrio.rank import AverageData as AverageDataV2
|
||||
from ....utils.render.schemas.v2.tetrio.rank import Data as DataV2
|
||||
from ....utils.render.schemas.v2.tetrio.rank import ItemData as ItemDataV2
|
||||
from ....utils.screenshot import screenshot
|
||||
from .. import alc
|
||||
from ..constant import GAME_TYPE
|
||||
from ..models import TETRIOLeagueStats
|
||||
@@ -41,6 +39,7 @@ async def _(event_session: Uninfo, template: Template | None = None):
|
||||
command_args=['--all'] + ([f'--template {template}'] if template is not None else []),
|
||||
):
|
||||
async with get_session() as session:
|
||||
# 获取最新记录
|
||||
latest_data = (
|
||||
await session.scalars(
|
||||
select(TETRIOLeagueStats)
|
||||
@@ -49,19 +48,42 @@ async def _(event_session: Uninfo, template: Template | None = None):
|
||||
.options(selectinload(TETRIOLeagueStats.fields))
|
||||
)
|
||||
).one()
|
||||
compare_data = (
|
||||
await session.scalars(
|
||||
|
||||
# 计算目标时间点 (24小时前)
|
||||
target_time = latest_data.update_time - timedelta(hours=24)
|
||||
|
||||
# 查询目标时间点之前的最近记录
|
||||
before = (
|
||||
await session.scalar(
|
||||
select(TETRIOLeagueStats)
|
||||
.order_by(
|
||||
func.abs(
|
||||
func.julianday(TETRIOLeagueStats.update_time)
|
||||
- func.julianday(latest_data.update_time - timedelta(hours=24))
|
||||
)
|
||||
)
|
||||
.where(TETRIOLeagueStats.update_time <= target_time)
|
||||
.order_by(TETRIOLeagueStats.update_time.desc())
|
||||
.limit(1)
|
||||
.options(selectinload(TETRIOLeagueStats.fields))
|
||||
)
|
||||
).one()
|
||||
or latest_data
|
||||
)
|
||||
|
||||
# 查询目标时间点之后的最近记录
|
||||
after = (
|
||||
await session.scalar(
|
||||
select(TETRIOLeagueStats)
|
||||
.where(TETRIOLeagueStats.update_time >= target_time) # 使用 >= 避免间隙
|
||||
.order_by(TETRIOLeagueStats.update_time.asc())
|
||||
.limit(1)
|
||||
.options(selectinload(TETRIOLeagueStats.fields))
|
||||
)
|
||||
or latest_data
|
||||
)
|
||||
|
||||
# 确定最接近的记录
|
||||
compare_data = (
|
||||
before
|
||||
if abs((target_time - before.update_time).total_seconds())
|
||||
< abs((target_time - after.update_time).total_seconds())
|
||||
else after
|
||||
)
|
||||
|
||||
match template:
|
||||
case 'v1' | None:
|
||||
await UniMessage.image(raw=await make_image_v1(latest_data, compare_data)).finish()
|
||||
@@ -70,49 +92,41 @@ async def _(event_session: Uninfo, template: Template | None = None):
|
||||
|
||||
|
||||
async def make_image_v1(latest_data: TETRIOLeagueStats, compare_data: TETRIOLeagueStats) -> bytes:
|
||||
async with HostPage(
|
||||
await render(
|
||||
'v1/tetrio/rank',
|
||||
DataV1(
|
||||
items={
|
||||
i[0].rank: ItemDataV1(
|
||||
trending=round(i[0].tr_line - i[1].tr_line, 2),
|
||||
require_tr=round(i[0].tr_line, 2),
|
||||
players=i[0].player_count,
|
||||
)
|
||||
for i in zip(latest_data.fields, compare_data.fields, strict=True)
|
||||
},
|
||||
updated_at=latest_data.update_time,
|
||||
lang=get_lang(),
|
||||
),
|
||||
)
|
||||
) as page_hash:
|
||||
return await screenshot(f'http://{get_self_netloc()}/host/{page_hash}.html')
|
||||
return await render_image(
|
||||
DataV1(
|
||||
items={
|
||||
i[0].rank: ItemDataV1(
|
||||
trending=round(i[0].tr_line - i[1].tr_line, 2),
|
||||
require_tr=round(i[0].tr_line, 2),
|
||||
players=i[0].player_count,
|
||||
)
|
||||
for i in zip(latest_data.fields, compare_data.fields, strict=True)
|
||||
},
|
||||
updated_at=latest_data.update_time,
|
||||
lang=get_lang(),
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
async def make_image_v2(latest_data: TETRIOLeagueStats, compare_data: TETRIOLeagueStats) -> bytes:
|
||||
async with HostPage(
|
||||
await render(
|
||||
'v2/tetrio/rank',
|
||||
DataV2(
|
||||
items={
|
||||
i[0].rank: ItemDataV2(
|
||||
require_tr=round(i[0].tr_line, 2),
|
||||
trending=round(i[0].tr_line - i[1].tr_line, 2),
|
||||
average_data=AverageDataV2(
|
||||
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.fields, compare_data.fields, strict=True)
|
||||
},
|
||||
updated_at=latest_data.update_time,
|
||||
lang=get_lang(),
|
||||
),
|
||||
)
|
||||
) as page_hash:
|
||||
return await screenshot(f'http://{get_self_netloc()}/host/{page_hash}.html')
|
||||
return await render_image(
|
||||
DataV2(
|
||||
items={
|
||||
i[0].rank: ItemDataV2(
|
||||
require_tr=round(i[0].tr_line, 2),
|
||||
trending=round(i[0].tr_line - i[1].tr_line, 2),
|
||||
average_data=AverageDataV2(
|
||||
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.fields, compare_data.fields, strict=True)
|
||||
},
|
||||
updated_at=latest_data.update_time,
|
||||
lang=get_lang(),
|
||||
),
|
||||
)
|
||||
|
||||
@@ -7,16 +7,14 @@ from nonebot_plugin_alconna import Option, 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 sqlalchemy import func, select
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.orm import selectinload
|
||||
|
||||
from ....db import trigger
|
||||
from ....utils.host import HostPage, get_self_netloc
|
||||
from ....utils.lang import get_lang
|
||||
from ....utils.metrics import get_metrics
|
||||
from ....utils.render import render
|
||||
from ....utils.render import render_image
|
||||
from ....utils.render.schemas.v2.tetrio.rank.detail import Data, SpecialData
|
||||
from ....utils.screenshot import screenshot
|
||||
from .. import alc
|
||||
from ..api.typedefs import ValidRank
|
||||
from ..constant import GAME_TYPE
|
||||
@@ -39,6 +37,7 @@ async def _(rank: ValidRank, event_session: Uninfo):
|
||||
command_args=[f'--detail {rank}'],
|
||||
):
|
||||
async with get_session() as session:
|
||||
# 获取最新记录
|
||||
latest_data = (
|
||||
await session.scalars(
|
||||
select(TETRIOLeagueStats)
|
||||
@@ -47,19 +46,41 @@ async def _(rank: ValidRank, event_session: Uninfo):
|
||||
.options(selectinload(TETRIOLeagueStats.fields))
|
||||
)
|
||||
).one()
|
||||
compare_data = (
|
||||
await session.scalars(
|
||||
|
||||
# 计算目标时间点 (24小时前)
|
||||
target_time = latest_data.update_time - timedelta(hours=24)
|
||||
|
||||
# 查询目标时间点之前的最近记录
|
||||
before = (
|
||||
await session.scalar(
|
||||
select(TETRIOLeagueStats)
|
||||
.order_by(
|
||||
func.abs(
|
||||
func.julianday(TETRIOLeagueStats.update_time)
|
||||
- func.julianday(latest_data.update_time - timedelta(hours=24))
|
||||
)
|
||||
)
|
||||
.where(TETRIOLeagueStats.update_time <= target_time)
|
||||
.order_by(TETRIOLeagueStats.update_time.desc())
|
||||
.limit(1)
|
||||
.options(selectinload(TETRIOLeagueStats.fields))
|
||||
)
|
||||
).one()
|
||||
or latest_data # 回退到最新记录
|
||||
)
|
||||
|
||||
# 查询目标时间点之后的最近记录
|
||||
after = (
|
||||
await session.scalar(
|
||||
select(TETRIOLeagueStats)
|
||||
.where(TETRIOLeagueStats.update_time >= target_time)
|
||||
.order_by(TETRIOLeagueStats.update_time.asc())
|
||||
.limit(1)
|
||||
.options(selectinload(TETRIOLeagueStats.fields))
|
||||
)
|
||||
or latest_data # 回退到最新记录
|
||||
)
|
||||
|
||||
# 确定最接近的记录
|
||||
compare_data = (
|
||||
before
|
||||
if abs((target_time - before.update_time).total_seconds())
|
||||
< abs((target_time - after.update_time).total_seconds())
|
||||
else after
|
||||
)
|
||||
await UniMessage.image(
|
||||
raw=await make_image(
|
||||
rank,
|
||||
@@ -91,40 +112,36 @@ async def make_image(rank: ValidRank, latest: TETRIOLeagueStats, compare: TETRIO
|
||||
max_vs = get_metrics(
|
||||
pps=latest_data.high_vs.league.pps, apm=latest_data.high_vs.league.apm, vs=latest_data.high_vs.league.vs
|
||||
)
|
||||
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=low_apm.apm,
|
||||
pps=low_pps.pps,
|
||||
lpm=low_pps.lpm,
|
||||
vs=low_vs.vs,
|
||||
adpm=low_vs.adpm,
|
||||
apm_holder=latest_data.low_apm.username.upper(),
|
||||
pps_holder=latest_data.low_pps.username.upper(),
|
||||
vs_holder=latest_data.low_vs.username.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=max_apm.apm,
|
||||
pps=max_pps.pps,
|
||||
lpm=max_pps.lpm,
|
||||
vs=max_vs.vs,
|
||||
adpm=max_vs.adpm,
|
||||
apm_holder=latest_data.high_apm.username.upper(),
|
||||
pps_holder=latest_data.high_pps.username.upper(),
|
||||
vs_holder=latest_data.high_vs.username.upper(),
|
||||
),
|
||||
updated_at=latest.update_time.replace(tzinfo=UTC).astimezone(ZoneInfo('Asia/Shanghai')),
|
||||
lang=get_lang(),
|
||||
return await render_image(
|
||||
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=low_apm.apm,
|
||||
pps=low_pps.pps,
|
||||
lpm=low_pps.lpm,
|
||||
vs=low_vs.vs,
|
||||
adpm=low_vs.adpm,
|
||||
apm_holder=latest_data.low_apm.username.upper(),
|
||||
pps_holder=latest_data.low_pps.username.upper(),
|
||||
vs_holder=latest_data.low_vs.username.upper(),
|
||||
),
|
||||
)
|
||||
) as page_hash:
|
||||
return await screenshot(f'http://{get_self_netloc()}/host/{page_hash}.html')
|
||||
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=max_apm.apm,
|
||||
pps=max_pps.pps,
|
||||
lpm=max_pps.lpm,
|
||||
vs=max_vs.vs,
|
||||
adpm=max_vs.adpm,
|
||||
apm_holder=latest_data.high_apm.username.upper(),
|
||||
pps_holder=latest_data.high_pps.username.upper(),
|
||||
vs_holder=latest_data.high_vs.username.upper(),
|
||||
),
|
||||
updated_at=latest.update_time.replace(tzinfo=UTC).astimezone(ZoneInfo('Asia/Shanghai')),
|
||||
lang=get_lang(),
|
||||
),
|
||||
)
|
||||
|
||||
@@ -15,14 +15,13 @@ from yarl import URL
|
||||
from ....db import query_bind_info, trigger
|
||||
from ....i18n import Lang
|
||||
from ....utils.exception import RecordNotFoundError
|
||||
from ....utils.host import HostPage, get_self_netloc
|
||||
from ....utils.host import get_self_netloc
|
||||
from ....utils.lang import get_lang
|
||||
from ....utils.metrics import get_metrics
|
||||
from ....utils.render import render
|
||||
from ....utils.render import render_image
|
||||
from ....utils.render.schemas.base import Avatar
|
||||
from ....utils.render.schemas.v2.tetrio.record.base import Finesse, Max, Mini, Tspins, User
|
||||
from ....utils.render.schemas.v2.tetrio.record.blitz import Record, Statistic
|
||||
from ....utils.screenshot import screenshot
|
||||
from ....utils.typedefs import Me
|
||||
from .. import alc
|
||||
from ..api.player import Player
|
||||
@@ -88,66 +87,62 @@ async def make_blitz_image(player: Player) -> bytes:
|
||||
duration = timedelta(milliseconds=stats.finaltime).total_seconds()
|
||||
metrics = get_metrics(pps=stats.piecesplaced / duration)
|
||||
netloc = get_self_netloc()
|
||||
async with HostPage(
|
||||
page=await render(
|
||||
'v2/tetrio/record/blitz',
|
||||
Record(
|
||||
type='best',
|
||||
user=User(
|
||||
id=user.ID,
|
||||
name=user.name.upper(),
|
||||
avatar=str(
|
||||
URL(f'http://{netloc}/host/resource/tetrio/avatars/{user.ID}') % {'revision': avatar_revision}
|
||||
)
|
||||
if (avatar_revision := (await player.avatar_revision)) is not None and avatar_revision != 0
|
||||
else Avatar(
|
||||
type='identicon',
|
||||
hash=md5(user.ID.encode()).hexdigest(), # noqa: S324
|
||||
),
|
||||
return await render_image(
|
||||
Record(
|
||||
type='best',
|
||||
user=User(
|
||||
id=user.ID,
|
||||
name=user.name.upper(),
|
||||
avatar=str(
|
||||
URL(f'http://{netloc}/host/resource/tetrio/avatars/{user.ID}') % {'revision': avatar_revision}
|
||||
)
|
||||
if (avatar_revision := (await player.avatar_revision)) is not None and avatar_revision != 0
|
||||
else Avatar(
|
||||
type='identicon',
|
||||
hash=md5(user.ID.encode()).hexdigest(), # noqa: S324
|
||||
),
|
||||
replay_id=blitz.data.record.replayid,
|
||||
rank=blitz.data.rank,
|
||||
personal_rank=1,
|
||||
statistic=Statistic(
|
||||
keys=stats.inputs,
|
||||
kpp=round(stats.inputs / stats.piecesplaced, 2),
|
||||
kps=round(stats.inputs / duration, 2),
|
||||
max=Max(
|
||||
combo=max((0, stats.topcombo - 1)),
|
||||
btb=max((0, stats.topbtb - 1)),
|
||||
),
|
||||
pieces=stats.piecesplaced,
|
||||
pps=metrics.pps,
|
||||
lines=stats.lines,
|
||||
lpm=metrics.lpm,
|
||||
holds=stats.holds,
|
||||
score=stats.score,
|
||||
spp=round(stats.score / stats.piecesplaced, 2),
|
||||
single=clears.singles,
|
||||
double=clears.doubles,
|
||||
triple=clears.triples,
|
||||
quad=clears.quads,
|
||||
tspins=Tspins(
|
||||
total=clears.realtspins,
|
||||
single=clears.tspinsingles,
|
||||
double=clears.tspindoubles,
|
||||
triple=clears.tspintriples,
|
||||
mini=Mini(
|
||||
total=clears.minitspins,
|
||||
single=clears.minitspinsingles,
|
||||
double=clears.minitspindoubles,
|
||||
),
|
||||
),
|
||||
all_clear=clears.allclear,
|
||||
finesse=Finesse(
|
||||
faults=stats.finesse.faults,
|
||||
accuracy=round(stats.finesse.perfectpieces / stats.piecesplaced * 100, 2),
|
||||
),
|
||||
level=stats.level,
|
||||
),
|
||||
play_at=blitz.data.record.ts,
|
||||
lang=get_lang(),
|
||||
),
|
||||
)
|
||||
) as page_hash:
|
||||
return await screenshot(f'http://{netloc}/host/{page_hash}.html')
|
||||
replay_id=blitz.data.record.replayid,
|
||||
rank=blitz.data.rank,
|
||||
personal_rank=1,
|
||||
statistic=Statistic(
|
||||
keys=stats.inputs,
|
||||
kpp=round(stats.inputs / stats.piecesplaced, 2),
|
||||
kps=round(stats.inputs / duration, 2),
|
||||
max=Max(
|
||||
combo=max((0, stats.topcombo - 1)),
|
||||
btb=max((0, stats.topbtb - 1)),
|
||||
),
|
||||
pieces=stats.piecesplaced,
|
||||
pps=metrics.pps,
|
||||
lines=stats.lines,
|
||||
lpm=metrics.lpm,
|
||||
holds=stats.holds,
|
||||
score=stats.score,
|
||||
spp=round(stats.score / stats.piecesplaced, 2),
|
||||
single=clears.singles,
|
||||
double=clears.doubles,
|
||||
triple=clears.triples,
|
||||
quad=clears.quads,
|
||||
tspins=Tspins(
|
||||
total=clears.realtspins,
|
||||
single=clears.tspinsingles,
|
||||
double=clears.tspindoubles,
|
||||
triple=clears.tspintriples,
|
||||
mini=Mini(
|
||||
total=clears.minitspins,
|
||||
single=clears.minitspinsingles,
|
||||
double=clears.minitspindoubles,
|
||||
),
|
||||
),
|
||||
all_clear=clears.allclear,
|
||||
finesse=Finesse(
|
||||
faults=stats.finesse.faults,
|
||||
accuracy=round(stats.finesse.perfectpieces / stats.piecesplaced * 100, 2),
|
||||
),
|
||||
level=stats.level,
|
||||
),
|
||||
play_at=blitz.data.record.ts,
|
||||
lang=get_lang(),
|
||||
),
|
||||
)
|
||||
|
||||
@@ -15,14 +15,13 @@ from yarl import URL
|
||||
from ....db import query_bind_info, trigger
|
||||
from ....i18n import Lang
|
||||
from ....utils.exception import RecordNotFoundError
|
||||
from ....utils.host import HostPage, get_self_netloc
|
||||
from ....utils.host import get_self_netloc
|
||||
from ....utils.lang import get_lang
|
||||
from ....utils.metrics import get_metrics
|
||||
from ....utils.render import render
|
||||
from ....utils.render import render_image
|
||||
from ....utils.render.schemas.base import Avatar
|
||||
from ....utils.render.schemas.v2.tetrio.record.base import Finesse, Max, Mini, Statistic, Tspins, User
|
||||
from ....utils.render.schemas.v2.tetrio.record.sprint import Record
|
||||
from ....utils.screenshot import screenshot
|
||||
from ....utils.typedefs import Me
|
||||
from .. import alc
|
||||
from ..api.player import Player
|
||||
@@ -89,65 +88,61 @@ async def make_sprint_image(player: Player) -> bytes:
|
||||
sprint_value = f'{duration:.3f}s' if duration < 60 else f'{duration // 60:.0f}m {duration % 60:.3f}s' # noqa: PLR2004
|
||||
metrics = get_metrics(pps=stats.piecesplaced / duration)
|
||||
netloc = get_self_netloc()
|
||||
async with HostPage(
|
||||
page=await render(
|
||||
'v2/tetrio/record/sprint',
|
||||
Record(
|
||||
type='best',
|
||||
user=User(
|
||||
id=user.ID,
|
||||
name=user.name.upper(),
|
||||
avatar=str(
|
||||
URL(f'http://{netloc}/host/resource/tetrio/avatars/{user.ID}') % {'revision': avatar_revision}
|
||||
)
|
||||
if (avatar_revision := (await player.avatar_revision)) is not None and avatar_revision != 0
|
||||
else Avatar(
|
||||
type='identicon',
|
||||
hash=md5(user.ID.encode()).hexdigest(), # noqa: S324
|
||||
),
|
||||
return await render_image(
|
||||
Record(
|
||||
type='best',
|
||||
user=User(
|
||||
id=user.ID,
|
||||
name=user.name.upper(),
|
||||
avatar=str(
|
||||
URL(f'http://{netloc}/host/resource/tetrio/avatars/{user.ID}') % {'revision': avatar_revision}
|
||||
)
|
||||
if (avatar_revision := (await player.avatar_revision)) is not None and avatar_revision != 0
|
||||
else Avatar(
|
||||
type='identicon',
|
||||
hash=md5(user.ID.encode()).hexdigest(), # noqa: S324
|
||||
),
|
||||
time=sprint_value,
|
||||
replay_id=sprint.data.record.replayid,
|
||||
rank=sprint.data.rank,
|
||||
personal_rank=1,
|
||||
statistic=Statistic(
|
||||
keys=stats.inputs,
|
||||
kpp=round(stats.inputs / stats.piecesplaced, 2),
|
||||
kps=round(stats.inputs / duration, 2),
|
||||
max=Max(
|
||||
combo=max((0, stats.topcombo - 1)),
|
||||
btb=max((0, stats.topbtb - 1)),
|
||||
),
|
||||
pieces=stats.piecesplaced,
|
||||
pps=metrics.pps,
|
||||
lines=stats.lines,
|
||||
lpm=metrics.lpm,
|
||||
holds=stats.holds,
|
||||
score=stats.score,
|
||||
single=clears.singles,
|
||||
double=clears.doubles,
|
||||
triple=clears.triples,
|
||||
quad=clears.quads,
|
||||
tspins=Tspins(
|
||||
total=clears.realtspins,
|
||||
single=clears.tspinsingles,
|
||||
double=clears.tspindoubles,
|
||||
triple=clears.tspintriples,
|
||||
mini=Mini(
|
||||
total=clears.minitspins,
|
||||
single=clears.minitspinsingles,
|
||||
double=clears.minitspindoubles,
|
||||
),
|
||||
),
|
||||
all_clear=clears.allclear,
|
||||
finesse=Finesse(
|
||||
faults=stats.finesse.faults,
|
||||
accuracy=round(stats.finesse.perfectpieces / stats.piecesplaced * 100, 2),
|
||||
),
|
||||
),
|
||||
play_at=sprint.data.record.ts,
|
||||
lang=get_lang(),
|
||||
),
|
||||
)
|
||||
) as page_hash:
|
||||
return await screenshot(f'http://{netloc}/host/{page_hash}.html')
|
||||
time=sprint_value,
|
||||
replay_id=sprint.data.record.replayid,
|
||||
rank=sprint.data.rank,
|
||||
personal_rank=1,
|
||||
statistic=Statistic(
|
||||
keys=stats.inputs,
|
||||
kpp=round(stats.inputs / stats.piecesplaced, 2),
|
||||
kps=round(stats.inputs / duration, 2),
|
||||
max=Max(
|
||||
combo=max((0, stats.topcombo - 1)),
|
||||
btb=max((0, stats.topbtb - 1)),
|
||||
),
|
||||
pieces=stats.piecesplaced,
|
||||
pps=metrics.pps,
|
||||
lines=stats.lines,
|
||||
lpm=metrics.lpm,
|
||||
holds=stats.holds,
|
||||
score=stats.score,
|
||||
single=clears.singles,
|
||||
double=clears.doubles,
|
||||
triple=clears.triples,
|
||||
quad=clears.quads,
|
||||
tspins=Tspins(
|
||||
total=clears.realtspins,
|
||||
single=clears.tspinsingles,
|
||||
double=clears.tspindoubles,
|
||||
triple=clears.tspintriples,
|
||||
mini=Mini(
|
||||
total=clears.minitspins,
|
||||
single=clears.minitspinsingles,
|
||||
double=clears.minitspindoubles,
|
||||
),
|
||||
),
|
||||
all_clear=clears.allclear,
|
||||
finesse=Finesse(
|
||||
faults=stats.finesse.faults,
|
||||
accuracy=round(stats.finesse.perfectpieces / stats.piecesplaced * 100, 2),
|
||||
),
|
||||
),
|
||||
play_at=sprint.data.record.ts,
|
||||
lang=get_lang(),
|
||||
),
|
||||
)
|
||||
|
||||
@@ -13,12 +13,12 @@ from yarl import URL
|
||||
|
||||
from ...config.config import global_config
|
||||
from ...db import query_bind_info, remove_bind, trigger
|
||||
from ...utils.host import HostPage, get_self_netloc
|
||||
from ...utils.host import get_self_netloc
|
||||
from ...utils.image import get_avatar
|
||||
from ...utils.lang import get_lang
|
||||
from ...utils.render import Bind, render
|
||||
from ...utils.render import render_image
|
||||
from ...utils.render.schemas.base import Avatar, People
|
||||
from ...utils.screenshot import screenshot
|
||||
from ...utils.render.schemas.bind import Bind
|
||||
from . import alc, command
|
||||
from .api import Player
|
||||
from .constant import GAME_TYPE
|
||||
@@ -51,9 +51,8 @@ async def _(nb_user: User, event_session: Uninfo, interface: QryItrface):
|
||||
player = Player(user_id=bind.game_account, trust=True)
|
||||
user = await player.user
|
||||
netloc = get_self_netloc()
|
||||
async with HostPage(
|
||||
await render(
|
||||
'v1/binding',
|
||||
await UniMessage.image(
|
||||
raw=await render_image(
|
||||
Bind(
|
||||
platform='TETR.IO',
|
||||
type='unlink',
|
||||
@@ -81,6 +80,5 @@ async def _(nb_user: User, event_session: Uninfo, interface: QryItrface):
|
||||
lang=get_lang(),
|
||||
),
|
||||
)
|
||||
) as page_hash:
|
||||
await UniMessage.image(raw=await screenshot(f'http://{netloc}/host/{page_hash}.html')).send()
|
||||
).send()
|
||||
await remove_bind(session=session, user=nb_user, game_platform=GAME_TYPE)
|
||||
|
||||
110
nonebot_plugin_tetris_stats/games/tetrio/verify.py
Normal file
110
nonebot_plugin_tetris_stats/games/tetrio/verify.py
Normal file
@@ -0,0 +1,110 @@
|
||||
from asyncio import gather
|
||||
from hashlib import md5
|
||||
from secrets import choice
|
||||
|
||||
from nonebot_plugin_alconna import Subcommand
|
||||
from nonebot_plugin_alconna.uniseg import UniMessage
|
||||
from nonebot_plugin_orm import get_session
|
||||
from nonebot_plugin_uninfo import QryItrface, Uninfo
|
||||
from nonebot_plugin_uninfo import User as UninfoUser
|
||||
from nonebot_plugin_uninfo.orm import get_session_persist_id
|
||||
from nonebot_plugin_user import User
|
||||
from yarl import URL
|
||||
|
||||
from ...config.config import global_config
|
||||
from ...db import create_or_update_bind, query_bind_info, trigger
|
||||
from ...utils.host import get_self_netloc
|
||||
from ...utils.image import get_avatar
|
||||
from ...utils.lang import get_lang
|
||||
from ...utils.render import render_image
|
||||
from ...utils.render.schemas.base import Avatar, People
|
||||
from ...utils.render.schemas.bind import Bind
|
||||
from . import alc, command
|
||||
from .api import Player
|
||||
from .constant import GAME_TYPE
|
||||
|
||||
command.add(Subcommand('verify', help_text='验证 TETR.IO 账号'))
|
||||
|
||||
alc.shortcut(
|
||||
'(?i:io)(?i:验证|verify)',
|
||||
command='tstats TETR.IO verify',
|
||||
humanized='io验证',
|
||||
)
|
||||
|
||||
try:
|
||||
from nonebot.adapters.discord import MessageCreateEvent
|
||||
|
||||
@alc.assign('TETRIO.verify')
|
||||
async def _(_: MessageCreateEvent, nb_user: User, event_session: Uninfo, interface: QryItrface):
|
||||
async with (
|
||||
trigger(
|
||||
session_persist_id=await get_session_persist_id(event_session),
|
||||
game_platform=GAME_TYPE,
|
||||
command_type='verify',
|
||||
command_args=[],
|
||||
),
|
||||
get_session() as session,
|
||||
):
|
||||
if (bind := await query_bind_info(session=session, user=nb_user, game_platform=GAME_TYPE)) is None:
|
||||
await UniMessage('您还未绑定 TETR.IO 账号').finish()
|
||||
if bind.verify is True:
|
||||
await UniMessage('您已经完成了验证.').finish()
|
||||
player = Player(user_id=bind.game_account, trust=True)
|
||||
user_info = await player.get_info()
|
||||
verify = (
|
||||
user_info.data.connections.discord is not None
|
||||
and user_info.data.connections.discord.id == event_session.user.id
|
||||
)
|
||||
if verify is False:
|
||||
await UniMessage('您未通过验证, 请确认目标 TETR.IO 账号绑定了当前 Discord 账号').finish()
|
||||
await create_or_update_bind(
|
||||
session=session,
|
||||
user=nb_user,
|
||||
game_platform=GAME_TYPE,
|
||||
game_account=bind.game_account,
|
||||
verify=verify,
|
||||
)
|
||||
user, avatar_revision = await gather(player.user, player.avatar_revision)
|
||||
await UniMessage.image(
|
||||
raw=await render_image(
|
||||
Bind(
|
||||
platform='TETR.IO',
|
||||
type='success',
|
||||
user=People(
|
||||
avatar=str(
|
||||
URL(f'http://{get_self_netloc()}/host/resource/tetrio/avatars/{user.ID}')
|
||||
% {'revision': avatar_revision}
|
||||
)
|
||||
if avatar_revision is not None and avatar_revision != 0
|
||||
else Avatar(type='identicon', hash=md5(user.ID.encode()).hexdigest()), # noqa: S324
|
||||
name=user.name.upper(),
|
||||
),
|
||||
bot=People(
|
||||
avatar=await get_avatar(
|
||||
(
|
||||
bot_user := await interface.get_user(event_session.self_id)
|
||||
or UninfoUser(id=event_session.self_id)
|
||||
),
|
||||
'Data URI',
|
||||
'../../static/logo/logo.svg',
|
||||
),
|
||||
name=bot_user.nick or bot_user.name or choice(list(global_config.nickname) or ['bot']),
|
||||
),
|
||||
prompt='io查我',
|
||||
lang=get_lang(),
|
||||
),
|
||||
)
|
||||
).finish()
|
||||
except ImportError:
|
||||
pass
|
||||
|
||||
|
||||
@alc.assign('TETRIO.verify')
|
||||
async def _(event_session: Uninfo):
|
||||
async with trigger(
|
||||
session_persist_id=await get_session_persist_id(event_session),
|
||||
game_platform=GAME_TYPE,
|
||||
command_type='verify',
|
||||
command_args=[],
|
||||
):
|
||||
await UniMessage('目前仅支持 Discord 账号验证').finish()
|
||||
@@ -9,12 +9,11 @@ from nonebot_plugin_user import User
|
||||
|
||||
from ...config.config import global_config
|
||||
from ...db import BindStatus, create_or_update_bind, trigger
|
||||
from ...utils.host import HostPage, get_self_netloc
|
||||
from ...utils.image import get_avatar
|
||||
from ...utils.lang import get_lang
|
||||
from ...utils.render import Bind, render
|
||||
from ...utils.render import render_image
|
||||
from ...utils.render.schemas.base import People
|
||||
from ...utils.screenshot import screenshot
|
||||
from ...utils.render.schemas.bind import Bind
|
||||
from . import alc
|
||||
from .api import Player
|
||||
from .constant import GAME_TYPE
|
||||
@@ -37,9 +36,8 @@ async def _(nb_user: User, account: Player, event_session: Uninfo, interface: Qr
|
||||
game_account=user.unique_identifier,
|
||||
)
|
||||
if bind_status in (BindStatus.SUCCESS, BindStatus.UPDATE):
|
||||
async with HostPage(
|
||||
await render(
|
||||
'v1/binding',
|
||||
await UniMessage.image(
|
||||
raw=await render_image(
|
||||
Bind(
|
||||
platform=GAME_TYPE,
|
||||
type='unknown',
|
||||
@@ -66,7 +64,4 @@ async def _(nb_user: User, account: Player, event_session: Uninfo, interface: Qr
|
||||
lang=get_lang(),
|
||||
),
|
||||
)
|
||||
) as page_hash:
|
||||
await UniMessage.image(
|
||||
raw=await screenshot(f'http://{get_self_netloc()}/host/{page_hash}.html')
|
||||
).finish()
|
||||
).finish()
|
||||
|
||||
@@ -10,15 +10,13 @@ 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.host import HostPage, get_self_netloc
|
||||
from ...utils.lang import get_lang
|
||||
from ...utils.metrics import TetrisMetricsBasicWithLPM, get_metrics
|
||||
from ...utils.render import render
|
||||
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.screenshot import screenshot
|
||||
from ...utils.typedefs import Me
|
||||
from . import alc
|
||||
from .api import Player
|
||||
@@ -75,32 +73,28 @@ async def make_query_image(profile: UserProfile) -> bytes:
|
||||
raise FallbackError
|
||||
today = get_metrics(lpm=profile.today.lpm, apm=profile.today.apm)
|
||||
history = get_avg_metrics(profile.total)
|
||||
async with HostPage(
|
||||
await render(
|
||||
'v1/top/info',
|
||||
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(),
|
||||
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,
|
||||
),
|
||||
)
|
||||
) as page_hash:
|
||||
return await screenshot(f'http://{get_self_netloc()}/host/{page_hash}.html')
|
||||
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:
|
||||
|
||||
@@ -10,12 +10,11 @@ from nonebot_plugin_waiter import suggest # type: ignore[import-untyped]
|
||||
|
||||
from ...config.config import global_config
|
||||
from ...db import query_bind_info, remove_bind, trigger
|
||||
from ...utils.host import HostPage, get_self_netloc
|
||||
from ...utils.image import get_avatar
|
||||
from ...utils.lang import get_lang
|
||||
from ...utils.render import Bind, render
|
||||
from ...utils.render import render_image
|
||||
from ...utils.render.schemas.base import People
|
||||
from ...utils.screenshot import screenshot
|
||||
from ...utils.render.schemas.bind import Bind
|
||||
from . import alc
|
||||
from .api import Player
|
||||
from .constant import GAME_TYPE
|
||||
@@ -43,10 +42,8 @@ async def _(
|
||||
return
|
||||
player = Player(user_name=bind.game_account, trust=True)
|
||||
user = await player.user
|
||||
netloc = get_self_netloc()
|
||||
async with HostPage(
|
||||
await render(
|
||||
'v1/binding',
|
||||
await UniMessage.image(
|
||||
raw=await render_image(
|
||||
Bind(
|
||||
platform='TOP',
|
||||
type='unlink',
|
||||
@@ -73,6 +70,5 @@ async def _(
|
||||
lang=get_lang(),
|
||||
),
|
||||
)
|
||||
) as page_hash:
|
||||
await UniMessage.image(raw=await screenshot(f'http://{netloc}/host/{page_hash}.html')).send()
|
||||
).send()
|
||||
await remove_bind(session=session, user=nb_user, game_platform=GAME_TYPE)
|
||||
|
||||
@@ -9,12 +9,11 @@ from nonebot_plugin_user import User
|
||||
|
||||
from ...config.config import global_config
|
||||
from ...db import BindStatus, create_or_update_bind, trigger
|
||||
from ...utils.host import HostPage, get_self_netloc
|
||||
from ...utils.image import get_avatar
|
||||
from ...utils.lang import get_lang
|
||||
from ...utils.render import Bind, render
|
||||
from ...utils.render import render_image
|
||||
from ...utils.render.schemas.base import People
|
||||
from ...utils.screenshot import screenshot
|
||||
from ...utils.render.schemas.bind import Bind
|
||||
from . import alc
|
||||
from .api import Player
|
||||
from .constant import GAME_TYPE
|
||||
@@ -43,9 +42,8 @@ async def _(
|
||||
)
|
||||
user_info = await account.get_info()
|
||||
if bind_status in (BindStatus.SUCCESS, BindStatus.UPDATE):
|
||||
async with HostPage(
|
||||
await render(
|
||||
'v1/binding',
|
||||
await UniMessage.image(
|
||||
raw=await render_image(
|
||||
Bind(
|
||||
platform=GAME_TYPE,
|
||||
type='unknown',
|
||||
@@ -72,7 +70,4 @@ async def _(
|
||||
lang=get_lang(),
|
||||
),
|
||||
)
|
||||
) as page_hash:
|
||||
await UniMessage.image(
|
||||
raw=await screenshot(f'http://{get_self_netloc()}/host/{page_hash}.html')
|
||||
).finish()
|
||||
).finish()
|
||||
|
||||
@@ -18,16 +18,14 @@ from ...db import query_bind_info, trigger
|
||||
from ...i18n import Lang
|
||||
from ...utils.chart import get_split, get_value_bounds, handle_history_data
|
||||
from ...utils.exception import RequestError
|
||||
from ...utils.host import HostPage, get_self_netloc
|
||||
from ...utils.image import get_avatar
|
||||
from ...utils.lang import get_lang
|
||||
from ...utils.metrics import TetrisMetricsProWithLPMADPM, get_metrics
|
||||
from ...utils.render import render
|
||||
from ...utils.render import render_image
|
||||
from ...utils.render.avatar import get_avatar as get_random_avatar
|
||||
from ...utils.render.schemas.base import HistoryData, People, Trending
|
||||
from ...utils.render.schemas.v1.base import History
|
||||
from ...utils.render.schemas.v1.tos.info import Info, Multiplayer, Singleplayer
|
||||
from ...utils.screenshot import screenshot
|
||||
from ...utils.time_it import time_it
|
||||
from ...utils.typedefs import Me, Number
|
||||
from . import alc
|
||||
@@ -264,52 +262,48 @@ async def make_query_image(user_info: UserInfoSuccess, game_data: GameData, even
|
||||
)
|
||||
data = handle_history_data(await get_historical_data(user_info.data.teaid))
|
||||
values = get_value_bounds([i.score for i in data])
|
||||
async with HostPage(
|
||||
await render(
|
||||
'v1/tos/info',
|
||||
Info(
|
||||
user=People(
|
||||
avatar=await get_avatar(event_user_info, 'Data URI', None)
|
||||
if event_user_info is not None
|
||||
else get_random_avatar(user_info.data.teaid),
|
||||
name=user_info.data.name,
|
||||
),
|
||||
multiplayer=Multiplayer(
|
||||
history=History(
|
||||
data=data,
|
||||
max_value=values.value_max,
|
||||
min_value=values.value_min,
|
||||
split_interval=(split := get_split(value_bound=values, min_value=0)).split_value,
|
||||
offset=split.offset,
|
||||
),
|
||||
rating=round(float(user_info.data.rating_now), 2),
|
||||
rd=round(float(user_info.data.rd_now), 2),
|
||||
lpm=metrics.lpm,
|
||||
pps=metrics.pps,
|
||||
lpm_trending=Trending.KEEP,
|
||||
apm=metrics.apm,
|
||||
apl=metrics.apl,
|
||||
apm_trending=Trending.KEEP,
|
||||
adpm=metrics.adpm,
|
||||
vs=metrics.vs,
|
||||
adpl=metrics.adpl,
|
||||
adpm_trending=Trending.KEEP,
|
||||
app=(app := (metrics.apm / (60 * metrics.pps))),
|
||||
or_=game_data.or_,
|
||||
dspp=game_data.dspp,
|
||||
ci=150 * game_data.dspp - 125 * app + 50 * (metrics.vs / metrics.apm) - 25,
|
||||
ge=game_data.ge,
|
||||
),
|
||||
singleplayer=Singleplayer(
|
||||
sprint=sprint_value,
|
||||
challenge=f'{int(user_info.data.pb_challenge):,}' if user_info.data.pb_challenge != '0' else 'N/A',
|
||||
marathon=f'{int(user_info.data.pb_marathon):,}' if user_info.data.pb_marathon != '0' else 'N/A',
|
||||
),
|
||||
lang=get_lang(),
|
||||
return await render_image(
|
||||
Info(
|
||||
user=People(
|
||||
avatar=await get_avatar(event_user_info, 'Data URI', None)
|
||||
if event_user_info is not None
|
||||
else get_random_avatar(user_info.data.teaid),
|
||||
name=user_info.data.name,
|
||||
),
|
||||
)
|
||||
) as page_hash:
|
||||
return await screenshot(f'http://{get_self_netloc()}/host/{page_hash}.html')
|
||||
multiplayer=Multiplayer(
|
||||
history=History(
|
||||
data=data,
|
||||
max_value=values.value_max,
|
||||
min_value=values.value_min,
|
||||
split_interval=(split := get_split(value_bound=values, min_value=0)).split_value,
|
||||
offset=split.offset,
|
||||
),
|
||||
rating=round(float(user_info.data.rating_now), 2),
|
||||
rd=round(float(user_info.data.rd_now), 2),
|
||||
lpm=metrics.lpm,
|
||||
pps=metrics.pps,
|
||||
lpm_trending=Trending.KEEP,
|
||||
apm=metrics.apm,
|
||||
apl=metrics.apl,
|
||||
apm_trending=Trending.KEEP,
|
||||
adpm=metrics.adpm,
|
||||
vs=metrics.vs,
|
||||
adpl=metrics.adpl,
|
||||
adpm_trending=Trending.KEEP,
|
||||
app=(app := (metrics.apm / (60 * metrics.pps))),
|
||||
or_=game_data.or_,
|
||||
dspp=game_data.dspp,
|
||||
ci=150 * game_data.dspp - 125 * app + 50 * (metrics.vs / metrics.apm) - 25,
|
||||
ge=game_data.ge,
|
||||
),
|
||||
singleplayer=Singleplayer(
|
||||
sprint=sprint_value,
|
||||
challenge=f'{int(user_info.data.pb_challenge):,}' if user_info.data.pb_challenge != '0' else 'N/A',
|
||||
marathon=f'{int(user_info.data.pb_marathon):,}' if user_info.data.pb_marathon != '0' else 'N/A',
|
||||
),
|
||||
lang=get_lang(),
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
def make_query_text(user_info: UserInfoSuccess, game_data: GameData | None) -> UniMessage:
|
||||
|
||||
@@ -10,12 +10,11 @@ from nonebot_plugin_waiter import suggest # type: ignore[import-untyped]
|
||||
|
||||
from ...config.config import global_config
|
||||
from ...db import query_bind_info, remove_bind, trigger
|
||||
from ...utils.host import HostPage, get_self_netloc
|
||||
from ...utils.image import get_avatar
|
||||
from ...utils.lang import get_lang
|
||||
from ...utils.render import Bind, render
|
||||
from ...utils.render import render_image
|
||||
from ...utils.render.schemas.base import People
|
||||
from ...utils.screenshot import screenshot
|
||||
from ...utils.render.schemas.bind import Bind
|
||||
from . import alc
|
||||
from .api import Player
|
||||
from .constant import GAME_TYPE
|
||||
@@ -43,10 +42,8 @@ async def _(
|
||||
return
|
||||
player = Player(user_name=bind.game_account, trust=True)
|
||||
user = await player.user
|
||||
netloc = get_self_netloc()
|
||||
async with HostPage(
|
||||
await render(
|
||||
'v1/binding',
|
||||
await UniMessage.image(
|
||||
raw=await render_image(
|
||||
Bind(
|
||||
platform='TOS',
|
||||
type='unlink',
|
||||
@@ -69,6 +66,5 @@ async def _(
|
||||
lang=get_lang(),
|
||||
),
|
||||
)
|
||||
) as page_hash:
|
||||
await UniMessage.image(raw=await screenshot(f'http://{netloc}/host/{page_hash}.html')).send()
|
||||
).send()
|
||||
await remove_bind(session=session, user=nb_user, game_platform=GAME_TYPE)
|
||||
|
||||
@@ -78,7 +78,7 @@ class BrowserManager:
|
||||
"""启动浏览器实例"""
|
||||
playwright = await async_playwright().start()
|
||||
cls._browser = await playwright.firefox.launch(
|
||||
headless=not config.tetris.development,
|
||||
headless=not config.tetris.dev.enabled,
|
||||
firefox_user_prefs={
|
||||
'network.http.max-persistent-connections-per-server': 64,
|
||||
},
|
||||
|
||||
@@ -45,7 +45,7 @@ class HostPage:
|
||||
async def __aenter__(self) -> str:
|
||||
return self.page_hash
|
||||
|
||||
if not config.tetris.development:
|
||||
if not config.tetris.dev.enabled:
|
||||
|
||||
async def __aexit__(self, exc_type, exc, tb) -> None: # noqa: ANN001
|
||||
self.pages.pop(self.page_hash, None)
|
||||
|
||||
@@ -1,22 +1,10 @@
|
||||
from typing import Literal, overload
|
||||
|
||||
from jinja2 import Environment, FileSystemLoader
|
||||
from nonebot.compat import PYDANTIC_V2
|
||||
|
||||
from ..host import HostPage, get_self_netloc
|
||||
from ..screenshot import screenshot
|
||||
from ..templates import TEMPLATES_DIR
|
||||
from .schemas.base import Base
|
||||
from .schemas.bind import Bind
|
||||
from .schemas.v1.tetrio.rank import Data as TETRIORankDataV1
|
||||
from .schemas.v1.tetrio.user.info import Info as TETRIOUserInfoV1
|
||||
from .schemas.v1.top.info import Info as TOPInfoV1
|
||||
from .schemas.v1.tos.info import Info as TOSInfoV1
|
||||
from .schemas.v2.tetrio.rank import Data as TETRIORankDataV2
|
||||
from .schemas.v2.tetrio.rank.detail import Data as TETRIORankDetailDataV2
|
||||
from .schemas.v2.tetrio.record.blitz import Record as TETRIORecordBlitzV2
|
||||
from .schemas.v2.tetrio.record.sprint import Record as TETRIORecordSprintV2
|
||||
from .schemas.v2.tetrio.tetra_league import Data as TETRIOTetraLeagueDataV2
|
||||
from .schemas.v2.tetrio.user.info import Info as TETRIOUserInfoV2
|
||||
from .schemas.v2.tetrio.user.list import List as TETRIOUserListV2
|
||||
|
||||
env = Environment(
|
||||
loader=FileSystemLoader(TEMPLATES_DIR),
|
||||
@@ -27,39 +15,19 @@ 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: TETRIOUserInfoV1) -> str: ...
|
||||
@overload
|
||||
async def render(render_type: Literal['v1/tetrio/rank'], data: TETRIORankDataV1) -> str: ...
|
||||
@overload
|
||||
async def render(render_type: Literal['v1/top/info'], data: TOPInfoV1) -> str: ...
|
||||
@overload
|
||||
async def render(render_type: Literal['v1/tos/info'], data: TOSInfoV1) -> str: ...
|
||||
@overload
|
||||
async def render(render_type: Literal['v2/tetrio/rank'], data: TETRIORankDataV2) -> str: ...
|
||||
@overload
|
||||
async def render(render_type: Literal['v2/tetrio/rank/detail'], data: TETRIORankDetailDataV2) -> str: ...
|
||||
@overload
|
||||
async def render(render_type: Literal['v2/tetrio/record/blitz'], data: TETRIORecordBlitzV2) -> str: ...
|
||||
@overload
|
||||
async def render(render_type: Literal['v2/tetrio/record/sprint'], data: TETRIORecordSprintV2) -> str: ...
|
||||
@overload
|
||||
async def render(render_type: Literal['v2/tetrio/tetra-league'], data: TETRIOTetraLeagueDataV2) -> 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: ...
|
||||
async def render(
|
||||
render_type: str,
|
||||
data: Base,
|
||||
) -> str:
|
||||
if PYDANTIC_V2:
|
||||
return await env.get_template('index.html').render_async(
|
||||
path=render_type, data=data.model_dump_json(by_alias=True)
|
||||
)
|
||||
return await env.get_template('index.html').render_async(path=render_type, data=data.json(by_alias=True))
|
||||
return await env.get_template('index.html').render_async(data=data.model_dump_json(by_alias=True))
|
||||
return await env.get_template('index.html').render_async(data=data.json(by_alias=True))
|
||||
|
||||
|
||||
async def render_image(
|
||||
data: Base,
|
||||
) -> bytes:
|
||||
async with HostPage(page=await render(data)) as page_hash:
|
||||
return await screenshot(f'http://{get_self_netloc()}/host/{page_hash}.html#/{data.path}')
|
||||
|
||||
|
||||
__all__ = ['render']
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
from abc import abstractmethod
|
||||
from datetime import datetime
|
||||
from typing import Literal
|
||||
|
||||
@@ -8,6 +9,11 @@ from ...typedefs import Lang, Number
|
||||
|
||||
|
||||
class Base(BaseModel):
|
||||
@property
|
||||
@abstractmethod
|
||||
def path(self) -> str:
|
||||
raise NotImplementedError
|
||||
|
||||
lang: Lang
|
||||
|
||||
|
||||
|
||||
@@ -1,9 +1,16 @@
|
||||
from typing import Literal
|
||||
|
||||
from typing_extensions import override
|
||||
|
||||
from .base import Base, People
|
||||
|
||||
|
||||
class Bind(Base):
|
||||
@property
|
||||
@override
|
||||
def path(self) -> str:
|
||||
return 'v1/binding'
|
||||
|
||||
platform: Literal['TETR.IO', 'TOP', 'TOS']
|
||||
type: Literal['success', 'unknown', 'unlink', 'unverified', 'error']
|
||||
user: People
|
||||
|
||||
@@ -1,9 +1,10 @@
|
||||
from pydantic import BaseModel
|
||||
from typing_extensions import override
|
||||
|
||||
from .......games.tetrio.api.typedefs import Rank
|
||||
from ......typedefs import Number
|
||||
from ....base import Base, People, Trending
|
||||
from ...base import History
|
||||
from ......games.tetrio.api.typedefs import Rank
|
||||
from .....typedefs import Number
|
||||
from ...base import Base, People, Trending
|
||||
from ..base import History
|
||||
|
||||
|
||||
class User(People):
|
||||
@@ -45,6 +46,11 @@ class Singleplayer(BaseModel):
|
||||
|
||||
|
||||
class Info(Base):
|
||||
@property
|
||||
@override
|
||||
def path(self) -> str:
|
||||
return 'v1/tetrio/info'
|
||||
|
||||
user: User
|
||||
multiplayer: Multiplayer
|
||||
singleplayer: Singleplayer
|
||||
@@ -1,6 +1,7 @@
|
||||
from datetime import datetime
|
||||
|
||||
from pydantic import BaseModel
|
||||
from typing_extensions import override
|
||||
|
||||
from ......games.tetrio.api.typedefs import ValidRank
|
||||
from ...base import Base
|
||||
@@ -13,5 +14,10 @@ class ItemData(BaseModel):
|
||||
|
||||
|
||||
class Data(Base):
|
||||
@property
|
||||
@override
|
||||
def path(self) -> str:
|
||||
return 'v1/tetrio/rank'
|
||||
|
||||
items: dict[ValidRank, ItemData]
|
||||
updated_at: datetime
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
from pydantic import BaseModel
|
||||
from typing_extensions import override
|
||||
|
||||
from .....typedefs import Number
|
||||
from ...base import Base, People, Trending
|
||||
@@ -14,6 +15,11 @@ class Data(BaseModel):
|
||||
|
||||
|
||||
class Info(Base):
|
||||
@property
|
||||
@override
|
||||
def path(self) -> str:
|
||||
return 'v1/top/info'
|
||||
|
||||
user: People
|
||||
today: Data
|
||||
historical: Data
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
from pydantic import BaseModel, Field
|
||||
from typing_extensions import override
|
||||
|
||||
from .....typedefs import Number
|
||||
from ...base import Base, People, Trending
|
||||
@@ -37,6 +38,11 @@ class Singleplayer(BaseModel):
|
||||
|
||||
|
||||
class Info(Base):
|
||||
@property
|
||||
@override
|
||||
def path(self) -> str:
|
||||
return 'v1/tos/info'
|
||||
|
||||
user: People
|
||||
multiplayer: Multiplayer
|
||||
singleplayer: Singleplayer
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
from datetime import datetime
|
||||
|
||||
from pydantic import BaseModel
|
||||
from typing_extensions import override
|
||||
|
||||
from .......games.tetrio.api.typedefs import ValidRank
|
||||
from ......typedefs import Number
|
||||
@@ -23,5 +24,10 @@ class ItemData(BaseModel):
|
||||
|
||||
|
||||
class Data(Base):
|
||||
@property
|
||||
@override
|
||||
def path(self) -> str:
|
||||
return 'v2/tetrio/rank'
|
||||
|
||||
items: dict[ValidRank, ItemData]
|
||||
updated_at: datetime
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
from datetime import datetime
|
||||
|
||||
from pydantic import BaseModel
|
||||
from typing_extensions import override
|
||||
|
||||
from .......games.tetrio.api.typedefs import ValidRank
|
||||
from ......typedefs import Number
|
||||
@@ -21,6 +22,11 @@ class SpecialData(BaseModel):
|
||||
|
||||
|
||||
class Data(Base):
|
||||
@property
|
||||
@override
|
||||
def path(self) -> str:
|
||||
return 'v2/tetrio/rank/detail'
|
||||
|
||||
name: ValidRank
|
||||
trending: Number
|
||||
require_tr: Number
|
||||
|
||||
@@ -1,3 +1,5 @@
|
||||
from typing_extensions import override
|
||||
|
||||
from .base import Record as BaseRecord
|
||||
from .base import Statistic as BaseStatistic
|
||||
|
||||
@@ -9,4 +11,9 @@ class Statistic(BaseStatistic):
|
||||
|
||||
|
||||
class Record(BaseRecord):
|
||||
@property
|
||||
@override
|
||||
def path(self) -> str:
|
||||
return 'v2/tetrio/record/blitz'
|
||||
|
||||
statistic: Statistic
|
||||
|
||||
@@ -1,7 +1,14 @@
|
||||
from typing_extensions import override
|
||||
|
||||
from .base import Record as BaseRecord
|
||||
from .base import Statistic
|
||||
|
||||
|
||||
class Record(BaseRecord):
|
||||
@property
|
||||
@override
|
||||
def path(self) -> str:
|
||||
return 'v2/tetrio/record/sprint'
|
||||
|
||||
statistic: Statistic
|
||||
time: str
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
from datetime import datetime
|
||||
|
||||
from pydantic import BaseModel
|
||||
from typing_extensions import override
|
||||
|
||||
from .....typedefs import Number
|
||||
from ...base import Base
|
||||
@@ -34,6 +35,11 @@ class Game(BaseModel):
|
||||
|
||||
|
||||
class Data(Base):
|
||||
@property
|
||||
@override
|
||||
def path(self) -> str:
|
||||
return 'v2/tetrio/tetra-league'
|
||||
|
||||
replay_id: str
|
||||
games: list[Game]
|
||||
play_at: datetime
|
||||
|
||||
@@ -2,6 +2,7 @@ from datetime import datetime
|
||||
from typing import Literal
|
||||
|
||||
from pydantic import BaseModel
|
||||
from typing_extensions import override
|
||||
|
||||
from .......games.tetrio.api.schemas.summaries.achievements import ArType, RankType
|
||||
from .......games.tetrio.api.schemas.summaries.achievements import Rank as AchievementRank
|
||||
@@ -132,6 +133,11 @@ class Zenith(BaseModel):
|
||||
|
||||
|
||||
class Info(Base):
|
||||
@property
|
||||
@override
|
||||
def path(self) -> str:
|
||||
return 'v2/tetrio/user/info'
|
||||
|
||||
user: User
|
||||
tetra_league: TetraLeague | None
|
||||
zenith: Zenith | None
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
from pydantic import BaseModel
|
||||
from typing_extensions import override
|
||||
|
||||
from .......games.tetrio.api.typedefs import Rank
|
||||
from ......typedefs import Number
|
||||
@@ -34,5 +35,10 @@ class Data(BaseModel):
|
||||
|
||||
|
||||
class List(Base):
|
||||
@property
|
||||
@override
|
||||
def path(self) -> str:
|
||||
return 'v2/tetrio/user/list'
|
||||
|
||||
show_index: bool
|
||||
data: list[Data]
|
||||
|
||||
@@ -17,7 +17,8 @@ from ..config.config import CACHE_PATH, DATA_PATH, config
|
||||
|
||||
driver = get_driver()
|
||||
|
||||
TEMPLATES_DIR = DATA_PATH / 'templates'
|
||||
TEMPLATES_DIR = config.tetris.dev.template_path or DATA_PATH / 'templates'
|
||||
|
||||
|
||||
alc = on_alconna(Alconna('更新模板', Option('--revision', Args['revision', str], alias={'-R'})), permission=SUPERUSER)
|
||||
|
||||
@@ -111,16 +112,6 @@ async def check_tag(tag: str) -> bool:
|
||||
).status_code != HTTPStatus.NOT_FOUND
|
||||
|
||||
|
||||
@driver.on_startup
|
||||
async def _():
|
||||
if (path := (TEMPLATES_DIR / 'hash.sha256')).is_file() and await check_hash(path):
|
||||
logger.success('模板验证成功')
|
||||
return
|
||||
if not await init_templates('latest'):
|
||||
msg = '模板初始化失败'
|
||||
raise RuntimeError(msg)
|
||||
|
||||
|
||||
@alc.handle()
|
||||
async def _(revision: str | None = None):
|
||||
if revision is not None and not await check_tag(revision):
|
||||
@@ -129,3 +120,17 @@ async def _(revision: str | None = None):
|
||||
if await init_templates(revision or 'latest'):
|
||||
await alc.finish('更新模板成功')
|
||||
await alc.finish('更新模板失败')
|
||||
|
||||
|
||||
if config.tetris.dev.enable_template_check:
|
||||
# !https://github.com/python/mypy/issues/19516
|
||||
# 只能放def后面了(
|
||||
|
||||
@driver.on_startup
|
||||
async def _():
|
||||
if (path := (TEMPLATES_DIR / 'hash.sha256')).is_file() and await check_hash(path):
|
||||
logger.success('模板验证成功')
|
||||
return
|
||||
if not await init_templates('latest'):
|
||||
msg = '模板初始化失败'
|
||||
raise RuntimeError(msg)
|
||||
|
||||
@@ -3,7 +3,7 @@ from typing import Literal, TypeAlias
|
||||
Number: TypeAlias = float | int
|
||||
GameType: TypeAlias = Literal['IO', 'TOP', 'TOS']
|
||||
BaseCommandType: TypeAlias = Literal['bind', 'unbind', 'query']
|
||||
TETRIOCommandType: TypeAlias = BaseCommandType | Literal['rank', 'config', 'list', 'record']
|
||||
TETRIOCommandType: TypeAlias = BaseCommandType | Literal['rank', 'config', 'list', 'record', 'verify']
|
||||
AllCommandType: TypeAlias = BaseCommandType | TETRIOCommandType
|
||||
Me: TypeAlias = Literal[
|
||||
'我',
|
||||
|
||||
@@ -2,7 +2,7 @@
|
||||
|
||||
[project]
|
||||
name = "nonebot-plugin-tetris-stats"
|
||||
version = "1.10.1"
|
||||
version = "1.11.0"
|
||||
description = "一款基于 NoneBot2 的用于查询 Tetris 相关游戏数据的插件"
|
||||
readme = "README.md"
|
||||
authors = [{ name = "shoucandanghehe", email = "wallfjjd@gmail.com" }]
|
||||
@@ -162,7 +162,7 @@ defineConstant = { PYDANTIC_V2 = true }
|
||||
typeCheckingMode = "standard"
|
||||
|
||||
[tool.bumpversion]
|
||||
current_version = "1.10.1"
|
||||
current_version = "1.11.0"
|
||||
tag = true
|
||||
sign_tags = true
|
||||
tag_name = "{new_version}"
|
||||
|
||||
Reference in New Issue
Block a user