14 KiB
sidebar_position | description |
---|---|
3 | 响应规则的使用 |
Alconna 响应规则
以下为一个使用示例:
from nonebot_plugin_alconna.adapters.onebot12 import Image
from nonebot_plugin_alconna import At, on_alconna
from arclet.alconna import Args, Option, Alconna, Arparma, MultiVar, Subcommand
alc = Alconna(
["/", "!"],
"role-group",
Subcommand(
"add",
Args["name", str],
Option("member", Args["target", MultiVar(At)]),
),
Option("list"),
Option("icon", Args["icon", Image])
)
rg = on_alconna(alc, auto_send_output=True)
@rg.handle()
async def _(result: Arparma):
if result.find("list"):
img = await ob12_gen_role_group_list_image()
await rg.finish(Image(img))
if result.find("add"):
group = await create_role_group(result.query[str]("add.name"))
if result.find("add.member"):
ats = result.query[tuple[At, ...]]("add.member.target")
group.extend(member.target for member in ats)
await rg.finish("添加成功")
响应器使用
on_alconna
的所有参数如下:
command: Alconna | str
: Alconna 命令skip_for_unmatch: bool = True
: 是否在命令不匹配时跳过该响应auto_send_output: bool = False
: 是否自动发送输出信息并跳过响应aliases: set[str | tuple[str, ...]] | None = None
: 命令别名,作用类似于on_command
中的 aliasescomp_config: CompConfig | None = None
: 补全会话配置,不传入则不启用补全会话extensions: list[type[Extension] | Extension] | None = None
: 需要加载的匹配扩展,可以是扩展类或扩展实例exclude_ext: list[type[Extension] | str] | None = None
: 需要排除的匹配扩展,可以是扩展类或扩展的 iduse_origin: bool = False
: 是否使用未经 to_me 等处理过的消息use_cmd_start: bool = False
: 是否使用 COMMAND_START 作为命令前缀use_cmd_sep: bool = False
: 是否使用 COMMAND_SEP 作为命令分隔符
on_alconna
返回的是 Matcher
的子类 AlconnaMatcher
,其拓展了如下方法:
.assign(path, value, or_not)
: 用于对包含多个选项/子命令的命令的分派处理.got_path(path, prompt, middleware)
: 在got
方法的基础上,会以 path 对应的参数为准,读取传入 message 的最后一个消息段并验证转换.set_path_arg(key, value)
,.get_path_arg(key)
: 类似set_arg
和got_arg
,为got_path
的特化版本.reject_path(path[, prompt, fallback])
: 类似于reject_arg
,对应got_path
.dispatch
: 同样的分派处理,但是是类似CommandGroup
一样返回新的AlconnaMatcher
.got
,send
,reject
, ...: 拓展了 prompt 类型,即支持使用UniMessage
作为 prompt
用例:
from arclet.alconna import Alconna, Option, Args
from nonebot_plugin_alconna import on_alconna, AlconnaMatch, Match, UniMessage
login = on_alconna(Alconna(["/"], "login", Args["password?", str], Option("-r|--recall")))
@login.assign("recall")
async def login_exit():
await login.finish("已退出")
@login.assign("password")
async def login_handle(pw: Match[str] = AlconnaMatch("password")):
if pw.available:
login.set_path_arg("password", pw.result)
@login.got_path("password", prompt=UniMessage.template("{:At(user, $event.get_user_id())} 请输入密码"))
async def login_got(password: str):
assert password
await login.send("登录成功")
依赖注入
Alconna
的解析结果会放入 Arparma
类中,或用户指定的 Duplication
类。
而 AlconnaMatcher
在原有 Matcher 的基础上拓展了允许的依赖注入:
@cmd.handle()
async def handle(
result: CommandResult,
arp: Arparma,
dup: Duplication, # 基类或子类都可以
ext: Extension,
source: Alconna,
abc: str, # 类似 Match, 但是若匹配结果不存在对应字段则跳过该 handler
foo: Match[str],
bar: Query[int] = Query("ttt.bar", 0) # Query 仍然需要一个默认值来传递 path 参数
):
...
可以看到,本插件提供了几类额外的模型:
CommandResult
: 解析结果,包括了源命令source: Alconna
,解析结果result: Arparma
,以及可能的输出信息output: str | None
字段Match
: 匹配项,表示参数是否存在于all_matched_args
内,可用Match.available
判断是否匹配,Match.result
获取匹配的值Query
: 查询项,表示参数是否可由Arparma.query
查询并获得结果,可用Query.available
判断是否查询成功,Query.result
获取查询结果
:::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 on_alconna, Match, Query, AlconnaQuery
from arclet.alconna import Alconna, Args, Option, Arparma
test = on_alconna(
Alconna(
"test",
Option("foo", Args["bar", int]),
Option("baz", Args["qux", bool, False])
),
auto_send_output=True
)
@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(bar: Match[int]):
if bar.available:
await test.send(f"foo={bar.result}")
@test.handle()
async def handle_test3(qux: Query[bool] = AlconnaQuery("baz.qux", False)):
if qux.available:
await test.send(f"baz.qux={qux.result}")
消息段标注
示例中使用了消息段标注,其中 At
属于通用标注,而 Image
属于 onebot12
适配器下的标注。
适配器下的消息段标注会匹配特定的 MessageSegment
:
而通用标注与适配器标注的区别在于,通用标注会匹配多个适配器中相似类型的消息段,并返回
nonebot_plugin_alconna.uniseg
中定义的 Segment
模型
例如:
...
ats = result.query[tuple[At, ...]]("add.member.target")
group.extend(member.target for member in ats)
这样插件使用者就不用考虑平台之间字段的差异
本插件为以下适配器提供了专门的适配器标注:
协议名称 | 路径 |
---|---|
OneBot 协议 | adapters.onebot11, adapters.onebot12 |
Telegram | adapters.telegram |
飞书 | adapters.feishu |
GitHub | adapters.github |
QQ bot | adapters.qq |
QQ 频道 bot | adapters.qqguild |
钉钉 | adapters.ding |
Console | adapters.console |
开黑啦 | adapters.kook |
Mirai | adapters.mirai |
Ntchat | adapters.ntchat |
MineCraft | adapters.minecraft |
BiliBili Live | adapters.bilibili |
Walle-Q | adapters.onebot12 |
Villa | adapters.villa |
Discord | adapters.discord |
Red 协议 | adapters.red |
Satori 协议 | adapters.satori |
条件控制
本插件可以通过 assign
来控制一个具体的响应函数是否在不满足条件时跳过响应。
...
from nonebot import require
require("nonebot_plugin_alconna")
...
from arclet.alconna import Alconna, Subcommand, Option, Args
from nonebot_plugin_alconna import on_alconna, CommandResult
pip = Alconna(
"pip",
Subcommand(
"install", Args["pak", str],
Option("--upgrade"),
Option("--force-reinstall")
),
Subcommand("list", Option("--out-dated"))
)
pip_cmd = on_alconna(pip)
# 仅在命令为 `pip install pip` 时响应
@pip_cmd.assign("install.pak", "pip")
async def update(res: CommandResult):
...
# 仅在命令为 `pip list` 时响应
@pip_cmd.assign("list")
async def list_(res: CommandResult):
...
# 在命令为 `pip install xxx` 时响应
@pip_cmd.assign("install")
async def install(res: CommandResult):
...
此外,使用 AlconnaMatcher.dispatch
还能像 CommandGroup
一样为每个分发设置独立的 matcher:
update_cmd = pip_cmd.dispatch("install.pak", "pip")
@update_cmd.handle()
async def update(arp: CommandResult):
...
另外,AlconnaMatcher
有类似于 got
的 got_path
:
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 才能进入处理函数。
:::tip
path
支持 ~XXX 语法,其会把 ~ 替换为可能的父级路径:
pip = Alconna(
"pip",
Subcommand(
"install",
Args["pak", str],
Option("--upgrade|-U"),
Option("--force-reinstall"),
),
Subcommand("list", Option("--out-dated")),
)
pipcmd = on_alconna(pip)
pip_install_cmd = pipcmd.dispatch("install")
@pip_install_cmd.assign("~upgrade")
async def pip1_u(pak: Query[str] = Query("~pak")):
await pip_install_cmd.finish(f"pip upgrading {pak.result}...")
:::
匹配拓展
本插件提供了一个 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
参数注入模型变量
目前 Extension
的功能有:
validate
: 对于事件的来源适配器或 bot 选择是否接受响应output_converter
: 输出信息的自定义转换方法message_provider
: 从传入事件中自定义提取消息的方法receive_provider
: 对传入的消息 (Message 或 UniMessage) 的额外处理permission_check
: 命令对消息解析并确认头部匹配(即确认选择响应)时对发送者的权限判断parse_wrapper
: 对命令解析结果的额外处理send_wrapper
: 对发送的消息 (Message 或 UniMessage) 的额外处理before_catch
: 自定义依赖注入的绑定确认函数catch
: 自定义依赖注入处理函数post_init
: 响应器创建后对命令对象的额外除了
例如内置的 DiscordSlashExtension
,其可自动将 Alconna 对象翻译成 slash 指令并注册,且将收到的指令交互事件转为指令供命令解析:
from nonebot_plugin_alconna import Match, on_alconna
from nonebot_plugin_alconna.adapters.discord import DiscordSlashExtension
alc = Alconna(
["/"],
"permission",
Subcommand("add", Args["plugin", str]["priority?", int]),
Option("remove", Args["plugin", str]["time?", int]),
meta=CommandMeta(description="权限管理"),
)
matcher = on_alconna(alc, extensions=[DiscordSlashExtension()])
@matcher.assign("add")
async def add(plugin: Match[str], priority: Match[int]):
await matcher.finish(f"added {plugin.result} with {priority.result if priority.available else 0}")
@matcher.assign("remove")
async def remove(plugin: Match[str], time: Match[int]):
await matcher.finish(f"removed {plugin.result} with {time.result if time.available else -1}")
:::tip
全局的 Extension 可延迟加载 (即若有全局拓展加载于部分 AlconnaMatcher 之后,这部分响应器会被追加拓展)
:::