Skip to content

Commit

Permalink
🐛 修复使用 ID 绑定账号错误的bug
Browse files Browse the repository at this point in the history
  • Loading branch information
@shoucandanghehe
shoucandanghehe committed Sep 21, 2023
1 parent 1219aed commit 86adf26
Showing 1 changed file with 57 additions and 46 deletions.
103 changes: 57 additions & 46 deletions nonebot_plugin_tetris_stats/game_data_processor/io_data_processor/processor.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,7 @@
class Processor:
@classmethod
async def handle_bind(cls, message: str, qq_number: int | None) -> str:
'''处理绑定消息'''
"""处理绑定消息"""
decoded_message = await handle_bind_message(message=message, game_type='IO')
if decoded_message[0] is None:
return decoded_message[1][0]
Expand All @@ -25,25 +25,27 @@ async def handle_bind(cls, message: str, qq_number: int | None) -> str:
if user_id_stats[0] is False:
return user_id_stats[1]
user_id = decoded_message[1][1]
if qq_number is None: # 理论上是不会有None出现的, ide快乐行属于是(
logger.error('获取QQ号失败')
return '获取QQ号失败'
return await DataBase.write_bind_info(
qq_number=qq_number, user=user_id, game_type='IO'
)
elif decoded_message[0] == 'Name':
user_data = await cls.get_user_data(user_name=decoded_message[1][1])
if user_data[0] is False:
return '用户信息请求失败'
if user_data[1] is False:
return f'用户信息请求错误:\n{user_data[2]["error"]}'
user_id = await cls.get_user_id(user_data[2])
if qq_number is None: # 理论上是不会有None出现的, ide快乐行属于是（
if qq_number is None: # 理论上是不会有None出现的, ide快乐行属于是(
logger.error('获取QQ号失败')
return '获取QQ号失败'
return (
await DataBase.write_bind_info(
qq_number=qq_number,
user=user_id,
game_type='IO'
)
return await DataBase.write_bind_info(
qq_number=qq_number, user=user_id, game_type='IO'
)
logger.error('预期外行为, 请上报GitHub')
return '出现预期外行为，请查看后台信息'
return '出现预期外行为, 请查看后台信息'

@classmethod
async def handle_rank(cls, message: str):
Expand Down Expand Up @@ -74,11 +76,17 @@ async def handle_rank(cls, message: str):
if query_rank.lower() not in (i for i in ranks_percentiles.keys()):
return '未知段位'

result = await Request.request('https://ch.tetr.io/api/users/lists/league/all')
result = await Request.request(
'https://ch.tetr.io/api/users/lists/league/all'
)
users: list = result[2]['data']['users']

def avg(rank_users: list, column: str, playercount: int | None = None) -> float:
return sum(i['league'][column] for i in rank_users) / (playercount or len(rank_users))
def avg(
rank_users: list, column: str, playercount: int | None = None
) -> float:
return sum(i['league'][column] for i in rank_users) / (
playercount or len(rank_users)
)

for rank, percentile in ranks_percentiles.items():
offset = math.floor((percentile / 100) * len(users)) - 1
Expand All @@ -97,8 +105,8 @@ def avg(rank_users: list, column: str, playercount: int | None = None) -> float:
playercount=playercount,
avgapm=avg_apm,
avgpps=avg_pps,
avgvs=avg_vs
)
avgvs=avg_vs,
)

return await Processor.handle_rank(message=message)
else:
Expand All @@ -123,20 +131,26 @@ def avg(rank_users: list, column: str, playercount: int | None = None) -> float:

@classmethod
async def handle_query(cls, message: str, qq_number: int | None):
'''处理查询消息'''
decoded_message = await handle_stats_query_message(message=message, game_type='IO')
"""处理查询消息"""
decoded_message = await handle_stats_query_message(
message=message, game_type='IO'
)
if decoded_message[0] is None:
return decoded_message[1][0]
if decoded_message[0] == 'AT': # 在入口处判断是否@bot本身
bind_info = await DataBase.query_bind_info(qq_number=decoded_message[1][1], game_type='IO')
bind_info = await DataBase.query_bind_info(
qq_number=decoded_message[1][1], game_type='IO'
)
if bind_info is None:
return '未查询到绑定信息'
return f'* 由于无法验证绑定信息, 不能保证查询到的用户为本人\n{await Processor.generate_message(user_id=bind_info)}'
if decoded_message[0] == 'ME':
if qq_number is None:
logger.error('获取QQ号失败')
return '获取QQ号失败, 请联系bot主人'
bind_info = await DataBase.query_bind_info(qq_number=qq_number, game_type='IO')
bind_info = await DataBase.query_bind_info(
qq_number=qq_number, game_type='IO'
)
if bind_info is None:
return '未查询到绑定信息'
return f'* 由于无法验证绑定信息, 不能保证查询到的用户为本人\n{await Processor.generate_message(user_id=bind_info)}'
Expand All @@ -147,11 +161,9 @@ async def handle_query(cls, message: str, qq_number: int | None):

@classmethod
async def get_user_data(
cls,
user_name: str | None = None,
user_id: str | None = None
cls, user_name: str | None = None, user_id: str | None = None
) -> tuple[bool, bool, dict[str, Any]]:
'''获取用户数据'''
"""获取用户数据"""
if user_name is not None and user_id is None:
user_data_url = f'https://ch.tetr.io/api/users/{user_name}'
elif user_name is None and user_id is not None:
Expand All @@ -162,11 +174,9 @@ async def get_user_data(

@classmethod
async def get_solo_data(
cls,
user_name: str | None = None,
user_id: str | None = None
cls, user_name: str | None = None, user_id: str | None = None
) -> tuple[bool, bool, dict[str, Any]]:
'''获取Solo数据'''
"""获取Solo数据"""
if user_name is not None and user_id is None:
user_solo_url = f'https://ch.tetr.io/api/users/{user_name}/records'
elif user_name is None and user_id is not None:
Expand All @@ -177,31 +187,32 @@ async def get_solo_data(

@classmethod
async def get_user_id(cls, user_data: dict) -> str:
'''获取用户ID'''
"""获取用户ID"""
return user_data['data']['user']['_id']

@classmethod
async def check_user_id(cls, user_id: str) -> tuple[bool, str]:
'''检查用户ID是否有效 返回值为tuple[bool, message]'''
"""检查用户ID是否有效 返回值为tuple[bool, message]"""
user_data = await cls.get_user_data(user_id=user_id)
if user_data[0] is False:
return False, '用户信息请求失败'
if user_data[1] is False:
return False, f'用户信息请求错误:\n{user_data[2]["error"]}'
if user_id == user_data[2]['data']['user']['_id']:
return True, ''
raise ValueError('服务器返回的userID和用户提供的不一致, 这种情况理论上不应该发生, 以防万一还是写一下（x')
raise ValueError('服务器返回的userID和用户提供的不一致, 这种情况理论上不应该发生, 以防万一还是写一下(x')

@classmethod
async def get_league_stats(cls, user_data: dict) -> dict[str, Any]:
'''获取排位统计数据'''
"""获取排位统计数据"""
league = user_data['data']['user']['league']
league_stats: dict[str, Any] = {}
if league['gamesplayed'] != 0:
league_stats['PPS'] = league['pps']
league_stats['APM'] = league['apm']
league_stats['VS'] = 0 if league['vs'] is None else league['vs']
league_stats['Rank'] = 'Z' if league['rank'] == 'z' else league['rank'].upper(
league_stats['Rank'] = (
'Z' if league['rank'] == 'z' else league['rank'].upper()
)
if league['rating'] == -1:
league_stats['Rank'] = None
Expand All @@ -211,28 +222,29 @@ async def get_league_stats(cls, user_data: dict) -> dict[str, Any]:
league_stats['RD'] = round(league['rd'], 2)
league_stats['Standing'] = league['standing']
league_stats['LPM'] = round((league['pps'] * 24), 2)
league_stats['APL'] = round(
(league_stats['APM'] / league_stats['LPM']), 2)
league_stats['APL'] = round((league_stats['APM'] / league_stats['LPM']), 2)
league_stats['ADPM'] = round((league_stats['VS'] * 0.6), 2)
league_stats['ADPL'] = round(
(league_stats['ADPM'] / league_stats['LPM']), 2)
(league_stats['ADPM'] / league_stats['LPM']), 2
)
return league_stats

@classmethod
async def get_sprint_stats(cls, solo_data: dict) -> dict[str, Any]:
'''获取40L统计数据'''
"""获取40L统计数据"""
sprint_stats = {}
solo = solo_data['data']['records']['40l']
if solo['record'] is not None:
sprint_stats['Time'] = round(
solo['record']['endcontext']['finalTime'] / 1000, 2)
solo['record']['endcontext']['finalTime'] / 1000, 2
)
if solo['rank'] is not None:
sprint_stats['Rank'] = solo['rank']
return sprint_stats

@classmethod
async def get_blitz_stats(cls, solo_data: dict) -> dict[str, Any]:
'''获取Blitz统计数据'''
"""获取Blitz统计数据"""
blitz_stats = {}
blitz = solo_data['data']['records']['blitz']
if blitz['record'] is not None:
Expand All @@ -243,14 +255,12 @@ async def get_blitz_stats(cls, solo_data: dict) -> dict[str, Any]:

@classmethod
async def generate_message(
cls,
user_name: str | None = None,
user_id: str | None = None
cls, user_name: str | None = None, user_id: str | None = None
) -> str:
'''生成消息'''
"""生成消息"""
user_data, solo_data = await gather(
cls.get_user_data(user_name=user_name, user_id=user_id),
cls.get_solo_data(user_name=user_name, user_id=user_id)
cls.get_solo_data(user_name=user_name, user_id=user_id),
)
if user_data[0] is False:
return '用户信息请求失败'
Expand All @@ -269,18 +279,19 @@ async def generate_message(
message += f'用户 {user_name} 暂无段位, {league_stats["Rating"]} TR'
else:
message += f'{league_stats["Rank"]} 段用户 {user_name} {league_stats["Rating"]} TR (#{league_stats["Standing"]})'
message += f', 段位分 {league_stats["Glicko"]}±{league_stats["RD"]}, 最近十场的数据:'
message += (
f', 段位分 {league_stats["Glicko"]}±{league_stats["RD"]}, 最近十场的数据:'
)
message += f'\nL\'PM: {league_stats["LPM"]} ( {league_stats["PPS"]} pps )'
message += f'\nAPM: {league_stats["APM"]} ( x{league_stats["APL"]} )'
if league_stats["VS"] != 0:
if league_stats['VS'] != 0:
message += f'\nADPM: {league_stats["ADPM"]} ( x{league_stats["ADPL"]} ) ( {league_stats["VS"]}vs )'
if solo_data[0] is False:
return f'{message}\nSolo统计数据请求失败'
if solo_data[1] is False:
return f'{message}\nSolo统计数据请求错误:\n{solo_data[2]["error"]}'
sprint_stats, blitz_stats = await gather(
cls.get_sprint_stats(solo_data[2]),
cls.get_blitz_stats(solo_data[2])
cls.get_sprint_stats(solo_data[2]), cls.get_blitz_stats(solo_data[2])
)
message += f'\n40L: {sprint_stats["Time"]}s' if 'Time' in sprint_stats else ''
message += f' ( #{sprint_stats["Rank"]} )' if 'Rank' in sprint_stats else ''
Expand Down

0 comments on commit 86adf26

Please sign in to comment.