Compare commits

...

6 Commits
1.6.0 ... 1.6.2

Author SHA1 Message Date
08a1a427b4 🔖 1.6.2 2024-10-29 00:44:55 +08:00
呵呵です
d4e91c8521 🐛 修复被 ban 的爆炸 (#510) 2024-10-28 16:43:38 +00:00
dbde1181ce 🔖 1.6.1 2024-10-28 22:06:30 +08:00
呵呵です
86fe4f0766 🐛 修复 handle_history_data 的索引越界问题 (#509) 2024-10-28 14:01:11 +00:00
呵呵です
381f2505d6 🐛 修复 leagueflow 没有有效数据爆炸 (#508) 2024-10-28 13:58:38 +00:00
b3a77f5296 💚 修复 basedpyright
https://github.com/DetachHead/basedpyright/issues/819
2024-10-27 21:28:19 +08:00
6 changed files with 140 additions and 101 deletions

View File

@@ -1,6 +1,6 @@
from datetime import datetime from datetime import datetime
from enum import IntEnum from enum import IntEnum
from typing import NamedTuple from typing import Literal, NamedTuple
from pydantic import BaseModel, Field from pydantic import BaseModel, Field
@@ -28,11 +28,16 @@ class Point(NamedTuple):
class Data(BaseModel): class Data(BaseModel):
start_time: datetime = Field(..., alias='startTime') start_time: datetime = Field(..., alias='startTime')
points: list[Point] points: list[Point] = Field(..., min_length=1)
class Empty(BaseModel):
start_time: Literal[9007199254740991] = Field(..., alias='startTime')
points: list = Field(..., max_length=0)
class LeagueFlowSuccess(BaseSuccessModel): class LeagueFlowSuccess(BaseSuccessModel):
data: Data data: Data | Empty
LeagueFlow = LeagueFlowSuccess | FailedModel LeagueFlow = LeagueFlowSuccess | FailedModel

View File

@@ -122,5 +122,9 @@ class RatedData(BaseData):
percentile_rank: str percentile_rank: str
class InvalidData(BaseModel):
"""I don't know what osk is doing, but the return value is an empty dictionary"""
class LeagueSuccessModel(SuccessModel): class LeagueSuccessModel(SuccessModel):
data: NeverPlayedData | NeverRatedData | RatedData data: NeverPlayedData | NeverRatedData | RatedData | InvalidData

View File

@@ -5,14 +5,16 @@ from zoneinfo import ZoneInfo
from ....utils.exception import FallbackError from ....utils.exception import FallbackError
from ....utils.render.schemas.tetrio.user.base import TetraLeagueHistoryData from ....utils.render.schemas.tetrio.user.base import TetraLeagueHistoryData
from ..api.schemas.labs.leagueflow import LeagueFlowSuccess from ..api.schemas.labs.leagueflow import Empty, LeagueFlowSuccess
from ..api.schemas.summaries.league import LeagueSuccessModel, NeverPlayedData, NeverRatedData, RatedData from ..api.schemas.summaries.league import InvalidData, LeagueSuccessModel, NeverPlayedData, NeverRatedData, RatedData
def flow_to_history( def flow_to_history(
leagueflow: LeagueFlowSuccess, leagueflow: LeagueFlowSuccess,
handle: Callable[[list[TetraLeagueHistoryData]], list[TetraLeagueHistoryData]] | None = None, handle: Callable[[list[TetraLeagueHistoryData]], list[TetraLeagueHistoryData]] | None = None,
) -> list[TetraLeagueHistoryData]: ) -> list[TetraLeagueHistoryData]:
if isinstance(leagueflow.data, Empty):
raise FallbackError
start_time = leagueflow.data.start_time.astimezone(ZoneInfo('Asia/Shanghai')) start_time = leagueflow.data.start_time.astimezone(ZoneInfo('Asia/Shanghai'))
ret = [ ret = [
TetraLeagueHistoryData( TetraLeagueHistoryData(
@@ -45,6 +47,8 @@ def get_league_data(
user_info: LeagueSuccessModel, league_type: type[L] | None = None user_info: LeagueSuccessModel, league_type: type[L] | None = None
) -> L | NeverPlayedData | NeverRatedData | RatedData: ) -> L | NeverPlayedData | NeverRatedData | RatedData:
league = user_info.data league = user_info.data
if isinstance(league, InvalidData):
raise FallbackError
if league_type is None: if league_type is None:
return league return league
if isinstance(league, league_type): if isinstance(league, league_type):

View File

@@ -6,7 +6,7 @@ from zoneinfo import ZoneInfo
from yarl import URL from yarl import URL
from ....utils.exception import FallbackError from ....utils.exception import FallbackError, WhatTheFuckError
from ....utils.host import HostPage, get_self_netloc from ....utils.host import HostPage, get_self_netloc
from ....utils.render import render from ....utils.render import render
from ....utils.render.schemas.base import Avatar, Ranking from ....utils.render.schemas.base import Avatar, Ranking
@@ -68,15 +68,21 @@ def get_specified_point(
) )
def handle_history_data(data: list[TetraLeagueHistoryData]) -> list[TetraLeagueHistoryData]: def handle_history_data(data: list[TetraLeagueHistoryData]) -> list[TetraLeagueHistoryData]: # noqa: C901, PLR0912
# 按照 记录时间 对数据进行排序
data.sort(key=lambda x: x.record_at) data.sort(key=lambda x: x.record_at)
right_border = datetime.now(ZoneInfo('Asia/Shanghai')).replace(hour=0, minute=0, second=0, microsecond=0) # 定义时间边界, 右边界为当前时间的当天零点, 左边界为右边界前推9天
left_border = right_border - timedelta(days=9) # 返回值的[0]和[-1]分别应满足left_borderright_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] = [] lefts: list[TetraLeagueHistoryData] = []
in_border: list[TetraLeagueHistoryData] = [] in_border: list[TetraLeagueHistoryData] = []
rights: list[TetraLeagueHistoryData] = [] rights: list[TetraLeagueHistoryData] = []
# 根据 记录时间 将数据分类到对应的列表中
for i in data: for i in data:
if i.record_at < left_border: if i.record_at < left_border:
lefts.append(i) lefts.append(i)
@@ -84,16 +90,35 @@ def handle_history_data(data: list[TetraLeagueHistoryData]) -> list[TetraLeagueH
in_border.append(i) in_border.append(i)
else: else:
rights.append(i) rights.append(i)
ret: list[TetraLeagueHistoryData] = [] ret: list[TetraLeagueHistoryData] = []
if lefts:
# 处理左边界的点
if lefts and in_border: # 如果边界左侧和边界内都有值则推算
ret.append(get_specified_point(lefts[-1], in_border[0], left_border)) ret.append(get_specified_point(lefts[-1], in_border[0], left_border))
else: 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)) 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) ret.extend(in_border)
if rights:
ret.append(get_specified_point(in_border[-1], rights[0], right_border.replace(microsecond=1000))) # 处理右边界的点
else: if in_border and rights: # 如果边界内和边界右侧都有值则推算
ret.append(TetraLeagueHistoryData(tr=in_border[-1].tr, record_at=right_border.replace(microsecond=1000))) 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 return ret

View File

@@ -4,6 +4,7 @@ from hashlib import md5
from yarl import URL from yarl import URL
from ....utils.exception import FallbackError
from ....utils.host import HostPage, get_self_netloc from ....utils.host import HostPage, get_self_netloc
from ....utils.metrics import get_metrics from ....utils.metrics import get_metrics
from ....utils.render import render from ....utils.render import render
@@ -21,7 +22,7 @@ from ....utils.render.schemas.tetrio.user.info_v2 import (
) )
from ....utils.screenshot import screenshot from ....utils.screenshot import screenshot
from ..api import Player from ..api import Player
from ..api.schemas.summaries.league import 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
@@ -49,89 +50,89 @@ async def make_query_image_v2(player: Player) -> bytes:
play_time = f'{game_time:.0f}s' play_time = f'{game_time:.0f}s'
else: else:
play_time = game_time play_time = game_time
try:
history = flow_to_history(leagueflow, lambda x: x[-100:])
except FallbackError:
history = None
netloc = get_self_netloc() netloc = get_self_netloc()
async with ( async with HostPage(
HostPage( await render(
await render( 'v2/tetrio/user/info',
'v2/tetrio/user/info', Info(
Info( user=User(
user=User( id=user.ID,
id=user.ID, name=user.name.upper(),
name=user.name.upper(), bio=user_info.data.bio,
bio=user_info.data.bio, banner=str(
banner=str( URL(f'http://{netloc}/host/resource/tetrio/banners/{user.ID}') % {'revision': banner_revision}
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,
if banner_revision is not None and banner_revision != 0 avatar=str(
else None, 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}') if avatar_revision is not None and avatar_revision != 0
% {'revision': avatar_revision} else Avatar(
) type='identicon',
if avatar_revision is not None and avatar_revision != 0 hash=md5(user.ID.encode()).hexdigest(), # noqa: S324
else Avatar(
type='identicon',
hash=md5(user.ID.encode()).hexdigest(), # noqa: S324
),
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
],
country=user_info.data.country,
role=user_info.data.role,
xp=user_info.data.xp,
friend_count=user_info.data.friend_count,
supporter_tier=user_info.data.supporter_tier,
bad_standing=user_info.data.badstanding or False,
playtime=play_time,
join_at=user_info.data.ts,
), ),
tetra_league=TetraLeague( badges=[
rank=league.data.rank, Badge(
highest_rank='z' if isinstance(league.data, NeverRatedData) else league.data.bestrank, id=i.id,
tr=round(league.data.tr, 2), description=i.label,
glicko=round(league.data.glicko, 2), group=i.group,
rd=round(league.data.rd, 2), receive_at=i.ts if isinstance(i.ts, datetime) else None,
global_rank=league.data.standing, )
country_rank=league.data.standing_local, for i in user_info.data.badges
pps=(metrics := get_metrics(pps=league.data.pps, apm=league.data.apm, vs=league.data.vs)).pps, ],
apm=metrics.apm, country=user_info.data.country,
apl=metrics.apl, role=user_info.data.role,
vs=metrics.vs, xp=user_info.data.xp,
adpl=metrics.adpl, friend_count=user_info.data.friend_count,
statistic=TetraLeagueStatistic(total=league.data.gamesplayed, wins=league.data.gameswon), supporter_tier=user_info.data.supporter_tier,
decaying=league.data.decaying, bad_standing=user_info.data.badstanding or False,
history=flow_to_history(leagueflow, lambda x: x[-100:]), playtime=play_time,
) join_at=user_info.data.ts,
if not isinstance(league.data, NeverPlayedData)
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,
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,
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),
), ),
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,
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,
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,
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),
), ),
) as page_hash ),
): ) as page_hash:
return await screenshot(f'http://{netloc}/host/{page_hash}.html') return await screenshot(f'http://{netloc}/host/{page_hash}.html')

View File

@@ -1,6 +1,6 @@
[project] [project]
name = "nonebot-plugin-tetris-stats" name = "nonebot-plugin-tetris-stats"
version = "1.6.0" version = "1.6.2"
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" }]
@@ -151,12 +151,12 @@ defineConstant = { PYDANTIC_V2 = true }
typeCheckingMode = "standard" typeCheckingMode = "standard"
[tool.bumpversion] [tool.bumpversion]
current_version = "1.6.0" current_version = "1.6.2"
tag = true tag = true
sign_tags = true sign_tags = true
tag_name = "{new_version}" tag_name = "{new_version}"
commit = true commit = true
message = "🔖 {new_version}" message = ":bookmark: {new_version}"
[[tool.bumpversion.files]] [[tool.bumpversion.files]]
filename = "pyproject.toml" filename = "pyproject.toml"