🛠️ 插件开发文档

开始开发插件前,请确保具备以下环境与知识储备:

一个标准插件文件由 六个部分 组成:导入区、模块级变量、Matcher 注册、 Handler 处理函数、生命周期钩子、菜单注册。下面是一个完整可运行的"打招呼"插件骨架:

# plugins/group_hello.py
# 打招呼插件 - 标准插件骨架示例

# ===== 1. 导入区 =====
# 标准库
from typing import Dict, Any

# core 框架(事件注册 / 生命周期 / 菜单 / 权限)
from core.matcher import on_command, on_message, on_notice, on_request
from core.plugin_loader import on_startup, on_shutdown, on_bot_connect
from core.menu_registry import menu_registry
from core.permission import SUPERUSER, GROUP_ADMIN, GROUP_OWNER

# OneBot 协议
from core.onebot.bot import Bot
from core.onebot.event import GroupMessageEvent, PrivateMessageEvent, NoticeEvent
from core.onebot.models import Message

# 配置与日志
from config_manager import config_manager
from log_manager import log_manager

# 插件工具(统一消息构造 / 数据持久化)
from plugins.utils import reply, JsonDataManager

logger = log_manager.get_logger("group_hello")


# ===== 2. 模块级变量 =====
plugin_name = "group_hello"
hello_count: Dict[str, int] = {}


# ===== 3. Matcher 注册 =====
# 命令注册:触发词"打招呼",优先级 1,执行后阻断后续 matcher
hello = on_command(
    "打招呼",
    priority=1,
    block=True,
    permission=GROUP_ADMIN | GROUP_OWNER | SUPERUSER,
    aliases={"hello", "你好"},
)

# 消息监听:priority=99 用于消息收集,不阻断
msg_collector = on_message(priority=99, block=False)

# 通知事件监听
notice_listener = on_notice(priority=1, block=False)

# 请求事件监听
request_listener = on_request(priority=1, block=False)


# ===== 4. Handler 处理函数 =====
@hello.handle()
async def handle_hello(bot: Bot, event: GroupMessageEvent, matcher, args: Message):
    """处理打招呼命令"""
    user_id = event.user_id
    user_name = event.sender.card or event.sender.nickname or str(user_id)

    # 统计打招呼次数
    count = hello_count.get(str(user_id), 0) + 1
    hello_count[str(user_id)] = count

    # 使用统一回复函数(自动识别群/私聊)
    await reply(event, f"你好,{user_name}!这是你第 {count} 次打招呼 👋")


@msg_collector.handle()
async def handle_collect(bot: Bot, event, matcher):
    """消息收集(仅统计,不回复)"""
    pass


@notice_listener.handle()
async def handle_notice(bot: Bot, event: NoticeEvent, matcher):
    """通知事件(如撤回、加群等)"""
    pass


@request_listener.handle()
async def handle_request(bot: Bot, event, matcher):
    """请求事件(如加好友、加群请求)"""
    pass


# ===== 5. 生命周期钩子 =====
@on_startup
async def _on_startup():
    logger.info("打招呼插件已加载")


@on_bot_connect
async def _on_bot_connect(bot: Bot):
    logger.info(f"Bot 已连接:{bot.self_id}")


@on_shutdown
async def _on_shutdown():
    logger.info("打招呼插件已卸载")


# ===== 6. 菜单注册(文件底部) =====
menu_registry.register(
    category="娱乐",
    item_name="打招呼",
    text="👋 打招呼 - 和机器人打招呼",
    category_title="🎉◇━娱乐功能━◇🎉",
    category_trigger="娱乐",
    category_description="群内娱乐互动功能",
)

框架提供四种事件注册函数,分别对应不同的事件类型:

on_command(cmd, priority, block, permission, aliases)

注册命令。当用户发送以命令前缀开头且匹配 cmd 或 aliases 的消息时触发。

# 基础命令
checkin = on_command("签到", priority=1, block=True)

# 带别名和权限
admin_cmd = on_command(
    "禁言",
    priority=1,
    block=True,
    permission=GROUP_ADMIN | GROUP_OWNER | SUPERUSER,
    aliases={"mute", "禁言用户"},
)

on_message(priority, block)

监听所有消息事件,常用于日志采集、消息统计、关键词匹配等场景。

# priority=-1 最先执行,用于日志采集
logger_matcher = on_message(priority=-1, block=False)

on_notice(priority, block)

监听通知事件,如群成员变动、消息撤回、戳一戳等。

notice = on_notice(priority=1, block=False)

on_request(priority, block)

监听请求事件,如加好友请求、加群请求/邀请。

request = on_request(priority=1, block=False)

priority 优先级说明

• 越小越先执行,多个 matcher 按优先级升序依次执行
• 1 - 常规命令的默认优先级
• -1 - 用于日志采集,确保最先执行
• 99 - 用于消息收集,确保最后执行

block 阻断说明

• True - 当前 matcher 执行完毕后,停止后续所有 matcher 的执行
• False - 当前 matcher 执行完毕后,继续执行后续 matcher

Handler 是实际处理事件的异步函数,通过 @matcher.handle() 装饰器注册。 框架的 EventBus._build_kwargs 会根据函数签名的参数名自动完成依赖注入,无需手动传参。

参数注入机制

Matcher 方法

@hello.handle()
async def handle_hello(bot: Bot, event: GroupMessageEvent, matcher, args: Message, state):
    # 发送消息但不结束 handler(可继续执行后续代码)
    await matcher.send("正在处理...")

    # 发送消息并结束当前 handler
    await matcher.finish("处理完成!")

    # 拒绝当前输入,重新等待用户输入
    await matcher.reject("输入有误,请重新输入:")

    # 暂停,等待用户下一条消息(用于多轮交互)
    await matcher.pause("请发送内容:")

    # 获取纯文本(剥离消息段)
    text = matcher.get_plaintext()

    # 停止执行(不发送消息,直接中断)
    matcher.stop()

plugins/utils.py 提供了一系列统一的消息构造函数。所有插件应优先使用这些函数, 以保证回复格式统一(如群消息自动引用+@、私聊直接发送等)。

from plugins.utils import (
    reply, reply_msg, reply_private,
    at_msg, text_msg, at, image, face, record,
)

# 群消息回复:引用 + @用户 + 内容
await reply_msg(event, "收到!")

# 私聊回复
await reply_private(event, "私聊消息")

# 通用回复:自动识别群/私聊(最常用)
await reply(event, "通用回复")

# @用户 + 内容(通知场景)
await bot.send(event, at_msg(user_id, "请注意!"))

# 纯文本
await bot.send(event, text_msg("纯文本消息"))

# @消息段
msg = at(123456789)

# 图片消息段(支持 url / file / path / base64)
msg = image(url="https://example.com/pic.jpg")
msg = image(file="pic.jpg")

# QQ 表情
msg = face(face_id=178)

# 语音消息
msg = record(file="voice.mp3")

框架内置三种权限等级,可用于命令鉴权:

权限组合

使用 | 运算符组合多个权限,满足任意一个即通过:

from core.permission import SUPERUSER, GROUP_ADMIN, GROUP_OWNER

# 组合权限:群管理员 或 群主 或 超级用户 均可使用
permission=GROUP_ADMIN | GROUP_OWNER | SUPERUSER

方式一:在 on_command 中声明(推荐)

restart = on_command(
    "重启",
    priority=1,
    block=True,
    permission=GROUP_ADMIN | GROUP_OWNER | SUPERUSER,
)

方式二:在 handler 内手动判断

@restart.handle()
async def handle_restart(bot: Bot, event, matcher):
    if not await SUPERUSER(bot, event):
        await matcher.finish("仅超级管理员可使用此命令")
    await matcher.send("正在重启...")

通过 menu_registry.register() 将命令注册到菜单系统,用户发送"菜单"或分类触发词时可查看。 建议在插件文件底部统一注册。

参数说明

批量注册示例

from core.menu_registry import menu_registry

# 同一分类的公共配置
GROUP_ADMIN_MENU = {
    "category": "群管理",
    "category_title": "🔧◇━群管理━◇🔧",
    "category_trigger": "群管理",
    "category_description": "群聊授权与管理功能",
}

# 批量注册该分类下的多个命令
for item_name, text in [
    ("授权群聊", "🔧 授权群聊"),
    ("禁言", "🔇 禁言用户"),
    ("踢人", "👢 踢出成员"),
    ("全体禁言", "🔕 全体禁言"),
]:
    menu_registry.register(
        category=GROUP_ADMIN_MENU["category"],
        item_name=item_name,
        text=text,
        category_title=GROUP_ADMIN_MENU["category_title"],
        category_trigger=GROUP_ADMIN_MENU["category_trigger"],
        category_description=GROUP_ADMIN_MENU["category_description"],
    )

plugins/utils.py 提供了 JsonDataManager 基类,封装了 JSON 文件的懒加载、 延迟保存、脏数据标记等机制,数据文件自动存放在 config/data/ 目录下。

JsonDataManager 方法

完整示例:计数器数据管理器

from plugins.utils import JsonDataManager

class CounterManager(JsonDataManager):
    """计数器数据管理器"""

    def __init__(self):
        super().__init__(
            filename="counter.json",
            default_data={"total": 0, "users": {}},
        )

    def add(self, user_id: str, count: int = 1):
        """增加计数(高频写入使用 mark_dirty 延迟保存)"""
        data = self.load()
        data["total"] += count
        data["users"][user_id] = data["users"].get(user_id, 0) + count
        self.mark_dirty()

    def get_total(self):
        return self.load()["total"]

    def get_user_count(self, user_id: str):
        return self.load()["users"].get(user_id, 0)


# 使用
counter = CounterManager()
counter.add("123456789")
print(counter.get_total())  # 1

Bot 实例封装了所有 OneBot v11 API,均以 async 方法形式提供。在 handler 中通过注入的 bot 参数调用。

方法	参数	说明
send	event, message	发送消息,自动识别群/私聊
send_group_msg	group_id, message	发送群消息
send_private_msg	user_id, message	发送私聊消息
delete_msg	message_id	撤回消息
get_group_list	-	获取群列表
get_group_info	group_id, no_cache	获取群信息
get_group_member_list	group_id, no_cache	获取群成员列表
get_group_member_info	group_id, user_id, no_cache	获取群成员信息
get_stranger_info	user_id, no_cache	获取陌生人信息
get_friend_list	-	获取好友列表
set_group_whole_ban	group_id, enable	全体禁言(enable=True 开启)
set_group_ban	group_id, user_id, duration	禁言用户(duration 单位:秒,0 表示解禁)
set_group_kick	group_id, user_id, reject_add_request	踢出群成员
set_group_admin	group_id, user_id, enable	设置/取消群管理员
set_group_leave	group_id	退出群聊
send_like	user_id, times	送赞(times:每天最多 10 次)
set_group_special_title	group_id, user_id, special_title	设置群专属头衔
set_group_name	group_id, group_name	设置群名
set_group_card	group_id, user_id, card	设置群名片
send_group_notice	group_id, content	发送群公告
delete_friend	user_id	删除好友
set_friend_add_request	flag, approve, remark	处理加好友请求
set_group_add_request	flag, sub_type, approve	处理加群请求/邀请
get_msg	message_id	获取消息详情
get_login_info	-	获取登录号信息(机器人自身)
set_essence_msg	message_id	设为精华消息
delete_essence_msg	message_id	移出精华消息
get_essence_msg_list	group_id	获取群精华消息列表
call_api	action, **params	通用 API 调用(未封装的 API 用这个)

调用示例

# 禁言某用户 60 秒
await bot.set_group_ban(group_id=event.group_id, user_id=123456, duration=60)

# 撤回消息
await bot.delete_msg(message_id=event.message_id)

# 通用 API 调用(如调用未封装的 API)
await bot.call_api(action="get_group_honor_info", group_id=123456, type="talkative")

下面是一个完整的"群签到增强版"插件,综合运用了 on_command 命令注册、 reply() 统一回复、JsonDataManager 数据持久化、 menu_registry 菜单注册、SUPERUSER 权限、on_shutdown 资源清理。

# plugins/group_checkin_pro.py
# 群签到增强版 - 综合示例插件

from typing import Dict
from datetime import datetime, date

from core.matcher import on_command
from core.plugin_loader import on_shutdown
from core.menu_registry import menu_registry
from core.permission import SUPERUSER
from core.onebot.bot import Bot
from core.onebot.event import GroupMessageEvent
from core.onebot.models import Message

from log_manager import log_manager
from plugins.utils import reply, JsonDataManager

logger = log_manager.get_logger("checkin_pro")


class CheckinDataManager(JsonDataManager):
    """签到数据管理器"""

    def __init__(self):
        super().__init__(
            filename="checkin_pro.json",
            default_data={"records": {}, "total_count": 0},
        )

    def checkin(self, user_id: str, today: str):
        """执行签到,返回结果字典"""
        data = self.load()
        records = data["records"]

        if today in records.get(user_id, {}):
            return {"ok": False, "msg": "今天已经签到过了~"}

        # 写入记录
        records.setdefault(user_id, {})[today] = datetime.now().strftime("%H:%M")
        data["total_count"] = data.get("total_count", 0) + 1
        self.mark_dirty()

        return {"ok": True, "msg": f"签到成功!本群已累计签到 {data['total_count']} 次"}

    def get_stats(self, user_id: str):
        """获取统计信息"""
        data = self.load()
        return {
            "total": data.get("total_count", 0),
            "user_days": len(data["records"].get(user_id, {})),
        }

    def reset(self):
        """重置全部数据"""
        self.data = {"records": {}, "total_count": 0}
        self.save()


checkin_data = CheckinDataManager()

# ===== 命令注册 =====
checkin_cmd = on_command("签到", priority=1, block=True, aliases={"打卡"})
checkin_stats = on_command("签到统计", priority=1, block=True)
reset_cmd = on_command("重置签到", priority=1, block=True, permission=SUPERUSER)


# ===== Handler =====
@checkin_cmd.handle()
async def handle_checkin(bot: Bot, event: GroupMessageEvent, matcher, args: Message):
    user_id = str(event.user_id)
    today = date.today().strftime("%Y-%m-%d")
    result = checkin_data.checkin(user_id, today)
    await reply(event, result["msg"])


@checkin_stats.handle()
async def handle_stats(bot: Bot, event: GroupMessageEvent, matcher):
    user_id = str(event.user_id)
    stats = checkin_data.get_stats(user_id)
    user_name = event.sender.card or event.sender.nickname or user_id
    await reply(
        event,
        f"📊 {user_name} 的签到统计\n"
        f"个人签到:{stats['user_days']} 天\n"
        f"全群总计:{stats['total']} 次",
    )


@reset_cmd.handle()
async def handle_reset(bot: Bot, event: GroupMessageEvent, matcher):
    # 双重校验:on_command 已声明 permission,这里再手动确认
    if not await SUPERUSER(bot, event):
        await matcher.finish("仅超级管理员可重置签到数据")
    checkin_data.reset()
    await matcher.finish("签到数据已重置 ✅")


# ===== 生命周期钩子 =====
@on_shutdown
async def _on_shutdown():
    """关闭时强制保存数据"""
    checkin_data.shutdown()
    logger.info("签到插件数据已保存")


# ===== 菜单注册 =====
menu_registry.register(
    category="签到",
    item_name="签到",
    text="📝 签到 - 每日打卡",
    category_title="📝◇━签到打卡━◇📝",
    category_trigger="签到",
    category_description="每日签到与统计",
)
menu_registry.register(
    category="签到",
    item_name="签到统计",
    text="📊 签到统计",
    category_title="📝◇━签到打卡━◇📝",
    category_trigger="签到",
    category_description="每日签到与统计",
)

框架支持用户自行开发插件,并通过在线更新机制保护自定义插件不被覆盖。

• 自定义插件目录: 用户开发的插件放在 plugins/ 目录下,框架启动时自动加载
• 内置插件清单: 框架内置插件清单定义在 core/updater.py 的 _BUILTIN_PLUGINS 中
• 在线更新安全: 在线更新时只覆盖内置插件,自定义插件原样保留,不会丢失
• 命名建议: 自定义插件建议加前缀避免与内置插件冲突,如 custom_xxx.py

# 推荐的自定义插件命名
plugins/custom_weather.py      ✅ 加 custom_ 前缀
plugins/custom_bilibili.py     ✅
plugins/my_tool.py             ✅ 普通命名

# 错误命名(会被加载器跳过)
plugins/_helper.py             ❌ 以下划线开头
plugins/utils.py               ❌ 与框架工具模块重名

print 大法

最直接的调试方式,输出会打印到控制台:

@hello.handle()
async def handle_hello(bot: Bot, event, matcher, args: Message):
    print(f"[DEBUG] user_id={event.user_id}, args={args}")
    print(f"[DEBUG] event type={type(event)}")

使用 log_manager 日志

推荐使用框架的 log_manager,日志带模块名、时间戳,便于过滤:

from log_manager import log_manager
logger = log_manager.get_logger("my_plugin")

logger.info("插件已加载")
logger.warning("配置项缺失,使用默认值")
logger.error("数据处理失败")
logger.debug(f"接收到的参数:{args}")

热重载

修改插件代码后,无需重启框架,通过 WebUI 插件管理页面即可重新加载指定插件,即时生效。

检查数据文件

插件的数据文件保存在 config/data/ 目录下。调试时可直接查看对应 JSON 文件, 确认数据是否正确读写:

config/data/
├── checkin_pro.json      # 签到数据
├── counter.json          # 计数器数据
├── group_admin.json      # 群管理数据
└── ...                   # 其他插件数据