An alternative for got-and-reject in Nonebot
Project description
nonebot-plugin-waiter
该插件提供一个 got-and-reject 会话控制的替代方案,可自由控制超时时间
安装
pip install nonebot-plugin-waiter
使用
导入
from nonebot_plugin_waiter import waiter
创建
...
@waiter()
async def check(event: Event):
...
waiter
装饰一个函数来创建一个 Waiter
对象用以等待预期事件
该函数可以使用依赖注入
函数内需要自行判断输入事件是否符合预期并返回相应结果以供外层 handler 继续处理
waiter
有如下参数:
- waits: 等待的事件类型列表,可以是
Event
的类型或事件的get_type()
返回值; 如果 waits 为空则继承matcher
参数的事件响应器类型 - matcher: 所属的
Matcher
对象,如果不指定则使用当前上下文的Matcher
- parameterless: 非参数类型依赖列表
- keep_session: 是否保持会话,即仅允许会话发起者响应
- rule: 事件响应规则,例如
to_me()
- block: waiter 成功响应后是否阻塞事件传递,默认为
True
等待
可直接用 Waiter.wait
等待函数返回结果:
resp = await check.wait(timeout=60, default=False)
参数:
- before: 等待前发送的消息
- default: 超时时返回的默认值
- timeout: 等待超时时间
或使用异步迭代器持续等待,直到满足结果才退出循环。适合多轮对话:
async for resp in check(timeout=60, default=False):
...
参数:
- default: 超时时返回的默认值
- timeout: 等待超时时间
- retry: 重试次数,不设置则无限重试
- prompt: 重试时发送的消息,若没有设置
retry
则不发送
便捷函数
插件提供了一个 prompt
函数用于直接等待用户输入,适用于等待用户输入文本消息:
from nonebot_plugin_waiter import prompt
resp = await prompt("请输入XXX", timeout=60)
相应的,同时提供了一个 prompt_until
函数用于可重试一定次数地等待用户输入。
from nonebot_plugin_waiter import prompt_until
resp = await prompt_until(
"请输入数字",
lambda msg: msg.extract_plain_text().isdigit(),
timeout=60,
retry=5,
retry_prompt="输入错误,请输入数字。剩余次数:{count}",
)
基于此,还有一个 suggest
函数,用于向用户展示候选项并等待输入。
from nonebot_plugin_waiter import suggest
resp = await suggest("请选择xxx", ["a", "b", "c", "d"])
suggest_not
函数用于等待候选项以外的用户输入
resp = await suggest_not("XX已存在,请输入新的XX", not_expect=["a", "b", "c", "d"])
>>> XX已存在,请输入新的XX
以下为非候选项
- a
- b
- c
- d
配置
建议阅读源码来确定各配置项生效的位置
配置项 | 必填 | 默认值 | 说明 |
---|---|---|---|
waiter_timeout | 否 | 120 | 默认等待超时时间 |
waiter_retry_prompt | 否 | 输入错误,请重新输入。剩余次数:{count} | 默认重试时的提示信息 |
waiter_timeout_prompt | 否 | 等待超时。 | 默认超时时的提示信息 |
waiter_limited_prompt | 否 | 重试次数已用完,输入失败。 | 默认重试次数用完时的提示信息 |
waiter_suggest_hint | 否 | - {suggest} | 默认建议信息的提示 |
waiter_suggest_sep | 否 | \n | 默认建议信息的分隔符 |
waiter_suggest_not_hint | 否 | 以下为非候选项 | 默认非候选项前的提示信息 |
示例
等待用户输入数字,超时时间为 60 秒,此时 waits 接收所有来自当前用户的消息事件。
from nonebot import on_command
from nonebot.adapters import Event
from nonebot_plugin_waiter import waiter, prompt
test = on_command("test")
@test.handle()
async def _():
await test.send("请输入数字")
@waiter(waits=["message"], keep_session=True)
async def check(event: Event):
return event.get_plaintext()
resp = await check.wait(timeout=60)
# 上面的代码等价于下面的代码
# resp = await prompt("请输入数字", timeout=60)
if resp is None:
await test.send("等待超时")
return
if not resp.isdigit():
await test.send("无效输入")
return
await test.finish(f"你输入了{resp}")
等待用户输入数字,超时时间为 30 秒,只允许重试 5 次,此时 waits 接收所有来自当前用户的消息事件。
from nonebot import on_command
from nonebot.adapters import Event
from nonebot_plugin_waiter import waiter
test = on_command("test")
@test.handle()
async def _():
await test.send("请输入数字")
@waiter(waits=["message"], keep_session=True)
async def check(event: Event):
return event.get_plaintext()
async for resp in check(timeout=30, retry=5, prompt="输入错误,请输入数字。剩余次数:{count}"):
if resp is None:
await test.send("等待超时")
break
if not resp.isdigit():
continue
await test.send(f"你输入了{resp}")
break
else:
await test.send("输入失败")
在 telegram 适配器下等待用户点击按钮,超时时间为 30 秒,此时 waits 接收 telegram 的 CallbackQueryEvent 事件。
from nonebot import on, on_command
from nonebot.adapters.telegram import Bot
from nonebot.adapters.telegram.event import (
MessageEvent,
CallbackQueryEvent,
)
from nonebot.adapters.telegram.model import (
InlineKeyboardButton,
InlineKeyboardMarkup,
InputTextMessageContent,
InlineQueryResultArticle,
)
inline = on_command("inline")
@inline.handle()
async def _(bot: Bot, event: MessageEvent):
await bot.send(
event,
"Hello InlineKeyboard !",
reply_markup=InlineKeyboardMarkup(
inline_keyboard=[
[
InlineKeyboardButton(
text="Say hello to me",
callback_data="hello",
)
],
]
),
)
@waiter(waits=[CallbackQueryEvent])
async def check(event1: CallbackQueryEvent):
if ...:
return event1.id, event1.message
resp = await check.wait(timeout=30)
if resp is None:
await inline.finish("等待超时")
_id, _message = resp
if _message:
await bot.edit_message_text(
"Hello CallbackQuery!", _message.chat.id, _message.message_id
)
await bot.answer_callback_query(_id, text="Hello CallbackQuery!")
await inline.finish()
配合 nonebot-plugin-session
插件使用,实现在群内的多轮会话游戏:
from nonebot import on_command
from nonebot.adapters import Event
from nonebot_plugin_waiter import waiter
from nonebot_plugin_session import EventSession, SessionIdType
game = on_command("game")
@game.handle()
async def main(event: Event, session: EventSession):
session_id = session.get_id(SessionIdType.GROUP)
gm = Game(event.get_user_id(), session_id)
if gm.already_start():
await game.finish("游戏已经开始了, 请不要重复开始")
gm.start()
await game.send(f"开始游戏\n输入 “取消” 结束游戏")
@waiter(waits=["message"], block=False)
async def listen(_event: Event, _session: EventSession):
if _session.get_id(SessionIdType.GROUP) != session_id:
return
text = _event.get_message().extract_plain_text()
if text == "取消":
return False
return await gm.handle(text)
async for resp in listen(timeout=60):
if resp is False:
await game.finish("游戏已取消")
gm.finish()
break
if resp is None:
await game.finish("游戏已超时, 请重新开始")
gm.finish()
break
await game.send(resp)
if gm.is_finish():
await game.finish("游戏结束")
break
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
Built Distribution
Hashes for nonebot_plugin_waiter-0.8.0.tar.gz
Algorithm | Hash digest | |
---|---|---|
SHA256 | b750f2a3b255ebe9ab7ce75056c1e829281d898c6da8d81552147df2b1686270 |
|
MD5 | eb658276d3218428490fb3e80733daf5 |
|
BLAKE2b-256 | ae168e25227cc0fd57872e40279a485c0be8994d38fb7d4ec7289a073d41d68c |
Hashes for nonebot_plugin_waiter-0.8.0-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | eabf284f783a1e534591a552139a747623e621453a790b554854bc8a386f783f |
|
MD5 | c61f1da606277e1ad1b46d16a9be5373 |
|
BLAKE2b-256 | de86573ab600ab252d5f8c58e90e6a09f0e0b443f39e1217ec5d1b3a60fda64b |