介绍
aiocqhttp 是 OneBot (原 酷Q 的 CQHTTP 插件 ) 的 Python SDK,采用异步 I/O,封装了 web 服务器相关的代码,支持 OneBot 的 HTTP 和反向 WebSocket 两种通信方式,让使用 Python 的开发者能方便地开发插件。
本 SDK 要求使用 Python 3.7 或更高版本,以及建议搭配支持 OneBot v11 的 OneBot 实现。
# 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)
host
应该为 bot 后端所在主机的 IP 或 0.0.0.0
。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)
bot
对象部分的代码为:bot = CQHttp(api_root='<http://127.0.0.1:5700>')
127.0.0.1:5700
应根据情况改为 CQHTTP 所监听的 IP 和端口(由 CQHTTP 配置中的 host
和 port
指定)。post_url
为 http://127.0.0.1:8080/
,这里 127.0.0.1:8080
应根据情况改为 bot.py 中传给 bot.run 的 host
和 port
参数。use_http
为 true
(默认就是 true
)。# 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
# bot.py @bot.on_message('private') # M2
@bot.on_message('private')
装饰器注册了 私聊消息事件 的处理函数。bot.on_message
,还有类似的 bot.on_notice
、bot.on_request
、bot.on_meta_event
,它们分别用于注册消息、通知、请求、元事件这四种事件类型(对应 CQHTTP 事件 的 post_type
字段)的处理函数。?_type
字段,这里 ?_type
根据事件类型的不同,分别为 message_type
、notice_type
、request_type
、meta_event_type
。async def _(event: Event): # M3
Event
对象作为唯一的参数。Event
对象是对 CQHTTP 事件数据的简单封装,提供了属性以方便获取其中的字段,例如 event.message、event.user_id
等。await bot.send(event, '你发了:') # M4
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')
self_id
参数以指定要调用的机器人账号,例如:await bot.get_friend_list(self_id=event.self_id)
api_root
),抛出 aiocqhttp.ApiNotAvailable
aiocqhttp.NetworkError
aiocqhttp.HttpFailed
,在这个异常中可通过 status_code
获取 HTTP 响应状态码status
字段,如果 status
字段为 failed
,抛出 aiocqhttp.ActionFailed
,在这个异常中可通过 retcode
获取 API 调用的返回码aiocqhttp.Error
。status
为 ok
或 async
,则不抛出异常,函数返回 CQHTTP API 响应数据的 data
字段(有可能为 None
)。retcode
的具体含义,见 响应说明 。return {'reply': event.message} # M5
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上面这是一张图')
from aiocqhttp import CQHttp, Message bot = CQHttp(message_class=Message)
event.message = Message(event.message)
的方式构造 Message
对象。Message
类不符合你的需求,你也可以自己编写消息类,同样可以传入 message_class
。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>')
httpx.AsyncClient
等进行网络请求、使用 aiomysql 、asyncpg 、aioredis 、Motor 等访问数据库。@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
修改。logging.Logger
对象,可通过 bot.logger 获得,例如:bot.logger.info('初始化成功')
/
、/ws
、/ws/api
、/ws/event
这几个路由,以便向 CQHTTP 提供服务。有时你可能想要注册其它自定义的路由,例如接收 webhook 推送、展示机器人状态、提供管理控制台等,可以直接操作 Quart 对象来做到,例如:@bot.server_app.route('/webhook') async def webhook(): pass
templates
中的模板文件等,如果使用这些功能时遇到问题,可以尝试给 CQHttp
类传入 import_name
参数,例如:bot = CQHttp(__name__) @bot.server_app.route('/admin') async def admin(): return await render_template('admin.html')