📝NoneBot2 接入 QQ 频道的几种方式

[MingxuanGame]

NoneBot2 接入 QQ 频道的几种方式

目前使用 NoneBot2 接入频道有两种方式

• 使用 QQ 频道适配器
• 使用 go-cqhttp + go-cqhttp 频道支持适配补丁

QQ 频道适配器

QQ 频道适配器（下面简称频道适配器）使用官方机器人接口，这意味着你需要拥有 QQ 开放平台 账号和机器人

申请机器人和机器人资料设置这里不会提到，下面将讲解 NoneBot 如何连接官方接口

新建频道机器人项目

使用 nb init 新建项目并选择频道适配器

配置机器人

2023/1/9更新：配置文档已经写进频道适配器的 README 内，可参考 README 来配置

需要修改下面的配置

• DRIVER
• QQGUILD_BOTS 频道机器人 List

此外，还有一个可选配置

• QQGUILD_IS_SANDBOX 是否为沙箱环境（bool）

沙箱环境只会收到测试频道的事件，且调用openapi仅能操作测试频道

在 .env 或 .env.* 中修改配置文件

DRIVER 需要同时拥有支持 HTTP 和支持正向 WebSocket 的驱动器，选择驱动器参见文档

至于为什么需要支持 HTTP 和支持正向 WebSocket 的驱动器，因为频道适配器会首先使用 HTTP 获取通用 WSS 接入点，然后使用正向 WebSocket 保持长连接，更多信息请参阅《QQ机器人文档》

我个人使用的驱动器组合如下

DRIVER=~fastapi+~httpx+~websockets

QQGUILD_BOTS 为一个频道机器人 List，其中每个元素均为一个 Dict，拥有三个 key

• id BotAppID（开发者ID）
• token 机器人令牌
• secret 机器人密钥

就像下面这样

[
    {
        "id": 100000000,
        "token": "abcdef...",
        "secret": "abcdef..."
    }
]

WARN 请确保配置项值为 JSON 格式的数据，如果配置项值解析失败将作为字符串处理

下方为我的配置

DRIVER=~fastapi+~websockets+~httpx
QQGUILD_BOTS=[{"id":101******,"token":"1dY*****************************","secret":"d3d********************************"}]
QQGUILD_IS_SANDBOX=true

对应开发平台的开发设置

启动 NoneBot，当出现下面的日志时即为连接成功

[INFO] nonebot | QQ Guild | Bot {BotAppID} connected

这时即可收到机器人相关¹的消息，效果如下

Footnotes

1. @机器人或回复机器人的消息等，可启用 GUILD_MESSAGES intents 来监听所有消息事件（仅私域） ↩

[MingxuanGame]

使用 go-cqhttp + go-cqhttp 频道支持适配补丁

对于拥有官方机器人的人来说，使用官方接口是一个好的选择，但是官方接口也有诸多限制，比如

• 上线机器人拥有严格的审核
• 只能响应机器人相关消息
• ...

如果不想被这些条条框框所限制，或者没有官方机器人，使用 go-cqhttp（下面简称 gocq）+ go-cqhttp 频道支持适配补丁（下面简称补丁）是一个好的选择，只需要你的机器人在频道内即可

gocq + 补丁基于 OneBot V11，这意味着你需要使用 OneBot V11 适配器而不是前文提到的 QQ 频道适配器

新建项目

同样使用 nb init 新建，不过这一次选择 OneBot V11 适配器

配置 gocq

gocq 的登录与配置与连接 NoneBot 这里不再赘述，详情请看 gocq 文档 和 OneBot 适配器文档

但是这里需要强调一点：使用频道功能需要将登录协议设置为 安卓手机（1）或者 IPAD（0或者5），登录协议修改参考 gocq 文档

安装补丁

补丁是一个 NoneBot 插件，因此可以通过脚手架安装

nb plugin install nonebot-plugin-guild-patch

也可以通过 pip/poetry 等安装

pip install nonebot_plugin_guild_patch
poetry add nonebot_plugin_guild_patch

---

在上面的步骤执行完毕后，启动 gocq 和 NoneBot，新建像下面这样的 echo 插件

from nonebot.rule import to_me
from nonebot.plugin import on_command
from nonebot.params import CommandArg
from nonebot.adapters.onebot.v11 import Message
from nonebot_plugin_guild_patch import GuildMessageEvent

matcher = on_command("echo", rule=to_me())


@matcher.handle()
async def _(args: Message = CommandArg()):
    await matcher.send(args)

第5行的导入虽然没有使用，但是这个导入不可删去，因为导入这个才会添加频道消息事件类型，这样才会将收到的消息转换成 GuildMessageEvent 而不是基类（源码参考），你也可以通过加载插件的方式来确保频道消息事件类型已经被添加

效果如下

接下来，你就可以开始使用 NoneBot 开发频道机器人了

相关链接：

• gocq 频道文档简介
• gocq 频道 API
• gocq 频道事件
• 补丁的 GitHub

[MingxuanGame]

发送图片

机器人发送图片是一件很平常的事情，接下来将讲解不同的接入方式如何发送图片

go-cqhttp + 补丁

这种方式发送图片的实现最简单，只需要像向 QQ 内发送图片那样发送即可，即使用 MessageSegment.image()

下面的 流汗 插件就是一个例子

from pathlib import Path

from nonebot.rule import to_me
from nonebot.plugin import on_command
from nonebot.adapters.onebot.v11 import MessageSegment
from nonebot_plugin_guild_patch import GuildMessageEvent  # DO NOT DELETE

matcher = on_command("流汗", rule=to_me())


@matcher.handle()
async def _():
    await matcher.finish(MessageSegment.image(Path(__file__).parent / "ddl.png"))

效果如下

更多用法详见 MessageSegment.image()

频道适配器

频道适配器的发送相对麻烦一点

发送网络图片

使用 MessageSegment.image()，但是只能填 url，即发送网络图片

不要想着本地起一个 HTTP Server 来发送图片，因为你填 url，并不是由你发送图片，而是由平台下载图片再发送。填写本地 url 会返回 ip ban（40013）错误，更多信息参考《QQ机器人文档》)

下面的 mix 插件就是一个例子

from nonebot import on_command
from nonebot.adapters.qqguild import MessageSegment

matcher = on_command("mix")


@matcher.handle()
async def _():
    msg = MessageSegment.image("https://avatars.githubusercontent.com/u/32300164?v=4")
    await matcher.finish(msg)

效果如下

发送本地图片

使用 MessageSegment.file_image() 发送，可填 bytes/BytesIO/Path

发送本地图片不能与 markdown message_reference ark embed 同时使用¹

from pathlib import Path

from nonebot import on_command
from nonebot.adapters.qqguild import MessageSegment

matcher = on_command("流汗")


@matcher.handle()
async def _():
    msg = MessageSegment.file_image(Path(__file__).parent / "ddl.png")
    await matcher.finish(msg)

以下内容已经过时，频道适配器已经支持 form-data 发送图片

WARN 这部分的内容相对较难理解

本地图片要稍微棘手一点，因为根据文档，发送本地图片需要以 form-data 形式，但是频道适配器是以 JSON 的方式发送的，所以这里我们需要魔改一下

NoneBot 发送消息（bot.send()）的流程

先让我们了解下你发送一条消息时，nonebot 在背后做了什么

在你发送一条消息，即调用了 bot.send() 时，bot.send() 通常会将你的调用转换为调用 bot.call_api()

bot.call_api() 会继续调用 adapter._call_api() 来执行调用发送消息 API 操作（源码参考）

然后将调用 API 的结果返回

用流程图表示，就是下面这样的

当然这里是省略了一些步骤的，比如 bot.call_api() 运行 API 调用相关钩子，但是大体流程是这样的

剖析相关方法

bot.send()（源码）

bot.send() 方法对 Message 进行处理，变成调用 API 的参数，然后调用 post_messages API

bot.__getattr__ 即为 bot.call_api，所以 bot.post_messages() 等于 bot.call_api("post_messages")

adapter._call_api()（源码）

adapter._call_api() 方法会在一个常量 API_HANDLERS 中寻找相应 API 的 handler

API post_messages 对应 _post_messages handler（源码）

在这个 handler 执行请求并返回结果

开始魔改

通过刚刚你已经知道了发送消息的过程，接下来让我们魔改 bot.send() 和 adapter._call_api() 方法

bot.send()

我们需要增加三行，以实现对本地图片参数支持

建议新建一个消息段类型实现支持，下文是以此方式实现的

首先新建一个新类型消息段，只需要继承 MessageSegment 即可

from nonebot.adapters.qqguild.message import MessageSegment


class LocalImage(MessageSegment):
    @overrides(MessageSegment)
    def __str__(self) -> str:
        return "<local_image>"

然后新增一个函数，用来构造 LocalImage 这个消息段

你可以自己决定形参类型，这里我选择 str | Path | bytes | BytesIO，前两者是打开文件读取为 bytes

from io import BytesIO
from pathlib import Path
from typing import Union


def local_image(data: Union[str, Path, bytes, BytesIO]) -> LocalImage:
    if isinstance(data, (str, Path)):
        with open(data, "rb") as f:
            b = f.read()
    elif isinstance(data, BytesIO):
        b = data.getvalue()
    else:
        b = data
    return LocalImage("local_image", data={"content": b})

然后重写一个新的 bot.send() 方法，增加下面这几行以实现对上面消息段的支持

local_image = message["local_image"] or None
if local_image:
    local_image = local_image[-1].data["content"]

以及 self.post_messages 添加 file_image 参数

return await self.post_messages(
    channel_id=event.channel_id,
    msg_id=event.id,
    content=content,
    embed=embed,
    ark=ark,
    image=image,
+    file_image=local_image,
)

最终成果是这样的

点击展开

from io import BytesIO
from pathlib import Path
from typing import Any, Union

from nonebot.typing import overrides
from nonebot.adapters.qqguild.bot import Bot
from nonebot.adapters.qqguild.event import Event, MessageEvent
from nonebot.adapters.qqguild.message import Message, MessageSegment


class LocalImage(MessageSegment):
    @overrides(MessageSegment)
    def __str__(self) -> str:
        return "<local_image>"


def local_image(data: Union[str, Path, bytes, BytesIO]) -> LocalImage:
    if isinstance(data, (str, Path)):
        with open(data, "rb") as f:
            b = f.read()
    elif isinstance(data, BytesIO):
        b = data.getvalue()
    else:
        b = data
    return LocalImage("local_image", data={"content": b})


@overrides(Bot)
async def send(
    self,
    event: Event,
    message: Union[str, Message, MessageSegment],
    **kwargs,
) -> Any:
    if (
        not isinstance(event, MessageEvent)
        or not event.channel_id
        or not event.id
    ):
        raise RuntimeError("Event cannot be replied to!")
    message = (
        MessageSegment.text(message) if isinstance(message, str) else message
    )
    message = message if isinstance(message, Message) else Message(message)

    content = message.extract_content() or None
    embed = message["embed"] or None
    if embed:
        embed = embed[-1].data["embed"]
    ark = message["ark"] or None
    if ark:
        ark = ark[-1].data["ark"]
    image = message["attachment"] or None
    if image:
        image = image[-1].data["url"]
    local_image = message["local_image"] or None
    if local_image:
        local_image = local_image[-1].data["content"]
    return await self.post_messages(
        channel_id=event.channel_id,
        msg_id=event.id,
        content=content,
        embed=embed,
        ark=ark,
        image=image,
        file_image=local_image,
    )


Bot.send = send

_post_messages

原来的 _post_messages handler 不支持 form-data，让我们重写一个 handler 代替原来的

首先重新写一个 Model，用于解析 file_image 参数

from pydantic import BaseModel

class MessageSend(BaseModel):
    content: Optional[str] = None
    embed: Optional[MessageEmbed] = None
    ark: Optional[MessageArk] = None
    message_reference: Optional[MessageReference] = None
    image: Optional[str] = None
+    file_image: Optional[bytes] = None
    msg_id: Optional[str] = None

然后重写 handler，在 file_image 在参数中时使用表单发送

from io import BytesIO
from functools import partial

from pydantic import parse_obj_as
from nonebot.drivers import Request
from nonebot.adapters.qqguild.bot import Bot
from nonebot.adapters.qqguild.adapter import Adapter
from nonebot.adapters.qqguild.api.request import _request
from nonebot.adapters.qqguild.api.handle import API_HANDLERS
from nonebot.adapters.qqguild.api.model import Message as APIMessage


async def post_messages(
    adapter: Adapter, bot: Bot, channel_id: int, **data
) -> APIMessage:
    model_data = MessageSend(**data).dict(exclude_none=True)

    if "file_image" in model_data.keys():
        Request_ = partial(
            Request,
            files={
                "file_image": BytesIO(model_data["file_image"]),
                "msg_id": model_data["msg_id"],
            },
            headers={
                "Authorization": adapter.get_authorization(bot.bot_info),
            },
        )
    else:
        Request_ = partial(
            Request,
            json=model_data,
            headers={"Authorization": adapter.get_authorization(bot.bot_info)},
        )
    request = Request_(
        "POST",
        adapter.get_api_base() / f"channels/{channel_id}/messages",
    )
    return parse_obj_as(
        APIMessage, await _request(adapter, bot, request)  # type: ignore
    )

Request 参数 files 即为 form-data 形式发送

由于 httpx 的 form-data 不支持 str，所以使用此代码请将驱动器设置为 aiohttp（等有时间我找下 httpx 发 str 的方法）

这里使用了 functools.partial() 用于复用代码，你也可以按照自己想要的方式去修改

最后将 API_HANDLERS 的 post_messages handler 指向自己重写的 handler 既可

from nonebot.adapters.qqguild.api.handle import API_HANDLERS


API_HANDLERS["post_messages"] = post_messages

最终成果是这样的

点击展开

from io import BytesIO
from typing import Optional
from functools import partial

from nonebot.drivers import Request
from nonebot.adapters.qqguild.bot import Bot
from pydantic import BaseModel, parse_obj_as
from nonebot.adapters.qqguild.adapter import Adapter
from nonebot.adapters.qqguild.api.request import _request
from nonebot.adapters.qqguild.api.handle import API_HANDLERS
from nonebot.adapters.qqguild.api.model import Message as APIMessage
from nonebot.adapters.qqguild.api.model import (
    MessageArk,
    MessageEmbed,
    MessageReference,
)


class MessageSend(BaseModel):
    content: Optional[str] = None
    embed: Optional[MessageEmbed] = None
    ark: Optional[MessageArk] = None
    message_reference: Optional[MessageReference] = None
    image: Optional[str] = None
    file_image: Optional[bytes] = None
    msg_id: Optional[str] = None


async def post_messages(
    adapter: Adapter, bot: Bot, channel_id: int, **data
) -> APIMessage:
    model_data = MessageSend(**data).dict(exclude_none=True)

    if "file_image" in model_data.keys():
        Request_ = partial(
            Request,
            files={
                "file_image": BytesIO(model_data["file_image"]),
                "msg_id": model_data["msg_id"],
            },
            headers={
                "Authorization": adapter.get_authorization(bot.bot_info),
            },
        )
    else:
        Request_ = partial(
            Request,
            json=model_data,
            headers={"Authorization": adapter.get_authorization(bot.bot_info)},
        )
    request = Request_(
        "POST",
        adapter.get_api_base() / f"channels/{channel_id}/messages",
    )
    return parse_obj_as(APIMessage, await _request(adapter, bot, request))


API_HANDLERS["post_messages"] = post_messages

---

接下来你就可以使用 bot.send() 发送本地图片了，例如下面的流汗插件

我的 local_image() 在 .api module 里

from pathlib import Path

from nonebot import on_command

from .api import local_image

matcher = on_command("流汗")


@matcher.handle()
async def _():
    msg = local_image(Path(__file__).parent / "ddl.png")
    await matcher.finish(msg)

效果如下

Footnotes

1. 我只测试了 message_reference 不可用，频道方会返回 500 invalid form body，如果可以发记得发在下面提醒我改 ↩

[MingxuanGame]

碎碎念

写这个是在发完上一篇教程后想起频道适配器发图问题，在与 @MiniGrayGay @KimigaiiWuyi 聊过之后才知道可以发本地图片了

在之前我以为本地发图是 base64 形式，因为之前的消息也是说以 base64 发图

后来变成了 form-data + file_image

倒腾了两天终于把这个教程写出来了（主要是把时间浪费在发 form-data 上了，一直 400 Bad Request）

总之现在是可以用了，不过只能 aiohttp，msg_id .encode() 后 httpx 和 aiohttp 发送都报错，暂时没搞 httpx 支持，等之后有时间再看看

---

NoneBot 发送消息（bot.send()）的流程 有点不太想讲这里，因为这里涉及到适配器开发知识，想在以后再出适配器开发教程，但是思索了后还是写了出来。因为不讲这里，在下面的 剖析相关方法 和 开始魔改 可能会无法理解这些方法的作用

[MingxuanGame]

补个圣经

[mnixry]

@MingxuanGame 你有兴趣来维护guild-patch吗, 我这边没精力维护了, 现在gocq都懒得登