简介:本文详细介绍Python接入QQ机器人的完整流程,涵盖协议选择、库安装、基础功能实现及高级应用场景,提供可复用的代码示例和调试建议。
接入QQ机器人需要理解其通信协议和接口限制。目前主流方案分为两类:官方API(如企业QQ、腾讯云QBot)和非官方协议(如基于WebQQ或SmartQQ的逆向协议)。由于腾讯官方未提供个人开发者使用的开放API,多数个人项目选择非官方协议实现。
推荐方案:对于Python开发者,优先选择OneBot协议,通过go-cqhttp作为后端,Python作为业务逻辑层,实现解耦和可扩展性。
aiohttp(异步HTTP请求)websockets(实时通信)loguru(简化日志管理)
pip install aiohttp websockets loguru# 若使用NoneBot2框架pip install nonebot2
config.yml:
account:uin: 你的QQ号password: "密码或加密字符串"servers:- ws:host: 127.0.0.1port: 6700
./go-cqhttp
import asyncioimport websocketsimport jsonasync def connect_qq_bot():uri = "ws://127.0.0.1:6700"async with websockets.connect(uri) as websocket:# 发送心跳包await websocket.send(json.dumps({"action": "send_private_msg","params": {"user_id": 123456, "message": "Hello"}}))while True:response = await websocket.recv()data = json.loads(response)if data["post_type"] == "message":handle_message(data)async def handle_message(data):user_id = data["user_id"]message = data["message"]# 回复逻辑if "你好" in message:await send_message(user_id, "你好,我是机器人!")async def send_message(user_id, content):# 实现消息发送(需通过WebSocket或HTTP API)passasyncio.get_event_loop().run_until_complete(connect_qq_bot())
NoneBot2是Python生态中成熟的QQ机器人框架,支持插件化和异步编程。
from nonebot import on_commandfrom nonebot.adapters.onebot.v11 import Bot, Event, Message# 定义命令触发器hello = on_command("hello")@hello.handle()async def handle_hello(bot: Bot, event: Event):await hello.send(Message("Hello, 世界!"))
pyproject.toml
[tool.nonebot]plugins = []plugin_dirs = ["src/plugins"][tool.nonebot.adapters.onebot.v11]self_id = 你的机器人QQ号
通过正则表达式或NLP库(如jieba)解析用户意图:
import refrom nonebot.adapters.onebot.v11 import Message@on_messageasync def handle_all_messages(bot: Bot, event: Event):msg = str(event.message)if re.search(r"^天气(.*)", msg):location = re.search(r"天气(.*)", msg).group(1)await bot.send(event, f"查询{location}的天气...")
使用nonebot-plugin-apscheduler实现定时任务:
from apscheduler.schedulers.asyncio import AsyncIOSchedulerfrom nonebot_plugin_apscheduler import scheduler@scheduler.scheduled_job("cron", hour="9", minute="30")async def morning_greeting():# 获取群列表并发送消息pass
使用SQLAlchemy或Peewee存储用户数据:
from peewee import *db = SqliteDatabase("qq_bot.db")class User(Model):qq_id = BigIntegerField(primary_key=True)name = CharField()class Meta:database = db# 初始化数据库db.create_tables([User])
go-cqhttp日志是否显示Listening on WS://0.0.0.0:6700。curl测试WebSocket连接:
curl --include --no-buffer --header "Connection: Upgrade" \--header "Upgrade: websocket" --header "Host: 127.0.0.1:6700" \http://127.0.0.1:6700
self_id是否与登录QQ号一致。通过Python接入QQ机器人,开发者可以快速构建自动化工具。建议从以下方向深入:
完整代码示例:参考NoneBot2官方文档和go-cqhttp GitHub仓库。