109 lines
3.4 KiB
Python
109 lines
3.4 KiB
Python
"""
|
||
阴阳师抽卡插件 - 通用工具函数
|
||
|
||
提供常用的辅助函数：
|
||
- 用户提及格式化
|
||
- 图片路径处理
|
||
"""
|
||
|
||
import os
|
||
from typing import Optional
|
||
from pathlib import Path
|
||
|
||
def get_image_path(file_path: str) -> str:
|
||
"""获取图片的绝对路径"""
|
||
return os.path.abspath(file_path)
|
||
|
||
def format_user_mention(user_id: str, user_name: Optional[str] = None) -> str:
|
||
"""格式化用户@信息"""
|
||
display_name = user_name if user_name else f"用户{user_id}"
|
||
return f"@{display_name}"
|
||
|
||
|
||
def get_luck_description(points: int) -> tuple[str, str]:
|
||
"""根据积分返回运气描述与emoji"""
|
||
if points <= 10:
|
||
return "非酋", "😭"
|
||
if points <= 30:
|
||
return "一般", "😐"
|
||
if points <= 60:
|
||
return "小欧", "😊"
|
||
if points <= 90:
|
||
return "大欧", "🎉"
|
||
return "欧皇", "👑"
|
||
|
||
|
||
def format_sign_in_message(
|
||
user_id: str,
|
||
user_name: str,
|
||
points: int,
|
||
balance: int,
|
||
) -> str:
|
||
"""格式化签到成功消息"""
|
||
luck_text, luck_emoji = get_luck_description(points)
|
||
mention = format_user_mention(user_id, user_name)
|
||
return (
|
||
f"{mention} 📅 每日签到成功！\n"
|
||
f"🎁 获得积分：{points}\n"
|
||
f"{luck_emoji} 今日运气：{luck_text}\n"
|
||
f"💰 当前积分：{balance}"
|
||
)
|
||
|
||
|
||
|
||
def get_user_name(event) -> str:
|
||
"""从消息事件中获取用户昵称，统一多处重复逻辑。
|
||
|
||
Args:
|
||
event: NoneBot MessageEvent 对象
|
||
|
||
Returns:
|
||
str: 用户昵称，优先使用群名片(card)，其次昵称(nickname)，兜底"未知用户"
|
||
"""
|
||
from nonebot.adapters.onebot.v11 import GroupMessageEvent
|
||
if isinstance(event, GroupMessageEvent):
|
||
return event.sender.card or event.sender.nickname or "未知用户"
|
||
return event.sender.nickname or "未知用户"
|
||
|
||
|
||
async def build_achievement_notify(user_id: str, unlocked_ids: list) -> "Message | None":
|
||
"""统一的成就通知编排：ID列表 → 详情查询 → 奖励领取 → 消息格式化。
|
||
|
||
供所有handler共用，消除成就通知逻辑的重复（原则#4 变化半径小 / #12 功能越多代码越短）。
|
||
|
||
Args:
|
||
user_id: 用户ID
|
||
unlocked_ids: 新解锁的成就ID列表
|
||
|
||
Returns:
|
||
Message对象，无有效成就时返回None
|
||
"""
|
||
from .api_utils import process_achievement_reward, get_achievement_by_id
|
||
from . import formatters
|
||
|
||
achievements_data = []
|
||
for achievement_id in unlocked_ids:
|
||
ach = get_achievement_by_id(achievement_id)
|
||
if not ach:
|
||
continue
|
||
success, reward_msg = await process_achievement_reward(user_id, achievement_id)
|
||
ach["reward_msg"] = reward_msg if success else ""
|
||
ach["claimed"] = success
|
||
achievements_data.append(ach)
|
||
|
||
if not achievements_data:
|
||
return None
|
||
return formatters.format_achievement_notify(achievements_data, user_id)
|
||
|
||
|
||
# ---- GachaSystem 单例（P1#3: 避免每次handler调用都new实例） ----
|
||
_gacha_system_instance = None
|
||
|
||
def get_gacha_system():
|
||
"""获取全局唯一的GachaSystem实例（lazy init）。"""
|
||
global _gacha_system_instance
|
||
if _gacha_system_instance is None:
|
||
from .gacha import GachaSystem
|
||
_gacha_system_instance = GachaSystem()
|
||
return _gacha_system_instance
|