mirror of
https://github.com/A-Minos/nonebot-plugin-tetris-stats.git
synced 2026-03-05 05:36:54 +08:00
Compare commits
6 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 08a1a427b4 | |||
|
|
d4e91c8521 | ||
| dbde1181ce | |||
|
|
86fe4f0766 | ||
|
|
381f2505d6 | ||
| b3a77f5296 |
@@ -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
|
||||||
|
|||||||
@@ -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
|
||||||
|
|||||||
@@ -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):
|
||||||
|
|||||||
@@ -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_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] = []
|
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
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
@@ -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')
|
||||||
|
|||||||
@@ -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"
|
||||||
|
|||||||
Reference in New Issue
Block a user