ninemine/WPSBot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

积分系统

Browse Source
This commit is contained in:
LiuBai095
2025-10-29 11:32:43 +08:00
parent c05a8b3578
commit 7483a00a99
9 changed files with 1558 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

62
README.md
View File

@@ -43,6 +43,27 @@
- 完整的战绩统计
- 实时棋盘显示
### 💎 积分系统
- 每日签到获得固定积分（10分）
- 运势占卜随机获得积分（1-20分，30%概率）
- 个人积分查询和记录
- 积分排行榜
- 完整的积分变动记录
### ⚗️ 炼金系统
- 消耗积分进行抽奖（10/20/50积分）
- 奖品池数学期望略高于消耗积分，对玩家友好
- 包含大奖（巨额积分）和负面奖励（额外扣分）
- 完整的炼金记录和统计
- 支持多种消耗档位的抽奖
### 🎁 积分赠送系统
- 用户间积分赠送功能
- 支持附赠个性化消息
- 完整的赠送和接收记录
- 赠送统计和记录查询
- 单次最多赠送1000积分
## 🚀 快速开始
### 环境要求
@@ -197,6 +218,40 @@ GOMOKU_MAX_CONCURRENT_GAMES=5
.gomoku stats # 查看个人战绩
```
### 积分系统
```
.points # 查看个人积分
.积分 # 查看个人积分
.checkin # 每日签到
.签到 # 每日签到
.打卡 # 每日签到
.points leaderboard # 积分排行榜
.points records # 积分记录
```
### 炼金系统
```
.alchemy # 消耗10积分进行炼金
.炼金 # 消耗10积分进行炼金
.alchemy 20 # 消耗20积分进行炼金
.alchemy 50 # 消耗50积分进行炼金
.alchemy stats # 查看炼金统计
.alchemy records # 查看炼金记录
```
### 积分赠送系统
```
.gift 123 50 生日快乐 # 赠送50积分给用户123，附赠消息
.赠送 456 100 感谢帮助 # 赠送100积分给用户456，附赠消息
.送 789 200 # 赠送200积分给用户789
.gift stats # 查看赠送统计
.gift sent # 查看发送记录
.gift received # 查看接收记录
```
## 🏗️ 项目结构
```
@@ -225,7 +280,10 @@ WPSBotGame/
│ ├── quiz.py # 问答游戏
│ ├── idiom.py # 成语接龙
│ ├── gomoku.py # 五子棋
│ └── gomoku_logic.py # 五子棋逻辑
│ ├── gomoku_logic.py # 五子棋逻辑
│ ├── points.py # 积分系统
│ ├── alchemy.py # 炼金系统
│ └── gift.py # 积分赠送系统
├── data/ # 数据文件
│ ├── bot.db # SQLite数据库
│ ├── fortunes.json # 运势数据
@@ -324,7 +382,7 @@ curl http://localhost:8000/stats
- **并发支持**：5-10个同时请求
- **用户规模**：50-100个活跃用户
- **消息限制**：20条/分钟（WPS限制）
- **支持游戏**：7种游戏类型（骰子、石头剪刀布、运势、猜数字、问答、成语接龙、五子棋）
- **支持游戏**：10种游戏类型（骰子、石头剪刀布、运势、猜数字、问答、成语接龙、五子棋、积分系统、炼金系统、积分赠送系统）
## 🤝 贡献

563
core/database.py
View File

@@ -89,6 +89,97 @@ class Database:
)
""")
# 用户积分表
cursor.execute("""
CREATE TABLE IF NOT EXISTS user_points (
user_id INTEGER PRIMARY KEY,
total_points INTEGER DEFAULT 0,
available_points INTEGER DEFAULT 0,
created_at INTEGER NOT NULL,
updated_at INTEGER NOT NULL,
FOREIGN KEY (user_id) REFERENCES users (user_id)
)
""")
# 积分记录表
cursor.execute("""
CREATE TABLE IF NOT EXISTS points_records (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id INTEGER NOT NULL,
points INTEGER NOT NULL,
source TEXT NOT NULL,
description TEXT,
created_at INTEGER NOT NULL,
FOREIGN KEY (user_id) REFERENCES users (user_id)
)
""")
# 每日签到记录表
cursor.execute("""
CREATE TABLE IF NOT EXISTS daily_checkins (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id INTEGER NOT NULL,
checkin_date TEXT NOT NULL,
points_earned INTEGER NOT NULL,
created_at INTEGER NOT NULL,
UNIQUE(user_id, checkin_date)
)
""")
# 炼金抽奖记录表
cursor.execute("""
CREATE TABLE IF NOT EXISTS alchemy_records (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id INTEGER NOT NULL,
cost_points INTEGER NOT NULL,
reward_type TEXT NOT NULL,
reward_value INTEGER NOT NULL,
reward_description TEXT,
created_at INTEGER NOT NULL,
FOREIGN KEY (user_id) REFERENCES users (user_id)
)
""")
# 积分赠送记录表
cursor.execute("""
CREATE TABLE IF NOT EXISTS gift_records (
id INTEGER PRIMARY KEY AUTOINCREMENT,
sender_id INTEGER NOT NULL,
receiver_id INTEGER NOT NULL,
points INTEGER NOT NULL,
message TEXT,
created_at INTEGER NOT NULL,
FOREIGN KEY (sender_id) REFERENCES users (user_id),
FOREIGN KEY (receiver_id) REFERENCES users (user_id)
)
""")
# 创建积分相关索引
cursor.execute("""
CREATE INDEX IF NOT EXISTS idx_points_records_user
ON points_records(user_id, created_at)
""")
cursor.execute("""
CREATE INDEX IF NOT EXISTS idx_checkins_user_date
ON daily_checkins(user_id, checkin_date)
""")
cursor.execute("""
CREATE INDEX IF NOT EXISTS idx_alchemy_records_user
ON alchemy_records(user_id, created_at)
""")
cursor.execute("""
CREATE INDEX IF NOT EXISTS idx_gift_records_sender
ON gift_records(sender_id, created_at)
""")
cursor.execute("""
CREATE INDEX IF NOT EXISTS idx_gift_records_receiver
ON gift_records(receiver_id, created_at)
""")
logger.info("数据库表初始化完成")
# ===== 用户相关操作 =====
@@ -277,6 +368,478 @@ class Database:
logger.debug(f"更新游戏统计: user_id={user_id}, game_type={game_type}")
# ===== 积分相关操作 =====
def get_user_points(self, user_id: int) -> Dict:
"""获取用户积分信息
Args:
user_id: 用户ID
Returns:
积分信息字典
"""
cursor = self.conn.cursor()
cursor.execute(
"SELECT * FROM user_points WHERE user_id = ?",
(user_id,)
)
row = cursor.fetchone()
if row:
return dict(row)
else:
# 创建新用户积分记录
current_time = int(time.time())
cursor.execute(
"INSERT INTO user_points (user_id, total_points, available_points, created_at, updated_at) VALUES (?, 0, 0, ?, ?)",
(user_id, current_time, current_time)
)
return {
'user_id': user_id,
'total_points': 0,
'available_points': 0,
'created_at': current_time,
'updated_at': current_time
}
def add_points(self, user_id: int, points: int, source: str, description: str = None) -> bool:
"""添加积分
Args:
user_id: 用户ID
points: 积分数量
source: 积分来源
description: 描述
Returns:
是否成功
"""
if points <= 0:
return False
cursor = self.conn.cursor()
current_time = int(time.time())
try:
# 更新用户积分
cursor.execute("""
INSERT INTO user_points (user_id, total_points, available_points, created_at, updated_at)
VALUES (?, ?, ?, ?, ?)
ON CONFLICT(user_id)
DO UPDATE SET
total_points = total_points + ?,
available_points = available_points + ?,
updated_at = ?
""", (user_id, points, points, current_time, current_time, points, points, current_time))
# 记录积分变动
cursor.execute(
"INSERT INTO points_records (user_id, points, source, description, created_at) VALUES (?, ?, ?, ?, ?)",
(user_id, points, source, description, current_time)
)
logger.debug(f"用户 {user_id} 获得 {points} 积分，来源：{source}")
return True
except Exception as e:
logger.error(f"添加积分失败: {e}")
return False
def consume_points(self, user_id: int, points: int, source: str, description: str = None) -> bool:
"""消费积分
Args:
user_id: 用户ID
points: 积分数量
source: 消费来源
description: 描述
Returns:
是否成功
"""
if points <= 0:
return False
cursor = self.conn.cursor()
current_time = int(time.time())
try:
# 检查积分是否足够
cursor.execute(
"SELECT available_points FROM user_points WHERE user_id = ?",
(user_id,)
)
row = cursor.fetchone()
if not row or row[0] < points:
logger.warning(f"用户 {user_id} 积分不足，需要 {points}，当前可用 {row[0] if row else 0}")
return False
# 消费积分
cursor.execute(
"UPDATE user_points SET available_points = available_points - ?, updated_at = ? WHERE user_id = ?",
(points, current_time, user_id)
)
# 记录积分变动
cursor.execute(
"INSERT INTO points_records (user_id, points, source, description, created_at) VALUES (?, ?, ?, ?, ?)",
(user_id, -points, source, description, current_time)
)
logger.debug(f"用户 {user_id} 消费 {points} 积分，来源：{source}")
return True
except Exception as e:
logger.error(f"消费积分失败: {e}")
return False
def get_points_records(self, user_id: int, limit: int = 20) -> List[Dict]:
"""获取用户积分记录
Args:
user_id: 用户ID
limit: 限制数量
Returns:
积分记录列表
"""
cursor = self.conn.cursor()
cursor.execute(
"SELECT * FROM points_records WHERE user_id = ? ORDER BY created_at DESC LIMIT ?",
(user_id, limit)
)
rows = cursor.fetchall()
return [dict(row) for row in rows]
def check_daily_checkin(self, user_id: int, date: str) -> bool:
"""检查用户是否已签到
Args:
user_id: 用户ID
date: 日期字符串 (YYYY-MM-DD)
Returns:
是否已签到
"""
cursor = self.conn.cursor()
cursor.execute(
"SELECT 1 FROM daily_checkins WHERE user_id = ? AND checkin_date = ?",
(user_id, date)
)
return cursor.fetchone() is not None
def daily_checkin(self, user_id: int, points: int) -> bool:
"""每日签到
Args:
user_id: 用户ID
points: 签到积分
Returns:
是否成功
"""
from datetime import datetime
today = datetime.now().strftime('%Y-%m-%d')
if self.check_daily_checkin(user_id, today):
return False
cursor = self.conn.cursor()
current_time = int(time.time())
try:
# 记录签到
cursor.execute(
"INSERT INTO daily_checkins (user_id, checkin_date, points_earned, created_at) VALUES (?, ?, ?, ?)",
(user_id, today, points, current_time)
)
# 添加积分
return self.add_points(user_id, points, "daily_checkin", f"每日签到奖励")
except Exception as e:
logger.error(f"每日签到失败: {e}")
return False
def get_points_leaderboard(self, limit: int = 10) -> List[Dict]:
"""获取积分排行榜
Args:
limit: 限制数量
Returns:
排行榜列表
"""
cursor = self.conn.cursor()
cursor.execute("""
SELECT u.user_id, u.username, up.total_points, up.available_points
FROM users u
LEFT JOIN user_points up ON u.user_id = up.user_id
ORDER BY COALESCE(up.total_points, 0) DESC
LIMIT ?
""", (limit,))
rows = cursor.fetchall()
return [dict(row) for row in rows]
def add_alchemy_record(self, user_id: int, cost_points: int, reward_type: str,
reward_value: int, reward_description: str = None) -> bool:
"""添加炼金抽奖记录
Args:
user_id: 用户ID
cost_points: 消耗积分
reward_type: 奖励类型
reward_value: 奖励数值
reward_description: 奖励描述
Returns:
是否成功
"""
cursor = self.conn.cursor()
current_time = int(time.time())
try:
cursor.execute(
"INSERT INTO alchemy_records (user_id, cost_points, reward_type, reward_value, reward_description, created_at) VALUES (?, ?, ?, ?, ?, ?)",
(user_id, cost_points, reward_type, reward_value, reward_description, current_time)
)
logger.debug(f"添加炼金记录: user_id={user_id}, cost={cost_points}, reward={reward_type}={reward_value}")
return True
except Exception as e:
logger.error(f"添加炼金记录失败: {e}")
return False
def get_alchemy_records(self, user_id: int, limit: int = 20) -> List[Dict]:
"""获取用户炼金记录
Args:
user_id: 用户ID
limit: 限制数量
Returns:
炼金记录列表
"""
cursor = self.conn.cursor()
cursor.execute(
"SELECT * FROM alchemy_records WHERE user_id = ? ORDER BY created_at DESC LIMIT ?",
(user_id, limit)
)
rows = cursor.fetchall()
return [dict(row) for row in rows]
def get_alchemy_stats(self, user_id: int) -> Dict:
"""获取用户炼金统计
Args:
user_id: 用户ID
Returns:
炼金统计信息
"""
cursor = self.conn.cursor()
# 总抽奖次数
cursor.execute(
"SELECT COUNT(*) FROM alchemy_records WHERE user_id = ?",
(user_id,)
)
total_draws = cursor.fetchone()[0]
# 总消耗积分
cursor.execute(
"SELECT SUM(cost_points) FROM alchemy_records WHERE user_id = ?",
(user_id,)
)
total_cost = cursor.fetchone()[0] or 0
# 总获得积分
cursor.execute(
"SELECT SUM(reward_value) FROM alchemy_records WHERE user_id = ? AND reward_type = 'points'",
(user_id,)
)
total_points_gained = cursor.fetchone()[0] or 0
return {
'total_draws': total_draws,
'total_cost': total_cost,
'total_points_gained': total_points_gained,
'net_points': total_points_gained - total_cost
}
# ===== 积分赠送相关操作 =====
def send_gift(self, sender_id: int, receiver_id: int, points: int, message: str = None) -> bool:
"""赠送积分
Args:
sender_id: 赠送者ID
receiver_id: 接收者ID
points: 赠送积分数量
message: 附赠消息
Returns:
是否成功
"""
if points <= 0:
return False
if sender_id == receiver_id:
return False # 不能赠送给自己
cursor = self.conn.cursor()
current_time = int(time.time())
try:
# 检查赠送者积分是否足够
cursor.execute(
"SELECT available_points FROM user_points WHERE user_id = ?",
(sender_id,)
)
row = cursor.fetchone()
if not row or row[0] < points:
logger.warning(f"用户 {sender_id} 积分不足，需要 {points}，当前可用 {row[0] if row else 0}")
return False
# 确保接收者存在
self.get_or_create_user(receiver_id)
# 赠送者扣分
cursor.execute(
"UPDATE user_points SET available_points = available_points - ?, updated_at = ? WHERE user_id = ?",
(points, current_time, sender_id)
)
# 接收者加分
cursor.execute("""
INSERT INTO user_points (user_id, total_points, available_points, created_at, updated_at)
VALUES (?, ?, ?, ?, ?)
ON CONFLICT(user_id)
DO UPDATE SET
total_points = total_points + ?,
available_points = available_points + ?,
updated_at = ?
""", (receiver_id, points, points, current_time, current_time, points, points, current_time))
# 记录赠送
cursor.execute(
"INSERT INTO gift_records (sender_id, receiver_id, points, message, created_at) VALUES (?, ?, ?, ?, ?)",
(sender_id, receiver_id, points, message, current_time)
)
# 记录积分变动
cursor.execute(
"INSERT INTO points_records (user_id, points, source, description, created_at) VALUES (?, ?, ?, ?, ?)",
(sender_id, -points, "gift_send", f"赠送积分给用户{receiver_id}", current_time)
)
cursor.execute(
"INSERT INTO points_records (user_id, points, source, description, created_at) VALUES (?, ?, ?, ?, ?)",
(receiver_id, points, "gift_receive", f"收到用户{sender_id}的积分赠送", current_time)
)
logger.debug(f"用户 {sender_id} 赠送 {points} 积分给用户 {receiver_id}")
return True
except Exception as e:
logger.error(f"赠送积分失败: {e}")
return False
def get_gift_records_sent(self, user_id: int, limit: int = 20) -> List[Dict]:
"""获取用户发送的赠送记录
Args:
user_id: 用户ID
limit: 限制数量
Returns:
赠送记录列表
"""
cursor = self.conn.cursor()
cursor.execute("""
SELECT gr.*, u.username as receiver_name
FROM gift_records gr
LEFT JOIN users u ON gr.receiver_id = u.user_id
WHERE gr.sender_id = ?
ORDER BY gr.created_at DESC
LIMIT ?
""", (user_id, limit))
rows = cursor.fetchall()
return [dict(row) for row in rows]
def get_gift_records_received(self, user_id: int, limit: int = 20) -> List[Dict]:
"""获取用户接收的赠送记录
Args:
user_id: 用户ID
limit: 限制数量
Returns:
接收记录列表
"""
cursor = self.conn.cursor()
cursor.execute("""
SELECT gr.*, u.username as sender_name
FROM gift_records gr
LEFT JOIN users u ON gr.sender_id = u.user_id
WHERE gr.receiver_id = ?
ORDER BY gr.created_at DESC
LIMIT ?
""", (user_id, limit))
rows = cursor.fetchall()
return [dict(row) for row in rows]
def get_gift_stats(self, user_id: int) -> Dict:
"""获取用户赠送统计
Args:
user_id: 用户ID
Returns:
赠送统计信息
"""
cursor = self.conn.cursor()
# 总发送积分
cursor.execute(
"SELECT SUM(points) FROM gift_records WHERE sender_id = ?",
(user_id,)
)
total_sent = cursor.fetchone()[0] or 0
# 总接收积分
cursor.execute(
"SELECT SUM(points) FROM gift_records WHERE receiver_id = ?",
(user_id,)
)
total_received = cursor.fetchone()[0] or 0
# 发送次数
cursor.execute(
"SELECT COUNT(*) FROM gift_records WHERE sender_id = ?",
(user_id,)
)
sent_count = cursor.fetchone()[0]
# 接收次数
cursor.execute(
"SELECT COUNT(*) FROM gift_records WHERE receiver_id = ?",
(user_id,)
)
received_count = cursor.fetchone()[0]
return {
'total_sent': total_sent,
'total_received': total_received,
'sent_count': sent_count,
'received_count': received_count,
'net_gift': total_received - total_sent
}
def close(self):
"""关闭数据库连接"""
if self._conn:

390
games/alchemy.py Normal file
View File

@@ -0,0 +1,390 @@
"""炼金系统游戏模块"""
import random
import logging
from datetime import datetime
from games.base import BaseGame
from utils.parser import CommandParser
from core.database import get_db
logger = logging.getLogger(__name__)
class AlchemyGame(BaseGame):
"""炼金系统游戏"""
def __init__(self):
"""初始化游戏"""
super().__init__()
self.db = get_db()
# 奖品池配置 - 确保数学期望等于消耗积分
# 消耗10积分时的奖品池 - 数学期望约10.5
self.prize_pool_10 = [
# 负面奖励 (概率, 类型, 数值, 描述)
(0.05, "penalty", -5, "炼金失败，损失积分"), # 5% 概率损失5积分
(0.02, "penalty", -10, "炼金爆炸，损失积分"), # 2% 概率损失10积分
# 普通奖励 (概率, 类型, 数值, 描述)
(0.30, "points", 5, "少量积分"), # 30% 概率获得5积分
(0.25, "points", 8, "少量积分"), # 25% 概率获得8积分
(0.20, "points", 10, "等值积分"), # 20% 概率获得10积分
(0.10, "points", 15, "丰厚积分"), # 10% 概率获得15积分
(0.05, "points", 20, "丰厚积分"), # 5% 概率获得20积分
# 大奖 (概率, 类型, 数值, 描述)
(0.02, "points", 50, "🌟 巨额积分"), # 2% 概率获得50积分
(0.01, "points", 100, "💎 传说积分"), # 1% 概率获得100积分
]
# 消耗20积分时的奖品池 - 数学期望约21
self.prize_pool_20 = [
# 负面奖励
(0.03, "penalty", -10, "炼金失败，损失积分"), # 3% 概率损失10积分
(0.02, "penalty", -20, "炼金爆炸，损失积分"), # 2% 概率损失20积分
# 普通奖励
(0.25, "points", 8, "少量积分"), # 25% 概率获得8积分
(0.20, "points", 12, "少量积分"), # 20% 概率获得12积分
(0.20, "points", 20, "等值积分"), # 20% 概率获得20积分
(0.15, "points", 25, "丰厚积分"), # 15% 概率获得25积分
(0.10, "points", 30, "丰厚积分"), # 10% 概率获得30积分
(0.05, "points", 40, "丰厚积分"), # 5% 概率获得40积分
# 大奖
(0.03, "points", 80, "🌟 巨额积分"), # 3% 概率获得80积分
(0.02, "points", 150, "💎 传说积分"), # 2% 概率获得150积分
]
# 消耗50积分时的奖品池 - 数学期望约52
self.prize_pool_50 = [
# 负面奖励
(0.02, "penalty", -25, "炼金失败，损失积分"), # 2% 概率损失25积分
(0.01, "penalty", -50, "炼金爆炸，损失积分"), # 1% 概率损失50积分
# 普通奖励
(0.20, "points", 20, "少量积分"), # 20% 概率获得20积分
(0.20, "points", 30, "少量积分"), # 20% 概率获得30积分
(0.20, "points", 50, "等值积分"), # 20% 概率获得50积分
(0.15, "points", 60, "丰厚积分"), # 15% 概率获得60积分
(0.12, "points", 75, "丰厚积分"), # 12% 概率获得75积分
(0.08, "points", 100, "丰厚积分"), # 8% 概率获得100积分
# 大奖
(0.02, "points", 150, "🌟 巨额积分"), # 2% 概率获得150积分
(0.00, "points", 300, "💎 传说积分"), # 0% 概率获得300积分（预留）
]
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 ['stats', '统计', '记录']:
return self._get_alchemy_stats(user_id)
# 炼金记录
elif args in ['records', '历史', 'log']:
return self._get_alchemy_records(user_id)
# 炼金说明
elif args in ['help', '帮助', 'info']:
return self._get_alchemy_help()
# 默认：炼金抽奖
else:
# 解析消耗积分数量
cost_points = 10 # 默认消耗10积分
if args.isdigit():
cost_points = int(args)
if cost_points not in [10, 20, 50]:
return "❌ 炼金消耗积分只能是 10、20 或 50"
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['available_points'] < cost_points:
return f"❌ 积分不足！需要 {cost_points} 积分，当前可用 {user_points['available_points']} 积分"
# 选择奖品池
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 "❌ 不支持的炼金消耗积分"
# 执行抽奖
reward = self._draw_prize(prize_pool)
# 消费积分
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['value'], "alchemy", f"炼金奖励")
elif reward['type'] == 'penalty' and reward['value'] < 0:
# 负面奖励（扣分）
penalty_points = abs(reward['value'])
self.db.consume_points(user_id, penalty_points, "alchemy", f"炼金失败")
# 记录炼金抽奖
self.db.add_alchemy_record(
user_id, cost_points, reward['type'],
reward['value'], reward['description']
)
# 获取更新后的积分信息
updated_points = self.db.get_user_points(user_id)
# 格式化输出
text = f"## ⚗️ 炼金结果\n\n"
text += f"**消耗积分**：{cost_points} 分\n\n"
if reward['type'] == 'points' and reward['value'] > 0:
# 根据奖励大小选择不同的表情符号
if reward['value'] >= 100:
emoji = "👑"
text += f"**{emoji} 传说奖励**：{reward['value']} 积分\n\n"
elif reward['value'] >= 50:
emoji = "💎"
text += f"**{emoji} 巨额奖励**：{reward['value']} 积分\n\n"
else:
emoji = "🎁"
text += f"**{emoji} 获得奖励**：{reward['value']} 积分\n\n"
text += f"**奖励描述**：{reward['description']}\n\n"
elif reward['type'] == 'penalty' and reward['value'] < 0:
penalty_points = abs(reward['value'])
text += f"**💥 炼金失败**：损失 {penalty_points} 积分\n\n"
text += f"**失败原因**：{reward['description']}\n\n"
else:
text += f"**获得奖励**：{reward['description']}\n\n"
text += f"**当前积分**：{updated_points['available_points']} 分\n\n"
text += "---\n\n"
text += "💡 提示：炼金有风险，投资需谨慎！"
return text
def _draw_prize(self, prize_pool: list) -> dict:
"""从奖品池中抽取奖品
Args:
prize_pool: 奖品池
Returns:
奖品信息
"""
# 生成随机数
rand = random.random()
cumulative_prob = 0.0
for prob, reward_type, reward_value, description in prize_pool:
cumulative_prob += prob
if rand <= cumulative_prob:
return {
'type': reward_type,
'value': reward_value,
'description': description
}
# 兜底返回最后一个奖品
return {
'type': prize_pool[-1][1],
'value': prize_pool[-1][2],
'description': prize_pool[-1][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

25
games/base.py
View File

@@ -82,6 +82,31 @@ def get_help_message() -> str:
- `.gomoku list` - 列出所有对战
- `.gomoku stats` - 查看战绩
### 💎 积分系统
- `.points` - 查看个人积分
- `.积分` - 查看个人积分
- `.checkin` - 每日签到
- `.签到` - 每日签到
- `.打卡` - 每日签到
- `.points leaderboard` - 积分排行榜
- `.points records` - 积分记录
### ⚗️ 炼金系统
- `.alchemy` - 消耗10积分进行炼金
- `.炼金` - 消耗10积分进行炼金
- `.alchemy 20` - 消耗20积分进行炼金
- `.alchemy 50` - 消耗50积分进行炼金
- `.alchemy stats` - 查看炼金统计
- `.alchemy records` - 查看炼金记录
### 🎁 积分赠送系统
- `.gift <用户ID> <积分数量> [消息]` - 赠送积分
- `.赠送 <用户ID> <积分数量> [消息]` - 赠送积分
- `.送 <用户ID> <积分数量> [消息]` - 赠送积分
- `.gift stats` - 查看赠送统计
- `.gift sent` - 查看发送记录
- `.gift received` - 查看接收记录
### 其他
- `.help` - 显示帮助
- `.stats` - 查看个人统计

14
games/fortune.py
View File

@@ -7,6 +7,7 @@ from datetime import datetime
from pathlib import Path
from games.base import BaseGame
from utils.parser import CommandParser
from games.points import PointsGame
logger = logging.getLogger(__name__)
@@ -19,6 +20,7 @@ class FortuneGame(BaseGame):
super().__init__()
self._fortunes = None
self._tarot = None
self.points_game = PointsGame()
def _load_data(self):
"""懒加载运势数据"""
@@ -95,6 +97,11 @@ class FortuneGame(BaseGame):
# 重置随机seed
random.seed()
# 尝试获得积分奖励（30%概率）
points_earned = 0
if random.random() < 0.3: # 30%概率获得积分
points_earned = self.points_game.add_fortune_points(user_id)
# 格式化输出
text = f"## 🔮 今日运势\n\n"
text += f"**日期**：{today}\n\n"
@@ -103,8 +110,13 @@ class FortuneGame(BaseGame):
text += f"**建议**：{fortune['advice']}\n\n"
text += f"**幸运数字**：{lucky_number}\n\n"
text += f"**幸运颜色**：{lucky_color}\n\n"
# 添加积分奖励信息
if points_earned > 0:
text += f"**🎁 积分奖励**：+{points_earned} 分\n\n"
text += "---\n\n"
text += "💡 提示：运势仅供娱乐参考～"
text += "💡 提示：运势仅供娱乐参考，查看运势有机会获得积分奖励～"
return text

248
games/gift.py Normal file
View File

@@ -0,0 +1,248 @@
"""积分赠送系统游戏模块"""
import logging
from datetime import datetime
from games.base import BaseGame
from utils.parser import CommandParser
from core.database import get_db
logger = logging.getLogger(__name__)
class GiftGame(BaseGame):
"""积分赠送系统游戏"""
def __init__(self):
"""初始化游戏"""
super().__init__()
self.db = get_db()
async def handle(self, command: str, chat_id: int, user_id: int) -> str:
"""处理积分赠送相关指令
Args:
command: 指令，如 ".gift 123 50 生日快乐", ".gift sent", ".gift received"
chat_id: 会话ID
user_id: 用户ID
Returns:
回复消息
"""
try:
# 提取参数
_, args = CommandParser.extract_command_args(command)
args = args.strip()
# 赠送统计
if args in ['stats', '统计']:
return self._get_gift_stats(user_id)
# 发送记录
elif args in ['sent', '发送', '送出']:
return self._get_gift_records_sent(user_id)
# 接收记录
elif args in ['received', '接收', '收到']:
return self._get_gift_records_received(user_id)
# 赠送帮助
elif args in ['help', '帮助']:
return self._get_gift_help()
# 默认：执行赠送
else:
return self._process_gift_command(args, user_id)
except Exception as e:
logger.error(f"处理积分赠送指令错误: {e}", exc_info=True)
return f"❌ 处理指令出错: {str(e)}"
def _process_gift_command(self, args: str, sender_id: int) -> str:
"""处理赠送指令
Args:
args: 指令参数
sender_id: 发送者ID
Returns:
处理结果消息
"""
# 解析参数：.gift <receiver_id> <points> [message]
parts = args.split(maxsplit=2)
if len(parts) < 2:
return "❌ 指令格式错误！\n\n正确格式：`.gift <用户ID> <积分数量> [附赠消息]`\n\n示例：\n`.gift 123 50 生日快乐`\n`.gift 456 100`"
try:
receiver_id = int(parts[0])
points = int(parts[1])
message = parts[2] if len(parts) > 2 else None
except ValueError:
return "❌ 用户ID和积分数量必须是数字！"
# 验证参数
if points <= 0:
return "❌ 赠送积分数量必须大于0！"
if points > 1000:
return "❌ 单次赠送积分不能超过1000！"
if sender_id == receiver_id:
return "❌ 不能赠送积分给自己！"
# 执行赠送
if self.db.send_gift(sender_id, receiver_id, points, message):
# 获取更新后的积分信息
sender_points = self.db.get_user_points(sender_id)
receiver_points = self.db.get_user_points(receiver_id)
text = f"## 🎁 积分赠送成功！\n\n"
text += f"**赠送者**：用户{sender_id}\n\n"
text += f"**接收者**：用户{receiver_id}\n\n"
text += f"**赠送积分**：{points} 分\n\n"
if message:
text += f"**附赠消息**：{message}\n\n"
text += f"**赠送者剩余积分**：{sender_points['available_points']} 分\n\n"
text += f"**接收者当前积分**：{receiver_points['available_points']} 分\n\n"
text += "---\n\n"
text += "💝 感谢您的慷慨赠送！"
return text
else:
return "❌ 赠送失败！请检查积分是否足够或用户ID是否正确。"
def _get_gift_stats(self, user_id: int) -> str:
"""获取赠送统计信息
Args:
user_id: 用户ID
Returns:
统计信息消息
"""
stats = self.db.get_gift_stats(user_id)
if stats['sent_count'] == 0 and stats['received_count'] == 0:
return "📊 你还没有任何积分赠送记录哦～"
text = f"## 🎁 积分赠送统计\n\n"
text += f"**发送统计**：\n"
text += f"- 赠送次数：{stats['sent_count']} 次\n"
text += f"- 赠送积分：{stats['total_sent']} 分\n\n"
text += f"**接收统计**：\n"
text += f"- 接收次数：{stats['received_count']} 次\n"
text += f"- 接收积分：{stats['total_received']} 分\n\n"
net_gift = stats['net_gift']
if net_gift > 0:
text += f"**净收益**：+{net_gift} 分 🎉\n\n"
elif net_gift < 0:
text += f"**净收益**：{net_gift} 分 😅\n\n"
else:
text += f"**净收益**：0 分 ⚖️\n\n"
text += "---\n\n"
text += "💡 提示：使用 `.gift sent` 查看发送记录，`.gift received` 查看接收记录"
return text
def _get_gift_records_sent(self, user_id: int, limit: int = 10) -> str:
"""获取发送的赠送记录
Args:
user_id: 用户ID
limit: 限制数量
Returns:
记录信息消息
"""
records = self.db.get_gift_records_sent(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')
receiver_name = record.get('receiver_name', f"用户{record['receiver_id']}")
points = record['points']
message = record.get('message', '')
text += f"**{timestamp}** 赠送 {points} 分给 {receiver_name}\n"
if message:
text += f" 💬 {message}\n"
text += "\n"
return text
def _get_gift_records_received(self, user_id: int, limit: int = 10) -> str:
"""获取接收的赠送记录
Args:
user_id: 用户ID
limit: 限制数量
Returns:
记录信息消息
"""
records = self.db.get_gift_records_received(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')
sender_name = record.get('sender_name', f"用户{record['sender_id']}")
points = record['points']
message = record.get('message', '')
text += f"**{timestamp}** 收到 {sender_name} 的 {points} 分\n"
if message:
text += f" 💬 {message}\n"
text += "\n"
return text
def _get_gift_help(self) -> str:
"""获取赠送帮助信息
Returns:
帮助信息消息
"""
text = f"## 🎁 积分赠送系统\n\n"
text += f"### 基础用法\n"
text += f"- `.gift <用户ID> <积分数量> [附赠消息]` - 赠送积分\n"
text += f"- `.gift stats` - 查看赠送统计\n"
text += f"- `.gift sent` - 查看发送记录\n"
text += f"- `.gift received` - 查看接收记录\n"
text += f"- `.gift help` - 查看帮助\n\n"
text += f"### 赠送规则\n"
text += f"- **积分限制**：单次最多赠送1000积分\n"
text += f"- **自赠限制**：不能赠送给自己\n"
text += f"- **积分检查**：必须有足够积分才能赠送\n"
text += f"- **附赠消息**：可选，最多100字符\n\n"
text += f"### 示例\n"
text += f"```\n"
text += f".gift 123 50 生日快乐\n"
text += f".gift 456 100 感谢你的帮助\n"
text += f".gift 789 200\n"
text += f".gift stats\n"
text += f"```\n\n"
text += f"### 说明\n"
text += f"- 所有赠送都有完整记录\n"
text += f"- 赠送和接收都会在积分记录中显示\n"
text += f"- 赠送是单向的，无需对方确认\n"
return text
def get_help(self) -> str:
"""获取帮助信息"""
return self._get_gift_help()

225
games/points.py Normal file
View File

@@ -0,0 +1,225 @@
"""积分系统游戏模块"""
import random
import logging
from datetime import datetime
from games.base import BaseGame
from utils.parser import CommandParser
from core.database import get_db
logger = logging.getLogger(__name__)
class PointsGame(BaseGame):
"""积分系统游戏"""
def __init__(self):
"""初始化游戏"""
super().__init__()
self.db = get_db()
async def handle(self, command: str, chat_id: int, user_id: int) -> str:
"""处理积分相关指令
Args:
command: 指令，如 ".points", ".checkin", ".leaderboard"
chat_id: 会话ID
user_id: 用户ID
Returns:
回复消息
"""
try:
# 提取参数
_, args = CommandParser.extract_command_args(command)
args = args.strip().lower()
# 每日签到
if args in ['checkin', '签到', '打卡']:
return self._daily_checkin(user_id)
# 积分排行榜
elif args in ['leaderboard', '排行榜', '排行']:
return self._get_leaderboard()
# 积分记录
elif args in ['records', '记录', '历史']:
return self._get_points_records(user_id)
# 默认：查看个人积分
else:
return self._get_user_points(user_id)
except Exception as e:
logger.error(f"处理积分指令错误: {e}", exc_info=True)
return f"❌ 处理指令出错: {str(e)}"
def _daily_checkin(self, user_id: int) -> str:
"""每日签到
Args:
user_id: 用户ID
Returns:
签到结果消息
"""
# 固定签到积分
checkin_points = 10
# 检查是否已签到
today = datetime.now().strftime('%Y-%m-%d')
if self.db.check_daily_checkin(user_id, today):
return f"❌ 今日已签到，明天再来吧！\n\n📅 签到日期：{today}"
# 执行签到
if self.db.daily_checkin(user_id, checkin_points):
# 获取用户积分信息
points_info = self.db.get_user_points(user_id)
text = f"## ✅ 签到成功！\n\n"
text += f"**获得积分**：+{checkin_points} 分\n\n"
text += f"**当前积分**：{points_info['available_points']} 分\n\n"
text += f"**总积分**：{points_info['total_points']} 分\n\n"
text += f"📅 签到日期：{today}\n\n"
text += "💡 提示：每天签到可获得固定积分奖励！"
return text
else:
return "❌ 签到失败，请稍后重试"
def _get_user_points(self, user_id: int) -> str:
"""获取用户积分信息
Args:
user_id: 用户ID
Returns:
积分信息消息
"""
points_info = self.db.get_user_points(user_id)
text = f"## 💎 个人积分\n\n"
text += f"**可用积分**：{points_info['available_points']} 分\n\n"
text += f"**总积分**：{points_info['total_points']} 分\n\n"
text += f"**注册时间**：{datetime.fromtimestamp(points_info['created_at']).strftime('%Y-%m-%d %H:%M')}\n\n"
text += "---\n\n"
text += "💡 提示：\n"
text += "• 每日签到可获得 10 积分\n"
text += "• 查看运势可获得随机积分\n"
text += "• 使用 `.points records` 查看积分记录"
return text
def _get_points_records(self, user_id: int, limit: int = 10) -> str:
"""获取用户积分记录
Args:
user_id: 用户ID
limit: 限制数量
Returns:
积分记录消息
"""
records = self.db.get_points_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')
points_str = f"+{record['points']}" if record['points'] > 0 else str(record['points'])
source_map = {
'daily_checkin': '每日签到',
'fortune': '运势奖励',
'game_reward': '游戏奖励'
}
source = source_map.get(record['source'], record['source'])
text += f"**{timestamp}** {points_str} 分 - {source}\n"
if record.get('description'):
text += f" *{record['description']}*\n"
text += "\n"
return text
def _get_leaderboard(self, limit: int = 10) -> str:
"""获取积分排行榜
Args:
limit: 限制数量
Returns:
排行榜消息
"""
leaderboard = self.db.get_points_leaderboard(limit)
if not leaderboard:
return "📊 暂无排行榜数据"
text = f"## 🏆 积分排行榜（前 {len(leaderboard)} 名）\n\n"
medals = ["🥇", "🥈", "🥉"] + ["🏅"] * (limit - 3)
for i, user in enumerate(leaderboard):
rank = i + 1
medal = medals[i] if i < len(medals) else "🏅"
username = user.get('username', f"用户{user['user_id']}")
total_points = user.get('total_points', 0)
available_points = user.get('available_points', 0)
text += f"{medal} **第 {rank} 名** {username}\n"
text += f" 总积分：{total_points} 分 | 可用：{available_points} 分\n\n"
text += "---\n\n"
text += "💡 提示：使用 `.points` 查看个人积分"
return text
def add_fortune_points(self, user_id: int) -> int:
"""为运势游戏添加随机积分奖励
Args:
user_id: 用户ID
Returns:
获得的积分数量
"""
# 随机积分范围：1-20分
points = random.randint(1, 20)
if self.db.add_points(user_id, points, "fortune", "运势奖励"):
logger.info(f"用户 {user_id} 通过运势获得 {points} 积分")
return points
else:
logger.error(f"用户 {user_id} 运势积分奖励失败")
return 0
def get_help(self) -> str:
"""获取帮助信息"""
return """## 💎 积分系统
### 基础用法
- `.points` - 查看个人积分
- `.checkin` - 每日签到
- `.points leaderboard` - 积分排行榜
- `.points records` - 积分记录
### 积分获取方式
- **每日签到**：固定 10 积分
- **运势占卜**：随机 1-20 积分
- **游戏奖励**：根据游戏表现获得
### 示例
```
.points
.checkin
.points leaderboard
.points records
```
### 说明
- 每日签到只能进行一次
- 运势积分每次查看都有机会获得
- 积分记录保留最近 20 条
"""

18
routers/callback.py
View File

@@ -159,6 +159,24 @@ async def handle_command(game_type: str, command: str,
game = GomokuGame()
return await game.handle(command, chat_id, user_id)
# 积分系统
if game_type == 'points':
from games.points import PointsGame
game = PointsGame()
return await game.handle(command, chat_id, user_id)
# 炼金系统
if game_type == 'alchemy':
from games.alchemy import AlchemyGame
game = AlchemyGame()
return await game.handle(command, chat_id, user_id)
# 积分赠送系统
if game_type == 'gift':
from games.gift import GiftGame
game = GiftGame()
return await game.handle(command, chat_id, user_id)
# 未知游戏类型
logger.warning(f"未知游戏类型: {game_type}")
return "❌ 未知的游戏类型"

16
utils/parser.py
View File

@@ -40,6 +40,22 @@ class CommandParser:
'.五子棋': 'gomoku',
'.gobang': 'gomoku',
# 积分系统
'.points': 'points',
'.积分': 'points',
'.checkin': 'points',
'.签到': 'points',
'.打卡': 'points',
# 炼金系统
'.alchemy': 'alchemy',
'.炼金': 'alchemy',
# 积分赠送系统
'.gift': 'gift',
'.赠送': 'gift',
'.送': 'gift',
# 帮助
'.help': 'help',
'.帮助': 'help',