插件系统

约 6984 字大约 23 分钟

本章介绍 EasyBot SDK 的插件系统，包括命令注册、预处理器、权限控制和热重载功能。

一、概述

模块: easybot.plugins
导入: from easybot import Plugins

EasyBot 提供两种命令注册方式：

方式	适用场景	特点
`@Plugins.on_command()`	插件文件、多模块项目	类方法装饰器，全局共享
`@bot.on_command()`	单文件程序	实例方法装饰器，快速开发

两种方式的本质区别：Plugins 是类级别的注册表，所有 Bot 实例共享同一份命令集合；bot.on_command 是实例级别的便捷写法，内部仍然委托给 Plugins 注册表。因此无论用哪种方式注册的命令，对所有 Bot 实例都可见。混合使用时无优先级差异，按注册顺序匹配。

二、命令注册

2.1 使用 Plugins 类装饰器（推荐）

from easybot import CommandValidScenes, Model, Plugins

@Plugins.on_command(command="/hello")
async def cmd_hello(
    msg: Model.GuildMessage | Model.GroupMessage | Model.C2CMessage | Model.DirectMessage,
) -> None:
    await msg.reply("Hello!")

@Plugins.on_command(regex=r"^计算\s+(.+)$")
async def cmd_calc(
    msg: Model.GuildMessage | Model.GroupMessage | Model.C2CMessage | Model.DirectMessage,
) -> None:
    expression = msg.treated_msg[0]
    result = eval(expression, {"__builtins__": {}})
    await msg.reply(f"= {result}")

2.2 使用 Bot 实例装饰器

from easybot import Model

@bot.on_command(command="/ping")
async def cmd_ping(
    msg: Model.GuildMessage | Model.GroupMessage | Model.C2CMessage | Model.DirectMessage,
) -> None:
    await msg.reply("Pong!")

@bot.on_command(command=["/help", "帮助"])
async def cmd_help(
    msg: Model.GuildMessage | Model.GroupMessage | Model.C2CMessage | Model.DirectMessage,
) -> None:
    await msg.reply("可用命令:\n/ping - 测试连接")

2.3 装饰器参数

参数	类型	默认值	说明
`command`	`str \| list[str]`	—	文本命令列表，与 regex 二选一
`regex`	`Pattern \| str \| Iterable[Pattern \| str]`	—	正则表达式或正则列表，与 command 二选一
`is_treat`	`bool`	`True`	是否覆写 treated_msg（详见下方说明）
`is_require_at`	`bool`	`False`	是否要求 @机器人
`is_short_circuit`	`bool`	`True`	触发后是否短路（详见下方短路机制）
`is_custom_short_circuit`	`bool`	`False`	根据回调返回值决定是否短路（优先于 is_short_circuit）
`is_require_admin`	`bool`	`False`	是否要求频道管理员（仅频道场景有效）
`admin_error_msg`	`str \| None`	`None`	频道管理员权限不足时的回复内容
`is_require_bot_admin`	`bool`	`False`	是否要求机器人管理员（全局白名单）
`bot_admin_error_msg`	`str \| None`	`None`	机器人管理员权限不足时的回复内容
`valid_scenes`	`CommandValidScenes`	ALL	有效场景（支持位运算组合，详见§2.4）
`enabled`	`bool`	`True`	是否启用

`is_treat` 参数详解

is_treat 控制命令匹配成功后是否覆写 msg.treated_msg 字段：

is_treat=True（默认）：
- 文本命令：treated_msg 被更新为去除当前命令前缀后的剩余内容
- 正则命令：treated_msg 被更新为正则的捕获组元组 tuple
is_treat=False：
- treated_msg 保持命令匹配前的原始值（即已去掉 @机器人 和开头 / 的消息内容）

简单记忆：大多数情况下保持默认 True 即可。当你需要获取原始消息全文而不想被命令前缀截断时，设为 False。

短路机制详解

SDK 的"短路"指的是：命令匹配成功并执行后，是否阻止后续命令继续匹配以及事件处理器执行。三种模式：

模式	参数配置	行为
默认短路	`is_short_circuit=True`, `is_custom_short_circuit=False`	命令执行完后立即短路，不执行后续命令和事件处理器
无短路	`is_short_circuit=False`, `is_custom_short_circuit=False`	命令执行完后不短路，继续尝试下一个候选命令，最后到达事件处理器
自定义短路	`is_custom_short_circuit=True`	命令函数的返回值决定是否短路：返回 `True` → 短路；返回 `False` → 不短路

注意：当 is_short_circuit 与 is_custom_short_circuit 同时为 True 时，is_custom_short_circuit 优先生效。
短路的边界：短路只阻止同一消息事件内的后续命令匹配和事件处理器调用。它不影响其他独立事件、定时器、会话等。

`is_custom_short_circuit` 使用示例

当 is_custom_short_circuit=True 时，命令函数的返回值决定是否短路：

@Plugins.on_command(command="check", is_custom_short_circuit=True)
async def check_command(msg: Message, treated_msg: str):
    """根据条件决定是否阻止后续命令"""
    if "pass" in treated_msg:
        await msg.reply("检查通过，放行")
        return True   # → 短路，后续命令和事件处理器不再执行
    else:
        await msg.reply("检查未通过，交给下一条命令处理")
        return False  # → 不短路，继续匹配下一个候选命令

典型使用场景：

条件门控：只有满足特定条件时才"消费"这条消息
日志/审计：记录所有匹配但不拦截的命令
链式命令：让多个命令依次对同一条消息进行处理

2.4 valid_scenes 有效场景

from easybot import CommandValidScenes, Model

@Plugins.on_command(
    command="/guild_cmd",
    valid_scenes=CommandValidScenes.GUILD
)
async def guild_only(msg: Model.GuildMessage) -> None:
    await msg.reply("仅频道可用")

场景常量：

常量	值	说明
`GUILD`	1	频道消息
`DM`	2	频道私信
`GROUP`	4	QQ 群聊消息
`C2C`	8	QQ 单聊消息
`ALL`	15	全部场景

位运算组合：CommandValidScenes 基于 IntEnum，常量值采用 2 的幂次设计，支持使用 |（按位或）组合多个场景。例如：
# 仅允许频道和群聊，不允许私信和单聊
valid_scenes=CommandValidScenes.GUILD | CommandValidScenes.GROUP  # 值为 5
# 排除单聊
valid_scenes=CommandValidScenes.ALL & ~CommandValidScenes.C2C     # 值为 7

2.5 treated_msg 属性

命令匹配后，消息对象的 treated_msg 字段会被更新：

匹配模式	`treated_msg` 值
文本命令 `is_treat=True`	去除命令前缀后的字符串
文本命令 `is_treat=False`	原始消息内容
正则命令	正则捕获组元组 `tuple`

补充说明：

在进入命令匹配前，SDK 会先对消息做一次基础清洗：去掉 @机器人 和开头的 /，并把结果写入 treated_msg
当命中以 / 开头的文本命令且 is_treat=True 时，treated_msg 会再次更新为"去掉当前命令前缀后的内容"
当命中正则命令且 is_treat=True 时，treated_msg 会更新为 match.groups()

from easybot import Model

@Plugins.on_command(command="/echo")
async def echo(
    msg: Model.GuildMessage | Model.GroupMessage | Model.C2CMessage | Model.DirectMessage,
) -> None:
    # 用户输入 "/echo hello world"
    # msg.treated_msg == "hello world"
    await msg.reply(msg.treated_msg)

@bot.on_command(regex=r"^计算\s+(.+)$")
async def calc(
    msg: Model.GuildMessage | Model.GroupMessage | Model.C2CMessage | Model.DirectMessage,
) -> None:
    # 用户输入 "计算 1+1"
    # msg.treated_msg == ("1+1",)
    expression = msg.treated_msg[0]
    await msg.reply(f"= {eval(expression)}")

2.6 `_raw_data` 属性（自定义数据透传）

msg._raw_data 是消息对象上的一个 dict | None 类型属性，初始值为 None，用于在预处理器和命令函数之间传递结构化数据。

设计目的：SDK 不预设任何字段名，完全由开发者自由定义。典型用法是在预处理器中解析/校验参数后写入 _raw_data，命令函数再从中读取（详见 §3.3 示例）。

使用注意事项：

使用前务必检查 msg._raw_data is not None
每条新消息的 _raw_data 都会重置为 None，不会跨消息残留
该属性以 _ 开头表示"内部约定"，但 SDK 官方支持此用法

三、预处理器

预处理器在命令匹配前执行，可用于日志记录、消息过滤、自定义权限检查等。

3.1 使用 Plugins 类装饰器

from easybot import Model, Plugins

@Plugins.before_command()
async def preprocess(
    msg: Model.GuildMessage | Model.GroupMessage | Model.C2CMessage | Model.DirectMessage,
) -> bool:
    if len(msg.content) > 2000:
        await msg.reply("消息过长，请缩短后再试")
        return False

    return True

3.2 使用 Bot 实例装饰器

from easybot import Model

@bot.before_command()
async def preprocess(
    msg: Model.GuildMessage | Model.GroupMessage | Model.C2CMessage | Model.DirectMessage,
) -> bool:
    bot.logger.info(f"收到消息: {msg.content}")
    return True

3.3 验证与转换输入（兼容权限检查与短路机制）

预处理器执行顺序在命令匹配之前，因此适合做两类事情：

统一清洗输入（例如去掉多余空白、兼容不同前缀）
在命令执行前进行参数校验（不通过则提示并阻止本次命令系统执行）

下面示例实现一个"仅管理员可用"的 /ban <user_id> <minutes> 命令：

预处理器负责解析并校验参数，把结构化结果写入 msg._raw_data
命令处理器只关注业务逻辑
权限检查依然由命令系统执行（is_require_admin=True）
短路机制依然由命令系统执行（默认 is_short_circuit=True）

from easybot import CommandValidScenes, Model

@bot.before_command(valid_scenes=CommandValidScenes.GUILD)
async def normalize_and_validate(
    msg: Model.GuildMessage,
) -> bool:
    content = msg.content.strip()
    if not content.startswith("/ban"):
        return True

    parts = content.split()
    if len(parts) != 3:
        await msg.reply("用法：/ban <user_id> <minutes>")
        return False

    user_id = parts[1].strip()
    minutes_str = parts[2].strip()
    if not user_id.isdigit() or not minutes_str.isdigit():
        await msg.reply("参数格式错误：user_id 和 minutes 都必须是数字")
        return False

    minutes = int(minutes_str)
    if minutes <= 0 or minutes > 24 * 60:
        await msg.reply("minutes 取值范围：1~1440")
        return False

    if msg._raw_data is not None:
        msg._raw_data["ban_user_id"] = user_id
        msg._raw_data["ban_minutes"] = minutes

    return True

@bot.on_command(
    command="/ban",
    valid_scenes=CommandValidScenes.GUILD,
    is_require_admin=True,
    admin_error_msg="⛔ 需要频道管理员权限",
)
async def ban_user(msg: Model.GuildMessage) -> None:
    user_id = None
    minutes = None
    if msg._raw_data is not None:
        user_id = msg._raw_data.get("ban_user_id")
        minutes = msg._raw_data.get("ban_minutes")

    if not user_id or not minutes:
        await msg.reply("缺少参数，请按：/ban <user_id> <minutes> 使用")
        return

    await msg.reply(f"将对 {user_id} 执行禁言 {minutes} 分钟（示例）")

注意事项：

执行顺序：注册了多个预处理器时，按注册顺序依次执行。任一返回 False 则立即终止链路，后续预处理器不再执行，命令匹配也被跳过。
预处理器 return False 会跳过本次"命令匹配与执行"，但不会阻止普通事件处理器（例如 @bot.on_guild_message）继续执行。
想阻止某一条命令而不影响其他命令时，建议把校验做得更"定向"（像上面只拦截 /ban），避免无条件 return False。

四、自动加载插件

Bot 构造函数支持自动加载插件目录：

bot = Bot(
    app_id="xxx",
    app_secret="xxx",
    auto_load_plugins=True,      # 启用自动加载
    plugins_dir="plugins",       # 插件目录路径
    plugins_recursive=False,     # 是否递归扫描子目录
)

4.1 最小项目示例（自动发现并加载）

项目结构示例：

bot.py
plugins/
├── admin.py
└── tools/
    └── echo.py

入口文件 bot.py：

from easybot import Bot

bot = Bot(
    app_id="你的AppID",
    app_secret="你的AppSecret",
    auto_load_plugins=True,
    plugins_dir="plugins",
    plugins_recursive=True,
)

bot.start()

说明：当 auto_load_plugins=True 时，Bot 会在初始化阶段预导入插件文件，用于注册命令、预处理器和所需 intents。load_plugins() 也可在运行时手动调用以扫描并加载新增的插件文件（已有插件会被安全跳过，不会重复注册）。

4.2.x 插件加载失败的处理与排查

容错策略

SDK 对插件加载采用部分容错策略：

语法错误（SyntaxError）：该插件加载失败，但不会导致 Bot 崩溃。SDK 会记录错误日志并跳过该插件，继续加载后续插件
ImportError / ModuleNotFoundError：同上，跳过并继续
插件加载失败时，已成功加载的插件不受影响

排查建议

查看启动日志：插件加载失败时会输出 WARNING 级别日志，包含失败原因和模块名
检查已加载列表：Bot 启动后可通过 bot.get_loaded_plugins() 查看哪些插件成功加载了
常见问题清单：
- 插件文件中有顶层代码抛异常（如连接数据库、读取不存在的配置文件）→ 改为延迟初始化或用 try/except 包裹
- 循环 import → 将共享依赖提取到单独的 utils 模块
- 缺少第三方依赖 → 确认已在 requirements.txt 中声明并安装
手动测试单个插件：python -c "import plugins.xxx" 快速验证是否能正常导入

4.2 插件目录结构与发现规则

插件系统会从 plugins_dir 中扫描 Python 文件并导入：

plugins_recursive=False：只扫描 plugins_dir 下的 *.py
plugins_recursive=True：递归扫描 plugins_dir/**.py
自动跳过 __init__.py 和以下划线开头的文件（例如 _utils.py）

4.3 插件编写方式（推荐）

方式一：类方法装饰器（推荐）

plugins/admin.py：

from easybot import CommandValidScenes, Model, Plugins

@Plugins.on_command(
    command="/admin",
    is_require_admin=True,
    admin_error_msg="此指令仅管理员可用",
    valid_scenes=CommandValidScenes.GUILD,
)
async def admin_cmd(msg: Model.GuildMessage) -> None:
    await msg.reply("欢迎管理员！")

plugins/tools/echo.py：

from easybot import Model, Plugins

@Plugins.on_command(command="/echo")
async def echo(
    msg: Model.GuildMessage | Model.GroupMessage | Model.C2CMessage | Model.DirectMessage,
) -> None:
    await msg.reply(str(msg.treated_msg or ""))

这种写法的要点是：插件文件在被导入时执行装饰器注册，从而把命令/预处理器注册到 Plugins 的全局注册表中。

4.4 插件系统如何发现和注册

自动加载插件的流程可以概括为：

Bot 初始化时扫描 plugins_dir（是否递归由 plugins_recursive 决定），收集所有可加载的 *.py 插件文件
为每个插件文件生成模块名（递归场景下子目录会转为点号路径，例如 tools/echo.py → tools.echo）
使用 importlib 按文件路径导入并执行插件模块代码（相当于把插件文件运行一遍）
插件模块导入期间，@Plugins.on_command / @Plugins.before_command 会把命令/预处理器注册到 Plugins 的全局表，并记录"这个注册来自哪个插件模块"，从而支持后续的卸载/热重载
预导入完成后，Bot 会根据这些已注册命令的 valid_scenes 计算并补齐 intents

4.5 插件目录结构

plugins/
├── admin.py          # 管理员插件
├── weather.py        # 天气查询插件
├── translate.py      # 翻译插件
└── utils/            # 子目录（需开启 plugins_recursive）
    └── helpers.py    # 辅助工具插件

4.6 插件生命周期钩子

插件模块可以定义可选的生命周期钩子函数，在加载和卸载时自动被 SDK 调用，用于执行初始化或清理操作。

支持的钩子函数

钩子函数	触发时机	参数	返回值
`on_plugin_load(bot)`	插件模块导入完成后	`Bot` 实例	无（同步）或 `Awaitable[None]`（异步）
`on_plugin_unload(bot)`	卸载插件前（`unload_plugin` / `reload_plugin` 时）以及 Bot 关闭时（`stop()` / `Ctrl+C`）	`Bot` 实例	无（同步）或 `Awaitable[None]`（异步）

注意：这两个函数名是 SDK 约定的特殊标识符，不会被注册为命令。它们是普通的模块级函数，SDK 在导入/卸载时通过 hasattr() 检测并调用。

基本用法

# plugins/my_plugin.py
from easybot import CommandValidScenes, Model, Plugins

# 连接池、缓存等资源
_db_connection = None
_cache = dict()


def on_plugin_load(bot):
    """插件加载时初始化资源"""
    global _db_connection
    _db_connection = create_database_connection()
    bot.logger.info("my_plugin: 数据库连接已建立")


def on_plugin_unload(bot):
    """插件卸载时清理资源"""
    global _db_connection, _cache
    if _db_connection:
        _db_connection.close()
        _db_connection = None
    _cache.clear()
    bot.logger.info("my_plugin: 资源已释放")


@Plugins.on_command(command="/query", valid_scenes=CommandValidScenes.GUILD)
async def query_cmd(msg: Model.GuildMessage) -> None:
    await msg.reply(f"查询结果: {_cache.get('data', '无数据')}")

异步钩子

如果钩子需要执行异步操作（如异步初始化数据库连接），可定义为 async def：

import asyncio

_async_resource = None


async def on_plugin_load(bot):
    """异步钩子：将在 Bot 启动完成后的 _trigger_startup 阶段调用"""
    global _async_resource
    _async_resource = await async_init_resource()
    bot.logger.info("my_plugin: 异步资源初始化完成")


async def on_plugin_unload(bot):
    """异步卸载钩子"""
    global _async_resource
    if _async_resource:
        await _async_resource.close()
        _async_resource = None

异步钩子的调用时机：同步钩子在插件导入后立即调用；异步钩子会延迟到 Bot 的 _trigger_startup 阶段（WebSocket 连接建立后）才执行。

典型使用场景

数据库/Redis 连接管理：在 on_plugin_load 中建立连接，on_plugin_unload 中关闭
定时任务注册/取消：加载时注册周期性任务，卸载时取消
配置热加载：加载时读取配置文件，卸载时保存状态
依赖检查：加载时验证外部服务可用性

on_plugin_unload 的调用保证：SDK 保证在以下两种场景都会触发 on_plugin_unload：
手动卸载/热重载时：调用 bot.unload_plugin() 或 bot.reload_plugin() 时
Bot 关闭时：bot.stop() 或收到 Ctrl+C 信号后，stop_async() 会自动遍历所有已加载插件并调用其 on_plugin_unload
因此你可以放心地在 on_plugin_unload 中释放资源（关闭连接、保存状态等），无需担心 Bot 异常退出导致资源泄漏（正常退出路径均有保障）。

五、权限控制

EasyBot 采用双层权限体系：

权限类型	参数	校验方式	适用场景
频道管理员	`is_require_admin`	QQ 平台身份组	频道内管理操作
机器人管理员	`is_require_bot_admin`	自定义白名单	全局业务管控

5.1 频道管理员权限

from easybot import Model

@Plugins.on_command(
    command="/ban",
    is_require_admin=True,
    admin_error_msg="⛔ 需要频道管理员权限"
)
async def ban_user(msg: Model.GuildMessage) -> None:
    await msg.reply("禁言成功")

注意: 此权限依赖 QQ 平台的 member.roles 数据，仅在频道消息场景有效。

最小示例：仅允许频道管理员执行

更常见的写法是直接使用 bot.on_command 注册命令：

from easybot import Bot, CommandValidScenes, Model

bot = Bot(app_id="你的AppID", app_secret="你的AppSecret")

@bot.on_command(
    command="/ban",
    valid_scenes=CommandValidScenes.GUILD,
    is_require_admin=True,
    admin_error_msg="⛔ 需要频道管理员权限",
)
async def ban_user(msg: Model.GuildMessage) -> None:
    await msg.reply("禁言成功")

bot.start()

如果你希望"权限不足时静默处理"，可以把 admin_error_msg 设为 None（默认值就是 None）。此时 SDK 不会回复权限不足提示；同时该最佳匹配命令不会继续执行。若该命令的 is_short_circuit=True，则会短路后续处理；否则本次命令系统返回，但不会回退尝试次优候选命令。

5.2 机器人管理员权限

from easybot import Model

@Plugins.on_command(
    command="/shutdown",
    is_require_bot_admin=True,
    bot_admin_error_msg="🚫 仅机器人管理员可操作"
)
async def shutdown(
    msg: Model.GuildMessage | Model.GroupMessage | Model.C2CMessage | Model.DirectMessage,
) -> None:
    await msg.reply("正在关闭...")

设置机器人管理员列表（全局白名单）

机器人管理员由 BotAdminManager 管理，支持持久化。最简单的方式是在启动后写入管理员列表：

from easybot import Bot, Model

bot = Bot(app_id="你的AppID", app_secret="你的AppSecret")

@bot.on_startup
async def init_admins(event: Model.StartupEvent) -> None:
    await bot.bot_admin_manager.set_bot_admins(["admin_1", "admin_2"])

5.3 组合使用

当同时设置 is_require_admin=True 和 is_require_bot_admin=True 时，两者为 AND 关系（必须同时满足两项权限才能通过校验）。校验顺序为先频道管理员 → 后机器人管理员。

from easybot import Model

@Plugins.on_command(
    command="/config",
    is_require_admin=True,          # 条件1: 必须是频道管理员
    is_require_bot_admin=True,      # 条件2: 必须是机器人管理员
)
async def config(msg: Model.GuildMessage) -> None:
    await msg.reply("配置管理")

错误消息策略（AND 场景下）：

仅频道管理员不通过 → 发送 admin_error_msg（如有）
频道管理员通过，但机器人管理员不通过 → 发送 bot_admin_error_msg（如有）
两者都不通过 → 只发送先检测到的 admin_error_msg（不会发送两条）

即两条错误消息不会同时发送——校验是短路式的，第一项不通过就直接返回。

5.4 权限校验流程

收到消息
    │
    ▼
执行当前场景的预处理器
    │
    ├─ 任一预处理器返回 False → 跳过本次命令系统
    │
    ▼
遍历已启用命令并做候选收集
    │
    ├─ ① at 检查（如果设置了 is_require_at）
    │      └─ 不包含 @机器人 → 不进入候选集
    │
    ├─ ② 命令/正则匹配
    │      └─ 匹配成功 → 记录为候选
    │
    ▼
按"最长匹配优先"选择最佳候选命令
    │
    ├─ ③ 频道管理员检查（如果设置了 is_require_admin）
    │      ├─ 权限不足 + admin_error_msg 有值 → 回复消息
    │      └─ 是否短路由 is_short_circuit 决定
    │
    ├─ ④ 机器人管理员检查（如果设置了 is_require_bot_admin）
    │      ├─ 权限不足 + bot_admin_error_msg 有值 → 回复消息
    │      └─ 是否短路由 is_short_circuit 决定
    │
    └─ ⑤ 执行最佳匹配命令

注意：权限校验发生在"最佳匹配命令"上，而不是对所有匹配命令逐个校验。最佳命令权限不足时，SDK 不会自动回退去尝试次优候选命令。

5.5 命令异常处理

当命令函数执行过程中抛出未捕获的异常时，SDK 的默认行为如下：

异常捕获：SDK 会在命令执行外层捕获所有 Exception
日志记录：通过 logging 模块输出 ERROR 级别日志（包含异常类型、消息内容和堆栈摘要）
静默失败：不会向用户回复错误信息（避免泄露内部细节）
短路行为：异常发生后按该命令的短路配置决定是否阻止后续处理

# SDK 内部伪代码示意
try:
    await cmd.func(msg)
except Exception as e:
    logger.error(f"命令执行异常: {cmd.func.__name__}, 消息: {msg.content}, 错误: {e}")
    # 不向用户发送任何内容

最佳实践建议：

在关键命令中使用 try/except 包裹业务逻辑，提供用户友好的错误提示
对于网络请求等可能超时的操作，务必设置合理的 timeout
开发环境可通过查看控制台日志定位异常

5.6 命令冲突与匹配优先级

当多条命令同时匹配同一条消息时，SDK 的候选选择策略如下：

1. 候选收集：遍历所有已启用且 valid_scenes 包含当前场景的命令，收集所有匹配的候选

2. 最长匹配优先：

文本命令：比较命中的命令字符串长度，越长优先级越高（如 /hello world 优先于 /hello）
正则命令：比较正则匹配结果的结束位置（match.end()），越靠后优先级越高

3. 同等长度的决胜：

当多个候选的匹配长度相同时，按注册顺序决定——先注册的优先
文本命令和正则命令混合时，文本命令不享有特殊优先权，统一按匹配长度排序

4. 权限不足时不回退：

最佳匹配命令因权限校验未通过时，SDK 不会回退尝试次优候选（详见 §5.4 流程图说明）

# 示例：注册顺序影响同等长度时的选择
@Plugins.on_command(command="/hi")       # 先注册
async def cmd_v1(msg): ...

@Plugins.on_command(command="/hi")       # 后注册
async def cmd_v2(msg): ...
# 用户输入 "/hi" → 两个都匹配，长度相同 → cmd_v1 获胜（先注册）

六、BotAdminManager

模块: easybot.plugins
导入: from easybot import BotAdminManager

BotAdminManager 用于管理机器人管理员（全局超管）。管理员数据自动持久化到 YAML 文件（默认路径 sdk_data/bot_admins.yaml）。

6.1 基本使用

from easybot import BotAdminManager

# 独立实例化（推荐）
admin_mgr = BotAdminManager()

# 如果希望先读取已有持久化数据，需先异步初始化
await admin_mgr.initialize()

# 设置管理员列表（合并式，立即持久化）- 异步方法
await admin_mgr.set_bot_admins(["admin_1", "admin_2"])

# 增量操作（每次自动持久化）- 异步方法，支持可变参数
await admin_mgr.add_admin("admin_3")
await admin_mgr.add_admin("admin_4", "admin_5")  # 一次添加多个
await admin_mgr.remove_admin("admin_1")
await admin_mgr.remove_admin("admin_2", "admin_3")  # 一次移除多个

# 查询 - 同步方法
if admin_mgr.is_admin("user_id"):
    print("是管理员")

# 获取所有管理员 - 同步方法
admins = admin_mgr.get_all_admins()

# 清空 - 异步方法
await admin_mgr.clear_admins()

关于初始化时机：
独立使用 BotAdminManager()：必须手动调用 await initialize() 才能读写持久化文件。在 initialize() 之前调用写入方法会抛出异常。
通过 bot 实例 bot.bot_admin_manager：Bot 在启动阶段会自动调用 initialize()，因此无需手动初始化。在 on_startup 回调中使用时可以安全地直接调用写入方法。
注意: set_bot_admins(), add_admin(), remove_admin(), clear_admins() 都是异步方法，需要在异步上下文中使用 await 调用。is_admin() 和 get_all_admins() 是同步方法。get_all_admins() 返回的是 set[str] 副本。

6.2 通过 Bot 实例访问

bot = Bot(app_id="xxx", app_secret="xxx")

# 通过属性访问 - 异步方法
await bot.bot_admin_manager.set_bot_admins(["admin_1", "admin_2"])
await bot.bot_admin_manager.add_admin("admin_3")
await bot.bot_admin_manager.remove_admin("admin_1")

# 查询 - 同步方法
if bot.bot_admin_manager.is_admin("user_id"):
    print("是管理员")

6.3 自定义存储目录

# 使用自定义存储目录
admin_mgr = BotAdminManager(data_dir="/path/to/my_data")

6.4 支持的操作符

if "user_id" in admin_mgr:
    print("是管理员")

print(f"管理员数量: {len(admin_mgr)}")

七、命令动态管理

Plugins 类提供了完整的命令动态管理功能。

以下方法都支持两种定位方式：

通过函数名，例如 cmd_hello
通过命令名，例如 /hello 或 hello

7.1 查询命令

# 获取所有命令（包括禁用的）
all_commands = Plugins.get_all_commands()
for cmd in all_commands:
    status = "✅" if cmd.enabled else "❌"
    print(f"{status} {cmd.func.__name__}")

# 查找命令对象
cmd = Plugins.find_command("cmd_hello")
if cmd:
    print(f"找到命令: {cmd.func.__name__}")

# 也可以通过命令名查找
cmd = Plugins.find_command("/hello")

命令对象（`BotCommandObject`）结构

find_command() 和 get_all_commands() 返回的 cmd 对象具有以下属性：

属性	类型	说明
`func`	`Coroutine`	命令处理函数（可调用对象）
`enabled`	`bool`	是否启用
`command`	`list[str]`	文本命令列表（如 `["/hello"]`）
`regex`	`list[Pattern]`	正则表达式列表
`is_treat`	`bool`	is_treat 参数值
`is_require_at`	`bool`	是否要求 @机器人
`is_short_circuit`	`bool`	短路配置
`is_custom_short_circuit`	`bool`	自定义短路配置
`is_require_admin`	`bool`	是否要求频道管理员
`admin_error_msg`	`str \| None`	管理员错误提示
`is_require_bot_admin`	`bool`	是否要求机器人管理员
`bot_admin_error_msg`	`str \| None`	机器人管理员错误提示
`valid_scenes`	`CommandValidScenes`	有效场景
`module_name`	`str \| None`	所属插件模块名（主程序注册为 `None`）

7.2 启用/禁用命令

# 禁用命令
Plugins.disable_command("cmd_hello")

# 启用命令
Plugins.enable_command("cmd_hello")

# 也支持通过命令名启用/禁用
Plugins.disable_command("/hello")
Plugins.enable_command("hello")

# 检查命令是否启用
if Plugins.is_command_enabled("cmd_hello"):
    print("命令已启用")
else:
    print("命令已禁用或不存在")

`enabled=False` vs `disable_command()` 的区别

维度	装饰器参数 `enabled=False`	运行时 `disable_command()`
生效时机	注册时即不参与匹配	运行时动态切换
典型用途	开发调试时临时禁用、条件性注册（如根据配置决定）	管理命令动态开关、权限变更后禁用
可逆性	需修改代码并重启/重载	可随时通过 `enable_command()` 恢复
热重载影响	插件重载后会重新按代码中的值生效	重载后状态会被重置为装饰器参数的值

简单记忆：写代码时就确定不需要的用 enabled=False；运行中需要动态控制开关的用 disable_command() / enable_command()。

7.3 移除命令

Plugins.remove_command("cmd_hello")

# 也可以通过命令名移除
Plugins.remove_command("/hello")

注意：remove_command() 只会移除当前注册表中的命令对象。如果该命令来自某个插件模块，插件维度的统计信息会在卸载/热重载时统一维护。

八、热重载插件

EasyBot SDK 支持插件热重载，无需重启机器人即可更新插件代码。

注意：插件级热重载/卸载支持通过 @Plugins.on_command 和 @Plugins.before_command 注册到插件模块中的内容；通过 @bot.on_command / @bot.before_command 在主程序中直接注册的内容不属于插件模块热重载范围。

8.1 通过 Bot 实例操作

# 热重载单个插件
result = bot.reload_plugin("admin")
print(f"重载结果: {result}")

# 热重载所有插件
results = bot.reload_all_plugins()

# 卸载插件
bot.unload_plugin("admin")

# 获取已加载插件列表
plugins = bot.get_loaded_plugins()

插件名称（`module_name`）格式规则

热重载/卸载方法中的字符串参数为模块名，其映射规则如下：

目录结构	模块名	调用示例
`plugins/admin.py`	`"admin"`	`reload_plugin("admin")`
`plugins/tools/echo.py`	`"tools.echo"`	`reload_plugin("tools.echo")`

即：去掉 plugins_dir 前缀和 .py 后缀，子目录层级用点号 . 分隔。递归扫描时 SDK 会自动将路径转为点号格式的模块名。

热重载的安全性与原子性

原子性保证：SDK 采用"先卸载后加载"策略——先将旧版本的所有命令和预处理器从注册表中移除，再重新导入新代码并注册。如果新代码加载失败（语法错误、import 错误等），旧版本的命令已被移除且不会回滚恢复
执行中命令的影响：如果某个命令的协程正在执行（尚未返回），卸载该插件不会中断正在执行的协程，但该命令对象已从注册表移除
错误处理建议：
1. 重载前先在本地验证代码无语法错误：python -m py_compile plugins/xxx.py
2. 检查 reload_plugin() 返回值的 success 和 error 字段判断是否成功
3. 生产环境建议使用 reload_all_plugins() 后逐一检查各插件的 success 状态

8.2 通过 Plugins 类操作

from easybot import Plugins

# 热重载指定插件
result = Plugins.reload_plugin("admin")

# 卸载插件
Plugins.unload_plugin("admin")

# 获取已加载插件列表
plugins = Plugins.get_loaded_plugins()

# 获取插件注册的命令函数名
commands = Plugins.get_plugin_commands("admin")

8.3 热重载返回值

{
    "module": "admin",           # 模块名
    "unloaded": {                # 卸载的内容
        "commands": 2,
        "preprocessors": 1
    },
    "loaded": {                  # 新加载的内容
        "commands": 2,
        "preprocessors": 1
    },
    "success": True,             # 是否成功
    "error": None                # 错误信息（如果失败）
}

九、最佳实践

9.1 敏感操作使用机器人管理员权限

from easybot import Model

@Plugins.on_command(command="/reload", is_require_bot_admin=True)
async def reload_config(
    msg: Model.GuildMessage | Model.GroupMessage | Model.C2CMessage | Model.DirectMessage,
) -> None:
    await msg.reply("重新加载配置")

9.2 管理操作使用频道管理员权限

from easybot import Model

@bot.on_command(command="/mute", is_require_admin=True)
async def mute_user(msg: Model.GuildMessage) -> None:
    await msg.reply("禁言成功")

9.3 使用预处理器实现复杂逻辑

from easybot import Model

@Plugins.before_command()
async def rate_limit(
    msg: Model.GuildMessage | Model.GroupMessage | Model.C2CMessage | Model.DirectMessage,
) -> bool:
    user_id = msg.author.id
    if is_rate_limited(user_id):
        return False  # 跳过命令匹配
    return True

9.4 插件文件使用类方法装饰器

# plugins/my_plugin.py
from easybot import CommandValidScenes, Model, Plugins

@Plugins.on_command(command="/help")
async def help_cmd(
    msg: Model.GuildMessage | Model.GroupMessage | Model.C2CMessage | Model.DirectMessage,
) -> None:
    await msg.reply("这是帮助信息")

@Plugins.before_command(valid_scenes=CommandValidScenes.ALL)
async def log_message(
    msg: Model.GuildMessage | Model.GroupMessage | Model.C2CMessage | Model.DirectMessage,
) -> bool:
    print(f"[日志] {msg.content}")
    return True

9.5 单文件程序使用实例方法装饰器

# main.py
from easybot import Bot, Model

bot = Bot(app_id="xxx", app_secret="xxx")

@bot.on_command(command="/ping")
async def ping(
    msg: Model.GuildMessage | Model.GroupMessage | Model.C2CMessage | Model.DirectMessage,
) -> None:
    await msg.reply("Pong!")

@bot.before_command()
async def log(
    msg: Model.GuildMessage | Model.GroupMessage | Model.C2CMessage | Model.DirectMessage,
) -> bool:
    print(f"收到消息: {msg.content}")
    return True

bot.start()

十、下一步

Session 会话管理器 — 多轮对话与状态管理

更新日志

2026/4/29 07:21

查看所有更新日志

b50e7-docs: 完善文档内容与格式于 2026/4/29
v1.0.3于 2026/4/10
30b8c-feat: 提升代码质量并优化文档示例于 2026/4/7
a8177-feat(plugins): 支持插件热重载功能并优化命令查找逻辑于 2026/4/7
fb931-feat: 增强插件热重载功能于 2026/4/6
v1.0.0于 2026/4/4

插件系统

📢 加入交流