"""炼金系统游戏模块""" import random 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 AlchemyGame(BaseGame): """炼金系统游戏""" def __init__(self): """初始化游戏""" super().__init__() self.db = get_db() # 奖品池配置 - 确保数学期望等于消耗积分 self.prize_pool: List[Tuple[int, str, float, str]] = [ # (权重, 类型, 倍率, 描述) (500, "penalty", 0, "炼金失败"), (100, "penalty", -1, "炼金爆炸"), (100, "points", 0.1, "少量积分"), (390, "points", 0.5, "少量积分"), (500, "points", 1, "等值积分"), (390, "points", 2, "丰厚积分"), (200, "points", 5, "丰厚积分"), (9, "points", 10, "🌟 巨额积分"), (1, "points", 100, "💎 传说积分"), ] 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: 指令，如 ".alchemy", ".alchemy 10", ".alchemy 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_alchemy_help() # 默认：炼金抽奖 else: # 解析消耗积分数量 cost_points = 10 # 默认消耗10积分 if args.isdigit(): cost_points = int(args) return self._perform_alchemy(user_id, cost_points) except Exception as e: logger.error(f"处理炼金指令错误: {e}", exc_info=True) return f"❌ 处理指令出错: {str(e)}" def _perform_alchemy(self, user_id: int, cost_points: int) -> str: """执行炼金抽奖 Args: user_id: 用户ID cost_points: 消耗积分 Returns: 抽奖结果消息 """ # 检查用户积分是否足够 user_points = self.db.get_user_points(user_id) if user_points['points'] < cost_points: return f"❌ 积分不足！需要 {cost_points} 积分，当前可用 {user_points['points']} 积分" # 选择奖品池 prize_pool = self.prize_pool # 执行抽奖 reward = self._draw_prize(prize_pool) reward_points = int(reward['value']*cost_points) # 消费积分 if not self.db.consume_points(user_id, cost_points, "alchemy", f"炼金抽奖消耗"): return "❌ 积分消费失败，请稍后重试" # 处理奖励 if reward['type'] == 'points' and reward['value'] > 0: # 获得积分奖励 self.db.add_points(user_id, reward_points, "alchemy", f"炼金奖励") elif reward['type'] == 'penalty' and reward['value'] < 0: # 负面奖励（扣分） penalty_points = abs(reward_points) self.db.consume_points(user_id, penalty_points, "alchemy", f"炼金失败") # 炼金系统已简化，不再记录历史 # 获取更新后的积分信息 updated_points = self.db.get_user_points(user_id) # 格式化输出 text = f"## ⚗️ 炼金结果\n\n" text += f"**消耗积分**：{cost_points} 分\n\n" if reward['type'] == 'points': text += f"**{reward['description']}**: 获得{reward_points} 积分\n\n" elif reward['type'] == 'penalty': text += f"**{reward['description']}**: 损失 {abs(reward_points)} 积分\n\n" text += f"**当前积分**：{updated_points['points']} 分\n\n" text += "---\n\n" text += "💡 提示：炼金有风险，投资需谨慎！" return text def _draw_prize(self, prize_pool: list) -> dict: """从奖品池中抽取奖品 Args: prize_pool: 奖品池 Returns: 奖品信息 """ # 生成随机数 rand = random.random()*self.total_weight cumulative_prob = 0.0 for weight, reward_type, reward_value, description in prize_pool: cumulative_prob += weight if rand <= cumulative_prob: return { 'type': reward_type, 'value': reward_value, 'description': description } # 兜底返回第一个奖品 return { 'type': prize_pool[0][1], 'value': prize_pool[0][2], '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: """获取炼金帮助信息 Returns: 帮助信息消息 """ text = f"## ⚗️ 炼金系统\n\n" text += f"### 基础用法\n" text += f"- `.alchemy` - 消耗10积分进行炼金\n" text += f"- `.alchemy 10` - 消耗10积分进行炼金\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