207 lines
7.5 KiB
Python
207 lines
7.5 KiB
Python
"""炼金系统游戏模块"""
|
||
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 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:
|
||
抽奖结果消息
|
||
"""
|
||
# 检查用户是否正在冒险中
|
||
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:
|
||
return f"❌ 你有待回收的冒险奖励！\n\n💡 请先使用 `.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:
|
||
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_help(self) -> str:
|
||
"""获取炼金帮助信息
|
||
|
||
Returns:
|
||
帮助信息消息
|
||
"""
|
||
text = f"## ⚗️ 炼金系统\n\n"
|
||
text += f"### 基础用法\n"
|
||
text += f"- `.alchemy` - 消耗10积分进行炼金\n"
|
||
text += f"- `.alchemy cost` - 消耗cost积分进行炼金, 最少1积分\n"
|
||
|
||
text += f"### 其他功能\n"
|
||
text += f"- `.alchemy help` - 查看帮助\n\n"
|
||
|
||
return text
|
||
|
||
def get_help(self) -> str:
|
||
"""获取帮助信息"""
|
||
return self._get_alchemy_help()
|
||
|