366 lines
14 KiB
Python
366 lines
14 KiB
Python
"""冒险系统游戏模块"""
|
||
import random
|
||
import time
|
||
import logging
|
||
import re
|
||
from datetime import datetime
|
||
from games.base import BaseGame
|
||
from utils.parser import CommandParser
|
||
from core.database import get_db
|
||
from typing import *
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
class AdventureGame(BaseGame):
|
||
"""冒险系统游戏"""
|
||
|
||
def __init__(self):
|
||
"""初始化游戏"""
|
||
super().__init__()
|
||
self.db = get_db()
|
||
|
||
# 奖品池配置
|
||
self.prize_pool: List[Tuple[int, float, str]] = [
|
||
# (权重, 倍率, 描述)
|
||
(300, 1, "少量积分"),
|
||
(250, 2, "中等积分"),
|
||
(200, 2, "大量积分"),
|
||
(150, 5, "丰厚积分"),
|
||
(100, 10, "丰厚积分"),
|
||
(50, 100, "🌟 巨额积分"),
|
||
(10, 1000, "💎 传说积分"),
|
||
]
|
||
|
||
self.total_weight: int = 0
|
||
for weight,_,_ in self.prize_pool:
|
||
self.total_weight += weight
|
||
|
||
def _parse_time_string(self, time_str: str) -> int:
|
||
"""解析时间字符串，支持 h/m/s 格式
|
||
|
||
支持的格式示例：
|
||
- "1h30m10s" -> 5410秒
|
||
- "30m" -> 1800秒
|
||
- "10s" -> 10秒
|
||
- "1h30m" -> 5400秒
|
||
- "3600" -> 3600秒（纯数字，按秒处理）
|
||
|
||
Args:
|
||
time_str: 时间字符串
|
||
|
||
Returns:
|
||
解析后的秒数，如果解析失败返回None
|
||
"""
|
||
if not time_str:
|
||
return None
|
||
|
||
# 如果是纯数字，直接返回
|
||
if time_str.isdigit():
|
||
return int(time_str)
|
||
|
||
# 使用正则表达式匹配 h/m/s 格式，确保整个字符串匹配
|
||
pattern = r'^(?:(\d+)h)?(?:(\d+)m)?(?:(\d+)s)?$'
|
||
match = re.match(pattern, time_str.lower())
|
||
|
||
if not match:
|
||
return None
|
||
|
||
hours = int(match.group(1) or 0)
|
||
minutes = int(match.group(2) or 0)
|
||
seconds = int(match.group(3) or 0)
|
||
|
||
# 如果所有值都是0，返回None
|
||
if hours == 0 and minutes == 0 and seconds == 0:
|
||
return None
|
||
|
||
total_seconds = hours * 3600 + minutes * 60 + seconds
|
||
return total_seconds
|
||
|
||
def _format_time(self, seconds: int) -> str:
|
||
"""将秒数格式化为 "X时X分X秒" 格式
|
||
|
||
Args:
|
||
seconds: 秒数
|
||
|
||
Returns:
|
||
格式化的时间字符串，如 "1时30分10秒"、"30分10秒"、"10秒"
|
||
"""
|
||
if seconds < 0:
|
||
seconds = 0
|
||
|
||
hours = seconds // 3600
|
||
minutes = (seconds % 3600) // 60
|
||
secs = seconds % 60
|
||
|
||
parts = []
|
||
if hours > 0:
|
||
parts.append(f"{hours}时")
|
||
if minutes > 0:
|
||
parts.append(f"{minutes}分")
|
||
if secs > 0 or not parts:
|
||
parts.append(f"{secs}秒")
|
||
|
||
return "".join(parts)
|
||
|
||
async def handle(self, command: str, chat_id: int, user_id: int) -> str:
|
||
"""处理冒险相关指令
|
||
|
||
Args:
|
||
command: 指令，如 ".adventure", ".adventure 10", ".adventure stats"
|
||
chat_id: 会话ID
|
||
user_id: 用户ID
|
||
|
||
Returns:
|
||
回复消息
|
||
"""
|
||
try:
|
||
# 提取参数
|
||
_, args = CommandParser.extract_command_args(command)
|
||
args = args.strip().lower()
|
||
|
||
|
||
# 冒险说明
|
||
if args in ['help', '帮助', 'info']:
|
||
return self._get_adventure_help()
|
||
|
||
# 放弃当前冒险（按最低倍率结算已冒险时间）
|
||
if args in ['abandon', '放弃']:
|
||
return await self._abandon_adventure(chat_id, user_id)
|
||
|
||
# 默认：冒险耗时1秒
|
||
else:
|
||
# 解析消耗时间
|
||
cost_time = 1 # 默认消耗1秒
|
||
if args:
|
||
parsed_time = self._parse_time_string(args)
|
||
if parsed_time is not None:
|
||
cost_time = parsed_time
|
||
else:
|
||
return f"❌ 时间格式错误！请使用以下格式：\n- 纯数字（秒）：`.adventure 60`\n- 时分秒格式：`.adventure 1h30m10s`\n- 分钟秒格式：`.adventure 30m10s`\n- 只有秒：`.adventure 10s`"
|
||
|
||
return await self._perform_adventure(chat_id, user_id, cost_time)
|
||
|
||
except Exception as e:
|
||
logger.error(f"处理冒险指令错误: {e}", exc_info=True)
|
||
return f"❌ 处理指令出错: {str(e)}"
|
||
|
||
async def _perform_adventure(self, chat_id: int, user_id: int, cost_time: int) -> str:
|
||
"""执行冒险耗时
|
||
|
||
Args:
|
||
chat_id: 会话ID（使用0作为用户级标识）
|
||
user_id: 用户ID
|
||
cost_time: 消耗时间（秒）
|
||
|
||
Returns:
|
||
抽奖结果消息
|
||
"""
|
||
# 参数验证
|
||
if cost_time < 1:
|
||
return "❌ 冒险时间至少需要1秒！"
|
||
|
||
# 查询冒险状态（使用chat_id=0表示用户级状态）
|
||
state = self.db.get_game_state(0, user_id, 'adventure')
|
||
current_time = int(time.time())
|
||
|
||
# 情况1：用户已有未完成的冒险
|
||
if state:
|
||
try:
|
||
state_data = state['state_data']
|
||
start_time = state_data.get('start_time', 0)
|
||
saved_cost_time = state_data.get('cost_time', 1)
|
||
end_time = start_time + saved_cost_time
|
||
remaining_seconds = end_time - current_time
|
||
|
||
# 情况1.1：冒险已完成（时间已到或过期）
|
||
if remaining_seconds <= 0:
|
||
# 选择奖品池
|
||
prize_pool = self.prize_pool
|
||
|
||
# 执行抽奖
|
||
reward = self._draw_prize(prize_pool)
|
||
reward_points = int(reward['value'] * saved_cost_time)
|
||
|
||
# 处理奖励
|
||
self.db.add_points(user_id, reward_points, "adventure", f"冒险奖励")
|
||
|
||
# 删除冒险状态
|
||
self.db.delete_game_state(0, user_id, 'adventure')
|
||
|
||
# 获取更新后的积分信息
|
||
updated_points = self.db.get_user_points(user_id)
|
||
|
||
# 格式化输出
|
||
time_str = self._format_time(saved_cost_time)
|
||
text = f"## ⚡️ 冒险结果\n\n"
|
||
text += f"**消耗时间**: {time_str}\n\n"
|
||
text += f"**{reward['description']}**: 获得 {reward_points} 积分\n\n"
|
||
text += f"**当前积分**: {updated_points['points']} 分\n\n"
|
||
text += "---\n\n"
|
||
text += "💡 提示：冒险进行时不能炼金！"
|
||
|
||
return text
|
||
|
||
# 情况1.2：冒险未完成，返回等待提示
|
||
wait_msg = self._format_time(remaining_seconds)
|
||
saved_time_str = self._format_time(saved_cost_time)
|
||
|
||
text = f"## ⚡️ 冒险进行中\n\n"
|
||
text += f"你正在进行一次冒险，还需等待 **{wait_msg}** 才能完成。\n\n"
|
||
text += f"**当前冒险时长**: {saved_time_str}\n\n"
|
||
text += "---\n\n"
|
||
text += "💡 提示：冒险期间无法进行炼金，请耐心等待！"
|
||
|
||
return text
|
||
|
||
except Exception as e:
|
||
# 状态数据异常，删除损坏状态并允许重新开始
|
||
logger.error(f"冒险状态数据异常: {e}", exc_info=True)
|
||
self.db.delete_game_state(0, user_id, 'adventure')
|
||
# 继续执行到情况3（开始新冒险）
|
||
|
||
# 情况3：用户没有冒险状态，开始新的冒险
|
||
state_data = {
|
||
'start_time': current_time,
|
||
'cost_time': cost_time
|
||
}
|
||
|
||
# 保存冒险状态
|
||
self.db.save_game_state(0, user_id, 'adventure', state_data)
|
||
|
||
# 计算预计完成时间
|
||
end_time = current_time + cost_time
|
||
end_datetime = datetime.fromtimestamp(end_time)
|
||
current_datetime = datetime.fromtimestamp(current_time)
|
||
# 判断是否跨天：如果完成日期和当前日期不同，或跨年，则显示完整日期时间
|
||
if (end_datetime.date() != current_datetime.date() or
|
||
end_datetime.year != current_datetime.year):
|
||
end_time_str = end_datetime.strftime('%Y-%m-%d %H:%M:%S')
|
||
else:
|
||
end_time_str = end_datetime.strftime('%H:%M:%S')
|
||
cost_time_str = self._format_time(cost_time)
|
||
|
||
text = f"## ⚡️ 冒险开始\n\n"
|
||
text += f"你已经开始了冒险之旅，本次冒险将持续 **{cost_time_str}**。\n\n"
|
||
text += f"**预计完成时间**: {end_time_str}\n\n"
|
||
text += "---\n\n"
|
||
text += "💡 提示：冒险期间无法进行炼金，完成后使用 `.adventure` 获取奖励！"
|
||
|
||
return text
|
||
|
||
async def _abandon_adventure(self, chat_id: int, user_id: int) -> str:
|
||
"""放弃当前冒险，按最低倍率结算已冒险时间
|
||
|
||
Args:
|
||
chat_id: 会话ID（使用0作为用户级标识）
|
||
user_id: 用户ID
|
||
|
||
Returns:
|
||
放弃结果消息
|
||
"""
|
||
try:
|
||
# 查询冒险状态
|
||
state = self.db.get_game_state(0, user_id, 'adventure')
|
||
if not state:
|
||
return "❌ 当前没有进行中的冒险，可使用 `.adventure` 开始新的冒险。"
|
||
|
||
state_data = state.get('state_data', {})
|
||
start_time = state_data.get('start_time')
|
||
cost_time = state_data.get('cost_time')
|
||
if start_time is None or cost_time is None:
|
||
# 状态异常，清理并提示
|
||
self.db.delete_game_state(0, user_id, 'adventure')
|
||
return "⚠️ 冒险状态异常已清理，请使用 `.adventure` 重新开始。"
|
||
|
||
current_time = int(time.time())
|
||
elapsed_seconds = max(0, current_time - int(start_time))
|
||
if elapsed_seconds < 1:
|
||
elapsed_seconds = 1
|
||
|
||
# 计算最低倍率
|
||
try:
|
||
min_multiplier = min(m for _, m, _ in self.prize_pool)
|
||
except Exception:
|
||
# 兜底：若奖池异常，按0.5处理
|
||
min_multiplier = 0.5
|
||
|
||
reward_points = int(min_multiplier * elapsed_seconds)
|
||
if reward_points < 0:
|
||
reward_points = 0
|
||
|
||
# 发放奖励并清理状态
|
||
if reward_points > 0:
|
||
self.db.add_points(user_id, reward_points, "adventure", "冒险放弃奖励")
|
||
self.db.delete_game_state(0, user_id, 'adventure')
|
||
|
||
# 查询当前积分
|
||
updated_points = self.db.get_user_points(user_id)
|
||
|
||
# 输出
|
||
elapsed_time_str = self._format_time(elapsed_seconds)
|
||
text = f"## ⚡️ 冒险放弃\n\n"
|
||
text += f"**已计入时间**: {elapsed_time_str}\n\n"
|
||
text += f"**最低倍率**: {min_multiplier} 倍\n\n"
|
||
text += f"**获得积分**: {reward_points} 分\n\n"
|
||
text += f"**当前积分**: {updated_points['points']} 分\n\n"
|
||
text += "---\n\n"
|
||
text += "💡 提示：可随时使用 `.adventure` 再次踏上冒险之旅！"
|
||
return text
|
||
except Exception as e:
|
||
logger.error(f"放弃冒险时出错: {e}", exc_info=True)
|
||
# 失败时不影响原状态，返回提示
|
||
return f"❌ 放弃冒险失败：{str(e)}"
|
||
|
||
def _draw_prize(self, prize_pool: list) -> dict:
|
||
"""从奖品池中抽取奖品
|
||
|
||
Args:
|
||
prize_pool: 奖品池，格式为 (权重, 倍率, 描述)
|
||
|
||
Returns:
|
||
奖品信息，格式为 {'value': 倍率, 'description': 描述}
|
||
"""
|
||
# 生成随机数
|
||
rand = random.random()*self.total_weight
|
||
cumulative_prob = 0.0
|
||
|
||
for weight, multiplier, description in prize_pool:
|
||
cumulative_prob += weight
|
||
if rand <= cumulative_prob:
|
||
return {
|
||
'value': multiplier,
|
||
'description': description
|
||
}
|
||
|
||
# 兜底返回第一个奖品
|
||
return {
|
||
'value': prize_pool[0][1],
|
||
'description': prize_pool[0][2]
|
||
}
|
||
|
||
def _get_adventure_help(self) -> str:
|
||
"""获取冒险帮助信息
|
||
|
||
Returns:
|
||
帮助信息消息
|
||
"""
|
||
text = f"## ⚡️ 冒险系统\n\n"
|
||
text += f"### 基础用法\n"
|
||
text += f"- `.adventure` - 消耗1秒进行冒险\n"
|
||
text += f"- `.adventure 60` - 消耗60秒进行冒险\n"
|
||
text += f"- `.adventure 1h30m10s` - 消耗1小时30分10秒进行冒险\n"
|
||
text += f"- `.adventure 30m` - 消耗30分钟进行冒险\n"
|
||
text += f"- `.adventure 10s` - 消耗10秒进行冒险\n\n"
|
||
text += f"**时间格式说明**：支持时分秒组合，如 `1h30m10s`、`30m`、`10s`，也支持纯数字（按秒计算）。\n\n"
|
||
text += f"### 其他功能\n"
|
||
text += f"- `.adventure abandon` - 放弃当前冒险，按最低倍率结算已冒险时间\n"
|
||
text += f"- `.adventure 放弃` - 放弃当前冒险，按最低倍率结算已冒险时间\n"
|
||
text += f"- `.adventure help` - 查看帮助\n\n"
|
||
|
||
return text
|
||
|
||
def get_help(self) -> str:
|
||
"""获取帮助信息"""
|
||
return self._get_adventure_help()
|