新增冒险系统

games/adventure.py

@@ -0,0 +1,218 @@
+"""冒险系统游戏模块"""
+import random
+import time
+import logging
+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]] = [
+            # (权重, 倍率, 描述) 
+            (500, 0.5, "少量积分"),       
+            (350, 1, "中等积分"),        
+            (200, 2, "大量积分"),      
+            (100, 5, "丰厚积分"),      
+            (50, 10, "丰厚积分"),    
+            (10, 100, "🌟 巨额积分"),   
+            (1, 1000, "💎 传说积分"),  
+        ]
+
+        self.total_weight: int = 0
+        for weight,_ in self.prize_pool:
+            self.total_weight += weight
+    
+    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()
+            
+            # 默认：冒险耗时1分钟
+            else:
+                # 解析消耗时间
+                cost_time = 1  # 默认消耗1分钟
+                if args.isdigit():
+                    cost_time = int(args)
+                
+                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 * 60
+                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)
+                    
+                    # 格式化输出
+                    text = f"## ⚡️ 冒险结果\n\n"
+                    text += f"**消耗时间**: {saved_cost_time} 分钟\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：冒险未完成，返回等待提示
+                remaining_minutes = remaining_seconds // 60
+                remaining_secs = remaining_seconds % 60
+                
+                if remaining_minutes > 0:
+                    wait_msg = f"{remaining_minutes} 分 {remaining_secs} 秒"
+                else:
+                    wait_msg = f"{remaining_secs} 秒"
+                
+                text = f"## ⚡️ 冒险进行中\n\n"
+                text += f"你正在进行一次冒险，还需等待 **{wait_msg}** 才能完成。\n\n"
+                text += f"**当前冒险时长**: {saved_cost_time} 分钟\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 * 60
+        end_datetime = datetime.fromtimestamp(end_time)
+        end_time_str = end_datetime.strftime('%H:%M:%S')
+        
+        text = f"## ⚡️ 冒险开始\n\n"
+        text += f"你已经开始了冒险之旅，本次冒险将持续 **{cost_time}** 分钟。\n\n"
+        text += f"**预计完成时间**: {end_time_str}\n\n"
+        text += "---\n\n"
+        text += "💡 提示：冒险期间无法进行炼金，完成后使用 `.adventure` 获取奖励！"
+        
+        return text
+    
+    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 time` - 消耗time分钟进行冒险, 最少一分钟\n"
+        
+        text += f"### 其他功能\n"
+        text += f"- `.adventure help` - 查看帮助\n\n"
+        
+        return text
+    
+    def get_help(self) -> str:
+        """获取帮助信息"""
+        return self._get_adventure_help()

games/alchemy.py

@@ -1,5 +1,6 @@
 """炼金系统游戏模块"""
 import random
+import time
 import logging
 from datetime import datetime
 from games.base import BaseGame
@@ -80,6 +81,36 @@ class AlchemyGame(BaseGame):
         Returns:
             抽奖结果消息
         """
+        # 检查用户是否正在冒险中
+        state = self.db.get_game_state(0, user_id, 'adventure')
+        if state:
+            try:
+                state_data = state['state_data']
+                start_time = state_data.get('start_time', 0)
+                cost_time = state_data.get('cost_time', 1)
+                current_time = int(time.time())
+                end_time = start_time + cost_time * 60
+                remaining_seconds = end_time - current_time
+                
+                # 如果冒险已完成，自动清理状态，允许炼金
+                if remaining_seconds <= 0:
+                    self.db.delete_game_state(0, user_id, 'adventure')
+                else:
+                    # 冒险未完成，返回错误提示
+                    remaining_minutes = remaining_seconds // 60
+                    remaining_secs = remaining_seconds % 60
+                    
+                    if remaining_minutes > 0:
+                        wait_msg = f"{remaining_minutes} 分 {remaining_secs} 秒"
+                    else:
+                        wait_msg = f"{remaining_secs} 秒"
+                    
+                    return f"❌ 冒险进行中，无法进行炼金！\n\n还需等待 **{wait_msg}** 才能完成冒险。\n\n💡 提示：冒险期间无法进行炼金，请先完成冒险！"
+            except Exception as e:
+                # 状态数据异常，删除损坏状态，允许继续
+                logger.error(f"冒险状态数据异常: {e}", exc_info=True)
+                self.db.delete_game_state(0, user_id, 'adventure')
+        
         # 检查用户积分是否足够
         user_points = self.db.get_user_points(user_id)
         if user_points['points'] < cost_points:
@@ -154,88 +185,6 @@ class AlchemyGame(BaseGame):
             'description': prize_pool[0][3]
         }
     
-    def _get_alchemy_stats(self, user_id: int) -> str:
-        """获取炼金统计信息
-        
-        Args:
-            user_id: 用户ID
-            
-        Returns:
-            统计信息消息
-        """
-        stats = self.db.get_alchemy_stats(user_id)
-        
-        if stats['total_draws'] == 0:
-            return "📊 你还没有进行过炼金抽奖哦～"
-        
-        text = f"## ⚗️ 炼金统计\n\n"
-        text += f"**总抽奖次数**：{stats['total_draws']} 次\n\n"
-        text += f"**总消耗积分**：{stats['total_cost']} 分\n\n"
-        text += f"**总获得积分**：{stats['total_points_gained']} 分\n\n"
-        
-        net_points = stats['net_points']
-        if net_points > 0:
-            text += f"**净收益**：+{net_points} 分 🎉\n\n"
-        elif net_points < 0:
-            text += f"**净收益**：{net_points} 分 😅\n\n"
-        else:
-            text += f"**净收益**：0 分 ⚖️\n\n"
-        
-        # 计算平均收益
-        avg_gain = stats['total_points_gained'] / stats['total_draws']
-        avg_cost = stats['total_cost'] / stats['total_draws']
-        
-        text += f"**平均消耗**：{avg_cost:.1f} 分/次\n\n"
-        text += f"**平均获得**：{avg_gain:.1f} 分/次\n\n"
-        text += "---\n\n"
-        text += "💡 提示：使用 `.alchemy records` 查看详细记录"
-        
-        return text
-    
-    def _get_alchemy_records(self, user_id: int, limit: int = 10) -> str:
-        """获取炼金记录
-        
-        Args:
-            user_id: 用户ID
-            limit: 限制数量
-            
-        Returns:
-            记录信息消息
-        """
-        records = self.db.get_alchemy_records(user_id, limit)
-        
-        if not records:
-            return "📝 暂无炼金记录"
-        
-        text = f"## ⚗️ 炼金记录（最近 {len(records)} 条）\n\n"
-        
-        for record in records:
-            timestamp = datetime.fromtimestamp(record['created_at']).strftime('%m-%d %H:%M')
-            cost = record['cost_points']
-            reward_type = record['reward_type']
-            reward_value = record['reward_value']
-            description = record['reward_description']
-            
-            text += f"**{timestamp}** 消耗 {cost} 分\n"
-            
-            if reward_type == 'points' and reward_value > 0:
-                if reward_value >= 100:
-                    emoji = "👑"
-                elif reward_value >= 50:
-                    emoji = "💎"
-                else:
-                    emoji = "🎁"
-                text += f"  {emoji} 获得 {reward_value} 积分 - {description}\n"
-            elif reward_type == 'penalty' and reward_value < 0:
-                penalty_points = abs(reward_value)
-                text += f"  💥 损失 {penalty_points} 积分 - {description}\n"
-            else:
-                text += f"  {description}\n"
-            
-            text += "\n"
-        
-        return text
-    
     def _get_alchemy_help(self) -> str:
         """获取炼金帮助信息
         
@@ -245,75 +194,14 @@ class AlchemyGame(BaseGame):
         text = f"## ⚗️ 炼金系统\n\n"
         text += f"### 基础用法\n"
         text += f"- `.alchemy` - 消耗10积分进行炼金\n"
-        text += f"- `.alchemy 10` - 消耗10积分进行炼金\n"
+        text += f"- `.alchemy cost` - 消耗cost积分进行炼金, 最少1积分\n"
-        text += f"- `.alchemy 20` - 消耗20积分进行炼金\n"
-        text += f"- `.alchemy 50` - 消耗50积分进行炼金\n\n"
         
         text += f"### 其他功能\n"
-        text += f"- `.alchemy stats` - 查看炼金统计\n"
-        text += f"- `.alchemy records` - 查看炼金记录\n"
         text += f"- `.alchemy help` - 查看帮助\n\n"
         
-        text += f"### 炼金说明\n"
-        text += f"- **消耗积分**：10、20、50 积分\n"
-        text += f"- **奖励类型**：积分奖励、负面惩罚\n"
-        text += f"- **大奖概率**：极小概率获得巨额积分\n"
-        text += f"- **风险提示**：小概率额外扣分\n"
-        text += f"- **数学期望**：略高于消耗积分，对玩家友好\n"
-        text += f"- **风险提示**：炼金有风险，投资需谨慎！\n\n"
-        
-        text += f"### 示例\n"
-        text += f"```\n"
-        text += f".alchemy\n"
-        text += f".alchemy 20\n"
-        text += f".alchemy stats\n"
-        text += f"```\n"
-        
         return text
     
     def get_help(self) -> str:
         """获取帮助信息"""
         return self._get_alchemy_help()
-    
+    
-    def calculate_expected_value(self, cost_points: int) -> float:
-        """计算奖品池的数学期望
-        
-        Args:
-            cost_points: 消耗积分
-            
-        Returns:
-            数学期望值
-        """
-        if cost_points == 10:
-            prize_pool = self.prize_pool_10
-        elif cost_points == 20:
-            prize_pool = self.prize_pool_20
-        elif cost_points == 50:
-            prize_pool = self.prize_pool_50
-        else:
-            return 0.0
-        
-        expected_value = 0.0
-        for prob, reward_type, reward_value, description in prize_pool:
-            if reward_type == 'points':
-                expected_value += prob * reward_value
-        
-        return expected_value
-    
-    def verify_prize_pools(self) -> dict:
-        """验证所有奖品池的数学期望
-        
-        Returns:
-            验证结果字典
-        """
-        results = {}
-        
-        for cost_points in [10, 20, 50]:
-            expected = self.calculate_expected_value(cost_points)
-            results[f"cost_{cost_points}"] = {
-                "expected_value": expected,
-                "is_fair": abs(expected - cost_points) < 0.01,  # 允许0.01的误差
-                "difference": expected - cost_points
-            }
-        
-        return results

routers/callback.py

@@ -175,6 +175,12 @@ async def handle_command(game_type: str, command: str,
             game = AlchemyGame()
             return await game.handle(command, chat_id, user_id)
         
+        # 冒险系统
+        if game_type == 'adventure':
+            from games.adventure import AdventureGame
+            game = AdventureGame()
+            return await game.handle(command, chat_id, user_id)
+        
         # 积分赠送系统
         if game_type == 'gift':
             from games.gift import GiftGame

utils/parser.py

@@ -55,6 +55,10 @@ class CommandParser:
         '.alchemy': 'alchemy',
         '.炼金': 'alchemy',
         
+        # 冒险系统
+        '.adventure': 'adventure',
+        '.冒险': 'adventure',
+        
         # 积分赠送系统
         '.gift': 'gift',
         '.赠送': 'gift',