功能：实现 Danding_Points 积分系统插件
- 新增积分系统插件，支持积分查询、签到、转账等核心功能 - 包含对应的测试脚本 Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
205
danding_bot/plugins/danding_points/README.md
Normal file
205
danding_bot/plugins/danding_points/README.md
Normal file
@@ -0,0 +1,205 @@
|
||||
# Danding Points 插件
|
||||
|
||||
全局积分/虚拟货币服务层，为其他插件提供统一的积分管理能力。用户在一个插件中获得的积分可以在另一个插件中消费。
|
||||
|
||||
本插件不包含任何用户交互命令（无 NoneBot matcher），纯 API 服务层，供其他插件直接 import 调用。
|
||||
|
||||
## 目录结构
|
||||
|
||||
```
|
||||
danding_points/
|
||||
├── __init__.py # 插件元数据 & points_api 单例导出
|
||||
├── config.py # 配置类
|
||||
├── database.py # SQLite 数据库操作
|
||||
└── api.py # PointsAPI 核心
|
||||
```
|
||||
|
||||
## 配置选项
|
||||
|
||||
在 `.env` 文件或 NoneBot 配置文件中添加（均带 `DANDING_` 前缀）：
|
||||
|
||||
| 配置项 | 类型 | 默认值 | 说明 |
|
||||
|--------|------|--------|------|
|
||||
| `DANDING_POINTS_DB_FILE` | str | `data/danding_points/points.db` | 数据库文件路径 |
|
||||
| `DANDING_POINTS_MAX_BALANCE` | int | `0` | 用户积分余额上限，`0` = 无限制 |
|
||||
| `DANDING_POINTS_MAX_PER_OPERATION` | int | `0` | 单次操作积分上限，`0` = 无限制 |
|
||||
| `DANDING_POINTS_LOG_RETENTION_DAYS` | int | `365` | 流水日志保留天数 |
|
||||
|
||||
## API 接口
|
||||
|
||||
所有方法均为 `async`，通过 `points_api` 单例调用。
|
||||
|
||||
### `get_balance(user_id: str) -> int`
|
||||
|
||||
查询用户积分余额。用户不存在时自动返回 `0`，不会创建账户。
|
||||
|
||||
```python
|
||||
balance = await points_api.get_balance("123456")
|
||||
```
|
||||
|
||||
### `add_points(user_id, amount, source, reason=None) -> Tuple[bool, int]`
|
||||
|
||||
为用户增加积分。用户不存在时自动建户。
|
||||
|
||||
- `amount`: 正整数，必须 > 0
|
||||
- `source`: 来源标识，不能为空（用于流水追踪）
|
||||
- `reason`: 变动原因，可选
|
||||
- 返回 `(success, new_balance)`，失败时 `success=False`
|
||||
|
||||
触发校验：操作上限 → 余额上限。任一校验失败均返回 `(False, 当前余额)`。
|
||||
|
||||
```python
|
||||
ok, balance = await points_api.add_points("123456", 100, "gacha", "抽卡奖励")
|
||||
if ok:
|
||||
print(f"充值成功，当前余额: {balance}")
|
||||
```
|
||||
|
||||
### `spend_points(user_id, amount, source, reason=None) -> Tuple[bool, int]`
|
||||
|
||||
消费用户积分。余额不足时返回 `(False, 当前余额)`。
|
||||
|
||||
- `amount`: 正整数，必须 > 0
|
||||
- 流水中 `amount` 记录为负数
|
||||
|
||||
```python
|
||||
ok, balance = await points_api.spend_points("123456", 30, "shop", "购买道具")
|
||||
if not ok:
|
||||
print("余额不足")
|
||||
```
|
||||
|
||||
### `set_points(user_id, amount, source, reason=None) -> Tuple[bool, int]`
|
||||
|
||||
直接设定用户积分。**绕过** 操作上限和余额上限约束。
|
||||
|
||||
- `amount`: 非负整数，必须 >= 0
|
||||
- 新值等于旧值时不写流水，直接返回 `(True, amount)`
|
||||
- `total_earned` 仅累加正向差额（设低不影响）
|
||||
|
||||
```python
|
||||
ok, balance = await points_api.set_points("123456", 500, "admin", "管理员调整")
|
||||
```
|
||||
|
||||
### `get_transactions(user_id, limit=20, offset=0) -> List[dict]`
|
||||
|
||||
查询用户积分流水记录，按时间倒序。
|
||||
|
||||
- `limit`: 1~100，超出范围自动裁剪
|
||||
- `offset`: 分页偏移量，>= 0
|
||||
|
||||
返回字段：`id, user_id, amount, balance_after, source, reason, created_at`
|
||||
|
||||
```python
|
||||
txs = await points_api.get_transactions("123456", limit=10, offset=0)
|
||||
for tx in txs:
|
||||
print(f"{tx['created_at']} {tx['amount']:+d} 余额:{tx['balance_after']}")
|
||||
```
|
||||
|
||||
### `get_ranking(limit=10, order_by="points") -> List[dict]`
|
||||
|
||||
查询积分排行榜。
|
||||
|
||||
- `limit`: 1~100
|
||||
- `order_by`: `"points"`（按余额）或 `"total_earned"`（按累计获得），其他值回退为 `"points"`
|
||||
- 使用 `RANK()` 窗口函数，同分并列，次级排序按 `user_id` 字母序
|
||||
|
||||
返回字段：`rank, user_id, points, total_earned, total_spent`
|
||||
|
||||
```python
|
||||
ranking = await points_api.get_ranking(limit=10, order_by="points")
|
||||
for r in ranking:
|
||||
print(f"#{r['rank']} {r['user_id']} {r['points']}分")
|
||||
```
|
||||
|
||||
## 其他插件对接
|
||||
|
||||
### 基本用法
|
||||
|
||||
在需要积分功能的插件中 import 单例即可：
|
||||
|
||||
```python
|
||||
from nonebot import require
|
||||
|
||||
require("danding_points")
|
||||
from danding_bot.plugins.danding_points import points_api
|
||||
```
|
||||
|
||||
### 示例：抽卡插件发放奖励
|
||||
|
||||
```python
|
||||
# 在 onmyoji_gacha 插件中
|
||||
from danding_bot.plugins.danding_points import points_api
|
||||
|
||||
async def reward_user(user_id: str):
|
||||
ok, balance = await points_api.add_points(
|
||||
user_id, 50, "onmyoji_gacha", "SSR 抽到奖励"
|
||||
)
|
||||
return balance
|
||||
```
|
||||
|
||||
### 示例：商店插件消费积分
|
||||
|
||||
```python
|
||||
# 在 shop 插件中
|
||||
from danding_bot.plugins.danding_points import points_api
|
||||
|
||||
async def buy_item(user_id: str, cost: int):
|
||||
ok, balance = await points_api.spend_points(
|
||||
user_id, cost, "shop", "购买商品"
|
||||
)
|
||||
if not ok:
|
||||
return "积分不足"
|
||||
return f"购买成功，剩余 {balance} 积分"
|
||||
```
|
||||
|
||||
### 示例：管理员插件调整积分
|
||||
|
||||
```python
|
||||
# 在 danding_api 插件中
|
||||
from danding_bot.plugins.danding_points import points_api
|
||||
|
||||
async def admin_set(user_id: str, amount: int):
|
||||
ok, balance = await points_api.set_points(
|
||||
user_id, amount, "danding_api", "管理员手动调整"
|
||||
)
|
||||
return balance
|
||||
```
|
||||
|
||||
### source 命名建议
|
||||
|
||||
`source` 参数用于标识积分变动来源，建议各插件使用自身插件名作为 source：
|
||||
|
||||
| 插件 | source 值 |
|
||||
|------|-----------|
|
||||
| onmyoji_gacha | `"onmyoji_gacha"` |
|
||||
| danding_api | `"danding_api"` |
|
||||
| shop | `"shop"` |
|
||||
| sign_in | `"sign_in"` |
|
||||
|
||||
## 数据库
|
||||
|
||||
使用 SQLite，数据文件位于 `data/danding_points/points.db`，无需额外配置。
|
||||
|
||||
### 表结构
|
||||
|
||||
**user_points** — 用户积分账户
|
||||
|
||||
| 字段 | 类型 | 说明 |
|
||||
|------|------|------|
|
||||
| user_id | TEXT PK | 用户 ID |
|
||||
| points | INTEGER | 当前余额，>= 0 |
|
||||
| total_earned | INTEGER | 累计获得 |
|
||||
| total_spent | INTEGER | 累计消费 |
|
||||
| created_at | TEXT | 创建时间 |
|
||||
| updated_at | TEXT | 最后更新时间 |
|
||||
|
||||
**point_transactions** — 积分变动流水
|
||||
|
||||
| 字段 | 类型 | 说明 |
|
||||
|------|------|------|
|
||||
| id | INTEGER PK | 自增 ID |
|
||||
| user_id | TEXT | 用户 ID |
|
||||
| amount | INTEGER | 变动数额（消费为负） |
|
||||
| balance_after | INTEGER | 变动后余额 |
|
||||
| source | TEXT | 来源标识 |
|
||||
| reason | TEXT | 变动原因 |
|
||||
| created_at | TEXT | 创建时间 |
|
||||
18
danding_bot/plugins/danding_points/__init__.py
Normal file
18
danding_bot/plugins/danding_points/__init__.py
Normal file
@@ -0,0 +1,18 @@
|
||||
from nonebot.plugin import PluginMetadata
|
||||
from .config import Config
|
||||
from .api import PointsAPI
|
||||
|
||||
__plugin_meta__ = PluginMetadata(
|
||||
name="Danding Points",
|
||||
description="Global points/virtual currency system for danding-bot",
|
||||
usage="Import points_api from this plugin to use the points API",
|
||||
type="service",
|
||||
homepage="https://github.com/danding-bot/danding-bot",
|
||||
config=Config,
|
||||
)
|
||||
|
||||
# Initialize configuration and API
|
||||
config = Config()
|
||||
points_api = PointsAPI(config)
|
||||
|
||||
__all__ = ["points_api", "PointsAPI", "Config"]
|
||||
294
danding_bot/plugins/danding_points/api.py
Normal file
294
danding_bot/plugins/danding_points/api.py
Normal file
@@ -0,0 +1,294 @@
|
||||
import asyncio
|
||||
import threading
|
||||
from datetime import datetime
|
||||
from typing import Tuple, List, Dict, Any
|
||||
from .config import Config
|
||||
from .database import PointsDatabase
|
||||
|
||||
|
||||
class PointsAPI:
|
||||
"""Points system API for managing user points."""
|
||||
|
||||
def __init__(self, config: Config):
|
||||
self.config = config
|
||||
self.db = PointsDatabase(config)
|
||||
self._lock = threading.Lock()
|
||||
|
||||
async def get_balance(self, user_id: str) -> int:
|
||||
"""Get user's current points balance."""
|
||||
return await asyncio.to_thread(self.db.get_user_balance, user_id)
|
||||
|
||||
async def add_points(
|
||||
self, user_id: str, amount: int, source: str, reason: str = None
|
||||
) -> Tuple[bool, int]:
|
||||
"""Add points to user account.
|
||||
|
||||
Returns: (success, new_balance)
|
||||
"""
|
||||
# Parameter validation
|
||||
if not isinstance(amount, int) or amount <= 0:
|
||||
return False, 0
|
||||
if not user_id or not source:
|
||||
return False, 0
|
||||
|
||||
# Operation limit validation
|
||||
if self.config.POINTS_MAX_PER_OPERATION > 0:
|
||||
if amount > self.config.POINTS_MAX_PER_OPERATION:
|
||||
return False, 0
|
||||
|
||||
def _add():
|
||||
with self._lock:
|
||||
conn = self.db.get_connection()
|
||||
cursor = conn.cursor()
|
||||
try:
|
||||
# Ensure user exists
|
||||
self.db.ensure_user_exists(user_id)
|
||||
|
||||
# Get current balance
|
||||
cursor.execute(
|
||||
"SELECT points FROM user_points WHERE user_id = ?",
|
||||
(user_id,),
|
||||
)
|
||||
row = cursor.fetchone()
|
||||
current_balance = row["points"] if row else 0
|
||||
|
||||
# Check balance limit
|
||||
new_balance = current_balance + amount
|
||||
if self.config.POINTS_MAX_BALANCE > 0:
|
||||
if new_balance > self.config.POINTS_MAX_BALANCE:
|
||||
conn.close()
|
||||
return False, current_balance
|
||||
|
||||
# Update balance and total_earned
|
||||
now = datetime.now().isoformat()
|
||||
cursor.execute(
|
||||
"""
|
||||
UPDATE user_points
|
||||
SET points = ?, total_earned = total_earned + ?, updated_at = ?
|
||||
WHERE user_id = ?
|
||||
""",
|
||||
(new_balance, amount, now, user_id),
|
||||
)
|
||||
|
||||
# Write transaction log
|
||||
cursor.execute(
|
||||
"""
|
||||
INSERT INTO point_transactions
|
||||
(user_id, amount, balance_after, source, reason, created_at)
|
||||
VALUES (?, ?, ?, ?, ?, ?)
|
||||
""",
|
||||
(user_id, amount, new_balance, source, reason, now),
|
||||
)
|
||||
|
||||
conn.commit()
|
||||
conn.close()
|
||||
return True, new_balance
|
||||
except Exception:
|
||||
conn.close()
|
||||
return False, 0
|
||||
|
||||
return await asyncio.to_thread(_add)
|
||||
|
||||
async def spend_points(
|
||||
self, user_id: str, amount: int, source: str, reason: str = None
|
||||
) -> Tuple[bool, int]:
|
||||
"""Spend points from user account.
|
||||
|
||||
Returns: (success, new_balance)
|
||||
"""
|
||||
# Parameter validation
|
||||
if not isinstance(amount, int) or amount <= 0:
|
||||
return False, 0
|
||||
if not user_id or not source:
|
||||
return False, 0
|
||||
|
||||
# Operation limit validation
|
||||
if self.config.POINTS_MAX_PER_OPERATION > 0:
|
||||
if amount > self.config.POINTS_MAX_PER_OPERATION:
|
||||
return False, 0
|
||||
|
||||
def _spend():
|
||||
with self._lock:
|
||||
conn = self.db.get_connection()
|
||||
cursor = conn.cursor()
|
||||
try:
|
||||
# Ensure user exists
|
||||
self.db.ensure_user_exists(user_id)
|
||||
|
||||
# Get current balance
|
||||
cursor.execute(
|
||||
"SELECT points FROM user_points WHERE user_id = ?",
|
||||
(user_id,),
|
||||
)
|
||||
row = cursor.fetchone()
|
||||
current_balance = row["points"] if row else 0
|
||||
|
||||
# Check sufficient balance
|
||||
if current_balance < amount:
|
||||
conn.close()
|
||||
return False, current_balance
|
||||
|
||||
# Update balance and total_spent
|
||||
new_balance = current_balance - amount
|
||||
now = datetime.now().isoformat()
|
||||
cursor.execute(
|
||||
"""
|
||||
UPDATE user_points
|
||||
SET points = ?, total_spent = total_spent + ?, updated_at = ?
|
||||
WHERE user_id = ?
|
||||
""",
|
||||
(new_balance, amount, now, user_id),
|
||||
)
|
||||
|
||||
# Write transaction log (amount as negative)
|
||||
cursor.execute(
|
||||
"""
|
||||
INSERT INTO point_transactions
|
||||
(user_id, amount, balance_after, source, reason, created_at)
|
||||
VALUES (?, ?, ?, ?, ?, ?)
|
||||
""",
|
||||
(user_id, -amount, new_balance, source, reason, now),
|
||||
)
|
||||
|
||||
conn.commit()
|
||||
conn.close()
|
||||
return True, new_balance
|
||||
except Exception:
|
||||
conn.close()
|
||||
return False, 0
|
||||
|
||||
return await asyncio.to_thread(_spend)
|
||||
|
||||
async def set_points(
|
||||
self, user_id: str, amount: int, source: str, reason: str = None
|
||||
) -> Tuple[bool, int]:
|
||||
"""Set user's points to exact amount.
|
||||
|
||||
Returns: (success, new_balance)
|
||||
"""
|
||||
# Parameter validation
|
||||
if not isinstance(amount, int) or amount < 0:
|
||||
return False, 0
|
||||
if not user_id or not source:
|
||||
return False, 0
|
||||
|
||||
def _set():
|
||||
with self._lock:
|
||||
conn = self.db.get_connection()
|
||||
cursor = conn.cursor()
|
||||
try:
|
||||
# Ensure user exists
|
||||
self.db.ensure_user_exists(user_id)
|
||||
|
||||
# Get current balance
|
||||
cursor.execute(
|
||||
"SELECT points, total_earned FROM user_points WHERE user_id = ?",
|
||||
(user_id,),
|
||||
)
|
||||
row = cursor.fetchone()
|
||||
current_balance = row["points"] if row else 0
|
||||
current_earned = row["total_earned"] if row else 0
|
||||
|
||||
# If new value equals old value, return without writing
|
||||
if current_balance == amount:
|
||||
conn.close()
|
||||
return True, amount
|
||||
|
||||
# Calculate difference for total_earned (only positive diff)
|
||||
diff = amount - current_balance
|
||||
earned_diff = max(0, diff)
|
||||
|
||||
# Update balance and total_earned
|
||||
now = datetime.now().isoformat()
|
||||
cursor.execute(
|
||||
"""
|
||||
UPDATE user_points
|
||||
SET points = ?, total_earned = total_earned + ?, updated_at = ?
|
||||
WHERE user_id = ?
|
||||
""",
|
||||
(amount, earned_diff, now, user_id),
|
||||
)
|
||||
|
||||
# Write transaction log
|
||||
cursor.execute(
|
||||
"""
|
||||
INSERT INTO point_transactions
|
||||
(user_id, amount, balance_after, source, reason, created_at)
|
||||
VALUES (?, ?, ?, ?, ?, ?)
|
||||
""",
|
||||
(user_id, diff, amount, source, reason, now),
|
||||
)
|
||||
|
||||
conn.commit()
|
||||
conn.close()
|
||||
return True, amount
|
||||
except Exception:
|
||||
conn.close()
|
||||
return False, 0
|
||||
|
||||
return await asyncio.to_thread(_set)
|
||||
|
||||
async def get_transactions(
|
||||
self, user_id: str, limit: int = 20, offset: int = 0
|
||||
) -> List[Dict[str, Any]]:
|
||||
"""Get transaction history for a user.
|
||||
|
||||
Returns: List of transaction dicts
|
||||
"""
|
||||
# Normalize parameters
|
||||
limit = max(1, min(100, limit))
|
||||
offset = max(0, offset)
|
||||
|
||||
def _get():
|
||||
conn = self.db.get_connection()
|
||||
cursor = conn.cursor()
|
||||
cursor.execute(
|
||||
"""
|
||||
SELECT id, user_id, amount, balance_after, source, reason, created_at
|
||||
FROM point_transactions
|
||||
WHERE user_id = ?
|
||||
ORDER BY id DESC
|
||||
LIMIT ? OFFSET ?
|
||||
""",
|
||||
(user_id, limit, offset),
|
||||
)
|
||||
rows = cursor.fetchall()
|
||||
conn.close()
|
||||
return [dict(row) for row in rows]
|
||||
|
||||
return await asyncio.to_thread(_get)
|
||||
|
||||
async def get_ranking(
|
||||
self, limit: int = 10, order_by: str = "points"
|
||||
) -> List[Dict[str, Any]]:
|
||||
"""Get points ranking.
|
||||
|
||||
Returns: List of ranking dicts with rank field
|
||||
"""
|
||||
# Normalize parameters
|
||||
limit = max(1, min(100, limit))
|
||||
if order_by not in ("points", "total_earned"):
|
||||
order_by = "points"
|
||||
|
||||
def _get():
|
||||
conn = self.db.get_connection()
|
||||
cursor = conn.cursor()
|
||||
|
||||
order_column = "points" if order_by == "points" else "total_earned"
|
||||
query = f"""
|
||||
SELECT
|
||||
RANK() OVER (ORDER BY {order_column} DESC) as rank,
|
||||
user_id,
|
||||
points,
|
||||
total_earned,
|
||||
total_spent
|
||||
FROM user_points
|
||||
ORDER BY {order_column} DESC, user_id ASC
|
||||
LIMIT ?
|
||||
"""
|
||||
cursor.execute(query, (limit,))
|
||||
rows = cursor.fetchall()
|
||||
conn.close()
|
||||
return [dict(row) for row in rows]
|
||||
|
||||
return await asyncio.to_thread(_get)
|
||||
27
danding_bot/plugins/danding_points/config.py
Normal file
27
danding_bot/plugins/danding_points/config.py
Normal file
@@ -0,0 +1,27 @@
|
||||
from pydantic import BaseSettings, validator
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
class Config(BaseSettings):
|
||||
"""Points system configuration."""
|
||||
|
||||
POINTS_DB_FILE: str = "data/danding_points/points.db"
|
||||
POINTS_MAX_BALANCE: int = 0 # 0 = unlimited
|
||||
POINTS_MAX_PER_OPERATION: int = 0 # 0 = unlimited
|
||||
POINTS_LOG_RETENTION_DAYS: int = 365
|
||||
|
||||
class Config:
|
||||
env_prefix = "DANDING_"
|
||||
case_sensitive = True
|
||||
|
||||
@validator("POINTS_MAX_BALANCE", "POINTS_MAX_PER_OPERATION", "POINTS_LOG_RETENTION_DAYS")
|
||||
def validate_non_negative(cls, v):
|
||||
if v < 0:
|
||||
raise ValueError("Value must be non-negative")
|
||||
return v
|
||||
|
||||
@validator("POINTS_DB_FILE")
|
||||
def validate_db_path(cls, v):
|
||||
if not v:
|
||||
raise ValueError("Database file path cannot be empty")
|
||||
return v
|
||||
100
danding_bot/plugins/danding_points/database.py
Normal file
100
danding_bot/plugins/danding_points/database.py
Normal file
@@ -0,0 +1,100 @@
|
||||
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()
|
||||
71
test_danding_points.py
Normal file
71
test_danding_points.py
Normal file
@@ -0,0 +1,71 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Manual test script for danding_points plugin."""
|
||||
|
||||
import asyncio
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
# Add project root to path
|
||||
sys.path.insert(0, str(Path(__file__).parent.parent.parent))
|
||||
|
||||
from danding_bot.plugins.danding_points import points_api
|
||||
|
||||
|
||||
async def test_basic_operations():
|
||||
"""Test basic points operations."""
|
||||
print("Testing basic operations...")
|
||||
|
||||
# Test 1: Get balance for non-existent user
|
||||
balance = await points_api.get_balance("test_user_1")
|
||||
assert balance == 0, f"Expected 0, got {balance}"
|
||||
print("✓ Non-existent user returns 0 balance")
|
||||
|
||||
# Test 2: Add points (auto-create user)
|
||||
success, new_balance = await points_api.add_points(
|
||||
"test_user_1", 100, "test_source", "test reason"
|
||||
)
|
||||
assert success and new_balance == 100, f"Add failed: {success}, {new_balance}"
|
||||
print("✓ Add points works and auto-creates user")
|
||||
|
||||
# Test 3: Get balance after add
|
||||
balance = await points_api.get_balance("test_user_1")
|
||||
assert balance == 100, f"Expected 100, got {balance}"
|
||||
print("✓ Balance updated correctly")
|
||||
|
||||
# Test 4: Spend points
|
||||
success, new_balance = await points_api.spend_points(
|
||||
"test_user_1", 30, "test_source", "spend reason"
|
||||
)
|
||||
assert success and new_balance == 70, f"Spend failed: {success}, {new_balance}"
|
||||
print("✓ Spend points works")
|
||||
|
||||
# Test 5: Spend more than balance (should fail)
|
||||
success, new_balance = await points_api.spend_points(
|
||||
"test_user_1", 100, "test_source", "should fail"
|
||||
)
|
||||
assert not success, "Should fail when spending more than balance"
|
||||
print("✓ Spend fails when insufficient balance")
|
||||
|
||||
# Test 6: Set points
|
||||
success, new_balance = await points_api.set_points(
|
||||
"test_user_1", 50, "test_source", "set reason"
|
||||
)
|
||||
assert success and new_balance == 50, f"Set failed: {success}, {new_balance}"
|
||||
print("✓ Set points works")
|
||||
|
||||
# Test 7: Get transactions
|
||||
transactions = await points_api.get_transactions("test_user_1", limit=10)
|
||||
assert len(transactions) > 0, "Should have transactions"
|
||||
print(f"✓ Get transactions works ({len(transactions)} transactions)")
|
||||
|
||||
# Test 8: Get ranking
|
||||
ranking = await points_api.get_ranking(limit=10)
|
||||
assert len(ranking) > 0, "Should have ranking entries"
|
||||
assert "rank" in ranking[0], "Ranking should have rank field"
|
||||
print(f"✓ Get ranking works ({len(ranking)} entries)")
|
||||
|
||||
print("\n✅ All tests passed!")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(test_basic_operations())
|
||||
Reference in New Issue
Block a user