Mr.Xia
0fd011fa1e
- 新增积分系统插件，支持积分查询、签到、转账等核心功能 - 包含对应的测试脚本 Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
101 lines
3.2 KiB
Python
101 lines
3.2 KiB
Python
import sqlite3
|
|
import os
|
|
from datetime import datetime
|
|
from typing import Optional, List, Dict, Any
|
|
from .config import Config
|
|
|
|
|
|
class PointsDatabase:
|
|
"""SQLite database handler for points system."""
|
|
|
|
def __init__(self, config: Config):
|
|
self.config = config
|
|
self.db_path = config.POINTS_DB_FILE
|
|
self._ensure_db_dir()
|
|
self._init_db()
|
|
|
|
def _ensure_db_dir(self):
|
|
"""Create database directory if it doesn't exist."""
|
|
db_dir = os.path.dirname(self.db_path)
|
|
if db_dir:
|
|
os.makedirs(db_dir, exist_ok=True)
|
|
|
|
def _init_db(self):
|
|
"""Initialize database tables."""
|
|
conn = sqlite3.connect(self.db_path, timeout=5.0)
|
|
cursor = conn.cursor()
|
|
|
|
# Create user_points table
|
|
cursor.execute(
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS user_points (
|
|
user_id TEXT PRIMARY KEY,
|
|
points INTEGER NOT NULL DEFAULT 0 CHECK(points >= 0),
|
|
total_earned INTEGER NOT NULL DEFAULT 0,
|
|
total_spent INTEGER NOT NULL DEFAULT 0,
|
|
created_at TEXT NOT NULL,
|
|
updated_at TEXT NOT NULL
|
|
)
|
|
"""
|
|
)
|
|
|
|
# Create point_transactions table
|
|
cursor.execute(
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS point_transactions (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
user_id TEXT NOT NULL,
|
|
amount INTEGER NOT NULL,
|
|
balance_after INTEGER NOT NULL,
|
|
source TEXT NOT NULL,
|
|
reason TEXT,
|
|
created_at TEXT NOT NULL
|
|
)
|
|
"""
|
|
)
|
|
|
|
# Create indexes
|
|
cursor.execute(
|
|
"CREATE INDEX IF NOT EXISTS idx_transactions_user_id ON point_transactions(user_id)"
|
|
)
|
|
cursor.execute(
|
|
"CREATE INDEX IF NOT EXISTS idx_transactions_source ON point_transactions(source)"
|
|
)
|
|
cursor.execute(
|
|
"CREATE INDEX IF NOT EXISTS idx_transactions_created_at ON point_transactions(created_at)"
|
|
)
|
|
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
def get_connection(self) -> sqlite3.Connection:
|
|
"""Get a database connection."""
|
|
conn = sqlite3.connect(self.db_path, timeout=5.0)
|
|
conn.row_factory = sqlite3.Row
|
|
return conn
|
|
|
|
def get_user_balance(self, user_id: str) -> int:
|
|
"""Get user's current points balance."""
|
|
conn = self.get_connection()
|
|
cursor = conn.cursor()
|
|
cursor.execute("SELECT points FROM user_points WHERE user_id = ?", (user_id,))
|
|
row = cursor.fetchone()
|
|
conn.close()
|
|
return row["points"] if row else 0
|
|
|
|
def ensure_user_exists(self, user_id: str) -> None:
|
|
"""Create user account if it doesn't exist."""
|
|
conn = self.get_connection()
|
|
cursor = conn.cursor()
|
|
now = datetime.now().isoformat()
|
|
cursor.execute(
|
|
"""
|
|
INSERT OR IGNORE INTO user_points
|
|
(user_id, points, total_earned, total_spent, created_at, updated_at)
|
|
VALUES (?, 0, 0, 0, ?, ?)
|
|
""",
|
|
(user_id, now, now),
|
|
)
|
|
conn.commit()
|
|
conn.close()
|