Skip to content

✨ 迁移到新模板 #536

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 13 commits into from
Apr 23, 2025
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -86,7 +86,7 @@ class Record(BaseModel):
pb: bool
oncepb: bool
ts: datetime
revolution: None
revolution: str | None
user: User
otherusers: list
leaderboards: list[str]
Expand All @@ -97,7 +97,7 @@ class Record(BaseModel):


class Best(BaseModel):
record: None # WTF
record: Record | None
rank: int


Expand Down
6 changes: 4 additions & 2 deletions nonebot_plugin_tetris_stats/games/tetrio/bind.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@
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.schemas.base import Avatar, People
from ...utils.screenshot import screenshot
Expand Down Expand Up @@ -65,7 +66,7 @@ async def _(nb_user: User, account: Player, event_session: EventSession, bot_inf
'v1/binding',
Bind(
platform='TETR.IO',
status='unknown',
type='unknown',
user=People(
avatar=str(
URL(f'http://{netloc}/host/resource/tetrio/avatars/{user.ID}')
Expand All @@ -79,7 +80,8 @@ async def _(nb_user: User, account: Player, event_session: EventSession, bot_inf
avatar=await get_avatar(bot_info, 'Data URI', '../../static/logo/logo.svg'),
name=bot_info.user_name,
),
command='io查我',
prompt='io查我',
_lang=get_lang(),
),
)
) as page_hash:
Expand Down
21 changes: 12 additions & 9 deletions nonebot_plugin_tetris_stats/games/tetrio/list.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,9 +5,10 @@

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.schemas.tetrio.user.list_v2 import 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 command
Expand Down Expand Up @@ -63,12 +64,15 @@ async def _(
'v2/tetrio/user/list',
List(
show_index=True,
users=[
User(
id=i.id,
name=i.username.upper(),
avatar=f'https://tetr.io/user-content/avatars/{i.id}.jpg',
country=i.country,
data=[
Data(
user=User(
id=i.id,
name=i.username.upper(),
avatar=f'https://tetr.io/user-content/avatars/{i.id}.jpg',
country=i.country,
xp=i.xp,
),
tetra_league=TetraLeague(
rank=i.league.rank,
tr=round(i.league.tr, 2),
Expand All @@ -81,12 +85,11 @@ async def _(
vs=metrics.vs,
adpl=metrics.adpl,
),
xp=i.xp,
join_at=None,
)
for i in league.data.entries
if isinstance(i, Entry)
],
_lang=get_lang(),
),
)
) as page_hash:
Expand Down
10 changes: 5 additions & 5 deletions nonebot_plugin_tetris_stats/games/tetrio/query/tools.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,22 +4,22 @@
from zoneinfo import ZoneInfo

from ....utils.exception import FallbackError
from ....utils.render.schemas.tetrio.user.base import TetraLeagueHistoryData
from ....utils.render.schemas.base import HistoryData
from ..api.schemas.labs.leagueflow import Empty, LeagueFlowSuccess
from ..api.schemas.summaries.league import InvalidData, LeagueSuccessModel, NeverPlayedData, NeverRatedData, RatedData


def flow_to_history(
leagueflow: LeagueFlowSuccess,
handle: Callable[[list[TetraLeagueHistoryData]], list[TetraLeagueHistoryData]] | None = None,
) -> list[TetraLeagueHistoryData]:
handle: Callable[[list[HistoryData]], list[HistoryData]] | None = None,
) -> list[HistoryData]:
if isinstance(leagueflow.data, Empty):
raise FallbackError
start_time = leagueflow.data.start_time.astimezone(ZoneInfo('Asia/Shanghai'))
ret = [
TetraLeagueHistoryData(
HistoryData(
record_at=start_time + timedelta(milliseconds=i.timestamp_offset),
tr=i.post_match_tr,
score=i.post_match_tr,
)
for i in leagueflow.data.points
if start_time + timedelta(milliseconds=i.timestamp_offset)
Expand Down
173 changes: 38 additions & 135 deletions nonebot_plugin_tetris_stats/games/tetrio/query/v1.py
Original file line number Diff line number Diff line change
@@ -1,127 +1,25 @@
from asyncio import gather
from datetime import datetime, timedelta
from datetime import timedelta
from hashlib import md5
from math import ceil, floor
from zoneinfo import ZoneInfo

from yarl import URL

from ....utils.exception import FallbackError, WhatTheFuckError
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.lang import get_lang
from ....utils.metrics import get_metrics
from ....utils.render import render
from ....utils.render.schemas.base import Avatar, Ranking
from ....utils.render.schemas.tetrio.user.base import TetraLeagueHistoryData
from ....utils.render.schemas.tetrio.user.info_v1 import Info, Radar, TetraLeague, TetraLeagueHistory, User
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 ..api import Player
from ..api.schemas.summaries.league import RatedData
from ..constant import TR_MAX, TR_MIN
from .tools import flow_to_history, get_league_data


def get_value_bounds(values: list[int | float]) -> tuple[int, int]:
value_max = 10 * ceil(max(values) / 10)
value_min = 10 * floor(min(values) / 10)
return value_max, value_min


def get_split(value_max: int, value_min: int) -> tuple[int, int]:
offset = 0
overflow = 0

while True:
if (new_max_value := value_max + offset + overflow) > TR_MAX:
overflow -= 1
continue
if (new_min_value := value_min - offset + overflow) < TR_MIN:
overflow += 1
continue
if ((new_max_value - new_min_value) / 40).is_integer():
split_value = int((value_max + offset - (value_min - offset)) / 4)
break
offset += 1
return split_value, offset + overflow


def get_specified_point(
previous_point: TetraLeagueHistoryData,
behind_point: TetraLeagueHistoryData,
point_time: datetime,
) -> TetraLeagueHistoryData:
"""根据给出的 previous_point 和 behind_point, 推算 point_time 点处的数据

Args:
previous_point (Data): 前面的数据点
behind_point (Data): 后面的数据点
point_time (datetime): 要推算的点的位置

Returns:
Data: 要推算的点的数据
"""
# 求两个点的斜率
slope = (behind_point.tr - previous_point.tr) / (
datetime.timestamp(behind_point.record_at) - datetime.timestamp(previous_point.record_at)
)
return TetraLeagueHistoryData(
record_at=point_time,
tr=previous_point.tr + slope * (datetime.timestamp(point_time) - datetime.timestamp(previous_point.record_at)),
)


def handle_history_data(data: list[TetraLeagueHistoryData]) -> list[TetraLeagueHistoryData]: # noqa: C901, PLR0912
# 按照 记录时间 对数据进行排序
data.sort(key=lambda x: x.record_at)

# 定义时间边界, 右边界为当前时间的当天零点, 左边界为右边界前推9天
# 返回值的[0]和[-1]分别应满足left_border和right_border
zero = datetime.now(ZoneInfo('Asia/Shanghai')).replace(hour=0, minute=0, second=0, microsecond=0)
left_border = zero - timedelta(days=9)
right_border = zero.replace(microsecond=1000)

lefts: list[TetraLeagueHistoryData] = []
in_border: list[TetraLeagueHistoryData] = []
rights: list[TetraLeagueHistoryData] = []

# 根据 记录时间 将数据分类到对应的列表中
for i in data:
if i.record_at < left_border:
lefts.append(i)
elif i.record_at < right_border:
in_border.append(i)
else:
rights.append(i)

ret: list[TetraLeagueHistoryData] = []

# 处理左边界的点
if lefts and in_border: # 如果边界左侧和边界内都有值则推算
ret.append(get_specified_point(lefts[-1], in_border[0], left_border))
elif lefts and not in_border: # 如果边界左侧有值但是边界内没有值则直接取左侧的最后一个值
ret.append(TetraLeagueHistoryData(tr=lefts[-1].tr, record_at=left_border))
elif not lefts and in_border: # 如果边界左侧没有值但是边界内有值则直接取边界内的第一个值
ret.append(TetraLeagueHistoryData(tr=in_border[0].tr, record_at=left_border))
elif not lefts and not in_border and rights: # 如果边界左侧和边界内都没有值但是边界右侧有值则直接取边界右侧的第一个值 # fmt: skip
ret.append(TetraLeagueHistoryData(tr=rights[0].tr, record_at=left_border))
else: # 暂时没想到其他情况
raise WhatTheFuckError

# 添加边界内数据
ret.extend(in_border)

# 处理右边界的点
if in_border and rights: # 如果边界内和边界右侧都有值则推算
ret.append(get_specified_point(in_border[-1], rights[0], right_border))
elif not in_border and rights: # 如果边界内没有值但是边界右侧有值则直接取右侧的第一个值
ret.append(TetraLeagueHistoryData(tr=rights[0].tr, record_at=right_border))
elif in_border and not rights: # 如果边界内有值但是边界右侧没有值则直接取边界内的最后一个值
ret.append(TetraLeagueHistoryData(tr=in_border[-1].tr, record_at=right_border))
elif not in_border and not rights and lefts: # 如果边界内和边界右侧都没有值但是边界左侧有值则直接取边界左侧的最后一个值 # fmt: skip
ret.append(TetraLeagueHistoryData(tr=lefts[-1].tr, record_at=right_border))
else: # 暂时没想到其他情况
raise WhatTheFuckError
return ret


async def make_query_image_v1(player: Player) -> bytes:
(
(user, user_info, league, sprint, blitz, leagueflow),
Expand All @@ -134,15 +32,18 @@
if league_data.vs is None:
raise FallbackError
histories = flow_to_history(leagueflow, handle_history_data)
value_max, value_min = get_value_bounds([i.tr for i in histories])
split_value, offset = get_split(value_max, value_min)
values = get_value_bounds([i.score for i in histories])
split_value, offset = get_split(values, TR_MAX, TR_MIN)

Check warning on line 36 in nonebot_plugin_tetris_stats/games/tetrio/query/v1.py

View check run for this annotation

Codecov / codecov/patch

nonebot_plugin_tetris_stats/games/tetrio/query/v1.py#L35-L36

Added lines #L35 - L36 were not covered by tests
Copy link
Preview

Copilot AI Apr 23, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ensure that the updated get_split function accepts three arguments as used here and that TR_MAX and TR_MIN are correctly defined for its intended usage.

Copilot uses AI. Check for mistakes.

if sprint.data.record is not None:
duration = timedelta(milliseconds=sprint.data.record.results.stats.finaltime).total_seconds()
sprint_value = f'{duration:.3f}s' if duration < 60 else f'{duration // 60:.0f}m {duration % 60:.3f}s' # noqa: PLR2004
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',
Expand All @@ -159,38 +60,40 @@
name=user.name.upper(),
bio=user_info.data.bio,
),
ranking=Ranking(
rating=round(league_data.glicko, 2),
multiplayer=Multiplayer(
glicko=f'{round(league_data.glicko, 2):,}',
rd=round(league_data.rd, 2),
),
tetra_league=TetraLeague(
rank=league_data.rank,
tr=round(league_data.tr, 2),
tr=f'{round(league_data.tr, 2):,}',
global_rank=league_data.standing,
pps=league_data.pps,
lpm=round(lpm := (league_data.pps * 24), 2),
apm=league_data.apm,
apl=round(league_data.apm / lpm, 2),
vs=league_data.vs,
adpm=round(adpm := (league_data.vs * 0.6), 2),
adpl=round(adpm / lpm, 2),
),
tetra_league_history=TetraLeagueHistory(
data=histories,
split_interval=split_value,
min_tr=value_min,
max_tr=value_max,
offset=offset,
),
radar=Radar(
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),
),
sprint=sprint_value,
blitz=blitz_value,
singleplayer=Singleplayer(
sprint=sprint_value,
blitz=blitz_value,
),
_lang=get_lang(),
),
)
) as page_hash:
Expand Down
Loading