feat: 添加积分查询插件,提供用户积分查询相关命令
- 新增积分查询插件,包含我的积分、积分查询、积分排行和积分历史查询命令 - 支持群组和私聊场景,排行榜功能仅限群组使用 - 实现用户显示名称优先级(群昵称 > 昵称 > 用户ID) - 添加详细的帮助文档和使用说明
This commit is contained in:
163
danding_bot/plugins/danding_points_query/commands.py
Normal file
163
danding_bot/plugins/danding_points_query/commands.py
Normal file
@@ -0,0 +1,163 @@
|
||||
from nonebot import on_command, require
|
||||
from nonebot.adapters.onebot.v11 import Bot, Event, GroupMessageEvent, Message, MessageSegment
|
||||
from nonebot.params import CommandArg
|
||||
|
||||
require("danding_bot.plugins.danding_points")
|
||||
from danding_bot.plugins.danding_points import points_api
|
||||
|
||||
# Command handlers
|
||||
help_cmd = on_command("积分帮助", priority=5)
|
||||
my_points_cmd = on_command("我的积分", priority=5)
|
||||
query_points_cmd = on_command("积分查询", priority=5)
|
||||
ranking_cmd = on_command("积分排行", priority=5)
|
||||
history_cmd = on_command("积分历史查询", priority=5)
|
||||
|
||||
|
||||
async def _get_user_name(bot: Bot, event: Event, user_id: str) -> str:
|
||||
"""Get user display name (group card > nickname > user_id)."""
|
||||
try:
|
||||
if isinstance(event, GroupMessageEvent):
|
||||
info = await bot.get_group_member_info(
|
||||
group_id=event.group_id, user_id=int(user_id)
|
||||
)
|
||||
return info.get("card") or info.get("nickname") or user_id
|
||||
except Exception:
|
||||
pass
|
||||
return user_id
|
||||
|
||||
|
||||
def _parse_at_user(message: Message) -> str | None:
|
||||
"""Extract user_id from @mention in message."""
|
||||
for segment in message:
|
||||
if segment.type == "at":
|
||||
return str(segment.data.get("qq"))
|
||||
return None
|
||||
|
||||
|
||||
@help_cmd.handle()
|
||||
async def handle_help():
|
||||
"""Show points system help."""
|
||||
help_text = """📚 积分系统帮助
|
||||
|
||||
【查询命令】
|
||||
• 我的积分
|
||||
查询你的积分余额
|
||||
|
||||
• 积分查询 @用户 / 积分查询 用户ID
|
||||
查询指定用户的积分余额
|
||||
例:积分查询 @张三 或 积分查询 123456789
|
||||
|
||||
• 积分排行
|
||||
查看积分排行榜前10名(仅群组可用)
|
||||
|
||||
• 积分历史查询 [@用户 / 用户ID]
|
||||
查询最近5条积分变动记录
|
||||
例:积分历史查询(查自己)
|
||||
积分历史查询 @李四
|
||||
积分历史查询 987654321
|
||||
|
||||
【积分来源】
|
||||
• 赛马参赛:获得参赛奖励
|
||||
• 赛马冠军:获得冠军奖励
|
||||
• 赛马下注:下注获胜可获得奖励
|
||||
|
||||
【积分用途】
|
||||
• 赛马下注:消费积分进行下注
|
||||
|
||||
【其他】
|
||||
• 积分帮助
|
||||
显示此帮助信息"""
|
||||
await help_cmd.finish(help_text)
|
||||
|
||||
|
||||
@my_points_cmd.handle()
|
||||
async def handle_my_points(bot: Bot, event: Event):
|
||||
"""Query current user's points."""
|
||||
user_id = str(event.user_id)
|
||||
balance = await points_api.get_balance(user_id)
|
||||
user_name = await _get_user_name(bot, event, user_id)
|
||||
await my_points_cmd.finish(f"{user_name} 的积分余额:{balance}")
|
||||
|
||||
|
||||
@query_points_cmd.handle()
|
||||
async def handle_query_points(bot: Bot, event: Event, arg: Message = CommandArg()):
|
||||
"""Query specific user's points."""
|
||||
# Try to parse @mention first
|
||||
user_id = _parse_at_user(arg)
|
||||
|
||||
# If no @mention, try to parse user_id from text
|
||||
if not user_id:
|
||||
text = arg.extract_plain_text().strip()
|
||||
if text.isdigit():
|
||||
user_id = text
|
||||
else:
|
||||
await query_points_cmd.finish("请输入用户ID或@用户")
|
||||
return
|
||||
|
||||
balance = await points_api.get_balance(user_id)
|
||||
user_name = await _get_user_name(bot, event, user_id)
|
||||
await query_points_cmd.finish(f"{user_name} 的积分余额:{balance}")
|
||||
|
||||
|
||||
@ranking_cmd.handle()
|
||||
async def handle_ranking(bot: Bot, event: Event):
|
||||
"""Query top 10 points ranking."""
|
||||
if not isinstance(event, GroupMessageEvent):
|
||||
await ranking_cmd.finish("此命令仅在群组中可用")
|
||||
return
|
||||
|
||||
ranking = await points_api.get_ranking(limit=10, order_by="points")
|
||||
|
||||
if not ranking:
|
||||
await ranking_cmd.finish("暂无排行数据")
|
||||
return
|
||||
|
||||
lines = ["🏆 积分排行榜 TOP 10\n"]
|
||||
for entry in ranking:
|
||||
user_id = entry["user_id"]
|
||||
user_name = await _get_user_name(bot, event, user_id)
|
||||
points = entry["points"]
|
||||
rank = entry["rank"]
|
||||
lines.append(f"#{rank:2d} {user_name} {points} 分")
|
||||
|
||||
await ranking_cmd.finish("\n".join(lines))
|
||||
|
||||
|
||||
@history_cmd.handle()
|
||||
async def handle_history(bot: Bot, event: Event, arg: Message = CommandArg()):
|
||||
"""Query user's recent 5 point transactions."""
|
||||
# Try to parse @mention first
|
||||
user_id = _parse_at_user(arg)
|
||||
|
||||
# If no @mention, try to parse user_id from text or use current user
|
||||
if not user_id:
|
||||
text = arg.extract_plain_text().strip()
|
||||
if text.isdigit():
|
||||
user_id = text
|
||||
else:
|
||||
user_id = str(event.user_id)
|
||||
|
||||
transactions = await points_api.get_transactions(user_id, limit=5, offset=0)
|
||||
|
||||
if not transactions:
|
||||
user_name = await _get_user_name(bot, event, user_id)
|
||||
await history_cmd.finish(f"{user_name} 暂无积分变动记录")
|
||||
return
|
||||
|
||||
user_name = await _get_user_name(bot, event, user_id)
|
||||
lines = [f"📊 {user_name} 的积分变动记录(最近5条)\n"]
|
||||
|
||||
for tx in transactions:
|
||||
amount = tx["amount"]
|
||||
balance_after = tx["balance_after"]
|
||||
source = tx["source"]
|
||||
reason = tx["reason"] or source
|
||||
created_at = tx["created_at"]
|
||||
|
||||
# Format amount with sign
|
||||
amount_str = f"{amount:+d}"
|
||||
lines.append(
|
||||
f"{created_at} {amount_str:>6s} 余额: {balance_after} {reason}"
|
||||
)
|
||||
|
||||
await history_cmd.finish("\n".join(lines))
|
||||
Reference in New Issue
Block a user