`on_alconna` 响应器

nonebot_plugin_alconna 插件本体的大部分功能都围绕着 on_alconna 响应器展开。

该响应器类似于 on_command，基于 Alconna 解析器来解析命令。

以下是一个简单的 on_alconna 响应器的例子：

from nonebot_plugin_alconna import At, Image, Match, on_alconna
from arclet.alconna import Args, Option, Alconna, MultiVar, Subcommand


alc = Alconna(
    "role-group",
    Subcommand(
        "add|添加",
        Args["name", str],
        Option("member", Args["target", MultiVar(At)]),
        dest="add",
        compact=True,
    ),
    Option("list"),
    Option("icon", Args["icon", Image])
)
rg = on_alconna(alc, use_command_start=True, aliases={"角色组"})


@rg.assign("list")
async def list_role_group():
    img: bytes = await gen_role_group_list_image()
    await rg.finish(Image(raw=img))

@rg.assign("add")
async def _(name: str, target: Match[tuple[At, ...]]):
    group = await create_role_group(name)
    if target.available:
        ats: tuple[At, ...] = target.result
        group.extend(member.target for member in ats)
    await rg.finish("添加成功")

NoneBot

/role-group list

[图片]

/角色组添加foo @bar @baz

添加成功

on_alconna 的参数如下:

def on_alconna(
    command: Alconna | str,
    rule: Rule | T_RuleChecker | None = None,
    skip_for_unmatch: bool = True,
    auto_send_output: bool | None = None,
    aliases: set[str] | tuple[str, ...] | None = None,
    comp_config: CompConfig | None = None,
    extensions: list[type[Extension] | Extension] | None = None,
    exclude_ext: list[type[Extension] | str] | None = None,
    use_origin: bool | None = None,
    use_cmd_start: bool | None = None,
    use_cmd_sep: bool | None = None,
    response_self: bool | None = None,
    **kwargs: Any,
) -> type[AlconnaMatcher]:
    ...

command: Alconna 命令或字符串，字符串将通过 AlconnaFormat 转换为 Alconna 命令
rule: 事件响应规则，详见响应器规则
skip_for_unmatch: 是否在命令不匹配时跳过该响应, 默认为 True
auto_send_output: 是否自动发送输出信息并跳过该响应。
- True：自动发送输出信息并跳过该响应
- False：不自动发送输出信息，而是传递进行处理
- None：跟随全局配置项 alconna_auto_send_output，默认值为 True
aliases: 命令别名，作用类似于 on_command 中的 aliases
comp_config: 补全会话配置，不传入则不启用补全会话
extensions: 需要加载的匹配扩展, 可以是扩展类或扩展实例
exclude_ext: 需要排除的匹配扩展, 可以是扩展类或扩展的id
use_origin: 是否使用未经 to_me 等处理过的消息。None 时跟随全局配置项 alconna_use_origin，默认值为 False
use_cmd_start: 是否使用 COMMAND_START 作为命令前缀。None 时跟随全局配置项 alconna_use_command_start，默认值为 False
use_cmd_sep: 是否使用 COMMAND_SEP 作为命令分隔符。None 时跟随全局配置项 alconna_use_command_sep，默认值为 False
response_self: 是否响应自身消息。None 时跟随全局配置项 alconna_response_self，默认值为 False

on_alconna 返回的是 Matcher 的子类 AlconnaMatcher ，其拓展了如下方法：

.assign(path, value, or_not): 用于对包含多个选项/子命令的命令的分派处理
.dispatch: 同样的分派处理，但是是类似 CommandGroup 一样返回新的 AlconnaMatcher
.got_path(path, prompt, middleware): 在 got 方法的基础上，会以 path 对应的参数为准，读取传入 message 的最后一个消息段并验证转换
.got, send, reject, ... : 拓展了 prompt 类型，即支持使用 UniMessage 作为 prompt
...

除了标准的创建方式，本插件也提供了 funcommand 和 Command 两种快捷方式来创建 AlconnaMatcher，详见快捷方式。

AlconnaMatcher 的特性之一是拓展了依赖注入的功能。

插件提供了几种用来处理解析结果的模型：

CommandResult: 用于快捷访问命令解析结果
- result (Arparma): 解析结果
- source (Alconna): 源命令
- matched (bool): 是否匹配
- context (dict): 命令的上下文
- output (str | None): 命令的输出
Match: 匹配项，表示参数是否存在于 Arparma.all_matched_args 内，可用 Match.available 判断是否匹配，Match.result 获取匹配的值
- Match 只能查找到 Arparma.all_matched_args 中的参数。对于特定选项/子命令的参数，需要使用 Query 来查询
Query: 查询项，表示参数是否可由 Arparma.query 查询并获得结果，可用 Query.available 判断是否查询成功，Query.result 获取查询结果
- Query 除了查询参数，也可以查询某个选项/子命令是否存在

async def handle(
    result: CommandResult,
    arp: Arparma,
    dup: Duplication,
    source: Alconna,
    ext: Extension,
    exts: SelectedExtensions,
    abc: str,
    foo: Match[str],
    bar: Query[int] = Query("ttt.bar", 0)
):
    ...

AlconnaMatcher 的依赖注入拓展支持以下情况：

xxx: CommandResult
xxx: Arparma：命令的解析结果
xxx: Duplication：命令的解析结果的 Duplication
xxx: Alconna：命令的源命令
<key>: Match[<type>]：上述的匹配项，使用 key 作为查询路径
xxx: Query[<type>] = Query(<path>, default)：上述的查询项，必需声明默认值以设置查询路径 path
- 当用来查询选项/子命令是否存在时，可不写 Query[<type>]
xxx: Extension：当前 AlconnaMatcher 使用的指定类型的匹配扩展
xxx: SelectedExtensions：当前 AlconnaMatcher 使用的所有可用的匹配扩展
<key>: <type>: 其他情况
- 当 key 的名称是 "ctx" 或 "context" 并且类型为 dict 时，会注入命令的上下文
- 当 key 存在于命令的上下文中时，会注入对应的值
- 当 key 存在于 Arparma 的 all_matched_args 中时，会注入对应的值, 类似于 Match 的用法，但当该值不存在时将跳过响应器。
- 当 key 属于 got_path 的参数时，会注入对应的值
- 当 key 被某个 Extension.before_catch 确认为需要注入的参数时，会调用 Extension.catch 来注入对应的值

note

如果你更喜欢 Depends 式的依赖注入，nonebot_plugin_alconna 同时提供了一系列的依赖注入函数，他们包括：

AlconnaResult: CommandResult 类型的依赖注入函数
AlconnaMatches: Arparma 类型的依赖注入函数
AlconnaDuplication: Duplication 类型的依赖注入函数
AlconnaMatch: Match 类型的依赖注入函数，其能够额外传入一个 middleware 函数来处理得到的参数
AlconnaQuery: Query 类型的依赖注入函数，其能够额外传入一个 middleware 函数来处理得到的参数
AlconnaExecResult: 提供挂载在命令上的 callback 的返回结果 (Dict[str, Any]) 的依赖注入函数
AlconnaExtension: 提供指定类型的 Extension 的依赖注入函数

示例：

from nonebot import require
require("nonebot_plugin_alconna")

from nonebot_plugin_alconna import AlconnaQuery, AlcResult, Match, Query, on_alconna
from arclet.alconna import Alconna, Args, Option, Arparma


test = on_alconna(
    Alconna(
        "test",
        Option("foo", Args["bar", int]),
        Option("baz", Args["qux", bool, False])
    )
)

@test.handle()
async def handle_test1(result: AlcResult):
    await test.send(f"matched: {result.matched}")
    await test.send(f"maybe output: {result.output}")

@test.handle()
async def handle_test2(result: Arparma):
    await test.send(f"head result: {result.header_result}")
    await test.send(f"args: {result.all_matched_args}")

@test.handle()
async def handle_test3(bar: Match[int]):
    if bar.available:
        await test.send(f"foo={bar.result}")

@test.handle()
async def handle_test4(qux: Query[bool] = AlconnaQuery("baz.qux", False)):
    if qux.available:
        await test.send(f"baz.qux={qux.result}")

`assign` 方法

AlconnaMatcher 的 assign 方法与 handle 类似，但是可以控制响应函数是否在不满足条件时跳过响应。

assign 方法的参数如下：

def assign(
    cls,
    path: str,
    value: Any = _seminal,
    or_not: bool = False,
    additional: CHECK | None = None,
    parameterless: Iterable[Any] | None = None,
):
    ...

path: 指定的查询路径
- "$main" 表示没有任何选项/子命令匹配的时候
- "~XX" 时会把 "~" 替换为父级路径
value: 可能的指定查询值
or_not: 是否同时处理没有查询成功的情况
additional: 额外的条件检查函数

例如：

# 处理没有任何选项/子命令匹配的情况
@rg.assign("$main")
async def handle_main(): ...

# 处理 list 选项
@rg.assign("list")
async def handle_list(): ...

# 处理 add 选项，且 name 为 admin
@rg.assign("add.name", "admin")
async def handle_add_admin(): ...

`dispatch` 方法

此外，使用 .dispatch 还能像 CommandGroup 一样为每个分发设置独立的 matcher：

rg_list_cmd = rg.dispatch("list")

@rg_list_cmd.handle()
async def handle_list(): ...

dispatch 的参数与 assign 相同。

当使用 dispatch 时，父级路径表示为传入 dispatch 的 path:

rg_add_cmd = rg.dispatch("add")

# 此时 ~name 表示 add.name
@rg_add_cmd.assign("~name", "admin")
async def handle_add_admin(): ...

tip

在 dispatch 下, Query 的 path 也同样支持 ~ 前缀来表示父级路径

@rg_add_cmd.assign("~name", "admin")
async def handle_add_admin(target: Query[tuple[At, ...]] = Query("~target")):
    if target.available:
        await rg.send(f"添加成功: {target.result}")

`got_path` 方法

另外，AlconnaMatcher 有类似于 got 的 got_path 与配套的 get_path_arg, set_path_arg：

from nonebot_plugin_alconna import At, Match, UniMessage, on_alconna


test_cmd = on_alconna(Alconna("test", Args["target?", Union[str, At]]))

@test_cmd.handle()
async def tt_h(target: Match[Union[str, At]]):
    if target.available:
        test_cmd.set_path_arg("target", target.result)

@test_cmd.got_path("target", prompt="请输入目标")
async def tt(target: Union[str, At]):
    await test_cmd.send(UniMessage(["ok\n", target]))

got_path 与 assign，Match，Query 等地方一样，都需要指明 path 参数 (即对应 Arg 验证的路径)

got_path 会获取消息的最后一个消息段并转为 path 对应的类型，例如示例中 target 对应的 Arg 里要求 str 或 At，则 got 后用户输入的消息只有为 text 或 at 才能进入处理函数。

got_path 中可以使用依赖注入函数 AlconnaArg, 类似于 Arg.

`prompt` 方法

基于 Waiter 插件，AlconnaMatcher 提供了 prompt 方法来实现更灵活的交互式提示。

from nonebot_plugin_alconna import At, Match, UniMessage, on_alconna


test_cmd = on_alconna(Alconna("test", Args["target?", Union[str, At]]))

@test_cmd.handle()
async def tt_h(target: Match[Union[str, At]]):
    if target.available:
        await test_cmd.finish(UniMessage(["ok\n", target]))
    resp = await test_cmd.prompt("请输入目标", timeout=30)  # 等待 30 秒
    if resp is None:
        await test_cmd.finish("超时")
    await test_cmd.finish(UniMessage(["ok\n", resp[-1]]))

在 AlconnaMatch，AlconnaQuery 或 got_path 中，你可以使用 middleware 参数来传入一个对返回值进行处理的函数：

from nonebot_plugin_alconna import image_fetch


mask_cmd = on_alconna(Alconna("search", Args["img?", Image]))


@mask_cmd.handle()
async def mask_h(matcher: AlconnaMatcher, img: Match[bytes] = AlconnaMatch("img", image_fetch)):
    result = await search_img(img.result)
    await matcher.send(result.content)

其中，image_fetch 是一个中间件，其接受一个 Image 对象，并提取图片的二进制数据返回。

本插件基于 tarina.lang 模块提供了 i18n 的支持，参见 Lang 用法。

当你编写完语言文件后，你便可以通过 AlconnaMatcher.i18n 来快速地将语言文件中的内容转为 UniMessage.

中文
英文

zh-CN.yml
# 中文语言文件
demo:
  command:
    role-group:
      add: 添加 {name} 成功!

NoneBot

/角色组添加 foo

添加 foo 成功!

en-US.yml
# 英文语言文件
demo:
  command:
    role-group:
      add: Add {name} successfully!

NoneBot

/role-group add foo

Add foo successfully!

使用 i18n
@rg.assign("add")
async def handle_add(name: str):
    await rg.i18n("demo", "command.role-group.add", name=name).finish()

AlconnaMatcher.test 方法允许你在 NoneBot 启动时对命令进行测试。

def test(
    cls,
    message: str | UniMessage,
    expected: dict[str, Any] | None = None,
    prefix: bool = True
): ...

message: 测试的消息
expected: 预期的解析结果，若为 None 则表示只测试是否匹配
prefix: 是否使用命令前缀，默认为 True

本插件提供了一个 Extension 类，其用于自定义 AlconnaMatcher 的部分行为

目前 Extension 的功能有：

validate: 对于事件的来源适配器或 bot 选择是否接受响应
output_converter: 输出信息的自定义转换方法
message_provider: 从传入事件中自定义提取消息的方法
receive_provider: 对传入的消息 (UniMessage) 的额外处理
context_provider: 对命令上下文的额外处理
permission_check: 命令对消息解析并确认头部匹配（即确认选择响应）时对发送者的权限判断
parse_wrapper: 对命令解析结果的额外处理
send_wrapper: 对发送的消息 (Message 或 UniMessage) 的额外处理
before_catch: 自定义依赖注入的绑定确认函数
catch: 自定义依赖注入处理函数
post_init: 响应器创建后对命令对象的额外处理

tip

Extension 可以通过 add_global_extension 方法来全局添加。

from nonebot_plugin_alconna import add_global_extension
from nonebot_plugin_alconna.builtins.extensions.telegram import TelegramSlashExtension

add_global_extension(TelegramSlashExtension)

全局的 Extension 可延迟加载 (即若有全局拓展加载于部分 AlconnaMatcher 之后，这部分响应器会被追加拓展)

例如一个 LLMExtension 可以如下实现 (仅举例)：

from nonebot_plugin_alconna import Extension, Alconna, on_alconna, Interface


class LLMExtension(Extension):
    @property
    def priority(self) -> int:
        return 10

    @property
    def id(self) -> str:
        return "LLMExtension"

    def __init__(self, llm):
      self.llm = llm

    def post_init(self, alc: Alconna) -> None:
        self.llm.add_context(alc.command, alc.meta.description)

    async def receive_wrapper(self, bot, event, receive):
        resp = await self.llm.input(str(receive))
        return receive.__class__(resp.content)

    def before_catch(self, name, annotation, default):
        return name == "llm"

    def catch(self, interface: Interface):
        if interface.name == "llm":
            return self.llm

matcher = on_alconna(
    Alconna(...),
    extensions=[LLMExtension(LLM)]
)
...

那么添加了 LLMExtension 的响应器便能接受任何能通过 llm 翻译为具体命令的自然语言消息，同时可以在响应器中为所有 llm 参数注入模型变量。

def validate(self, bot: Bot, event: Event) -> bool: ...

默认情况下，validate 方法会筛选 event.get_type() 为 message 的情况，表示接受消息事件。

async def output_converter(self, output_type: OutputType, content: str) -> UniMessage: ...

依据输出信息的类型，将字符串转换为消息对象以便发送。

其中 OutputType 为 "help", "shortcut", "completion", "error" 其中之一。

该方法只会调用一次，即对于多个 Extension，选择优先级靠前且实现了该方法的 Extension。

async def message_provider(
    self, event: Event, state: T_State, bot: Bot, use_origin: bool = False
) -> UniMessage | None:...

该方法用于从事件中提取消息，默认情况下会使用 event.get_message() 来获取消息。

该方法可能会调用多次，即对于多个 Extension，选择优先级靠前且实现了该方法的 Extension，若调用的返回值不为 None 则作为结果。

caution

该方法的默认实现对结果 (UniMessage) 会进行缓存。Extension 的实现也应尽量实现缓存机制。

async def receive_provider(self, bot: Bot, event: Event, command: Alconna, receive: UniMessage) -> UniMessage: ...

该方法用于对传入的消息 (UniMessage) 进行额外处理，默认情况下会返回原始消息。

该方法会调用多次，即对于多个 Extension，前一个 Extension 的返回值会作为下一个 Extension 的输入。

async def context_provider(self, ctx: dict[str, Any], bot: Bot, event: Event, state: T_State) -> dict[str, Any]:

该方法用于提取命令上下文，默认情况下会返回 ctx 本身。

该方法会调用多次，即对于多个 Extension，前一个 Extension 的返回值会作为下一个 Extension 的输入。

async def permission_check(self, bot: Bot, event: Event, command: Alconna) -> bool: ...

该方法用于对发送者的权限进行检查，默认情况下会返回 True。

该方法可能会调用多次，即对于多个 Extension，若调用的返回值不为 True 则结束判断。

async def parse_wrapper(self, bot: Bot, state: T_State, event: Event, res: Arparma) -> None: ...

该方法用于对命令解析结果进行额外处理。

该方法会调用多次，即对于多个 Extension，会并发地调用该方法。

async def send_wrapper(self, bot: Bot, event: Event, send: TMessage) -> TMessage: ...

该方法用于对 AlconnaMatcher.send 或 UniMessage.send 发送的消息 (str 或 Message 或 UniMessage) 进行额外处理，默认情况下会返回原始消息。

该方法会调用多次，即对于多个 Extension，前一个 Extension 的返回值会作为下一个 Extension 的输入。

由于需要保证输入与输出的类型一致，该方法内需要自行判断类型。

def before_catch(self, name: str, annotation: type, default: Any) -> bool: ...

该方法用于响应函数中某个参数是否需要绑定到该 Extension 上。

async def catch(self, interface: Interface) -> Any: ...

该方法用于注入经过 before_catch 确认的参数。其中 Interface 的定义为

class Interface(Generic[TE]):
    event: TE
    state: T_State
    name: str
    annotation: Any
    default: Any

补全会话基于 半自动补全，用于指令参数缺失或参数错误时给予交互式提示，类似于 got-reject：

from nonebot_plugin_alconna import Alconna, Args, Field, At, on_alconna

alc = Alconna(
    "添加教师",
    Args["name", str, Field(completion=lambda: "请输入姓名")],
    Args["phone", int, Field(completion=lambda: "请输入手机号")],
    Args["at", [str, At], Field(completion=lambda: "请输入教师号")],
)

cmd = on_alconna(alc, comp_config={"lite": True}, skip_for_unmatch=False)

@cmd.handle()
async def handle(result: Arparma):
    cmd.finish("添加成功")

此时，当用户输入 添加教师 时，会自动提示用户输入姓名，手机号和教师号，用户输入后会自动进入下一个提示：

NoneBot

添加教师

以下是建议的输入：
- name: 请输入姓名

foo

以下是建议的输入：
- phone: 请输入手机号

12345

以下是建议的输入：
- at: 请输入教师号

@me

添加成功

补全会话配置如下：

class CompConfig(TypedDict):
    tab: NotRequired[str]
    """用于切换提示的指令的名称"""
    enter: NotRequired[str]
    """用于输入提示的指令的名称"""
    exit: NotRequired[str]
    """用于退出会话的指令的名称"""
    timeout: NotRequired[int]
    """超时时间"""
    hide_tabs: NotRequired[bool]
    """是否隐藏所有提示"""
    hides: NotRequired[Set[Literal["tab", "enter", "exit"]]]
    """隐藏的指令"""
    disables: NotRequired[Set[Literal["tab", "enter", "exit"]]]
    """禁用的指令"""
    lite: NotRequired[bool]
    """是否使用简洁版本的补全会话（相当于同时配置 disables、hides、hide_tabs）"""
    block: NotRequired[bool]
    """进行补全会话时是否阻塞响应器"""

`on_alconna` 响应器

声明

依赖注入

注入模型

编写

条件控制

`assign` 方法

`dispatch` 方法

`got_path` 方法

`prompt` 方法

返回值中间件

i18n

匹配测试

匹配拓展

validate

output_converter

message_provider

receive_provider

context_provider

permission_check

parse_wrapper

send_wrapper

before_catch

catch

补全会话

2.4.2Stable

NextDevelopment

2.4.2Stable

NextDevelopment

2.4.2Stable

NextDevelopment

声明​

依赖注入​

注入模型​

编写​

条件控制​

assign 方法​

dispatch 方法​

got_path 方法​

prompt 方法​

返回值中间件​

i18n​

匹配测试​

匹配拓展​

validate​

output_converter​

message_provider​

receive_provider​

context_provider​

permission_check​

parse_wrapper​

send_wrapper​

before_catch​

catch​

补全会话​

声明

依赖注入

注入模型

编写

条件控制

`assign` 方法

`dispatch` 方法

`got_path` 方法

`prompt` 方法

返回值中间件

i18n

匹配测试

匹配拓展

validate

output_converter

message_provider

receive_provider

context_provider

permission_check

parse_wrapper

send_wrapper

before_catch

catch

补全会话