使用Docker快速部署一个QQ机器人
简介
介绍如何使用Docker快速部署一个QQ机器人并对接Nonebot实现Memos机器人的功能:
- 绑定memos账号
- 转发消息发送到memos
步骤
部署QQ机器人
这里使用的项目是基于QQNT的无头机器人方案,使用webui登录,相对于之前我部署的Go-cqhttp的方案的好处是不会被风控掉线. 稳定性很nice
使用的项目地址 使用的项目文档: https://llonebot.github.io/zh-CN/guide/getting-started
Windows系统
在windows下非常简单,下载QQNT版本的QQ,登录你的QQ机器人账号
https://github.com/super1207/install_llob/releases
下载 exe,双击运行即可,之后打开 QQ 的设置,看到了 LLOneBot 就代表安装成功了。
Linux系统
在linux下 我选择使用 的项目 NapCatQQ
地址 : https://github.com/NapNeko/NapCatQQ
使用Docker部署 docker-compose.yaml
内容如下
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
services:
napcat:
environment:
- ACCOUNT=153985848 #QQ机器人号码
- WS_ENABLE=true
- NAPCAT_UID=0
- NAPCAT_GID=0
ports:
- 3001:3001 #上传端口
- 6099:6099 #webui端口
- 3000:3000 #http端口
restart: always
image: mlikiowa/napcat-docker:latest
volumes:
- "./napcat/app/.config/QQ:/app/.config/QQ"
- "./napcat/app/napcat/config:/app/napcat/config"
network_mode: host #使用host的原因是为了方便对接宿主机的nonebot框架
启动
1
docker-compose up -d
访问 http://ip:6099/webui/login.html
注意 : 登录所使用的 token
在docker-compose.yaml 所在目录下的 /napcat/app/napcat/config
中的webui.json
中
扫码登录
在设置页面中添加反向 WS 地址,地址为 ws://127.0.0.1:8080/onebot/v11/ws, 这里的 8080 是 NoneBot 输出的端口号,/onebot/v11/ws 是 NoneBot onebot 适配器默认的路径
部署nonebot
要求环境Python 版本 >= 3.9
按照文档操作 https://llonebot.github.io/zh-CN/guide/nonebot2
Memos转发机器人的实现
在nonebot 项目中
新建 bot.py 内容为
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
import nonebot
from nonebot.adapters.onebot.v11 import Adapter as ONEBOT_V11Adapter
# 初始化 NoneBot
nonebot.init()
# 注册适配器
driver = nonebot.get_driver()
driver.register_adapter(ONEBOT_V11Adapter)
# 在这里加载插件
nonebot.load_builtin_plugins("echo") # 加载内置插件
nonebot.load_from_toml("pyproject.toml") # 从 toml 文件加载插件
# 如果有额外的插件目录,可以这样加载
# nonebot.load_plugins("src/plugins")
if __name__ == "__main__":
nonebot.run()
新建 memos/plugins 文件夹 , 在其下创建 qq_to_memos.py 内容为
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
from nonebot import on_command, on_message, get_driver
from nonebot.rule import to_me
from nonebot.adapters.onebot.v11 import Bot, Event, Message
from nonebot.params import CommandArg
import json
import os
import httpx
from typing import Dict, Any
import logging
# 配置日志
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
filename='memos_bot.log',
filemode='a'
)
logger = logging.getLogger(__name__)
# 文件路径
JSON_FILE = "users_data.json"
# 读取 JSON 数据
def read_json() -> Dict[str, Any]:
if os.path.exists(JSON_FILE):
with open(JSON_FILE, 'r', encoding='utf-8') as f:
return json.load(f)
return {}
# 写入 JSON 数据
def write_json(data: Dict[str, Any]):
with open(JSON_FILE, 'w', encoding='utf-8') as f:
json.dump(data, f, ensure_ascii=False, indent=4)
# 初始化函数
async def init():
if not os.path.exists(JSON_FILE):
write_json({})
logger.info(f"Created new JSON file: {JSON_FILE}")
# 注册命令
start = on_command("start", rule=to_me(), priority=5)
@start.handle()
async def handle_start(bot: Bot, event: Event, args: Message = CommandArg()):
user_id = event.get_user_id()
token = args.extract_plain_text().strip()
if not token:
await start.finish(" 请提供 Token,格式:/start <token>")
logger.warning(f"User {user_id} failed to start due to missing token")
return
users_data = read_json()
users_data[user_id] = {"token": token}
write_json(users_data)
logger.info(f"User {user_id} started successfully")
await start.finish(" 绑定成功!现在您可以直接发送消息,我会将其保存到 Memos。")
# 处理所有消息
memo = on_message(priority=5)
@memo.handle()
async def handle_memo(bot: Bot, event: Event):
user_id = event.get_user_id()
message = event.get_message()
users_data = read_json()
user_info = users_data.get(user_id)
if not user_info:
await memo.finish(" 您还未绑定,请先使用 /start <token> 命令绑定。")
logger.warning(f"Unstarted user {user_id} attempted to send a memo")
return
token = user_info["token"]
text_content = message.extract_plain_text()
# 如果消息为空,不处理
if not text_content.strip():
return
# 发送到 Memos
async with httpx.AsyncClient() as client:
try:
payload = {
"content": text_content.strip(),
"visibility": "PUBLIC"
}
response = await client.post(
"https://memos.ee/api/v1/memos",
json=payload,
headers={"Authorization": f"Bearer {token}"}
)
response.raise_for_status()
logger.info(f"Memo sent successfully for user {user_id}")
except httpx.HTTPStatusError as e:
logger.error(f"HTTP error occurred for user {user_id}: {e}")
logger.error(f"Response content: {e.response.text}")
await memo.finish(f" 发送失败,错误代码:{e.response.status_code},请检查您的 Token 和网络连接。")
return
except Exception as e:
logger.error(f"Unexpected error occurred for user {user_id}: {e}")
await memo.finish(" 发送过程中发生意外错误,请稍后重试。")
return
await memo.finish(" 已成功发送到 Memos!")
# 获取驱动器并注册启动事件
driver = get_driver()
driver.on_startup(init)
logger.info("Memos bot plugin initialized")
API 端点按自己需求更改即可.
在 pyproject.toml 中加入
1
2
plugins = []
plugin_dirs = ["memos/plugins"]
运行机器人
1
nb run
使用机器人
在聊天界面 使用命令
1
/start token
token 为 Memos 后台获取的对应的token 绑定成功后如下图效果
然后直接发送消息
此时发送消息即可转发到 memos. 且默认为公开的 memos.
如需默认为其他状态 需修改 qq_to_memos.py 中 “visibility” 的 值 .
结尾
如果你想尝试一下此功能可以添加 QQ 机器人
153985848 需添加为好友且 在 https://memos.ee 注册获取 token 便可以使用.