171 lines
6.2 KiB
Python
171 lines
6.2 KiB
Python
"""
|
||
阴阳师抽卡插件 - NoneBot2插件
|
||
|
||
提供阴阳师主题的抽卡功能，包括：
|
||
- 单次抽卡和三连抽
|
||
- 用户统计和排行榜
|
||
- 成就系统
|
||
- SSR/SP奖励发放
|
||
- 每日签到
|
||
|
||
模块结构：
|
||
- config.py: 配置管理
|
||
- gacha.py: 抽卡核心逻辑
|
||
- utils.py: 工具函数
|
||
- rules.py: 匹配规则
|
||
- formatters.py: 消息格式化
|
||
- handlers/: 命令处理器
|
||
- api_utils.py: 外部API调用
|
||
- web_api.py: Web接口
|
||
"""
|
||
|
||
import os
|
||
import logging
|
||
import random
|
||
from pathlib import Path
|
||
|
||
from nonebot import on_command, on_startswith
|
||
from nonebot.adapters.onebot.v11 import Bot, GroupMessageEvent, MessageEvent, Message
|
||
from nonebot.adapters.onebot.v11.message import MessageSegment
|
||
from nonebot.typing import T_State
|
||
|
||
from .config import Config
|
||
from .gacha import GachaSystem
|
||
from .rules import check_permission, check_rank_permission
|
||
from .utils import format_user_mention, get_image_path, format_sign_in_message
|
||
from danding_bot.plugins.danding_points import points_api
|
||
from . import formatters
|
||
from . import handlers
|
||
|
||
# 初始化配置
|
||
config = Config()
|
||
gacha_system = GachaSystem()
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# 签到积分配置
|
||
SIGN_IN_MIN_POINTS = 10
|
||
SIGN_IN_MAX_POINTS = 50
|
||
SIGN_IN_SOURCE = "gacha"
|
||
SIGN_IN_REASON = "每日抽卡签到"
|
||
|
||
# 命令别名配置
|
||
GACHA_COMMANDS = {"抽卡", "阴阳师抽卡", "十连抽"}
|
||
STATS_COMMANDS = {"我的抽卡统计", "抽卡统计"}
|
||
DAILY_STATS_COMMANDS = {"今日抽卡", "今日抽卡统计"}
|
||
TRIPLE_GACHA_COMMANDS = {"三连抽", "三次抽", "三连"}
|
||
ACHIEVEMENT_COMMANDS = {"查询成就", "抽卡成就", "成就"}
|
||
INTRO_COMMANDS = {"抽卡介绍", "抽卡帮助"}
|
||
|
||
# 定义匹配器
|
||
gacha_matcher = on_command("抽卡", aliases=set(GACHA_COMMANDS), priority=10, rule=check_permission())
|
||
stats_matcher = on_command("我的抽卡", aliases=set(STATS_COMMANDS), priority=5, rule=check_permission())
|
||
daily_stats_matcher = on_command("今日抽卡", aliases=set(DAILY_STATS_COMMANDS), priority=5, rule=check_permission())
|
||
triple_gacha_matcher = on_command("三连抽", aliases=set(TRIPLE_GACHA_COMMANDS), priority=10, rule=check_permission())
|
||
achievement_matcher = on_command("查询成就", aliases=set(ACHIEVEMENT_COMMANDS), priority=5, rule=check_permission())
|
||
query_matcher = on_command("查询抽卡", aliases={"查询抽奖"}, priority=5, rule=check_permission())
|
||
rank_matcher = on_startswith(("抽卡排行", "抽卡榜"), priority=1, rule=check_rank_permission())
|
||
intro_matcher = on_command("抽卡介绍", aliases=set(INTRO_COMMANDS), priority=5, rule=check_permission())
|
||
|
||
|
||
async def try_handle_daily_sign_in(matcher, user_id: str, user_name: str) -> None:
|
||
"""
|
||
处理抽卡成功后的每日签到，不影响主流程。
|
||
|
||
Args:
|
||
matcher: NoneBot匹配器实例，用于发送消息
|
||
user_id: 用户ID
|
||
user_name: 用户昵称
|
||
|
||
Returns:
|
||
None
|
||
|
||
Side Effects:
|
||
- 检查用户今日是否已签到
|
||
- 如未签到，随机发放积分奖励
|
||
- 记录签到状态
|
||
- 发送签到通知消息
|
||
"""
|
||
try:
|
||
if gacha_system.data_manager.has_signed_in_today(user_id):
|
||
return
|
||
|
||
points = random.randint(SIGN_IN_MIN_POINTS, SIGN_IN_MAX_POINTS)
|
||
success, new_balance = await points_api.add_points(
|
||
user_id,
|
||
points,
|
||
SIGN_IN_SOURCE,
|
||
SIGN_IN_REASON,
|
||
)
|
||
if not success:
|
||
logger.error("抽卡签到积分发放失败 user_id=%s points=%s", user_id, points)
|
||
return
|
||
|
||
if not gacha_system.data_manager.record_sign_in(user_id, points):
|
||
logger.warning("抽卡签到落库冲突，积分已发放但签到记录重复 user_id=%s", user_id)
|
||
return
|
||
|
||
await matcher.send(format_sign_in_message(user_id, user_name, points, new_balance))
|
||
except Exception:
|
||
logger.exception("处理抽卡签到失败 user_id=%s", user_id)
|
||
|
||
|
||
# 注册命令处理器
|
||
@gacha_matcher.handle()
|
||
async def handle_gacha_wrapper(bot: Bot, event: MessageEvent, state: T_State) -> None:
|
||
"""单次抽卡命令处理器"""
|
||
await handlers.handle_gacha(bot, event, state)
|
||
# 签到处理：逻辑从handlers/gacha.py移至matcher层（遵循职责边界：matcher层负责编排，handler层负责业务）
|
||
user_id = str(event.user_id)
|
||
user_name = event.sender.card or event.sender.nickname or "未知用户"
|
||
await try_handle_daily_sign_in(gacha_matcher, user_id, user_name)
|
||
|
||
|
||
@triple_gacha_matcher.handle()
|
||
async def handle_triple_gacha_wrapper(bot: Bot, event: MessageEvent, state: T_State) -> None:
|
||
"""三连抽命令处理器"""
|
||
await handlers.handle_triple_gacha(bot, event, state)
|
||
|
||
|
||
@stats_matcher.handle()
|
||
async def handle_stats_wrapper(bot: Bot, event: MessageEvent, state: T_State) -> None:
|
||
"""个人统计查询命令处理器"""
|
||
await handlers.handle_stats(bot, event, state)
|
||
|
||
|
||
@query_matcher.handle()
|
||
async def handle_query_wrapper(bot: Bot, event: MessageEvent, state: T_State) -> None:
|
||
"""他人统计查询命令处理器"""
|
||
await handlers.handle_query(bot, event, state)
|
||
|
||
|
||
@rank_matcher.handle()
|
||
async def handle_rank_wrapper(bot: Bot, event: MessageEvent, state: T_State) -> None:
|
||
"""排行榜查询命令处理器"""
|
||
await handlers.handle_rank(bot, event, state)
|
||
|
||
|
||
@daily_stats_matcher.handle()
|
||
async def handle_daily_stats_wrapper(bot: Bot, event: MessageEvent, state: T_State) -> None:
|
||
"""今日统计查询命令处理器"""
|
||
await handlers.handle_daily_stats(bot, event, state)
|
||
|
||
|
||
@achievement_matcher.handle()
|
||
async def handle_achievement_wrapper(bot: Bot, event: MessageEvent, state: T_State) -> None:
|
||
"""成就查询命令处理器"""
|
||
await handlers.handle_achievement(bot, event, state)
|
||
|
||
|
||
@intro_matcher.handle()
|
||
async def handle_intro_wrapper(bot: Bot, event: MessageEvent, state: T_State) -> None:
|
||
"""插件介绍命令处理器"""
|
||
await handlers.handle_intro(bot, event, state)
|
||
|
||
|
||
# 注册Web API路由
|
||
try:
|
||
from . import web_api
|
||
web_api.register_web_routes()
|
||
except Exception as e:
|
||
logger.error(f"注册 onmyoji_gacha Web 路由失败: {e}")
|