Compare commits

..

8 Commits

Author SHA1 Message Date
a57811b0d3 🔖 1.11.0
Some checks failed
Code Coverage / Test (macos-latest, 3.10) (push) Has been cancelled
Code Coverage / Test (macos-latest, 3.11) (push) Has been cancelled
Code Coverage / Test (macos-latest, 3.12) (push) Has been cancelled
Code Coverage / Test (ubuntu-latest, 3.10) (push) Has been cancelled
Code Coverage / Test (ubuntu-latest, 3.11) (push) Has been cancelled
Code Coverage / Test (ubuntu-latest, 3.12) (push) Has been cancelled
Code Coverage / Test (windows-latest, 3.10) (push) Has been cancelled
Code Coverage / Test (windows-latest, 3.11) (push) Has been cancelled
Code Coverage / Test (windows-latest, 3.12) (push) Has been cancelled
TypeCheck / TypeCheck (push) Has been cancelled
CodeQL / Analyze (python) (push) Has been cancelled
Code Coverage / check (push) Has been cancelled
2025-07-28 01:41:59 +08:00
呵呵です
7a5170936b 🧑‍💻添加更多实用开发配置项 (#555) 2025-07-28 01:31:55 +08:00
呵呵です
068c508f57 IO添加重新验证命令 (#554)
* 🐛 修复更新绑定时验证状态没有正确更新的bug

*  IO添加重新验证命令

* Apply suggestion from @Copilot

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

---------

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
2025-07-28 01:11:41 +08:00
renovate[bot]
0648ca021b ⬆️ Upgrade re-actors/alls-green digest to 2765efe (#551)
Co-authored-by: renovate[bot] <29139614+renovate[bot]@users.noreply.github.com>
2025-07-27 15:00:04 +00:00
呵呵です
65e7fed32b ♻️ 重构模板截图部分以解决导航导致的报错 (#553)
* ♻️ 把 path 放到数据模型里

* ♻️ 使用通用函数来生成模板图片

* 🎨 同步模板项目结构

* 🐛 修正导入路径
2025-07-27 22:58:41 +08:00
呵呵です
fdbb2f3f6e 通过io账号绑定的discord验证归属权 #64 (#552)
Some checks are pending
Code Coverage / Test (macos-latest, 3.10) (push) Waiting to run
Code Coverage / Test (macos-latest, 3.11) (push) Waiting to run
Code Coverage / Test (macos-latest, 3.12) (push) Waiting to run
Code Coverage / Test (ubuntu-latest, 3.10) (push) Waiting to run
Code Coverage / Test (ubuntu-latest, 3.11) (push) Waiting to run
Code Coverage / Test (ubuntu-latest, 3.12) (push) Waiting to run
Code Coverage / Test (windows-latest, 3.10) (push) Waiting to run
Code Coverage / Test (windows-latest, 3.11) (push) Waiting to run
Code Coverage / Test (windows-latest, 3.12) (push) Waiting to run
Code Coverage / check (push) Blocked by required conditions
TypeCheck / TypeCheck (push) Waiting to run
CodeQL / Analyze (python) (push) Waiting to run
* 🗃️ 添加 verify 字段到 Bind 模型,并在创建或更新绑定时支持该字段

*  通过io账号绑定的discord验证归属权
2025-07-27 05:01:33 +08:00
144c223fe9 🔖 1.10.2
Some checks failed
Code Coverage / Test (macos-latest, 3.10) (push) Has been cancelled
Code Coverage / Test (macos-latest, 3.11) (push) Has been cancelled
Code Coverage / Test (macos-latest, 3.12) (push) Has been cancelled
Code Coverage / Test (ubuntu-latest, 3.10) (push) Has been cancelled
Code Coverage / Test (ubuntu-latest, 3.11) (push) Has been cancelled
Code Coverage / Test (ubuntu-latest, 3.12) (push) Has been cancelled
Code Coverage / Test (windows-latest, 3.10) (push) Has been cancelled
Code Coverage / Test (windows-latest, 3.11) (push) Has been cancelled
Code Coverage / Test (windows-latest, 3.12) (push) Has been cancelled
TypeCheck / TypeCheck (push) Has been cancelled
CodeQL / Analyze (python) (push) Has been cancelled
Code Coverage / check (push) Has been cancelled
2025-07-19 22:40:47 +08:00
呵呵です
52a6d95434 🐛 移除 julianday 的使用,兼容更多数据库 (#550) 2025-07-19 22:40:04 +08:00
42 changed files with 893 additions and 652 deletions

View File

@@ -53,6 +53,6 @@ jobs:
runs-on: ubuntu-latest runs-on: ubuntu-latest
steps: steps:
- name: Decide whether the needed jobs succeeded or failed - name: Decide whether the needed jobs succeeded or failed
uses: re-actors/alls-green@223e4bb7a751b91f43eda76992bcfbf23b8b0302 uses: re-actors/alls-green@2765efec08f0fd63e83ad900f5fd75646be69ff6
with: with:
jobs: ${{ toJSON(needs) }} jobs: ${{ toJSON(needs) }}

View File

@@ -1,3 +1,5 @@
from pathlib import Path
from nonebot import get_driver, get_plugin_config from nonebot import get_driver, get_plugin_config
from nonebot_plugin_localstore import get_plugin_cache_dir, get_plugin_data_dir from nonebot_plugin_localstore import get_plugin_cache_dir, get_plugin_data_dir
from pydantic import BaseModel, Field from pydantic import BaseModel, Field
@@ -14,11 +16,17 @@ class Proxy(BaseModel):
top: str | None = None top: str | None = None
class Dev(BaseModel):
enabled: bool = False
template_path: Path | None = None
enable_template_check: bool = True
class ScopedConfig(BaseModel): class ScopedConfig(BaseModel):
request_timeout: float = 30.0 request_timeout: float = 30.0
screenshot_quality: float = 2 screenshot_quality: float = 2
proxy: Proxy = Field(default_factory=Proxy) proxy: Proxy = Field(default_factory=Proxy)
development: bool = False dev: Dev = Field(default_factory=Dev)
class Config(BaseModel): class Config(BaseModel):

View File

@@ -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 ###

View File

@@ -42,6 +42,8 @@ async def create_or_update_bind(
user: User, user: User,
game_platform: GameType, game_platform: GameType,
game_account: str, game_account: str,
*,
verify: bool = False,
) -> BindStatus: ) -> BindStatus:
bind = await query_bind_info( bind = await query_bind_info(
session=session, session=session,
@@ -53,11 +55,13 @@ async def create_or_update_bind(
user_id=user.id, user_id=user.id,
game_platform=game_platform, game_platform=game_platform,
game_account=game_account, game_account=game_account,
verify=verify,
) )
session.add(bind) session.add(bind)
status = BindStatus.SUCCESS status = BindStatus.SUCCESS
else: else:
bind.game_account = game_account bind.game_account = game_account
bind.verify = verify
status = BindStatus.UPDATE status = BindStatus.UPDATE
await session.commit() await session.commit()
return status return status

View File

@@ -71,6 +71,7 @@ class Bind(MappedAsDataclass, Model):
user_id: Mapped[int] = mapped_column(index=True) user_id: Mapped[int] = mapped_column(index=True)
game_platform: Mapped[GameType] = mapped_column(String(32)) game_platform: Mapped[GameType] = mapped_column(String(32))
game_account: Mapped[str] game_account: Mapped[str]
verify: Mapped[bool]
class TriggerHistoricalDataV2(MappedAsDataclass, Model): class TriggerHistoricalDataV2(MappedAsDataclass, Model):

View File

@@ -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) main_command.add(command)
@@ -36,4 +36,5 @@ __all__ = [
'rank', 'rank',
'record', 'record',
'unbind', 'unbind',
'verify',
] ]

View File

@@ -1,3 +1,4 @@
from asyncio import gather
from hashlib import md5 from hashlib import md5
from secrets import choice from secrets import choice
@@ -13,12 +14,12 @@ from yarl import URL
from ...config.config import global_config from ...config.config import global_config
from ...db import BindStatus, create_or_update_bind, trigger 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.image import get_avatar
from ...utils.lang import get_lang 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.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 . import alc, command, get_player
from .api import Player from .api import Player
from .constant import GAME_TYPE from .constant import GAME_TYPE
@@ -44,6 +45,42 @@ alc.shortcut(
humanized='io绑定', 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') @alc.assign('TETRIO.bind')
async def _(nb_user: User, account: Player, event_session: Uninfo, interface: QryItrface): 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, game_account=user.unique_identifier,
) )
if bind_status in (BindStatus.SUCCESS, BindStatus.UPDATE): if bind_status in (BindStatus.SUCCESS, BindStatus.UPDATE):
netloc = get_self_netloc() await UniMessage.image(
async with HostPage( raw=await make_bind_image(
await render( player=account,
'v1/binding', event_session=event_session,
Bind( interface=interface,
platform='TETR.IO', verify=None,
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(),
),
) )
) as page_hash: ).finish()
await UniMessage.image(raw=await screenshot(f'http://{netloc}/host/{page_hash}.html')).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(),
),
)

View File

@@ -4,12 +4,10 @@ from nonebot_plugin_uninfo import Uninfo
from nonebot_plugin_uninfo.orm import get_session_persist_id from nonebot_plugin_uninfo.orm import get_session_persist_id
from ...db import trigger from ...db import trigger
from ...utils.host import HostPage, get_self_netloc
from ...utils.lang import get_lang from ...utils.lang import get_lang
from ...utils.metrics import get_metrics 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.render.schemas.v2.tetrio.user.list import Data, List, TetraLeague, User
from ...utils.screenshot import screenshot
from .. import alc from .. import alc
from . import command from . import command
from .api.leaderboards import by from .api.leaderboards import by
@@ -59,9 +57,8 @@ async def _(
country=country, country=country,
) )
league = await by('league', parameter) league = await by('league', parameter)
async with HostPage( await UniMessage.image(
await render( raw=await render_image(
'v2/tetrio/user/list',
List( List(
show_index=True, show_index=True,
data=[ data=[
@@ -92,5 +89,4 @@ async def _(
lang=get_lang(), lang=get_lang(),
), ),
) )
) as page_hash: ).finish()
await UniMessage.image(raw=await screenshot(f'http://{get_self_netloc()}/host/{page_hash}.html')).finish()

View File

@@ -6,14 +6,13 @@ from yarl import URL
from ....utils.chart import get_split, get_value_bounds, handle_history_data from ....utils.chart import get_split, get_value_bounds, handle_history_data
from ....utils.exception import FallbackError 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.lang import get_lang
from ....utils.metrics import get_metrics 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.base import Avatar, Trending
from ....utils.render.schemas.v1.base import History from ....utils.render.schemas.v1.base import History
from ....utils.render.schemas.v1.tetrio.user.info import Info, Multiplayer, Singleplayer, User from ....utils.render.schemas.v1.tetrio.info import Info, Multiplayer, Singleplayer, User
from ....utils.screenshot import screenshot
from ..api import Player from ..api import Player
from ..api.schemas.summaries.league import RatedData from ..api.schemas.summaries.league import RatedData
from ..constant import TR_MAX, TR_MIN from ..constant import TR_MAX, TR_MIN
@@ -40,61 +39,57 @@ async def make_query_image_v1(player: Player) -> bytes:
else: else:
sprint_value = 'N/A' sprint_value = 'N/A'
blitz_value = f'{blitz.data.record.results.stats.score:,}' if blitz.data.record is not None else '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 dsps: float
dspp: float dspp: float
# make mypy happy # make mypy happy
async with HostPage( return await render_image(
page=await render( Info(
'v1/tetrio/info', user=User(
Info( avatar=str(
user=User( URL(f'http://{get_self_netloc()}/host/resource/tetrio/avatars/{user.ID}')
avatar=str( % {'revision': avatar_revision}
URL(f'http://{netloc}/host/resource/tetrio/avatars/{user.ID}') % {'revision': avatar_revision} )
) if avatar_revision is not None and avatar_revision != 0
if avatar_revision is not None and avatar_revision != 0 else Avatar(
else Avatar( type='identicon',
type='identicon', hash=md5(user.ID.encode()).hexdigest(), # noqa: S324
hash=md5(user.ID.encode()).hexdigest(), # noqa: S324
),
name=user.name.upper(),
bio=user_info.data.bio,
), ),
multiplayer=Multiplayer( name=user.name.upper(),
glicko=f'{round(league_data.glicko, 2):,}', bio=user_info.data.bio,
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(),
), ),
) multiplayer=Multiplayer(
) as page_hash: glicko=f'{round(league_data.glicko, 2):,}',
return await screenshot(f'http://{netloc}/host/{page_hash}.html') 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(),
),
)

View File

@@ -5,10 +5,10 @@ from hashlib import md5
from yarl import URL from yarl import URL
from ....utils.exception import FallbackError 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.lang import get_lang
from ....utils.metrics import get_metrics 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.base import Avatar
from ....utils.render.schemas.v2.tetrio.user.info import ( from ....utils.render.schemas.v2.tetrio.user.info import (
Achievement, Achievement,
@@ -25,7 +25,6 @@ from ....utils.render.schemas.v2.tetrio.user.info import (
Zen, Zen,
Zenith, Zenith,
) )
from ....utils.screenshot import screenshot
from ..api import Player from ..api import Player
from ..api.schemas.summaries.league import InvalidData, NeverPlayedData, NeverRatedData from ..api.schemas.summaries.league import InvalidData, NeverPlayedData, NeverRatedData
from .tools import flow_to_history, handling_special_value from .tools import flow_to_history, handling_special_value
@@ -74,137 +73,133 @@ async def make_query_image_v2(player: Player) -> bytes:
except FallbackError: except FallbackError:
history = None history = None
netloc = get_self_netloc() netloc = get_self_netloc()
async with HostPage( return await render_image(
await render( Info(
'v2/tetrio/user/info', user=User(
Info( id=user.ID,
user=User( name=user.name.upper(),
id=user.ID, country=user_info.data.country,
name=user.name.upper(), role=user_info.data.role,
country=user_info.data.country, botmaster=user_info.data.botmaster,
role=user_info.data.role, avatar=str(
botmaster=user_info.data.botmaster, URL(f'http://{netloc}/host/resource/tetrio/avatars/{user.ID}') % {'revision': avatar_revision}
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,
) )
if not isinstance(league.data, NeverPlayedData | InvalidData) if avatar_revision is not None and avatar_revision != 0
else None, else Avatar(
zenith=Zenith( type='identicon',
week=Week( hash=md5(user.ID.encode()).hexdigest(), # noqa: S324
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( banner=str(
week=Week( URL(f'http://{netloc}/host/resource/tetrio/banners/{user.ID}') % {'revision': banner_revision}
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 if banner_revision is not None and banner_revision != 0
else None, else None,
blitz=Blitz( bio=user_info.data.bio,
score=blitz.data.record.results.stats.score, friend_count=user_info.data.friend_count,
global_rank=blitz.data.rank, supporter_tier=user_info.data.supporter_tier,
country_rank=blitz.data.rank_local, bad_standing=user_info.data.badstanding or False,
play_at=blitz.data.record.ts, badges=[
) Badge(
if blitz.data.record is not None id=i.id,
else None, description=i.label,
zen=Zen(level=zen.data.level, score=zen.data.score), group=i.group,
lang=get_lang(), 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')

View File

@@ -137,7 +137,7 @@ async def get_tetra_league_data() -> None:
await session.commit() await session.commit()
if not config.tetris.development: if not config.tetris.dev.enabled:
@driver.on_startup @driver.on_startup
async def _() -> None: async def _() -> None:

View File

@@ -5,20 +5,18 @@ from nonebot_plugin_alconna import Option, Subcommand, UniMessage
from nonebot_plugin_orm import get_session from nonebot_plugin_orm import get_session
from nonebot_plugin_uninfo import Uninfo from nonebot_plugin_uninfo import Uninfo
from nonebot_plugin_uninfo.orm import get_session_persist_id 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 sqlalchemy.orm import selectinload
from ....db import trigger from ....db import trigger
from ....utils.host import HostPage, get_self_netloc
from ....utils.lang import get_lang from ....utils.lang import get_lang
from ....utils.metrics import get_metrics 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 Data as DataV1
from ....utils.render.schemas.v1.tetrio.rank import ItemData as ItemDataV1 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 AverageData as AverageDataV2
from ....utils.render.schemas.v2.tetrio.rank import Data as DataV2 from ....utils.render.schemas.v2.tetrio.rank import Data as DataV2
from ....utils.render.schemas.v2.tetrio.rank import ItemData as ItemDataV2 from ....utils.render.schemas.v2.tetrio.rank import ItemData as ItemDataV2
from ....utils.screenshot import screenshot
from .. import alc from .. import alc
from ..constant import GAME_TYPE from ..constant import GAME_TYPE
from ..models import TETRIOLeagueStats 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 []), command_args=['--all'] + ([f'--template {template}'] if template is not None else []),
): ):
async with get_session() as session: async with get_session() as session:
# 获取最新记录
latest_data = ( latest_data = (
await session.scalars( await session.scalars(
select(TETRIOLeagueStats) select(TETRIOLeagueStats)
@@ -49,19 +48,42 @@ async def _(event_session: Uninfo, template: Template | None = None):
.options(selectinload(TETRIOLeagueStats.fields)) .options(selectinload(TETRIOLeagueStats.fields))
) )
).one() ).one()
compare_data = (
await session.scalars( # 计算目标时间点 (24小时前)
target_time = latest_data.update_time - timedelta(hours=24)
# 查询目标时间点之前的最近记录
before = (
await session.scalar(
select(TETRIOLeagueStats) select(TETRIOLeagueStats)
.order_by( .where(TETRIOLeagueStats.update_time <= target_time)
func.abs( .order_by(TETRIOLeagueStats.update_time.desc())
func.julianday(TETRIOLeagueStats.update_time)
- func.julianday(latest_data.update_time - timedelta(hours=24))
)
)
.limit(1) .limit(1)
.options(selectinload(TETRIOLeagueStats.fields)) .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: match template:
case 'v1' | None: case 'v1' | None:
await UniMessage.image(raw=await make_image_v1(latest_data, compare_data)).finish() 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 def make_image_v1(latest_data: TETRIOLeagueStats, compare_data: TETRIOLeagueStats) -> bytes:
async with HostPage( return await render_image(
await render( DataV1(
'v1/tetrio/rank', items={
DataV1( i[0].rank: ItemDataV1(
items={ trending=round(i[0].tr_line - i[1].tr_line, 2),
i[0].rank: ItemDataV1( require_tr=round(i[0].tr_line, 2),
trending=round(i[0].tr_line - i[1].tr_line, 2), players=i[0].player_count,
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)
) },
for i in zip(latest_data.fields, compare_data.fields, strict=True) updated_at=latest_data.update_time,
}, lang=get_lang(),
updated_at=latest_data.update_time, ),
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 def make_image_v2(latest_data: TETRIOLeagueStats, compare_data: TETRIOLeagueStats) -> bytes:
async with HostPage( return await render_image(
await render( DataV2(
'v2/tetrio/rank', items={
DataV2( i[0].rank: ItemDataV2(
items={ require_tr=round(i[0].tr_line, 2),
i[0].rank: ItemDataV2( trending=round(i[0].tr_line - i[1].tr_line, 2),
require_tr=round(i[0].tr_line, 2), average_data=AverageDataV2(
trending=round(i[0].tr_line - i[1].tr_line, 2), pps=(metrics := get_metrics(pps=i[0].avg_pps, apm=i[0].avg_apm, vs=i[0].avg_vs)).pps,
average_data=AverageDataV2( apm=metrics.apm,
pps=(metrics := get_metrics(pps=i[0].avg_pps, apm=i[0].avg_apm, vs=i[0].avg_vs)).pps, apl=metrics.apl,
apm=metrics.apm, vs=metrics.vs,
apl=metrics.apl, adpl=metrics.adpl,
vs=metrics.vs, ),
adpl=metrics.adpl, players=i[0].player_count,
), )
players=i[0].player_count, for i in zip(latest_data.fields, compare_data.fields, strict=True)
) },
for i in zip(latest_data.fields, compare_data.fields, strict=True) updated_at=latest_data.update_time,
}, lang=get_lang(),
updated_at=latest_data.update_time, ),
lang=get_lang(), )
),
)
) as page_hash:
return await screenshot(f'http://{get_self_netloc()}/host/{page_hash}.html')

View File

@@ -7,16 +7,14 @@ from nonebot_plugin_alconna import Option, UniMessage
from nonebot_plugin_orm import get_session from nonebot_plugin_orm import get_session
from nonebot_plugin_uninfo import Uninfo from nonebot_plugin_uninfo import Uninfo
from nonebot_plugin_uninfo.orm import get_session_persist_id 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 sqlalchemy.orm import selectinload
from ....db import trigger from ....db import trigger
from ....utils.host import HostPage, get_self_netloc
from ....utils.lang import get_lang from ....utils.lang import get_lang
from ....utils.metrics import get_metrics 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.render.schemas.v2.tetrio.rank.detail import Data, SpecialData
from ....utils.screenshot import screenshot
from .. import alc from .. import alc
from ..api.typedefs import ValidRank from ..api.typedefs import ValidRank
from ..constant import GAME_TYPE from ..constant import GAME_TYPE
@@ -39,6 +37,7 @@ async def _(rank: ValidRank, event_session: Uninfo):
command_args=[f'--detail {rank}'], command_args=[f'--detail {rank}'],
): ):
async with get_session() as session: async with get_session() as session:
# 获取最新记录
latest_data = ( latest_data = (
await session.scalars( await session.scalars(
select(TETRIOLeagueStats) select(TETRIOLeagueStats)
@@ -47,19 +46,41 @@ async def _(rank: ValidRank, event_session: Uninfo):
.options(selectinload(TETRIOLeagueStats.fields)) .options(selectinload(TETRIOLeagueStats.fields))
) )
).one() ).one()
compare_data = (
await session.scalars( # 计算目标时间点 (24小时前)
target_time = latest_data.update_time - timedelta(hours=24)
# 查询目标时间点之前的最近记录
before = (
await session.scalar(
select(TETRIOLeagueStats) select(TETRIOLeagueStats)
.order_by( .where(TETRIOLeagueStats.update_time <= target_time)
func.abs( .order_by(TETRIOLeagueStats.update_time.desc())
func.julianday(TETRIOLeagueStats.update_time)
- func.julianday(latest_data.update_time - timedelta(hours=24))
)
)
.limit(1) .limit(1)
.options(selectinload(TETRIOLeagueStats.fields)) .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( await UniMessage.image(
raw=await make_image( raw=await make_image(
rank, rank,
@@ -91,40 +112,36 @@ async def make_image(rank: ValidRank, latest: TETRIOLeagueStats, compare: TETRIO
max_vs = get_metrics( 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 pps=latest_data.high_vs.league.pps, apm=latest_data.high_vs.league.apm, vs=latest_data.high_vs.league.vs
) )
async with HostPage( return await render_image(
await render( Data(
'v2/tetrio/rank/detail', name=latest_data.rank,
Data( trending=round(latest_data.tr_line - compare_data.tr_line, 2),
name=latest_data.rank, require_tr=round(latest_data.tr_line, 2),
trending=round(latest_data.tr_line - compare_data.tr_line, 2), players=latest_data.player_count,
require_tr=round(latest_data.tr_line, 2), minimum_data=SpecialData(
players=latest_data.player_count, apm=low_apm.apm,
minimum_data=SpecialData( pps=low_pps.pps,
apm=low_apm.apm, lpm=low_pps.lpm,
pps=low_pps.pps, vs=low_vs.vs,
lpm=low_pps.lpm, adpm=low_vs.adpm,
vs=low_vs.vs, apm_holder=latest_data.low_apm.username.upper(),
adpm=low_vs.adpm, pps_holder=latest_data.low_pps.username.upper(),
apm_holder=latest_data.low_apm.username.upper(), vs_holder=latest_data.low_vs.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(),
), ),
) average_data=SpecialData(
) as page_hash: apm=avg.apm, pps=avg.pps, lpm=avg.lpm, vs=avg.vs, adpm=avg.adpm, apl=avg.apl, adpl=avg.adpl
return await screenshot(f'http://{get_self_netloc()}/host/{page_hash}.html') ),
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(),
),
)

View File

@@ -15,14 +15,13 @@ from yarl import URL
from ....db import query_bind_info, trigger from ....db import query_bind_info, trigger
from ....i18n import Lang from ....i18n import Lang
from ....utils.exception import RecordNotFoundError 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.lang import get_lang
from ....utils.metrics import get_metrics 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.base import Avatar
from ....utils.render.schemas.v2.tetrio.record.base import Finesse, Max, Mini, Tspins, User 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.render.schemas.v2.tetrio.record.blitz import Record, Statistic
from ....utils.screenshot import screenshot
from ....utils.typedefs import Me from ....utils.typedefs import Me
from .. import alc from .. import alc
from ..api.player import Player from ..api.player import Player
@@ -88,66 +87,62 @@ async def make_blitz_image(player: Player) -> bytes:
duration = timedelta(milliseconds=stats.finaltime).total_seconds() duration = timedelta(milliseconds=stats.finaltime).total_seconds()
metrics = get_metrics(pps=stats.piecesplaced / duration) metrics = get_metrics(pps=stats.piecesplaced / duration)
netloc = get_self_netloc() netloc = get_self_netloc()
async with HostPage( return await render_image(
page=await render( Record(
'v2/tetrio/record/blitz', type='best',
Record( user=User(
type='best', id=user.ID,
user=User( name=user.name.upper(),
id=user.ID, avatar=str(
name=user.name.upper(), URL(f'http://{netloc}/host/resource/tetrio/avatars/{user.ID}') % {'revision': avatar_revision}
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(
if (avatar_revision := (await player.avatar_revision)) is not None and avatar_revision != 0 type='identicon',
else Avatar( hash=md5(user.ID.encode()).hexdigest(), # noqa: S324
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(),
), ),
) replay_id=blitz.data.record.replayid,
) as page_hash: rank=blitz.data.rank,
return await screenshot(f'http://{netloc}/host/{page_hash}.html') 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(),
),
)

View File

@@ -15,14 +15,13 @@ from yarl import URL
from ....db import query_bind_info, trigger from ....db import query_bind_info, trigger
from ....i18n import Lang from ....i18n import Lang
from ....utils.exception import RecordNotFoundError 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.lang import get_lang
from ....utils.metrics import get_metrics 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.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.base import Finesse, Max, Mini, Statistic, Tspins, User
from ....utils.render.schemas.v2.tetrio.record.sprint import Record from ....utils.render.schemas.v2.tetrio.record.sprint import Record
from ....utils.screenshot import screenshot
from ....utils.typedefs import Me from ....utils.typedefs import Me
from .. import alc from .. import alc
from ..api.player import Player 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 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) metrics = get_metrics(pps=stats.piecesplaced / duration)
netloc = get_self_netloc() netloc = get_self_netloc()
async with HostPage( return await render_image(
page=await render( Record(
'v2/tetrio/record/sprint', type='best',
Record( user=User(
type='best', id=user.ID,
user=User( name=user.name.upper(),
id=user.ID, avatar=str(
name=user.name.upper(), URL(f'http://{netloc}/host/resource/tetrio/avatars/{user.ID}') % {'revision': avatar_revision}
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(
if (avatar_revision := (await player.avatar_revision)) is not None and avatar_revision != 0 type='identicon',
else Avatar( hash=md5(user.ID.encode()).hexdigest(), # noqa: S324
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(),
), ),
) time=sprint_value,
) as page_hash: replay_id=sprint.data.record.replayid,
return await screenshot(f'http://{netloc}/host/{page_hash}.html') 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(),
),
)

View File

@@ -13,12 +13,12 @@ from yarl import URL
from ...config.config import global_config from ...config.config import global_config
from ...db import query_bind_info, remove_bind, trigger 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.image import get_avatar
from ...utils.lang import get_lang 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.render.schemas.base import Avatar, People
from ...utils.screenshot import screenshot from ...utils.render.schemas.bind import Bind
from . import alc, command from . import alc, command
from .api import Player from .api import Player
from .constant import GAME_TYPE 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) player = Player(user_id=bind.game_account, trust=True)
user = await player.user user = await player.user
netloc = get_self_netloc() netloc = get_self_netloc()
async with HostPage( await UniMessage.image(
await render( raw=await render_image(
'v1/binding',
Bind( Bind(
platform='TETR.IO', platform='TETR.IO',
type='unlink', type='unlink',
@@ -81,6 +80,5 @@ async def _(nb_user: User, event_session: Uninfo, interface: QryItrface):
lang=get_lang(), lang=get_lang(),
), ),
) )
) as page_hash: ).send()
await UniMessage.image(raw=await screenshot(f'http://{netloc}/host/{page_hash}.html')).send()
await remove_bind(session=session, user=nb_user, game_platform=GAME_TYPE) await remove_bind(session=session, user=nb_user, game_platform=GAME_TYPE)

View 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()

View File

@@ -9,12 +9,11 @@ from nonebot_plugin_user import User
from ...config.config import global_config from ...config.config import global_config
from ...db import BindStatus, create_or_update_bind, trigger 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.image import get_avatar
from ...utils.lang import get_lang 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.render.schemas.base import People
from ...utils.screenshot import screenshot from ...utils.render.schemas.bind import Bind
from . import alc from . import alc
from .api import Player from .api import Player
from .constant import GAME_TYPE 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, game_account=user.unique_identifier,
) )
if bind_status in (BindStatus.SUCCESS, BindStatus.UPDATE): if bind_status in (BindStatus.SUCCESS, BindStatus.UPDATE):
async with HostPage( await UniMessage.image(
await render( raw=await render_image(
'v1/binding',
Bind( Bind(
platform=GAME_TYPE, platform=GAME_TYPE,
type='unknown', type='unknown',
@@ -66,7 +64,4 @@ async def _(nb_user: User, account: Player, event_session: Uninfo, interface: Qr
lang=get_lang(), lang=get_lang(),
), ),
) )
) as page_hash: ).finish()
await UniMessage.image(
raw=await screenshot(f'http://{get_self_netloc()}/host/{page_hash}.html')
).finish()

View File

@@ -10,15 +10,13 @@ from nonebot_plugin_user import get_user
from ...db import query_bind_info, trigger from ...db import query_bind_info, trigger
from ...i18n import Lang from ...i18n import Lang
from ...utils.exception import FallbackError from ...utils.exception import FallbackError
from ...utils.host import HostPage, get_self_netloc
from ...utils.lang import get_lang from ...utils.lang import get_lang
from ...utils.metrics import TetrisMetricsBasicWithLPM, get_metrics 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.avatar import get_avatar
from ...utils.render.schemas.base import People, Trending 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 Data as InfoData
from ...utils.render.schemas.v1.top.info import Info from ...utils.render.schemas.v1.top.info import Info
from ...utils.screenshot import screenshot
from ...utils.typedefs import Me from ...utils.typedefs import Me
from . import alc from . import alc
from .api import Player from .api import Player
@@ -75,32 +73,28 @@ async def make_query_image(profile: UserProfile) -> bytes:
raise FallbackError raise FallbackError
today = get_metrics(lpm=profile.today.lpm, apm=profile.today.apm) today = get_metrics(lpm=profile.today.lpm, apm=profile.today.apm)
history = get_avg_metrics(profile.total) history = get_avg_metrics(profile.total)
async with HostPage( return await render_image(
await render( Info(
'v1/top/info', user=People(avatar=get_avatar(profile.user_name), name=profile.user_name),
Info( today=InfoData(
user=People(avatar=get_avatar(profile.user_name), name=profile.user_name), pps=today.pps,
today=InfoData( lpm=today.lpm,
pps=today.pps, lpm_trending=Trending.KEEP,
lpm=today.lpm, apm=today.apm,
lpm_trending=Trending.KEEP, apl=today.apl,
apm=today.apm, apm_trending=Trending.KEEP,
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(),
), ),
) historical=InfoData(
) as page_hash: pps=history.pps,
return await screenshot(f'http://{get_self_netloc()}/host/{page_hash}.html') 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: def make_query_text(profile: UserProfile) -> UniMessage:

View File

@@ -10,12 +10,11 @@ from nonebot_plugin_waiter import suggest # type: ignore[import-untyped]
from ...config.config import global_config from ...config.config import global_config
from ...db import query_bind_info, remove_bind, trigger 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.image import get_avatar
from ...utils.lang import get_lang 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.render.schemas.base import People
from ...utils.screenshot import screenshot from ...utils.render.schemas.bind import Bind
from . import alc from . import alc
from .api import Player from .api import Player
from .constant import GAME_TYPE from .constant import GAME_TYPE
@@ -43,10 +42,8 @@ async def _(
return return
player = Player(user_name=bind.game_account, trust=True) player = Player(user_name=bind.game_account, trust=True)
user = await player.user user = await player.user
netloc = get_self_netloc() await UniMessage.image(
async with HostPage( raw=await render_image(
await render(
'v1/binding',
Bind( Bind(
platform='TOP', platform='TOP',
type='unlink', type='unlink',
@@ -73,6 +70,5 @@ async def _(
lang=get_lang(), lang=get_lang(),
), ),
) )
) as page_hash: ).send()
await UniMessage.image(raw=await screenshot(f'http://{netloc}/host/{page_hash}.html')).send()
await remove_bind(session=session, user=nb_user, game_platform=GAME_TYPE) await remove_bind(session=session, user=nb_user, game_platform=GAME_TYPE)

View File

@@ -9,12 +9,11 @@ from nonebot_plugin_user import User
from ...config.config import global_config from ...config.config import global_config
from ...db import BindStatus, create_or_update_bind, trigger 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.image import get_avatar
from ...utils.lang import get_lang 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.render.schemas.base import People
from ...utils.screenshot import screenshot from ...utils.render.schemas.bind import Bind
from . import alc from . import alc
from .api import Player from .api import Player
from .constant import GAME_TYPE from .constant import GAME_TYPE
@@ -43,9 +42,8 @@ async def _(
) )
user_info = await account.get_info() user_info = await account.get_info()
if bind_status in (BindStatus.SUCCESS, BindStatus.UPDATE): if bind_status in (BindStatus.SUCCESS, BindStatus.UPDATE):
async with HostPage( await UniMessage.image(
await render( raw=await render_image(
'v1/binding',
Bind( Bind(
platform=GAME_TYPE, platform=GAME_TYPE,
type='unknown', type='unknown',
@@ -72,7 +70,4 @@ async def _(
lang=get_lang(), lang=get_lang(),
), ),
) )
) as page_hash: ).finish()
await UniMessage.image(
raw=await screenshot(f'http://{get_self_netloc()}/host/{page_hash}.html')
).finish()

View File

@@ -18,16 +18,14 @@ from ...db import query_bind_info, trigger
from ...i18n import Lang from ...i18n import Lang
from ...utils.chart import get_split, get_value_bounds, handle_history_data from ...utils.chart import get_split, get_value_bounds, handle_history_data
from ...utils.exception import RequestError from ...utils.exception import RequestError
from ...utils.host import HostPage, get_self_netloc
from ...utils.image import get_avatar from ...utils.image import get_avatar
from ...utils.lang import get_lang from ...utils.lang import get_lang
from ...utils.metrics import TetrisMetricsProWithLPMADPM, get_metrics 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.avatar import get_avatar as get_random_avatar
from ...utils.render.schemas.base import HistoryData, People, Trending from ...utils.render.schemas.base import HistoryData, People, Trending
from ...utils.render.schemas.v1.base import History from ...utils.render.schemas.v1.base import History
from ...utils.render.schemas.v1.tos.info import Info, Multiplayer, Singleplayer 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.time_it import time_it
from ...utils.typedefs import Me, Number from ...utils.typedefs import Me, Number
from . import alc 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)) data = handle_history_data(await get_historical_data(user_info.data.teaid))
values = get_value_bounds([i.score for i in data]) values = get_value_bounds([i.score for i in data])
async with HostPage( return await render_image(
await render( Info(
'v1/tos/info', user=People(
Info( avatar=await get_avatar(event_user_info, 'Data URI', None)
user=People( if event_user_info is not None
avatar=await get_avatar(event_user_info, 'Data URI', None) else get_random_avatar(user_info.data.teaid),
if event_user_info is not None name=user_info.data.name,
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(),
), ),
) multiplayer=Multiplayer(
) as page_hash: history=History(
return await screenshot(f'http://{get_self_netloc()}/host/{page_hash}.html') 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: def make_query_text(user_info: UserInfoSuccess, game_data: GameData | None) -> UniMessage:

View File

@@ -10,12 +10,11 @@ from nonebot_plugin_waiter import suggest # type: ignore[import-untyped]
from ...config.config import global_config from ...config.config import global_config
from ...db import query_bind_info, remove_bind, trigger 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.image import get_avatar
from ...utils.lang import get_lang 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.render.schemas.base import People
from ...utils.screenshot import screenshot from ...utils.render.schemas.bind import Bind
from . import alc from . import alc
from .api import Player from .api import Player
from .constant import GAME_TYPE from .constant import GAME_TYPE
@@ -43,10 +42,8 @@ async def _(
return return
player = Player(user_name=bind.game_account, trust=True) player = Player(user_name=bind.game_account, trust=True)
user = await player.user user = await player.user
netloc = get_self_netloc() await UniMessage.image(
async with HostPage( raw=await render_image(
await render(
'v1/binding',
Bind( Bind(
platform='TOS', platform='TOS',
type='unlink', type='unlink',
@@ -69,6 +66,5 @@ async def _(
lang=get_lang(), lang=get_lang(),
), ),
) )
) as page_hash: ).send()
await UniMessage.image(raw=await screenshot(f'http://{netloc}/host/{page_hash}.html')).send()
await remove_bind(session=session, user=nb_user, game_platform=GAME_TYPE) await remove_bind(session=session, user=nb_user, game_platform=GAME_TYPE)

View File

@@ -78,7 +78,7 @@ class BrowserManager:
"""启动浏览器实例""" """启动浏览器实例"""
playwright = await async_playwright().start() playwright = await async_playwright().start()
cls._browser = await playwright.firefox.launch( cls._browser = await playwright.firefox.launch(
headless=not config.tetris.development, headless=not config.tetris.dev.enabled,
firefox_user_prefs={ firefox_user_prefs={
'network.http.max-persistent-connections-per-server': 64, 'network.http.max-persistent-connections-per-server': 64,
}, },

View File

@@ -45,7 +45,7 @@ class HostPage:
async def __aenter__(self) -> str: async def __aenter__(self) -> str:
return self.page_hash 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 async def __aexit__(self, exc_type, exc, tb) -> None: # noqa: ANN001
self.pages.pop(self.page_hash, None) self.pages.pop(self.page_hash, None)

View File

@@ -1,22 +1,10 @@
from typing import Literal, overload
from jinja2 import Environment, FileSystemLoader from jinja2 import Environment, FileSystemLoader
from nonebot.compat import PYDANTIC_V2 from nonebot.compat import PYDANTIC_V2
from ..host import HostPage, get_self_netloc
from ..screenshot import screenshot
from ..templates import TEMPLATES_DIR from ..templates import TEMPLATES_DIR
from .schemas.base import Base 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( env = Environment(
loader=FileSystemLoader(TEMPLATES_DIR), 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( async def render(
render_type: str,
data: Base, data: Base,
) -> str: ) -> str:
if PYDANTIC_V2: if PYDANTIC_V2:
return await env.get_template('index.html').render_async( return await env.get_template('index.html').render_async(data=data.model_dump_json(by_alias=True))
path=render_type, data=data.model_dump_json(by_alias=True) return await env.get_template('index.html').render_async(data=data.json(by_alias=True))
)
return await env.get_template('index.html').render_async(path=render_type, 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'] __all__ = ['render']

View File

@@ -1,3 +1,4 @@
from abc import abstractmethod
from datetime import datetime from datetime import datetime
from typing import Literal from typing import Literal
@@ -8,6 +9,11 @@ from ...typedefs import Lang, Number
class Base(BaseModel): class Base(BaseModel):
@property
@abstractmethod
def path(self) -> str:
raise NotImplementedError
lang: Lang lang: Lang

View File

@@ -1,9 +1,16 @@
from typing import Literal from typing import Literal
from typing_extensions import override
from .base import Base, People from .base import Base, People
class Bind(Base): class Bind(Base):
@property
@override
def path(self) -> str:
return 'v1/binding'
platform: Literal['TETR.IO', 'TOP', 'TOS'] platform: Literal['TETR.IO', 'TOP', 'TOS']
type: Literal['success', 'unknown', 'unlink', 'unverified', 'error'] type: Literal['success', 'unknown', 'unlink', 'unverified', 'error']
user: People user: People

View File

@@ -1,9 +1,10 @@
from pydantic import BaseModel from pydantic import BaseModel
from typing_extensions import override
from .......games.tetrio.api.typedefs import Rank from ......games.tetrio.api.typedefs import Rank
from ......typedefs import Number from .....typedefs import Number
from ....base import Base, People, Trending from ...base import Base, People, Trending
from ...base import History from ..base import History
class User(People): class User(People):
@@ -45,6 +46,11 @@ class Singleplayer(BaseModel):
class Info(Base): class Info(Base):
@property
@override
def path(self) -> str:
return 'v1/tetrio/info'
user: User user: User
multiplayer: Multiplayer multiplayer: Multiplayer
singleplayer: Singleplayer singleplayer: Singleplayer

View File

@@ -1,6 +1,7 @@
from datetime import datetime from datetime import datetime
from pydantic import BaseModel from pydantic import BaseModel
from typing_extensions import override
from ......games.tetrio.api.typedefs import ValidRank from ......games.tetrio.api.typedefs import ValidRank
from ...base import Base from ...base import Base
@@ -13,5 +14,10 @@ class ItemData(BaseModel):
class Data(Base): class Data(Base):
@property
@override
def path(self) -> str:
return 'v1/tetrio/rank'
items: dict[ValidRank, ItemData] items: dict[ValidRank, ItemData]
updated_at: datetime updated_at: datetime

View File

@@ -1,4 +1,5 @@
from pydantic import BaseModel from pydantic import BaseModel
from typing_extensions import override
from .....typedefs import Number from .....typedefs import Number
from ...base import Base, People, Trending from ...base import Base, People, Trending
@@ -14,6 +15,11 @@ class Data(BaseModel):
class Info(Base): class Info(Base):
@property
@override
def path(self) -> str:
return 'v1/top/info'
user: People user: People
today: Data today: Data
historical: Data historical: Data

View File

@@ -1,4 +1,5 @@
from pydantic import BaseModel, Field from pydantic import BaseModel, Field
from typing_extensions import override
from .....typedefs import Number from .....typedefs import Number
from ...base import Base, People, Trending from ...base import Base, People, Trending
@@ -37,6 +38,11 @@ class Singleplayer(BaseModel):
class Info(Base): class Info(Base):
@property
@override
def path(self) -> str:
return 'v1/tos/info'
user: People user: People
multiplayer: Multiplayer multiplayer: Multiplayer
singleplayer: Singleplayer singleplayer: Singleplayer

View File

@@ -1,6 +1,7 @@
from datetime import datetime from datetime import datetime
from pydantic import BaseModel from pydantic import BaseModel
from typing_extensions import override
from .......games.tetrio.api.typedefs import ValidRank from .......games.tetrio.api.typedefs import ValidRank
from ......typedefs import Number from ......typedefs import Number
@@ -23,5 +24,10 @@ class ItemData(BaseModel):
class Data(Base): class Data(Base):
@property
@override
def path(self) -> str:
return 'v2/tetrio/rank'
items: dict[ValidRank, ItemData] items: dict[ValidRank, ItemData]
updated_at: datetime updated_at: datetime

View File

@@ -1,6 +1,7 @@
from datetime import datetime from datetime import datetime
from pydantic import BaseModel from pydantic import BaseModel
from typing_extensions import override
from .......games.tetrio.api.typedefs import ValidRank from .......games.tetrio.api.typedefs import ValidRank
from ......typedefs import Number from ......typedefs import Number
@@ -21,6 +22,11 @@ class SpecialData(BaseModel):
class Data(Base): class Data(Base):
@property
@override
def path(self) -> str:
return 'v2/tetrio/rank/detail'
name: ValidRank name: ValidRank
trending: Number trending: Number
require_tr: Number require_tr: Number

View File

@@ -1,3 +1,5 @@
from typing_extensions import override
from .base import Record as BaseRecord from .base import Record as BaseRecord
from .base import Statistic as BaseStatistic from .base import Statistic as BaseStatistic
@@ -9,4 +11,9 @@ class Statistic(BaseStatistic):
class Record(BaseRecord): class Record(BaseRecord):
@property
@override
def path(self) -> str:
return 'v2/tetrio/record/blitz'
statistic: Statistic statistic: Statistic

View File

@@ -1,7 +1,14 @@
from typing_extensions import override
from .base import Record as BaseRecord from .base import Record as BaseRecord
from .base import Statistic from .base import Statistic
class Record(BaseRecord): class Record(BaseRecord):
@property
@override
def path(self) -> str:
return 'v2/tetrio/record/sprint'
statistic: Statistic statistic: Statistic
time: str time: str

View File

@@ -1,6 +1,7 @@
from datetime import datetime from datetime import datetime
from pydantic import BaseModel from pydantic import BaseModel
from typing_extensions import override
from .....typedefs import Number from .....typedefs import Number
from ...base import Base from ...base import Base
@@ -34,6 +35,11 @@ class Game(BaseModel):
class Data(Base): class Data(Base):
@property
@override
def path(self) -> str:
return 'v2/tetrio/tetra-league'
replay_id: str replay_id: str
games: list[Game] games: list[Game]
play_at: datetime play_at: datetime

View File

@@ -2,6 +2,7 @@ from datetime import datetime
from typing import Literal from typing import Literal
from pydantic import BaseModel 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 ArType, RankType
from .......games.tetrio.api.schemas.summaries.achievements import Rank as AchievementRank from .......games.tetrio.api.schemas.summaries.achievements import Rank as AchievementRank
@@ -132,6 +133,11 @@ class Zenith(BaseModel):
class Info(Base): class Info(Base):
@property
@override
def path(self) -> str:
return 'v2/tetrio/user/info'
user: User user: User
tetra_league: TetraLeague | None tetra_league: TetraLeague | None
zenith: Zenith | None zenith: Zenith | None

View File

@@ -1,4 +1,5 @@
from pydantic import BaseModel from pydantic import BaseModel
from typing_extensions import override
from .......games.tetrio.api.typedefs import Rank from .......games.tetrio.api.typedefs import Rank
from ......typedefs import Number from ......typedefs import Number
@@ -34,5 +35,10 @@ class Data(BaseModel):
class List(Base): class List(Base):
@property
@override
def path(self) -> str:
return 'v2/tetrio/user/list'
show_index: bool show_index: bool
data: list[Data] data: list[Data]

View File

@@ -17,7 +17,8 @@ from ..config.config import CACHE_PATH, DATA_PATH, config
driver = get_driver() 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) 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 ).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() @alc.handle()
async def _(revision: str | None = None): async def _(revision: str | None = None):
if revision is not None and not await check_tag(revision): 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'): if await init_templates(revision or 'latest'):
await alc.finish('更新模板成功') await alc.finish('更新模板成功')
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)

View File

@@ -3,7 +3,7 @@ from typing import Literal, TypeAlias
Number: TypeAlias = float | int Number: TypeAlias = float | int
GameType: TypeAlias = Literal['IO', 'TOP', 'TOS'] GameType: TypeAlias = Literal['IO', 'TOP', 'TOS']
BaseCommandType: TypeAlias = Literal['bind', 'unbind', 'query'] 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 AllCommandType: TypeAlias = BaseCommandType | TETRIOCommandType
Me: TypeAlias = Literal[ Me: TypeAlias = Literal[
'', '',

View File

@@ -2,7 +2,7 @@
[project] [project]
name = "nonebot-plugin-tetris-stats" name = "nonebot-plugin-tetris-stats"
version = "1.10.1" version = "1.11.0"
description = "一款基于 NoneBot2 的用于查询 Tetris 相关游戏数据的插件" description = "一款基于 NoneBot2 的用于查询 Tetris 相关游戏数据的插件"
readme = "README.md" readme = "README.md"
authors = [{ name = "shoucandanghehe", email = "wallfjjd@gmail.com" }] authors = [{ name = "shoucandanghehe", email = "wallfjjd@gmail.com" }]
@@ -162,7 +162,7 @@ defineConstant = { PYDANTIC_V2 = true }
typeCheckingMode = "standard" typeCheckingMode = "standard"
[tool.bumpversion] [tool.bumpversion]
current_version = "1.10.1" current_version = "1.11.0"
tag = true tag = true
sign_tags = true sign_tags = true
tag_name = "{new_version}" tag_name = "{new_version}"