mirror of
https://github.com/nonebot/nonebot2.git
synced 2024-11-25 01:25:04 +08:00
591 lines
17 KiB
Plaintext
591 lines
17 KiB
Plaintext
---
|
||
sidebar_position: 5
|
||
description: 通用消息组件
|
||
---
|
||
|
||
import Tabs from "@theme/Tabs";
|
||
import TabItem from "@theme/TabItem";
|
||
|
||
# 通用消息组件
|
||
|
||
`uniseg` 模块属于 `nonebot-plugin-alconna` 的子插件，其提供了一套通用的消息组件，用于在 `nonebot-plugin-alconna` 下构建通用消息。
|
||
|
||
## 通用消息段
|
||
|
||
适配器下的消息段标注会匹配适配器特定的 `MessageSegment`， 而通用消息段与适配器消息段的区别在于：
|
||
通用消息段会匹配多个适配器中相似类型的消息段，并返回 `uniseg` 模块中定义的 [`Segment` 模型](https://nonebot.dev/docs/next/best-practice/alconna/utils#%E9%80%9A%E7%94%A8%E6%B6%88%E6%81%AF%E6%AE%B5), 以达到**跨平台接收消息**的作用。
|
||
|
||
`nonebot-plugin-alconna.uniseg` 提供了类似 `MessageSegment` 的通用消息段，并可在 `Alconna` 下直接标注使用：
|
||
|
||
```python
|
||
class Segment:
|
||
"""基类标注"""
|
||
children: List["Segment"]
|
||
|
||
class Text(Segment):
|
||
"""Text对象, 表示一类文本元素"""
|
||
text: str
|
||
styles: Dict[Tuple[int, int], List[str]]
|
||
|
||
class At(Segment):
|
||
"""At对象, 表示一类提醒某用户的元素"""
|
||
flag: Literal["user", "role", "channel"]
|
||
target: str
|
||
display: Optional[str]
|
||
|
||
class AtAll(Segment):
|
||
"""AtAll对象, 表示一类提醒所有人的元素"""
|
||
here: bool
|
||
|
||
class Emoji(Segment):
|
||
"""Emoji对象, 表示一类表情元素"""
|
||
id: str
|
||
name: Optional[str]
|
||
|
||
class Media(Segment):
|
||
url: Optional[str]
|
||
id: Optional[str]
|
||
path: Optional[Union[str, Path]]
|
||
raw: Optional[Union[bytes, BytesIO]]
|
||
mimetype: Optional[str]
|
||
name: str
|
||
|
||
to_url: ClassVar[Optional[MediaToUrl]]
|
||
|
||
class Image(Media):
|
||
"""Image对象, 表示一类图片元素"""
|
||
|
||
class Audio(Media):
|
||
"""Audio对象, 表示一类音频元素"""
|
||
duration: Optional[int]
|
||
|
||
class Voice(Media):
|
||
"""Voice对象, 表示一类语音元素"""
|
||
duration: Optional[int]
|
||
|
||
class Video(Media):
|
||
"""Video对象, 表示一类视频元素"""
|
||
|
||
class File(Segment):
|
||
"""File对象, 表示一类文件元素"""
|
||
id: str
|
||
name: Optional[str]
|
||
|
||
class Reply(Segment):
|
||
"""Reply对象，表示一类回复消息"""
|
||
id: str
|
||
"""此处不一定是消息ID，可能是其他ID，如消息序号等"""
|
||
msg: Optional[Union[Message, str]]
|
||
origin: Optional[Any]
|
||
|
||
class Reference(Segment):
|
||
"""Reference对象，表示一类引用消息。转发消息 (Forward) 也属于此类"""
|
||
id: Optional[str]
|
||
"""此处不一定是消息ID，可能是其他ID，如消息序号等"""
|
||
children: List[Union[RefNode, CustomNode]]
|
||
|
||
class Hyper(Segment):
|
||
"""Hyper对象，表示一类超级消息。如卡片消息、ark消息、小程序等"""
|
||
format: Literal["xml", "json"]
|
||
raw: Optional[str]
|
||
content: Optional[Union[dict, list]]
|
||
|
||
class Other(Segment):
|
||
"""其他 Segment"""
|
||
origin: MessageSegment
|
||
|
||
```
|
||
|
||
:::tips
|
||
|
||
或许你注意到了 `Segment` 上有一个 `children` 属性。
|
||
|
||
这是因为在 [`Satori`](https://satori.js.org/zh-CN/) 协议的规定下，一类元素可以用其子元素来代表一类兼容性消息
|
||
（例如，qq 的商场表情在某些平台上可以用图片代替）。
|
||
|
||
为此，本插件提供了两种方式来表达 "获取子元素" 的方法：
|
||
|
||
```python
|
||
from nonebot_plugin_alconna.builtins.uniseg.chronocat import MarketFace
|
||
from nonebot_plugin_alconna import Args, Image, Alconna, select, select_first
|
||
|
||
# 表示这个指令需要的图片要么直接是 Image 要么是在 MarketFace 元素内的 Image
|
||
alc1 = Alconna("make_meme", Args["img", [Image, Image.from_(MarketFace)]])
|
||
|
||
# 表示这个指令需要的图片会在目标元素下进行搜索，将所有符合 Image 的元素选出来并将第一个作为结果
|
||
alc2 = Alconna("make_meme", Args["img", select(Image, index=0)]) # 也可以使用 select_first(Image)
|
||
```
|
||
|
||
:::
|
||
|
||
## 通用消息序列
|
||
|
||
`nonebot-plugin-alconna.uniseg` 同时提供了一个类似于 `Message` 的 `UniMessage` 类型，其元素为经过通用标注转换后的通用消息段。
|
||
|
||
你可以用如下方式获取 `UniMessage`：
|
||
|
||
<Tabs groupId="get_unimsg">
|
||
<TabItem value="depend" label="使用依赖注入">
|
||
|
||
通过提供的 `UniversalMessage` 或 `UniMsg` 依赖注入器来获取 `UniMessage`。
|
||
|
||
```python
|
||
from nonebot_plugin_alconna.uniseg import UniMsg, At, Reply
|
||
|
||
|
||
matcher = on_xxx(...)
|
||
|
||
@matcher.handle()
|
||
async def _(msg: UniMsg):
|
||
reply = msg[Reply, 0]
|
||
print(reply.origin)
|
||
if msg.has(At):
|
||
ats = msg.get(At)
|
||
print(ats)
|
||
...
|
||
```
|
||
|
||
</TabItem>
|
||
<TabItem value="method" label="使用 UniMessage.generate">
|
||
|
||
注意，`generate` 方法在响应器以外的地方如果不传入 `event` 与 `bot` 则无法处理 reply。
|
||
|
||
```python
|
||
from nonebot import Message, EventMessage
|
||
from nonebot_plugin_alconna.uniseg import UniMessage
|
||
|
||
|
||
matcher = on_xxx(...)
|
||
|
||
@matcher.handle()
|
||
async def _(message: Message = EventMessage()):
|
||
msg = await UniMessage.generate(message=message)
|
||
msg1 = UniMessage.generate_without_reply(message=message)
|
||
```
|
||
|
||
</TabItem>
|
||
</Tabs>
|
||
|
||
不仅如此，你还可以通过 `UniMessage` 的 `export` 与 `send` 方法来**跨平台发送消息**。
|
||
|
||
`UniMessage.export` 会通过传入的 `bot: Bot` 参数，或上下文中的 `Bot` 对象读取适配器信息，并使用对应的生成方法把通用消息转为适配器对应的消息序列：
|
||
|
||
```python
|
||
from nonebot import Bot, on_command
|
||
from nonebot_plugin_alconna.uniseg import Image, UniMessage
|
||
|
||
|
||
test = on_command("test")
|
||
|
||
@test.handle()
|
||
async def handle_test():
|
||
await test.send(await UniMessage(Image(path="path/to/img")).export())
|
||
```
|
||
|
||
除此之外 `UniMessage.send` 方法基于 `UniMessage.export` 并调用各适配器下的发送消息方法，返回一个 `Receipt` 对象，用于修改/撤回消息：
|
||
|
||
```python
|
||
from nonebot import Bot, on_command
|
||
from nonebot_plugin_alconna.uniseg import UniMessage
|
||
|
||
|
||
test = on_command("test")
|
||
|
||
@test.handle()
|
||
async def handle():
|
||
receipt = await UniMessage.text("hello!").send(at_sender=True, reply_to=True)
|
||
await receipt.recall(delay=1)
|
||
```
|
||
|
||
而在 `AlconnaMatcher` 下，`got`, `send`, `reject` 等可以发送消息的方法皆支持使用 `UniMessage`，不需要手动调用 export 方法：
|
||
|
||
```python
|
||
from arclet.alconna import Alconna, Args
|
||
from nonebot_plugin_alconna import Match, AlconnaMatcher, on_alconna
|
||
from nonebot_plugin_alconna.uniseg import At, UniMessage
|
||
|
||
|
||
test_cmd = on_alconna(Alconna("test", Args["target?", At]))
|
||
|
||
@test_cmd.handle()
|
||
async def tt_h(matcher: AlconnaMatcher, target: Match[At]):
|
||
if target.available:
|
||
matcher.set_path_arg("target", target.result)
|
||
|
||
@test_cmd.got_path("target", prompt="请输入目标")
|
||
async def tt(target: At):
|
||
await test_cmd.send(UniMessage([target, "\ndone."]))
|
||
```
|
||
|
||
:::caution
|
||
|
||
在响应器以外的地方，除非启用了 `alconna_apply_fetch_targets` 配置项，否则 `bot` 参数必须手动传入。
|
||
|
||
:::
|
||
|
||
### 构造
|
||
|
||
如同 `Message`, `UniMessage` 可以传入单个字符串/消息段，或可迭代的字符串/消息段：
|
||
|
||
```python
|
||
from nonebot_plugin_alconna.uniseg import UniMessage, At
|
||
|
||
|
||
msg = UniMessage("Hello")
|
||
msg1 = UniMessage(At("user", "124"))
|
||
msg2 = UniMessage(["Hello", At("user", "124")])
|
||
```
|
||
|
||
`UniMessage` 上同时存在便捷方法，令其可以链式地添加消息段：
|
||
|
||
```python
|
||
from nonebot_plugin_alconna.uniseg import UniMessage, At, Image
|
||
|
||
|
||
msg = UniMessage.text("Hello").at("124").image(path="/path/to/img")
|
||
assert msg == UniMessage(
|
||
["Hello", At("user", "124"), Image(path="/path/to/img")]
|
||
)
|
||
```
|
||
|
||
### 拼接消息
|
||
|
||
`str`、`UniMessage`、`Segment` 对象之间可以直接相加，相加均会返回一个新的 `UniMessage` 对象：
|
||
|
||
```python
|
||
# 消息序列与消息段相加
|
||
UniMessage("text") + Text("text")
|
||
# 消息序列与字符串相加
|
||
UniMessage([Text("text")]) + "text"
|
||
# 消息序列与消息序列相加
|
||
UniMessage("text") + UniMessage([Text("text")])
|
||
# 字符串与消息序列相加
|
||
"text" + UniMessage([Text("text")])
|
||
# 消息段与消息段相加
|
||
Text("text") + Text("text")
|
||
# 消息段与字符串相加
|
||
Text("text") + "text"
|
||
# 消息段与消息序列相加
|
||
Text("text") + UniMessage([Text("text")])
|
||
# 字符串与消息段相加
|
||
"text" + Text("text")
|
||
```
|
||
|
||
如果需要在当前消息序列后直接拼接新的消息段，可以使用 `Message.append`、`Message.extend` 方法，或者使用自加：
|
||
|
||
```python
|
||
msg = UniMessage([Text("text")])
|
||
# 自加
|
||
msg += "text"
|
||
msg += Text("text")
|
||
msg += UniMessage([Text("text")])
|
||
# 附加
|
||
msg.append(Text("text"))
|
||
# 扩展
|
||
msg.extend([Text("text")])
|
||
```
|
||
|
||
### 使用消息模板
|
||
|
||
`UniMessage.template` 同样类似于 `Message.template`，可以用于格式化消息，大体用法参考 [消息模板](../../tutorial/message#使用消息模板)。
|
||
|
||
这里额外说明 `UniMessage.template` 的拓展控制符
|
||
|
||
相比 `Message`，UniMessage 对于 {:XXX} 做了另一类拓展。其能够识别例如 At(xxx, yyy) 或 Emoji(aaa, bbb)的字符串并执行
|
||
|
||
以 At(...) 为例：
|
||
|
||
```python title=使用通用消息段的拓展控制符
|
||
>>> from nonebot_plugin_alconna.uniseg import UniMessage
|
||
>>> UniMessage.template("{:At(user, target)}").format(target="123")
|
||
UniMessage(At("user", "123"))
|
||
>>> UniMessage.template("{:At(type=user, target=id)}").format(id="123")
|
||
UniMessage(At("user", "123"))
|
||
>>> UniMessage.template("{:At(type=user, target=123)}").format()
|
||
UniMessage(At("user", "123"))
|
||
```
|
||
|
||
而在 `AlconnaMatcher` 中，{:XXX} 更进一步地提供了获取 `event` 和 `bot` 中的属性的功能：
|
||
|
||
```python title=在AlconnaMatcher中使用通用消息段的拓展控制符
|
||
from arclet.alconna import Alconna, Args
|
||
from nonebot_plugin_alconna import At, Match, UniMessage, AlconnaMatcher, on_alconna
|
||
|
||
|
||
test_cmd = on_alconna(Alconna("test", Args["target?", At]))
|
||
|
||
@test_cmd.handle()
|
||
async def tt_h(matcher: AlconnaMatcher, target: Match[At]):
|
||
if target.available:
|
||
matcher.set_path_arg("target", target.result)
|
||
|
||
@test_cmd.got_path(
|
||
"target",
|
||
prompt=UniMessage.template("{:At(user, $event.get_user_id())} 请确认目标")
|
||
)
|
||
async def tt():
|
||
await test_cmd.send(
|
||
UniMessage.template("{:At(user, $event.get_user_id())} 已确认目标为 {target}")
|
||
)
|
||
```
|
||
|
||
另外也有 `$message_id` 与 `$target` 两个特殊值。
|
||
|
||
### 检查消息段
|
||
|
||
我们可以通过 `in` 运算符或消息序列的 `has` 方法来：
|
||
|
||
```python
|
||
# 是否存在消息段
|
||
At("user", "1234") in message
|
||
# 是否存在指定类型的消息段
|
||
At in message
|
||
```
|
||
|
||
我们还可以使用 `only` 方法来检查消息中是否仅包含指定的消息段：
|
||
|
||
```python
|
||
# 是否都为 "test"
|
||
message.only("test")
|
||
# 是否仅包含指定类型的消息段
|
||
message.only(Text)
|
||
```
|
||
|
||
### 获取消息纯文本
|
||
|
||
类似于 `Message.extract_plain_text()`，用于获取通用消息的纯文本：
|
||
|
||
```python
|
||
from nonebot_plugin_alconna.uniseg import UniMessage, At
|
||
|
||
|
||
# 提取消息纯文本字符串
|
||
assert UniMessage(
|
||
[At("user", "1234"), "text"]
|
||
).extract_plain_text() == "text"
|
||
```
|
||
|
||
### 遍历
|
||
|
||
通用消息序列继承自 `List[Segment]` ，因此可以使用 `for` 循环遍历消息段：
|
||
|
||
```python
|
||
for segment in message: # type: Segment
|
||
...
|
||
```
|
||
|
||
### 过滤、索引与切片
|
||
|
||
消息序列对列表的索引与切片进行了增强，在原有列表 `int` 索引与 `slice` 切片的基础上，支持 `type` 过滤索引与切片：
|
||
|
||
```python
|
||
from nonebot_plugin_alconna.uniseg import UniMessage, At, Text, Reply
|
||
|
||
|
||
message = UniMessage(
|
||
[
|
||
Reply(...),
|
||
"text1",
|
||
At("user", "1234"),
|
||
"text2"
|
||
]
|
||
)
|
||
# 索引
|
||
message[0] == Reply(...)
|
||
# 切片
|
||
message[0:2] == UniMessage([Reply(...), Text("text1")])
|
||
# 类型过滤
|
||
message[At] == Message([At("user", "1234")])
|
||
# 类型索引
|
||
message[At, 0] == At("user", "1234")
|
||
# 类型切片
|
||
message[Text, 0:2] == UniMessage([Text("text1"), Text("text2")])
|
||
```
|
||
|
||
我们也可以通过消息序列的 `include`、`exclude` 方法进行类型过滤：
|
||
|
||
```python
|
||
message.include(Text, At)
|
||
message.exclude(Reply)
|
||
```
|
||
|
||
同样的，消息序列对列表的 `index`、`count` 方法也进行了增强，可以用于索引指定类型的消息段：
|
||
|
||
```python
|
||
# 指定类型首个消息段索引
|
||
message.index(Text) == 1
|
||
# 指定类型消息段数量
|
||
message.count(Text) == 2
|
||
```
|
||
|
||
此外，消息序列添加了一个 `get` 方法，可以用于获取指定类型指定个数的消息段：
|
||
|
||
```python
|
||
# 获取指定类型指定个数的消息段
|
||
message.get(Text, 1) == UniMessage([Text("test1")])
|
||
```
|
||
|
||
## 消息发送
|
||
|
||
前面提到，通用消息可用 `UniMessage.send` 发送自身：
|
||
|
||
```python
|
||
async def send(
|
||
self,
|
||
target: Union[Event, Target, None] = None,
|
||
bot: Optional[Bot] = None,
|
||
fallback: bool = True,
|
||
at_sender: Union[str, bool] = False,
|
||
reply_to: Union[str, bool] = False,
|
||
) -> Receipt:
|
||
```
|
||
|
||
实际上，`UniMessage` 同时提供了获取消息事件 id 与消息发送对象的方法:
|
||
|
||
<Tabs groupId="get_unimsg">
|
||
<TabItem value="depend" label="使用依赖注入">
|
||
|
||
通过提供的 `MessageTarget`, `MessageId` 或 `MsgTarget`, `MsgId` 依赖注入器来获取消息事件 id 与消息发送对象。
|
||
|
||
```python
|
||
from nonebot_plugin_alconna.uniseg import MessageId, MsgTarget
|
||
|
||
|
||
matcher = on_xxx(...)
|
||
|
||
@matcher.handle()
|
||
asycn def _(target: MsgTarget, msg_id: MessageId):
|
||
...
|
||
```
|
||
|
||
</TabItem>
|
||
<TabItem value="method" label="使用 UniMessage 的方法">
|
||
|
||
```python
|
||
from nonebot import Event, Bot
|
||
from nonebot_plugin_alconna.uniseg import UniMessage, Target
|
||
|
||
|
||
matcher = on_xxx(...)
|
||
|
||
@matcher.handle()
|
||
asycn def _(bot: Bot, event: Event):
|
||
target: Target = UniMessage.get_target(event, bot)
|
||
msg_id: str = UniMessage.get_message_id(event, bot)
|
||
|
||
```
|
||
|
||
</TabItem>
|
||
</Tabs>
|
||
|
||
`send`, `get_target`, `get_message_id` 中与 `event`, `bot` 相关的参数都会尝试从上下文中获取对象。
|
||
|
||
### 消息发送对象
|
||
|
||
消息发送对象是用来描述响应消息时的发送对象或者主动发送消息时的目标对象的对象，它包含了以下属性：
|
||
|
||
```python
|
||
class Target:
|
||
id: str
|
||
"""目标id；若为群聊则为group_id或者channel_id，若为私聊则为user_id"""
|
||
parent_id: str
|
||
"""父级id；若为频道则为guild_id，其他情况下可能为空字符串（例如 Feishu 下可作为部门 id）"""
|
||
channel: bool
|
||
"""是否为频道，仅当目标平台符合频道概念时"""
|
||
private: bool
|
||
"""是否为私聊"""
|
||
source: str
|
||
"""可能的事件id"""
|
||
self_id: Union[str, None]
|
||
"""机器人id，若为 None 则 Bot 对象会随机选择"""
|
||
selector: Union[Callable[[Bot], Awaitable[bool]], None]
|
||
"""选择器，用于在多个 Bot 对象中选择特定 Bot"""
|
||
extra: Dict[str, Any]
|
||
"""额外信息，用于适配器扩展"""
|
||
```
|
||
|
||
其构造时需要如下参数：
|
||
|
||
- `id` 为目标id；若为群聊则为 group_id 或者 channel_id，若为私聊则为user_id
|
||
- `parent_id` 为父级id；若为频道则为 guild_id，其他情况下可能为空字符串（例如 Feishu 下可作为部门 id）
|
||
- `channel` 为是否为频道，仅当目标平台符合频道概念时
|
||
- `private` 为是否为私聊
|
||
- `source` 为可能的事件id
|
||
- `self_id` 为机器人id，若为 None 则 Bot 对象会随机选择
|
||
- `selector` 为选择器，用于在多个 Bot 对象中选择特定 Bot
|
||
- `scope` 为适配器范围，用于传入内置的特定选择器
|
||
- `adapter` 为适配器名称，若为 None 则需要明确指定 Bot 对象
|
||
- `platform` 为平台名称，仅当目标适配器存在多个平台时使用
|
||
- `extra` 为额外信息，用于适配器扩展
|
||
|
||
通过 `Target` 对象，我们可以在 `UniMessage.send` 中指定发送对象：
|
||
|
||
```python
|
||
from nonebot_plugin_alconna.uniseg import UniMessage, MsgTarget, Target, SupportScope
|
||
|
||
|
||
matcher = on_xxx(...)
|
||
|
||
@matcher.handle()
|
||
async def _(target: MsgTarget):
|
||
await UniMessage("Hello!").send(target=target)
|
||
target1 = Target("xxxx", scope=SupportScope.qq_client)
|
||
await UniMessage("Hello!").send(target=target1)
|
||
```
|
||
|
||
### 主动发送消息
|
||
|
||
`UniMessage.send` 也可以用于主动发送消息：
|
||
|
||
```python
|
||
from nonebot_plugin_alconna.uniseg import UniMessage, Target, SupportScope
|
||
from nonebot import get_driver
|
||
|
||
|
||
driver = get_driver()
|
||
|
||
@driver.on_startup
|
||
async def on_startup():
|
||
target = Target("xxxx", scope=SupportScope.qq_client)
|
||
await UniMessage("Hello!").send(target=target)
|
||
```
|
||
|
||
## 自定义消息段
|
||
|
||
`uniseg` 提供了部分方法来允许用户自定义 Segment 的序列化和反序列化：
|
||
|
||
```python
|
||
from dataclasses import dataclass
|
||
|
||
from nonebot.adapters import Bot
|
||
from nonebot.adapters import MessageSegment as BaseMessageSegment
|
||
from nonebot.adapters.satori import Custom, Message, MessageSegment
|
||
|
||
from nonebot_plugin_alconna.uniseg.builder import MessageBuilder
|
||
from nonebot_plugin_alconna.uniseg.exporter import MessageExporter
|
||
from nonebot_plugin_alconna.uniseg import Segment, custom_handler, custom_register
|
||
|
||
|
||
@dataclass
|
||
class MarketFace(Segment):
|
||
tabId: str
|
||
faceId: str
|
||
key: str
|
||
|
||
|
||
@custom_register(MarketFace, "chronocat:marketface")
|
||
def mfbuild(builder: MessageBuilder, seg: BaseMessageSegment):
|
||
if not isinstance(seg, Custom):
|
||
raise ValueError("MarketFace can only be built from Satori Message")
|
||
return MarketFace(**seg.data)(*builder.generate(seg.children))
|
||
|
||
|
||
@custom_handler(MarketFace)
|
||
async def mfexport(exporter: MessageExporter, seg: MarketFace, bot: Bot, fallback: bool):
|
||
if exporter.get_message_type() is Message:
|
||
return MessageSegment("chronocat:marketface", seg.data)(await exporter.export(seg.children, bot, fallback))
|
||
|
||
```
|
||
|
||
具体而言，你可以使用 `custom_register` 来增加一个从 MessageSegment 到 Segment 的处理方法；使用 `custom_handler` 来增加一个从 Segment 到 MessageSegment 的处理方法。
|