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

1.尝试修复积分系统2.大幅度简化积分系统

Browse Source
This commit is contained in:
LiuBai095
2025-10-29 12:29:18 +08:00
parent 7b72ed9f34
commit 0c2638e948
2 changed files with 39 additions and 451 deletions
Download Patch File Download Diff File Expand all files Collapse all files

437
core/database.py
View File

@@ -102,97 +102,18 @@ 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,
points INTEGER DEFAULT 0,
last_checkin_date TEXT,
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("数据库表初始化完成")
# ===== 用户相关操作 =====
@@ -405,13 +326,13 @@ class Database:
# 创建新用户积分记录
current_time = int(time.time())
cursor.execute(
"INSERT INTO user_points (user_id, total_points, available_points, created_at, updated_at) VALUES (?, 0, 0, ?, ?)",
"INSERT INTO user_points (user_id, points, created_at, updated_at) VALUES (?, 0, ?, ?)",
(user_id, current_time, current_time)
)
return {
'user_id': user_id,
'total_points': 0,
'available_points': 0,
'points': 0,
'last_checkin_date': None,
'created_at': current_time,
'updated_at': current_time
}
@@ -441,29 +362,15 @@ class Database:
# 更新用户积分
cursor.execute("""
INSERT INTO user_points (user_id, total_points, available_points, created_at, updated_at)
VALUES (?, ?, ?, ?, ?)
INSERT INTO user_points (user_id, points, created_at, updated_at)
VALUES (?, ?, ?, ?)
ON CONFLICT(user_id)
DO UPDATE SET
total_points = total_points + ?,
available_points = available_points + ?,
points = points + ?,
updated_at = ?
""", (user_id, points, points, current_time, current_time, points, points, current_time))
""", (user_id, points, current_time, current_time, points, current_time))
# 验证积分是否真的更新了
cursor.execute("SELECT available_points FROM user_points WHERE user_id = ?", (user_id,))
updated_points = cursor.fetchone()
if not updated_points:
logger.error(f"积分更新后查询失败: user_id={user_id}")
return False
# 记录积分变动
cursor.execute(
"INSERT INTO points_records (user_id, points, source, description, created_at) VALUES (?, ?, ?, ?, ?)",
(user_id, points, source, description, current_time)
)
logger.info(f"用户 {user_id} 成功获得 {points} 积分，来源：{source}，当前积分：{updated_points[0]}")
logger.info(f"用户 {user_id} 成功获得 {points} 积分，来源：{source}")
return True
except Exception as e:
@@ -491,7 +398,7 @@ class Database:
try:
# 检查积分是否足够
cursor.execute(
"SELECT available_points FROM user_points WHERE user_id = ?",
"SELECT points FROM user_points WHERE user_id = ?",
(user_id,)
)
row = cursor.fetchone()
@@ -502,16 +409,10 @@ class Database:
# 消费积分
cursor.execute(
"UPDATE user_points SET available_points = available_points - ?, updated_at = ? WHERE user_id = ?",
"UPDATE user_points SET points = 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
@@ -519,24 +420,6 @@ class Database:
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:
"""检查用户是否已签到
@@ -549,10 +432,11 @@ class Database:
"""
cursor = self.conn.cursor()
cursor.execute(
"SELECT 1 FROM daily_checkins WHERE user_id = ? AND checkin_date = ?",
(user_id, date)
"SELECT last_checkin_date FROM user_points WHERE user_id = ?",
(user_id,)
)
return cursor.fetchone() is not None
row = cursor.fetchone()
return row and row[0] == date
def daily_checkin(self, user_id: int, points: int) -> bool:
"""每日签到
@@ -580,23 +464,25 @@ class Database:
# 获取签到前积分
points_before = self.get_user_points(user_id)
logger.info(f"用户 {user_id} 签到前积分: {points_before['available_points']}")
logger.info(f"用户 {user_id} 签到前积分: {points_before['points']}")
# 记录签到
cursor.execute(
"INSERT INTO daily_checkins (user_id, checkin_date, points_earned, created_at) VALUES (?, ?, ?, ?)",
(user_id, today, points, current_time)
)
# 添加积分
result = self.add_points(user_id, points, "daily_checkin", f"每日签到奖励")
# 更新积分和签到日期
cursor.execute("""
INSERT INTO user_points (user_id, points, last_checkin_date, created_at, updated_at)
VALUES (?, ?, ?, ?, ?)
ON CONFLICT(user_id)
DO UPDATE SET
points = points + ?,
last_checkin_date = ?,
updated_at = ?
""", (user_id, points, today, current_time, current_time, points, today, current_time))
# 验证积分是否真的增加了
points_after = self.get_user_points(user_id)
logger.info(f"用户 {user_id} 签到后积分: {points_after['available_points']}")
logger.info(f"用户 {user_id} 签到后积分: {points_after['points']}")
if points_after['available_points'] > points_before['available_points']:
logger.info(f"用户 {user_id} 签到成功，积分增加: {points_after['available_points'] - points_before['available_points']}")
if points_after['points'] > points_before['points']:
logger.info(f"用户 {user_id} 签到成功，积分增加: {points_after['points'] - points_before['points']}")
return True
else:
logger.error(f"用户 {user_id} 签到失败，积分未增加")
@@ -617,272 +503,15 @@ class Database:
"""
cursor = self.conn.cursor()
cursor.execute("""
SELECT u.user_id, u.username, up.total_points, up.available_points
SELECT u.user_id, u.username, up.points
FROM users u
LEFT JOIN user_points up ON u.user_id = up.user_id
ORDER BY COALESCE(up.total_points, 0) DESC
ORDER BY COALESCE(up.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:

53
games/points.py
View File

@@ -41,9 +41,6 @@ class PointsGame(BaseGame):
elif args in ['leaderboard', '排行榜', '排行']:
return self._get_leaderboard()
# 积分记录
elif args in ['records', '记录', '历史']:
return self._get_points_records(user_id)
# 默认：查看个人积分
else:
@@ -79,8 +76,7 @@ class PointsGame(BaseGame):
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"**当前积分**：{points_info['points']} 分\n\n"
text += f"📅 签到日期：{today}\n\n"
text += "💡 提示：每天签到可获得固定积分奖励！"
@@ -103,49 +99,15 @@ class PointsGame(BaseGame):
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"**当前积分**：{points_info['points']} 分\n\n"
text += "---\n\n"
text += "💡 提示：\n"
text += "• 每日签到可获得 10 积分\n"
text += "• 查看运势可获得随机积分\n"
text += "• 使用 `.points records` 查看积分记录"
text += "• 使用 `.points leaderboard` 查看排行榜"
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:
"""获取积分排行榜
@@ -169,11 +131,10 @@ class PointsGame(BaseGame):
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)
points = user.get('points', 0)
text += f"{medal} **第 {rank} 名** {username}\n"
text += f" 总积分：{total_points} 分 | 可用：{available_points} 分\n\n"
text += f" 积分：{points} 分\n\n"
text += "---\n\n"
text += "💡 提示：使用 `.points` 查看个人积分"
@@ -207,7 +168,6 @@ class PointsGame(BaseGame):
- `.points` - 查看个人积分
- `.checkin` - 每日签到
- `.points leaderboard` - 积分排行榜
- `.points records` - 积分记录
### 积分获取方式
- **每日签到**：固定 10 积分
@@ -219,11 +179,10 @@ class PointsGame(BaseGame):
.points
.checkin
.points leaderboard
.points records
```
### 说明
- 每日签到只能进行一次
- 运势积分每次查看都有机会获得
- 积分记录保留最近 20 条
- 积分系统已简化，不再保留历史记录
"""