14 KiB
sidebar_position, description
| 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 之后,这部分响应器会被追加拓展)
:::