refactor(plugins): comprehensive code review - ~35 fixes across 14 plugins
Phase 1 - Plugin code review (14/14 plugins): - Security: 3x token leak in print→logger.debug, Bearer prefix handling - Bug: bare except→specific exceptions, HorseState type safety, sync→async - Critical: response_model undefined, route dead code, sync blocking event loop - Quality: 11x print()→logger, variable name shadowing, consistent logging Phase 2 - Deep analysis: - Fix: payout int truncation→max(1, round(amount*odds)) - Fix: room_store get_lock race condition→dict.setdefault() - Verify: data_manager f-string SQL is safe (uses ? placeholders) Infrastructure: review reports generated for all plugins.
This commit is contained in:
File diff suppressed because it is too large
Load Diff
@@ -1,247 +1,251 @@
|
||||
import requests
|
||||
import json
|
||||
from typing import Dict, Optional, Tuple
|
||||
from nonebot import logger
|
||||
from .config import Config
|
||||
|
||||
def mask_username(username: str) -> str:
|
||||
"""
|
||||
对用户名进行脱敏处理，只显示前两位和后两位，中间用*号隐藏
|
||||
|
||||
Args:
|
||||
username: 原始用户名
|
||||
|
||||
Returns:
|
||||
脱敏后的用户名
|
||||
"""
|
||||
if not username:
|
||||
return username
|
||||
|
||||
# 如果用户名长度小于等于4，直接显示前两位和后两位（可能重叠）
|
||||
if len(username) <= 4:
|
||||
return username
|
||||
|
||||
# 显示前两位和后两位，中间用*号填充
|
||||
return f"{username[:2]}{'*' * (len(username) - 4)}{username[-2:]}"
|
||||
|
||||
# 获取配置
|
||||
config = Config()
|
||||
|
||||
# API 端点配置
|
||||
DD_API_HOST = "https://api.danding.vip/DD/" # 蛋定服务器连接地址
|
||||
BOT_TOKEN = "3340e353a49447f1be640543cbdcd937" # 对接服务器的Token
|
||||
BOT_USER_ID = "1424473282" # 机器人用户ID
|
||||
|
||||
async def query_qq_binding(qq: str) -> Tuple[bool, Optional[str], Optional[str]]:
|
||||
"""
|
||||
查询QQ号是否绑定了蛋定用户名
|
||||
|
||||
Args:
|
||||
qq: 要查询的QQ号
|
||||
|
||||
Returns:
|
||||
Tuple[是否绑定, 用户名, VIP到期时间]
|
||||
"""
|
||||
try:
|
||||
url = f"{DD_API_HOST}query_qq_binding"
|
||||
data = {"qq": qq}
|
||||
|
||||
response = requests.post(url=url, json=data)
|
||||
logger.debug(f"查询QQ绑定状态响应: {response}")
|
||||
|
||||
if response.status_code != 200:
|
||||
logger.error(f"查询QQ绑定状态失败，状态码: {response.status_code}")
|
||||
return False, None, None
|
||||
|
||||
result = response.json()
|
||||
logger.debug(f"查询QQ绑定状态结果: {result}")
|
||||
|
||||
if result.get("code") == 200:
|
||||
data = result.get("data", {})
|
||||
is_bound = data.get("is_bound", False)
|
||||
|
||||
if is_bound:
|
||||
username = data.get("username")
|
||||
vip_time = data.get("vip_time")
|
||||
return True, username, vip_time
|
||||
else:
|
||||
return False, None, None
|
||||
else:
|
||||
logger.error(f"查询QQ绑定状态失败，错误信息: {result.get('message')}")
|
||||
return False, None, None
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"查询QQ绑定状态异常: {str(e)}")
|
||||
return False, None, None
|
||||
|
||||
async def add_user_viptime(username: str, time_class: str = "Day", count: int = 1) -> Tuple[bool, str]:
|
||||
"""
|
||||
为用户添加VIP时间
|
||||
|
||||
Args:
|
||||
username: 蛋定用户名
|
||||
time_class: 时间类型 (Hour/Day/Week/Month/Season/Year)
|
||||
count: 添加次数（默认为1）
|
||||
|
||||
Returns:
|
||||
Tuple[是否成功, 响应消息]
|
||||
"""
|
||||
try:
|
||||
url = f"{DD_API_HOST}bot_add_user_viptime"
|
||||
|
||||
# 如果count大于1，需要多次调用API
|
||||
success_count = 0
|
||||
last_message = ""
|
||||
|
||||
for i in range(count):
|
||||
data = {
|
||||
"user": BOT_USER_ID,
|
||||
"token": BOT_TOKEN,
|
||||
"username": username,
|
||||
"classes": time_class
|
||||
}
|
||||
|
||||
response = requests.post(url=url, json=data)
|
||||
logger.debug(f"添加VIP时间响应({i+1}/{count}): {response}")
|
||||
|
||||
if response.status_code != 200:
|
||||
error_msg = f"添加VIP时间失败({i+1}/{count})，状态码: {response.status_code}"
|
||||
logger.error(error_msg)
|
||||
continue
|
||||
|
||||
result = response.json()
|
||||
logger.debug(f"添加VIP时间结果({i+1}/{count}): {result}")
|
||||
|
||||
if result.get("code") == 200:
|
||||
success_count += 1
|
||||
last_message = result.get("msg", "添加VIP时间成功")
|
||||
else:
|
||||
error_msg = result.get("msg", "添加VIP时间失败")
|
||||
logger.error(f"添加VIP时间失败({i+1}/{count}): {error_msg}")
|
||||
|
||||
if success_count == count:
|
||||
return True, f"成功添加{count}次{time_class}时长。{last_message}"
|
||||
elif success_count > 0:
|
||||
return False, f"仅成功添加{success_count}/{count}次{time_class}时长。{last_message}"
|
||||
else:
|
||||
return False, f"添加{count}次{time_class}时长全部失败。"
|
||||
|
||||
except Exception as e:
|
||||
error_msg = f"添加VIP时间异常: {str(e)}"
|
||||
logger.error(error_msg)
|
||||
return False, error_msg
|
||||
|
||||
async def process_ssr_sp_reward(user_id: str, count: int = 1) -> Tuple[bool, str]:
|
||||
"""
|
||||
处理SSR/SP奖励发放
|
||||
|
||||
Args:
|
||||
user_id: QQ用户ID
|
||||
count: 奖励数量（默认为1）
|
||||
|
||||
Returns:
|
||||
Tuple[是否自动发放成功, 消息内容]
|
||||
"""
|
||||
# 查询QQ绑定状态
|
||||
is_bound, username, vip_time = await query_qq_binding(user_id)
|
||||
|
||||
if not is_bound:
|
||||
# 用户未绑定，返回提示信息
|
||||
if count == 1:
|
||||
msg = (f"🎉恭喜您抽中了SSR/SP稀有度式神！🎉\n"
|
||||
f"获得奖励：蛋定助手天卡一张\n"
|
||||
f"获取奖励请联系管理员，或前往蛋定云服务中绑定QQ号即可体验自动加时！")
|
||||
else:
|
||||
msg = (f"🎉恭喜您抽中了{count}张SSR/SP稀有度式神！🎉\n"
|
||||
f"获得奖励：蛋定助手天卡{count}张\n"
|
||||
f"获取奖励请联系管理员，或前往蛋定云服务中绑定QQ号即可体验自动加时！")
|
||||
return False, msg
|
||||
else:
|
||||
# 用户已绑定，自动加时
|
||||
success, message = await add_user_viptime(username, "Day", count)
|
||||
|
||||
if success:
|
||||
masked_username = mask_username(username)
|
||||
if count == 1:
|
||||
msg = (f"🎉恭喜您抽中了SSR/SP稀有度式神！🎉\n"
|
||||
f"🎁已自动为您的蛋定账号({masked_username})添加天卡时长！\n"
|
||||
f"📅到期时间: {message.split('到期时间为:')[-1] if '到期时间为:' in message else '请查看您的账号详情'}")
|
||||
else:
|
||||
msg = (f"🎉恭喜您抽中了{count}张SSR/SP稀有度式神！🎉\n"
|
||||
f"🎁已自动为您的蛋定账号({masked_username})添加{count}天卡时长！\n"
|
||||
f"📅到期时间: {message.split('到期时间为:')[-1] if '到期时间为:' in message else '请查看您的账号详情'}")
|
||||
return True, msg
|
||||
else:
|
||||
# 自动加时失败，返回错误信息和手动领取提示
|
||||
if count == 1:
|
||||
msg = (f"🎉恭喜您抽中了SSR/SP稀有度式神！🎉\n"
|
||||
f"获得奖励：蛋定助手天卡一张\n"
|
||||
f"⚠️自动加时失败: {message}\n"
|
||||
f"请联系管理员手动领取奖励！")
|
||||
else:
|
||||
msg = (f"🎉恭喜您抽中了{count}张SSR/SP稀有度式神！🎉\n"
|
||||
f"获得奖励：蛋定助手天卡{count}张\n"
|
||||
f"⚠️自动加时失败: {message}\n"
|
||||
f"请联系管理员手动领取奖励！")
|
||||
return False, msg
|
||||
|
||||
async def process_achievement_reward(user_id: str, achievement_id: str) -> Tuple[bool, str]:
|
||||
"""
|
||||
处理成就奖励发放
|
||||
|
||||
Args:
|
||||
user_id: QQ用户ID
|
||||
achievement_id: 成就ID
|
||||
|
||||
Returns:
|
||||
Tuple[是否自动发放成功, 消息内容]
|
||||
"""
|
||||
# 获取成就配置
|
||||
achievement_config = config.ACHIEVEMENTS.get(achievement_id)
|
||||
if not achievement_config:
|
||||
# 检查是否是重复奖励
|
||||
if "_repeat_" in achievement_id:
|
||||
base_achievement_id = achievement_id.split("_repeat_")[0]
|
||||
base_config = config.ACHIEVEMENTS.get(base_achievement_id)
|
||||
if base_config:
|
||||
reward_type = base_config.get("repeat_reward", "天卡")
|
||||
else:
|
||||
reward_type = "天卡"
|
||||
else:
|
||||
return False, f"未找到成就配置: {achievement_id}"
|
||||
else:
|
||||
reward_type = achievement_config.get("reward", "天卡")
|
||||
|
||||
# 查询QQ绑定状态
|
||||
is_bound, username, vip_time = await query_qq_binding(user_id)
|
||||
|
||||
if not is_bound:
|
||||
# 用户未绑定，返回提示信息
|
||||
msg = (f"🏆 恭喜解锁成就奖励！\n"
|
||||
f"获得奖励：蛋定助手{reward_type}一张\n"
|
||||
f"获取奖励请联系管理员，或前往蛋定云服务中绑定QQ号即可体验自动加时！")
|
||||
return False, msg
|
||||
else:
|
||||
# 用户已绑定，自动加时
|
||||
# 将奖励类型转换为API需要的时间类型
|
||||
time_class = "Day" # 默认为天卡
|
||||
if "周卡" in reward_type:
|
||||
time_class = "Week"
|
||||
elif "月卡" in reward_type:
|
||||
time_class = "Month"
|
||||
|
||||
success, message = await add_user_viptime(username, time_class)
|
||||
|
||||
if success:
|
||||
masked_username = mask_username(username)
|
||||
msg = (f"🏆 恭喜解锁成就奖励！\n"
|
||||
f"🎁已自动为您的蛋定账号({masked_username})添加{reward_type}时长！\n"
|
||||
f"📅到期时间: {message.split('到期时间为:')[-1] if '到期时间为:' in message else '请查看您的账号详情'}")
|
||||
return True, msg
|
||||
else:
|
||||
# 自动加时失败，返回错误信息和手动领取提示
|
||||
msg = (f"🏆 恭喜解锁成就奖励！\n"
|
||||
f"获得奖励：蛋定助手{reward_type}一张\n"
|
||||
f"⚠️自动加时失败: {message}\n"
|
||||
f"请联系管理员手动领取奖励！")
|
||||
import asyncio
|
||||
import requests
|
||||
import json
|
||||
import logging
|
||||
from typing import Dict, Optional, Tuple
|
||||
from nonebot import logger
|
||||
from .config import Config
|
||||
|
||||
_sync_logger = logging.getLogger("onmyoji_gacha.api_utils")
|
||||
|
||||
def mask_username(username: str) -> str:
|
||||
"""
|
||||
对用户名进行脱敏处理，只显示前两位和后两位，中间用*号隐藏
|
||||
|
||||
Args:
|
||||
username: 原始用户名
|
||||
|
||||
Returns:
|
||||
脱敏后的用户名
|
||||
"""
|
||||
if not username:
|
||||
return username
|
||||
|
||||
# 如果用户名长度小于等于4，直接显示前两位和后两位（可能重叠）
|
||||
if len(username) <= 4:
|
||||
return username
|
||||
|
||||
# 显示前两位和后两位，中间用*号填充
|
||||
return f"{username[:2]}{'*' * (len(username) - 4)}{username[-2:]}"
|
||||
|
||||
# 获取配置
|
||||
config = Config()
|
||||
|
||||
# API 端点配置
|
||||
DD_API_HOST = "https://api.danding.vip/DD/" # 蛋定服务器连接地址
|
||||
BOT_TOKEN = "3340e353a49447f1be640543cbdcd937" # 对接服务器的Token
|
||||
BOT_USER_ID = "1424473282" # 机器人用户ID
|
||||
|
||||
async def query_qq_binding(qq: str) -> Tuple[bool, Optional[str], Optional[str]]:
|
||||
"""
|
||||
查询QQ号是否绑定了蛋定用户名
|
||||
|
||||
Args:
|
||||
qq: 要查询的QQ号
|
||||
|
||||
Returns:
|
||||
Tuple[是否绑定, 用户名, VIP到期时间]
|
||||
"""
|
||||
try:
|
||||
url = f"{DD_API_HOST}query_qq_binding"
|
||||
data = {"qq": qq}
|
||||
|
||||
response = await asyncio.to_thread(requests.post, url=url, json=data)
|
||||
logger.debug(f"查询QQ绑定状态响应: {response}")
|
||||
|
||||
if response.status_code != 200:
|
||||
logger.error(f"查询QQ绑定状态失败，状态码: {response.status_code}")
|
||||
return False, None, None
|
||||
|
||||
result = response.json()
|
||||
logger.debug(f"查询QQ绑定状态结果: {result}")
|
||||
|
||||
if result.get("code") == 200:
|
||||
data = result.get("data", {})
|
||||
is_bound = data.get("is_bound", False)
|
||||
|
||||
if is_bound:
|
||||
username = data.get("username")
|
||||
vip_time = data.get("vip_time")
|
||||
return True, username, vip_time
|
||||
else:
|
||||
return False, None, None
|
||||
else:
|
||||
logger.error(f"查询QQ绑定状态失败，错误信息: {result.get('message')}")
|
||||
return False, None, None
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"查询QQ绑定状态异常: {str(e)}")
|
||||
return False, None, None
|
||||
|
||||
async def add_user_viptime(username: str, time_class: str = "Day", count: int = 1) -> Tuple[bool, str]:
|
||||
"""
|
||||
为用户添加VIP时间
|
||||
|
||||
Args:
|
||||
username: 蛋定用户名
|
||||
time_class: 时间类型 (Hour/Day/Week/Month/Season/Year)
|
||||
count: 添加次数（默认为1）
|
||||
|
||||
Returns:
|
||||
Tuple[是否成功, 响应消息]
|
||||
"""
|
||||
try:
|
||||
url = f"{DD_API_HOST}bot_add_user_viptime"
|
||||
|
||||
# 如果count大于1，需要多次调用API
|
||||
success_count = 0
|
||||
last_message = ""
|
||||
|
||||
for i in range(count):
|
||||
data = {
|
||||
"user": BOT_USER_ID,
|
||||
"token": BOT_TOKEN,
|
||||
"username": username,
|
||||
"classes": time_class
|
||||
}
|
||||
|
||||
response = await asyncio.to_thread(requests.post, url=url, json=data)
|
||||
logger.debug(f"添加VIP时间响应({i+1}/{count}): {response}")
|
||||
|
||||
if response.status_code != 200:
|
||||
error_msg = f"添加VIP时间失败({i+1}/{count})，状态码: {response.status_code}"
|
||||
logger.error(error_msg)
|
||||
continue
|
||||
|
||||
result = response.json()
|
||||
logger.debug(f"添加VIP时间结果({i+1}/{count}): {result}")
|
||||
|
||||
if result.get("code") == 200:
|
||||
success_count += 1
|
||||
last_message = result.get("msg", "添加VIP时间成功")
|
||||
else:
|
||||
error_msg = result.get("msg", "添加VIP时间失败")
|
||||
logger.error(f"添加VIP时间失败({i+1}/{count}): {error_msg}")
|
||||
|
||||
if success_count == count:
|
||||
return True, f"成功添加{count}次{time_class}时长。{last_message}"
|
||||
elif success_count > 0:
|
||||
return False, f"仅成功添加{success_count}/{count}次{time_class}时长。{last_message}"
|
||||
else:
|
||||
return False, f"添加{count}次{time_class}时长全部失败。"
|
||||
|
||||
except Exception as e:
|
||||
error_msg = f"添加VIP时间异常: {str(e)}"
|
||||
logger.error(error_msg)
|
||||
return False, error_msg
|
||||
|
||||
async def process_ssr_sp_reward(user_id: str, count: int = 1) -> Tuple[bool, str]:
|
||||
"""
|
||||
处理SSR/SP奖励发放
|
||||
|
||||
Args:
|
||||
user_id: QQ用户ID
|
||||
count: 奖励数量（默认为1）
|
||||
|
||||
Returns:
|
||||
Tuple[是否自动发放成功, 消息内容]
|
||||
"""
|
||||
# 查询QQ绑定状态
|
||||
is_bound, username, vip_time = await query_qq_binding(user_id)
|
||||
|
||||
if not is_bound:
|
||||
# 用户未绑定，返回提示信息
|
||||
if count == 1:
|
||||
msg = (f"🎉恭喜您抽中了SSR/SP稀有度式神！🎉\n"
|
||||
f"获得奖励：蛋定助手天卡一张\n"
|
||||
f"获取奖励请联系管理员，或前往蛋定云服务中绑定QQ号即可体验自动加时！")
|
||||
else:
|
||||
msg = (f"🎉恭喜您抽中了{count}张SSR/SP稀有度式神！🎉\n"
|
||||
f"获得奖励：蛋定助手天卡{count}张\n"
|
||||
f"获取奖励请联系管理员，或前往蛋定云服务中绑定QQ号即可体验自动加时！")
|
||||
return False, msg
|
||||
else:
|
||||
# 用户已绑定，自动加时
|
||||
success, message = await add_user_viptime(username, "Day", count)
|
||||
|
||||
if success:
|
||||
masked_username = mask_username(username)
|
||||
if count == 1:
|
||||
msg = (f"🎉恭喜您抽中了SSR/SP稀有度式神！🎉\n"
|
||||
f"🎁已自动为您的蛋定账号({masked_username})添加天卡时长！\n"
|
||||
f"📅到期时间: {message.split('到期时间为:')[-1] if '到期时间为:' in message else '请查看您的账号详情'}")
|
||||
else:
|
||||
msg = (f"🎉恭喜您抽中了{count}张SSR/SP稀有度式神！🎉\n"
|
||||
f"🎁已自动为您的蛋定账号({masked_username})添加{count}天卡时长！\n"
|
||||
f"📅到期时间: {message.split('到期时间为:')[-1] if '到期时间为:' in message else '请查看您的账号详情'}")
|
||||
return True, msg
|
||||
else:
|
||||
# 自动加时失败，返回错误信息和手动领取提示
|
||||
if count == 1:
|
||||
msg = (f"🎉恭喜您抽中了SSR/SP稀有度式神！🎉\n"
|
||||
f"获得奖励：蛋定助手天卡一张\n"
|
||||
f"⚠️自动加时失败: {message}\n"
|
||||
f"请联系管理员手动领取奖励！")
|
||||
else:
|
||||
msg = (f"🎉恭喜您抽中了{count}张SSR/SP稀有度式神！🎉\n"
|
||||
f"获得奖励：蛋定助手天卡{count}张\n"
|
||||
f"⚠️自动加时失败: {message}\n"
|
||||
f"请联系管理员手动领取奖励！")
|
||||
return False, msg
|
||||
|
||||
async def process_achievement_reward(user_id: str, achievement_id: str) -> Tuple[bool, str]:
|
||||
"""
|
||||
处理成就奖励发放
|
||||
|
||||
Args:
|
||||
user_id: QQ用户ID
|
||||
achievement_id: 成就ID
|
||||
|
||||
Returns:
|
||||
Tuple[是否自动发放成功, 消息内容]
|
||||
"""
|
||||
# 获取成就配置
|
||||
achievement_config = config.ACHIEVEMENTS.get(achievement_id)
|
||||
if not achievement_config:
|
||||
# 检查是否是重复奖励
|
||||
if "_repeat_" in achievement_id:
|
||||
base_achievement_id = achievement_id.split("_repeat_")[0]
|
||||
base_config = config.ACHIEVEMENTS.get(base_achievement_id)
|
||||
if base_config:
|
||||
reward_type = base_config.get("repeat_reward", "天卡")
|
||||
else:
|
||||
reward_type = "天卡"
|
||||
else:
|
||||
return False, f"未找到成就配置: {achievement_id}"
|
||||
else:
|
||||
reward_type = achievement_config.get("reward", "天卡")
|
||||
|
||||
# 查询QQ绑定状态
|
||||
is_bound, username, vip_time = await query_qq_binding(user_id)
|
||||
|
||||
if not is_bound:
|
||||
# 用户未绑定，返回提示信息
|
||||
msg = (f"🏆 恭喜解锁成就奖励！\n"
|
||||
f"获得奖励：蛋定助手{reward_type}一张\n"
|
||||
f"获取奖励请联系管理员，或前往蛋定云服务中绑定QQ号即可体验自动加时！")
|
||||
return False, msg
|
||||
else:
|
||||
# 用户已绑定，自动加时
|
||||
# 将奖励类型转换为API需要的时间类型
|
||||
time_class = "Day" # 默认为天卡
|
||||
if "周卡" in reward_type:
|
||||
time_class = "Week"
|
||||
elif "月卡" in reward_type:
|
||||
time_class = "Month"
|
||||
|
||||
success, message = await add_user_viptime(username, time_class)
|
||||
|
||||
if success:
|
||||
masked_username = mask_username(username)
|
||||
msg = (f"🏆 恭喜解锁成就奖励！\n"
|
||||
f"🎁已自动为您的蛋定账号({masked_username})添加{reward_type}时长！\n"
|
||||
f"📅到期时间: {message.split('到期时间为:')[-1] if '到期时间为:' in message else '请查看您的账号详情'}")
|
||||
return True, msg
|
||||
else:
|
||||
# 自动加时失败，返回错误信息和手动领取提示
|
||||
msg = (f"🏆 恭喜解锁成就奖励！\n"
|
||||
f"获得奖励：蛋定助手{reward_type}一张\n"
|
||||
f"⚠️自动加时失败: {message}\n"
|
||||
f"请联系管理员手动领取奖励！")
|
||||
return False, msg
|
||||
@@ -1,202 +1,200 @@
|
||||
"""
|
||||
onmyoji_gacha 插件的 Web API 接口
|
||||
使用 NoneBot 内置的 FastAPI 适配器提供管理员后台接口
|
||||
"""
|
||||
import os
|
||||
from typing import Dict, List, Any, Optional
|
||||
from fastapi import APIRouter, Depends, HTTPException, Header, Request
|
||||
from fastapi.responses import HTMLResponse, JSONResponse
|
||||
from fastapi.staticfiles import StaticFiles
|
||||
from fastapi.templating import Jinja2Templates
|
||||
from pydantic import BaseModel
|
||||
|
||||
from nonebot import get_driver
|
||||
from .config import Config
|
||||
from .gacha import GachaSystem
|
||||
|
||||
# 创建配置实例
|
||||
config = Config()
|
||||
gacha_system = GachaSystem()
|
||||
|
||||
# 创建 FastAPI 路由
|
||||
router = APIRouter(prefix="/onmyoji_gacha", tags=["onmyoji_gacha"])
|
||||
|
||||
# 设置模板目录
|
||||
templates = Jinja2Templates(directory="danding_bot/plugins/onmyoji_gacha/templates")
|
||||
|
||||
# 依赖：验证管理员权限
|
||||
async def verify_admin_token(authorization: Optional[str] = Header(None)):
|
||||
"""验证管理员权限"""
|
||||
print(f"🔐 验证管理员令牌: {authorization}")
|
||||
|
||||
if not authorization:
|
||||
print("❌ 缺少认证令牌")
|
||||
raise HTTPException(status_code=401, detail="缺少认证令牌")
|
||||
|
||||
token = authorization.replace("Bearer ", "")
|
||||
print(f"🔑 提取的令牌: {token}")
|
||||
print(f"🎯 期望的令牌: {config.WEB_ADMIN_TOKEN}")
|
||||
|
||||
if token != config.WEB_ADMIN_TOKEN:
|
||||
print("❌ 令牌验证失败")
|
||||
raise HTTPException(status_code=403, detail="无效的认证令牌")
|
||||
|
||||
print("✅ 令牌验证成功")
|
||||
return True
|
||||
|
||||
# API 响应模型
|
||||
class DailyStatsResponse(BaseModel):
|
||||
success: bool
|
||||
date: str
|
||||
stats: Dict[str, Any]
|
||||
|
||||
class UserStatsResponse(BaseModel):
|
||||
success: bool
|
||||
user_id: str
|
||||
total_draws: int
|
||||
R_count: int
|
||||
SR_count: int
|
||||
SSR_count: int
|
||||
SP_count: int
|
||||
recent_draws: List[Dict[str, str]]
|
||||
|
||||
class RankListResponse(BaseModel):
|
||||
success: bool
|
||||
data: List[Dict[str, Any]]
|
||||
|
||||
class AchievementResponse(BaseModel):
|
||||
success: bool
|
||||
user_id: str
|
||||
achievements: Dict[str, Any]
|
||||
progress: Dict[str, Any]
|
||||
|
||||
class DailyDetailedRecordsResponse(BaseModel):
|
||||
success: bool
|
||||
date: str
|
||||
records: List[Dict[str, Any]]
|
||||
total_count: int
|
||||
|
||||
# 管理后台页面
|
||||
@router.get("/admin", response_class=HTMLResponse)
|
||||
async def admin_page(request: Request):
|
||||
"""管理后台页面"""
|
||||
return templates.TemplateResponse("admin.html", {"request": request})
|
||||
|
||||
# API 端点
|
||||
@router.get("/api/stats/daily", response_model=DailyStatsResponse, dependencies=[Depends(verify_admin_token)])
|
||||
async def get_daily_stats():
|
||||
"""获取今日抽卡统计"""
|
||||
result = gacha_system.get_daily_stats()
|
||||
if not result["success"]:
|
||||
return result
|
||||
|
||||
return {
|
||||
"success": True,
|
||||
"date": result["date"],
|
||||
"stats": result["stats"]
|
||||
}
|
||||
|
||||
@router.get("/api/stats/user/{user_id}", response_model=UserStatsResponse, dependencies=[Depends(verify_admin_token)])
|
||||
async def get_user_stats(user_id: str):
|
||||
"""获取用户抽卡统计"""
|
||||
result = gacha_system.get_user_stats(user_id)
|
||||
if not result["success"]:
|
||||
return {
|
||||
"success": False,
|
||||
"user_id": user_id,
|
||||
"total_draws": 0,
|
||||
"R_count": 0,
|
||||
"SR_count": 0,
|
||||
"SSR_count": 0,
|
||||
"SP_count": 0,
|
||||
"recent_draws": []
|
||||
}
|
||||
|
||||
return {
|
||||
"success": True,
|
||||
"user_id": user_id,
|
||||
"total_draws": result["total_draws"],
|
||||
"R_count": result["R_count"],
|
||||
"SR_count": result["SR_count"],
|
||||
"SSR_count": result["SSR_count"],
|
||||
"SP_count": result["SP_count"],
|
||||
"recent_draws": result["recent_draws"]
|
||||
}
|
||||
|
||||
@router.get("/api/stats/rank", response_model=RankListResponse, dependencies=[Depends(verify_admin_token)])
|
||||
async def get_rank_list():
|
||||
"""获取抽卡排行榜"""
|
||||
rank_data = gacha_system.get_rank_list()
|
||||
|
||||
# 转换数据格式
|
||||
formatted_data = []
|
||||
for user_id, stats in rank_data:
|
||||
formatted_data.append({
|
||||
"user_id": user_id,
|
||||
"total_draws": stats["total_draws"],
|
||||
"R_count": stats["R_count"],
|
||||
"SR_count": stats["SR_count"],
|
||||
"SSR_count": stats["SSR_count"],
|
||||
"SP_count": stats["SP_count"],
|
||||
"ssr_sp_total": stats["SSR_count"] + stats["SP_count"]
|
||||
})
|
||||
|
||||
return {
|
||||
"success": True,
|
||||
"data": formatted_data
|
||||
}
|
||||
|
||||
@router.get("/api/achievements/{user_id}", response_model=AchievementResponse, dependencies=[Depends(verify_admin_token)])
|
||||
async def get_user_achievements(user_id: str):
|
||||
"""获取用户成就信息"""
|
||||
result = gacha_system.get_user_achievements(user_id)
|
||||
if not result["success"]:
|
||||
return {
|
||||
"success": False,
|
||||
"user_id": user_id,
|
||||
"achievements": {},
|
||||
"progress": {}
|
||||
}
|
||||
|
||||
return {
|
||||
"success": True,
|
||||
"user_id": user_id,
|
||||
"achievements": result["achievements"],
|
||||
"progress": result["progress"]
|
||||
}
|
||||
|
||||
@router.get("/api/records/daily", response_model=DailyDetailedRecordsResponse, dependencies=[Depends(verify_admin_token)])
|
||||
async def get_daily_detailed_records(date: Optional[str] = None):
|
||||
"""获取每日详细抽卡记录"""
|
||||
result = gacha_system.get_daily_detailed_records(date)
|
||||
if not result["success"]:
|
||||
return {
|
||||
"success": False,
|
||||
"date": date or gacha_system.data_manager.get_today_date(),
|
||||
"records": [],
|
||||
"total_count": 0
|
||||
}
|
||||
|
||||
return {
|
||||
"success": True,
|
||||
"date": result["date"],
|
||||
"records": result["records"],
|
||||
"total_count": result["total_count"]
|
||||
}
|
||||
|
||||
# 注册路由到 NoneBot 的 FastAPI 应用
|
||||
# 将在插件加载时由 __init__.py 调用
|
||||
def register_web_routes():
|
||||
"""注册 Web 路由到 NoneBot 的 FastAPI 应用"""
|
||||
try:
|
||||
from nonebot import get_driver
|
||||
driver = get_driver()
|
||||
# 获取 FastAPI 应用实例
|
||||
app = driver.server_app
|
||||
# 注册路由
|
||||
app.include_router(router)
|
||||
print("✅ onmyoji_gacha Web API 路由注册成功")
|
||||
return True
|
||||
except Exception as e:
|
||||
print(f"❌ 注册 Web 路由时出错: {e}")
|
||||
"""
|
||||
onmyoji_gacha 插件的 Web API 接口
|
||||
使用 NoneBot 内置的 FastAPI 适配器提供管理员后台接口
|
||||
"""
|
||||
import os
|
||||
from typing import Dict, List, Any, Optional
|
||||
from fastapi import APIRouter, Depends, HTTPException, Header, Request
|
||||
from fastapi.responses import HTMLResponse, JSONResponse
|
||||
from fastapi.staticfiles import StaticFiles
|
||||
from fastapi.templating import Jinja2Templates
|
||||
from pydantic import BaseModel
|
||||
|
||||
from nonebot import get_driver, logger
|
||||
from .config import Config
|
||||
from .gacha import GachaSystem
|
||||
|
||||
# 创建配置实例
|
||||
config = Config()
|
||||
gacha_system = GachaSystem()
|
||||
|
||||
# 创建 FastAPI 路由
|
||||
router = APIRouter(prefix="/onmyoji_gacha", tags=["onmyoji_gacha"])
|
||||
|
||||
# 设置模板目录
|
||||
templates = Jinja2Templates(directory="danding_bot/plugins/onmyoji_gacha/templates")
|
||||
|
||||
# 依赖：验证管理员权限
|
||||
async def verify_admin_token(authorization: Optional[str] = Header(None)):
|
||||
"""验证管理员权限"""
|
||||
if not authorization:
|
||||
raise HTTPException(status_code=401, detail="缺少认证令牌")
|
||||
|
||||
# 支持 "Bearer xxx" 和直接 "xxx" 两种格式
|
||||
if authorization.startswith("Bearer "):
|
||||
token = authorization[7:]
|
||||
else:
|
||||
token = authorization
|
||||
|
||||
if token != config.WEB_ADMIN_TOKEN:
|
||||
logger.warning("管理员令牌验证失败")
|
||||
raise HTTPException(status_code=403, detail="无效的认证令牌")
|
||||
|
||||
return True
|
||||
|
||||
# API 响应模型
|
||||
class DailyStatsResponse(BaseModel):
|
||||
success: bool
|
||||
date: str
|
||||
stats: Dict[str, Any]
|
||||
|
||||
class UserStatsResponse(BaseModel):
|
||||
success: bool
|
||||
user_id: str
|
||||
total_draws: int
|
||||
R_count: int
|
||||
SR_count: int
|
||||
SSR_count: int
|
||||
SP_count: int
|
||||
recent_draws: List[Dict[str, str]]
|
||||
|
||||
class RankListResponse(BaseModel):
|
||||
success: bool
|
||||
data: List[Dict[str, Any]]
|
||||
|
||||
class AchievementResponse(BaseModel):
|
||||
success: bool
|
||||
user_id: str
|
||||
achievements: Dict[str, Any]
|
||||
progress: Dict[str, Any]
|
||||
|
||||
class DailyDetailedRecordsResponse(BaseModel):
|
||||
success: bool
|
||||
date: str
|
||||
records: List[Dict[str, Any]]
|
||||
total_count: int
|
||||
|
||||
# 管理后台页面
|
||||
@router.get("/admin", response_class=HTMLResponse)
|
||||
async def admin_page(request: Request):
|
||||
"""管理后台页面"""
|
||||
return templates.TemplateResponse("admin.html", {"request": request})
|
||||
|
||||
# API 端点
|
||||
@router.get("/api/stats/daily", response_model=DailyStatsResponse, dependencies=[Depends(verify_admin_token)])
|
||||
async def get_daily_stats():
|
||||
"""获取今日抽卡统计"""
|
||||
result = gacha_system.get_daily_stats()
|
||||
if not result["success"]:
|
||||
return result
|
||||
|
||||
return {
|
||||
"success": True,
|
||||
"date": result["date"],
|
||||
"stats": result["stats"]
|
||||
}
|
||||
|
||||
@router.get("/api/stats/user/{user_id}", response_model=UserStatsResponse, dependencies=[Depends(verify_admin_token)])
|
||||
async def get_user_stats(user_id: str):
|
||||
"""获取用户抽卡统计"""
|
||||
result = gacha_system.get_user_stats(user_id)
|
||||
if not result["success"]:
|
||||
return {
|
||||
"success": False,
|
||||
"user_id": user_id,
|
||||
"total_draws": 0,
|
||||
"R_count": 0,
|
||||
"SR_count": 0,
|
||||
"SSR_count": 0,
|
||||
"SP_count": 0,
|
||||
"recent_draws": []
|
||||
}
|
||||
|
||||
return {
|
||||
"success": True,
|
||||
"user_id": user_id,
|
||||
"total_draws": result["total_draws"],
|
||||
"R_count": result["R_count"],
|
||||
"SR_count": result["SR_count"],
|
||||
"SSR_count": result["SSR_count"],
|
||||
"SP_count": result["SP_count"],
|
||||
"recent_draws": result["recent_draws"]
|
||||
}
|
||||
|
||||
@router.get("/api/stats/rank", response_model=RankListResponse, dependencies=[Depends(verify_admin_token)])
|
||||
async def get_rank_list():
|
||||
"""获取抽卡排行榜"""
|
||||
rank_data = gacha_system.get_rank_list()
|
||||
|
||||
# 转换数据格式
|
||||
formatted_data = []
|
||||
for user_id, stats in rank_data:
|
||||
formatted_data.append({
|
||||
"user_id": user_id,
|
||||
"total_draws": stats["total_draws"],
|
||||
"R_count": stats["R_count"],
|
||||
"SR_count": stats["SR_count"],
|
||||
"SSR_count": stats["SSR_count"],
|
||||
"SP_count": stats["SP_count"],
|
||||
"ssr_sp_total": stats["SSR_count"] + stats["SP_count"]
|
||||
})
|
||||
|
||||
return {
|
||||
"success": True,
|
||||
"data": formatted_data
|
||||
}
|
||||
|
||||
@router.get("/api/achievements/{user_id}", response_model=AchievementResponse, dependencies=[Depends(verify_admin_token)])
|
||||
async def get_user_achievements(user_id: str):
|
||||
"""获取用户成就信息"""
|
||||
result = gacha_system.get_user_achievements(user_id)
|
||||
if not result["success"]:
|
||||
return {
|
||||
"success": False,
|
||||
"user_id": user_id,
|
||||
"achievements": {},
|
||||
"progress": {}
|
||||
}
|
||||
|
||||
return {
|
||||
"success": True,
|
||||
"user_id": user_id,
|
||||
"achievements": result["achievements"],
|
||||
"progress": result["progress"]
|
||||
}
|
||||
|
||||
@router.get("/api/records/daily", response_model=DailyDetailedRecordsResponse, dependencies=[Depends(verify_admin_token)])
|
||||
async def get_daily_detailed_records(date: Optional[str] = None):
|
||||
"""获取每日详细抽卡记录"""
|
||||
result = gacha_system.get_daily_detailed_records(date)
|
||||
if not result["success"]:
|
||||
return {
|
||||
"success": False,
|
||||
"date": date or gacha_system.data_manager.get_today_date(),
|
||||
"records": [],
|
||||
"total_count": 0
|
||||
}
|
||||
|
||||
return {
|
||||
"success": True,
|
||||
"date": result["date"],
|
||||
"records": result["records"],
|
||||
"total_count": result["total_count"]
|
||||
}
|
||||
|
||||
# 注册路由到 NoneBot 的 FastAPI 应用
|
||||
# 将在插件加载时由 __init__.py 调用
|
||||
def register_web_routes():
|
||||
"""注册 Web 路由到 NoneBot 的 FastAPI 应用"""
|
||||
try:
|
||||
from nonebot import get_driver
|
||||
driver = get_driver()
|
||||
# 获取 FastAPI 应用实例
|
||||
app = driver.server_app
|
||||
# 注册路由
|
||||
app.include_router(router)
|
||||
logger.info("✅ onmyoji_gacha Web API 路由注册成功")
|
||||
return True
|
||||
except Exception as e:
|
||||
logger.error(f"❌ 注册 Web 路由时出错: {e}")
|
||||
return False
|
||||
Reference in New Issue
Block a user