NOISE | 知识效率集 NOISE | 知识效率集 /BOT开发BOT开发/
📖aiocqh
Search
📖

aiocqh

Tags
aiocqh
URL
官方地址：aiocqhttp (nonebot.dev) nonebot/aiocqhttp: A Python SDK with async I/O for CQHTTP (OneBot). (github.com)

介绍

aiocqhttp 是 OneBot (原 酷Q 的 CQHTTP 插件 ) 的 Python SDK，采用异步 I/O，封装了 web 服务器相关的代码，支持 OneBot 的 HTTP 和反向 WebSocket 两种通信方式，让使用 Python 的开发者能方便地开发插件。
本 SDK 要求使用 Python 3.7 或更高版本，以及建议搭配支持 OneBot v11 的 OneBot 实现。

特点

  • 基于 asyncio，异步 I/O 使程序运行效率更高
  • 支持反向 WebSocket 通信方式，允许同时作为多个机器人账号的后端
  • 接口类似 Flask，直观易懂，上手成本低
  • 封装了消息类，便于进行消息处理和构造

开始使用

安装

首先安装 aiocqhttp 包：
pip install aiocqhttp
note:
可能需要将 pip 换成 pip3。
默认情况下将安装必要依赖 Quart 和 aiocqhttp。若要安装可选依赖，可使用：
pip install aiocqhttp[all]
这将会额外安装 ujson。

最小实例

新建 Python 文件（这里假设名为 bot.py），内容如下：
# bot.py

from aiocqhttp import CQHttp, Event

bot = CQHttp()


@bot.on_message('private')
async def _(event: Event):
    await bot.send(event, '你发了：')
    return {'reply': event.message}


bot.run(host='127.0.0.1', port=8080)
note:
如果需要在不同主机上运行此 bot 后端和 酷Q（CQHTTP），则此处 host 应该为 bot 后端所在主机的 IP 或 0.0.0.0。
如果你在使用 VPS，应确保在安全组中开放 8080 端口。
运行该文件，看到如下输出即为启动成功：
Running on <http://127.0.0.1:8080> (CTRL + C to quit)
[2020-01-29 19:27:57,133] Running on 127.0.0.1:8080 over http (CTRL + C to quit)
note:
如果没有启动成功，建议检查 Python 版本是否过旧、端口是否被占用等。

配置 CQHTTP

使用反向 WebSocket

在 CQHTTP 配置文件中，填写 ws_reverse_url 值为 ws://127.0.0.1:8080/ws/，这里 127.0.0.1:8080 应根据情况改为 bot.py 中传给 bot.run 的 host 和 port 参数。
然后，如果有的话，删掉 ws_reverse_event_url 和 ws_reverse_api_url 这两个配置项。
接着设置 use_ws_reverse 为 true。
最后重启 CQHTTP。

使用 HTTP

修改 bot.py 中创建 bot 对象部分的代码为：
bot = CQHttp(api_root='<http://127.0.0.1:5700>')
这里 127.0.0.1:5700 应根据情况改为 CQHTTP 所监听的 IP 和端口（由 CQHTTP 配置中的 host 和 port 指定）。
然后在 CQHTTP 配置文件中，填写 post_url 为 http://127.0.0.1:8080/，这里 127.0.0.1:8080 应根据情况改为 bot.py 中传给 bot.run 的 host 和 port 参数。
接着设置 use_http 为 true（默认就是 true）。
最后重启 CQHTTP。

测试对话

给你的机器人发一段私聊消息，如果一切正常，ta 应该会回复你。
note:
如果没有回复，请检查 bot.py 运行是否报错、酷Q 日志是否报错。如果都没有报错，则可能是机器人账号被腾讯风控，需要在同一环境中多登录一段时间。

发生了什么

本节围绕 开始使用 中的最小实例，来解释它如何工作。
先贴出代码：
# bot.py

from aiocqhttp import CQHttp, Event

bot = CQHttp()  # M1


@bot.on_message('private')  # M2
async def _(event: Event):  # M3
    await bot.send(event, '你发了：')  # M4
    return {'reply': event.message}  # M5


bot.run(host='127.0.0.1', port=8080)  # M6

CQHttp 类

# bot.py
bot = CQHttp()  # M1
M1 处首先创建了 aiocqhttp.CQHttp 类的对象 bot。
该类是本 SDK 的主体，内部封装了 Quart 对象作为 web 服务器。其中添加了 /、/ws/ 等路由，从而使 CQHTTP 能够通过 HTTP 或 WebSocket 协议连接 bot。
使用 HTTP 通信时，需要传入 api_root 参数是因为 bot.send 需要主动调用 CQHTTP API，它需要知道 CQHTTP「在哪」。
使用反向 WebSocket 通信时，可以有多个 CQHTTP 同时连接到一个 bot，bot.send 会自动选择对应的账号发送。

事件处理

# bot.py
@bot.on_message('private')  # M2
M2 处通过 @bot.on_message('private') 装饰器注册了 私聊消息事件 的处理函数。
除了 bot.on_message，还有类似的 bot.on_notice、bot.on_request、bot.on_meta_event，它们分别用于注册消息、通知、请求、元事件这四种事件类型（对应 CQHTTP 事件 的 post_type 字段）的处理函数。
这些装饰器可以带参数，也可以不带参数，参数可以有多个，对应 CQHTTP 事件的 ?_type 字段，这里 ?_type 根据事件类型的不同，分别为 message_type、notice_type、request_type、meta_event_type。
async def _(event: Event):  # M3
M3 处定义了事件处理函数，它必须接受一个 Event 对象作为唯一的参数。Event 对象是对 CQHTTP 事件数据的简单封装，提供了属性以方便获取其中的字段，例如 event.message、event.user_id 等。

API 调用

    await bot.send(event, '你发了：')  # M4
M4 处调用了 bot.send 方法，该方法是对 CQHTTP API 中 send_msg 的简单封装，它会向 event 对应的主体发送消息（由第二个参数指定），本例中这个主体是「发私聊消息来的人」。
除此之外，你可以在 bot 对象上直接调用任何 CQHTTP API，见 API 列表 ，所需参数通过命名参数传递，例如：
friends = await bot.get_friend_list()

await bot.set_group_ban(group_id=10010, user_id=10001000)

credentials = await bot.get_credentials(domain='qun.qq.com')
note:
如果有多个 CQHTTP 连接，可能需要在调用 API 时增加 self_id 参数以指定要调用的机器人账号，例如：
await bot.get_friend_list(self_id=event.self_id)
调用 API 时需向 CQHTTP 发出请求，这一步可能出错：
  • 如果 API 当前不可用（例如没有任何连接了的 CQHTTP、或未配置 api_root），抛出 aiocqhttp.ApiNotAvailable
  • 如果 API 可用，但网络无法连接或连接出现错误，抛出 aiocqhttp.NetworkError
一旦请求成功，SDK 会判断 HTTP 响应状态码是否为 2xx：
  • 如果不是，抛出 aiocqhttp.HttpFailed，在这个异常中可通过 status_code 获取 HTTP 响应状态码
  • 如果是，则进一步查看响应 JSON 的 status 字段，如果 status 字段为 failed，抛出 aiocqhttp.ActionFailed，在这个异常中可通过 retcode 获取 API 调用的返回码
以上异常全都继承自 aiocqhttp.Error。
如果 status 为 ok 或 async，则不抛出异常，函数返回 CQHTTP API 响应数据的 data 字段（有可能为 None）。
HTTP 响应状态码和 retcode 的具体含义，见 响应说明 。

快速操作

    return {'reply': event.message}  # M5
M5 处事件处理函数返回了一个字典，这会被 SDK 序列化为 JSON 并返回给 CQHTTP，作为 CQHTTP 事件上报的响应 （通过 HTTP 响应正文或 WebSocket 传送）。这称为「快速操作」，可用于对事件进行一些简单的操作，本例中对事件进行了「回复」操作，对于群聊等事件，快速操作还包括「禁言」「撤回」等，具体请见 事件列表 的「响应数据」。
快速操作不是必须的，事件处理函数可以不返回任何值。

运行

bot.run(host='127.0.0.1', port=8080)  # M6
M6 处调用 bot.run 运行了 bot 后端，该方法是 Quart 对象的 run 方法的简单封装，可直接传入更多参数，参数会直接进入 Quart.run 。

更多

更丰富的例子见 demo.py 。
本节所提到的类、方法、函数等，均可在右上角「模块 API」中找到更详细的说明。

常见主题

消息处理

为了方便地处理消息，本 SDK 提供了 Message 和 MessageSegment 类，用于解析和构造消息中的 CQ 码，例如：
from aiocqhttp import Message, MessageSegment

@bot.on_message
async def handle_msg(event):
    msg = Message(event.message)
    for seg in msg:
        if seg == MessageSegment.at(event.self_id):
            await bot.send(event, 'at 我干啥')
            break

    img = MessageSegment.image('<http://example.com/somepic.png>')
    await bot.send(event, img + '\\n上面这是一张图')
如果觉得手动从 event.message 构造 Message 对象不够方便，可以在 bot 对象初始化时传入 message_class 参数，例如：
from aiocqhttp import CQHttp, Message
bot = CQHttp(message_class=Message)
这会使 SDK 在收到消息事件后，使用形如 event.message = Message(event.message) 的方式构造 Message 对象。
当然，如果内置的 Message 类不符合你的需求，你也可以自己编写消息类，同样可以传入 message_class。

默认 bot 对象

如果你只是开发一些简单的功能，或临时做测试等，可以不用自己创建 bot 对象，而直接使用 SDK 内置的默认 bot 对象，例如：
from aiocqhttp.default import on_message, send, api, run

@on_message
async def handle_msg(event):
    await send(event, event.message)
    await api.send_private_msg(user_id=event.user_id, message='。。。')

run(host='127.0.0.1', port=8080)
如需修改 CQHttp 初始化的参数，可使用 reconfigure_default_bot 函数，例如：
from aiocqhttp.default import reconfigure_default_bot
reconfigure_default_bot(api_root='<http://127.0.0.1:8080>')

同步和异步

通常情况下，建议在 bot 中全部使用异步操作，例如使用 aiofiles 读写文件、使用 aiohttp 、httpx.AsyncClient 等进行网络请求、使用 aiomysql 、asyncpg 、aioredis 、Motor 等访问数据库。
但如果出于某些原因，你更偏好或不得不使用同步函数，SDK 也提供了原生支持，例如：
@bot.on_message
def sync_handle_msg(event):
    time.sleep(5)  # 模拟耗时 I/O
    # 使用 bot.sync 进行 API 调用
    bot.sync.send_private_msg(user_id=event.user_id, message='处理完毕')
sync_handle_msg 会在 asyncio loop 的默认 executor（多线程，需注意线程安全）里运行，可通过 loop.set_default_executor 修改。

日志

本 SDK 直接使用了内部 Quart 对象的日志器，是一个 logging.Logger 对象，可通过 bot.logger 获得，例如：
bot.logger.info('初始化成功')
如果你需要对 logger 进行配置，直接修改它即可（但不可给它赋值），但更建议的方式是在你的项目中自己创建新的 logger。

部署

默认情况下，bot.run 以 debug 模式运行，可通过 bot.run(..., debug=False) 关闭。
在实际部署环境中，不建议使用 bot.run，而应该使用专业的 ASGI 服务器，例如 Uvicorn 、Daphne 、Hypercorn 等。这里给出使用 Uvicorn 部署的例子：
# main.py
from aiocqhttp import CQHttp
bot = CQHttp()

uvicorn --host 127.0.0.1 --port 8080 main:bot.asgi

添加路由

SDK 注册了 /、/ws、/ws/api、/ws/event 这几个路由，以便向 CQHTTP 提供服务。有时你可能想要注册其它自定义的路由，例如接收 webhook 推送、展示机器人状态、提供管理控制台等，可以直接操作 Quart 对象来做到，例如：
@bot.server_app.route('/webhook')
async def webhook():
    pass
使用 Quart 时，有时需要用到当前模块的相对路径，例如加载当前模块下的 templates 中的模板文件等，如果使用这些功能时遇到问题，可以尝试给 CQHttp 类传入 import_name 参数，例如：
bot = CQHttp(__name__)

@bot.server_app.route('/admin')
async def admin():
    return await render_template('admin.html')

在已有事件循环中运行

通过 CQHttp.run_task 方法可以将 bot 运行在已有的事件循环中，参数同 CQHttp.run，例如：
bot = CQHttp()

loop = asyncio.new_event_loop()
loop.create_task(bot.run_task(host='127.0.0.1', port=8080))
loop.run_forever()