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,19 +99,30 @@ 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',
|
||||
await UniMessage.image(
|
||||
raw=await make_bind_image(
|
||||
player=account,
|
||||
event_session=event_session,
|
||||
interface=interface,
|
||||
verify=None,
|
||||
)
|
||||
).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',
|
||||
type='unknown' if verify is None else 'success' if verify else 'unverified',
|
||||
user=People(
|
||||
avatar=str(
|
||||
URL(f'http://{netloc}/host/resource/tetrio/avatars/{user.ID}')
|
||||
URL(f'http://{get_self_netloc()}/host/resource/tetrio/avatars/{user.ID}')
|
||||
% {'revision': avatar_revision}
|
||||
)
|
||||
if (avatar_revision := (await account.avatar_revision)) is not None and avatar_revision != 0
|
||||
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(),
|
||||
),
|
||||
@@ -93,5 +141,3 @@ 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://{netloc}/host/{page_hash}.html')).finish()
|
||||
|
||||
@@ -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,17 +39,15 @@ 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',
|
||||
return await render_image(
|
||||
Info(
|
||||
user=User(
|
||||
avatar=str(
|
||||
URL(f'http://{netloc}/host/resource/tetrio/avatars/{user.ID}') % {'revision': avatar_revision}
|
||||
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(
|
||||
@@ -96,5 +93,3 @@ async def make_query_image_v1(player: Player) -> bytes:
|
||||
lang=get_lang(),
|
||||
),
|
||||
)
|
||||
) as page_hash:
|
||||
return await screenshot(f'http://{netloc}/host/{page_hash}.html')
|
||||
|
||||
@@ -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,9 +73,7 @@ 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',
|
||||
return await render_image(
|
||||
Info(
|
||||
user=User(
|
||||
id=user.ID,
|
||||
@@ -205,6 +202,4 @@ async def make_query_image_v2(player: Player) -> bytes:
|
||||
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,9 +92,7 @@ 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',
|
||||
return await render_image(
|
||||
DataV1(
|
||||
items={
|
||||
i[0].rank: ItemDataV1(
|
||||
@@ -86,14 +106,10 @@ async def make_image_v1(latest_data: TETRIOLeagueStats, compare_data: TETRIOLeag
|
||||
lang=get_lang(),
|
||||
),
|
||||
)
|
||||
) as page_hash:
|
||||
return await screenshot(f'http://{get_self_netloc()}/host/{page_hash}.html')
|
||||
|
||||
|
||||
async def make_image_v2(latest_data: TETRIOLeagueStats, compare_data: TETRIOLeagueStats) -> bytes:
|
||||
async with HostPage(
|
||||
await render(
|
||||
'v2/tetrio/rank',
|
||||
return await render_image(
|
||||
DataV2(
|
||||
items={
|
||||
i[0].rank: ItemDataV2(
|
||||
@@ -114,5 +130,3 @@ async def make_image_v2(latest_data: TETRIOLeagueStats, compare_data: TETRIOLeag
|
||||
lang=get_lang(),
|
||||
),
|
||||
)
|
||||
) as page_hash:
|
||||
return await screenshot(f'http://{get_self_netloc()}/host/{page_hash}.html')
|
||||
|
||||
@@ -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,9 +112,7 @@ 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',
|
||||
return await render_image(
|
||||
Data(
|
||||
name=latest_data.rank,
|
||||
trending=round(latest_data.tr_line - compare_data.tr_line, 2),
|
||||
@@ -126,5 +145,3 @@ async def make_image(rank: ValidRank, latest: TETRIOLeagueStats, compare: TETRIO
|
||||
lang=get_lang(),
|
||||
),
|
||||
)
|
||||
) as page_hash:
|
||||
return await screenshot(f'http://{get_self_netloc()}/host/{page_hash}.html')
|
||||
|
||||
@@ -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,9 +87,7 @@ 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',
|
||||
return await render_image(
|
||||
Record(
|
||||
type='best',
|
||||
user=User(
|
||||
@@ -149,5 +146,3 @@ async def make_blitz_image(player: Player) -> bytes:
|
||||
lang=get_lang(),
|
||||
),
|
||||
)
|
||||
) as page_hash:
|
||||
return await screenshot(f'http://{netloc}/host/{page_hash}.html')
|
||||
|
||||
@@ -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,9 +88,7 @@ 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',
|
||||
return await render_image(
|
||||
Record(
|
||||
type='best',
|
||||
user=User(
|
||||
@@ -149,5 +146,3 @@ async def make_sprint_image(player: Player) -> bytes:
|
||||
lang=get_lang(),
|
||||
),
|
||||
)
|
||||
) as page_hash:
|
||||
return await screenshot(f'http://{netloc}/host/{page_hash}.html')
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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,9 +73,7 @@ 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',
|
||||
return await render_image(
|
||||
Info(
|
||||
user=People(avatar=get_avatar(profile.user_name), name=profile.user_name),
|
||||
today=InfoData(
|
||||
@@ -99,8 +95,6 @@ async def make_query_image(profile: UserProfile) -> bytes:
|
||||
lang=get_lang(),
|
||||
),
|
||||
)
|
||||
) as page_hash:
|
||||
return await screenshot(f'http://{get_self_netloc()}/host/{page_hash}.html')
|
||||
|
||||
|
||||
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()
|
||||
|
||||
@@ -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,9 +262,7 @@ 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',
|
||||
return await render_image(
|
||||
Info(
|
||||
user=People(
|
||||
avatar=await get_avatar(event_user_info, 'Data URI', None)
|
||||
@@ -308,8 +304,6 @@ async def make_query_image(user_info: UserInfoSuccess, game_data: GameData, even
|
||||
lang=get_lang(),
|
||||
),
|
||||
)
|
||||
) as page_hash:
|
||||
return await screenshot(f'http://{get_self_netloc()}/host/{page_hash}.html')
|
||||
|
||||
|
||||
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