分享 390 0

    使用Docker快速部署一个QQ机器人

    AI摘要:本文介绍了如何使用Docker快速部署一个QQ机器人,并通过对接Nonebot实现Memos机器人的功能。文章详细介绍了在Windows和Linux系统下的部署步骤,并提供了相应的代码示例。通过绑定Memos账号和转发消息发送到Memos,实现了将QQ机器人的消息转发到Memos的功能。最后,文章提供了一个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内容如下

    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框架

    启动

    docker-compose up -d

    访问 http://ip:6099/webui/login.html

    注意 : 登录所使用的 token 在docker-compose.yaml 所在目录下的
    /napcat/app/napcat/config中的webui.json
    QQ20240912-083331.png

    扫码登录

    在设置页面中添加反向 WS 地址,地址为 ws://127.0.0.1:8080/onebot/v11/ws, 这里的 8080 是 NoneBot 输出的端口号,/onebot/v11/ws 是 NoneBot onebot 适配器默认的路径
    2024-09-12T00:36:18.png

    部署nonebot

    要求环境​Python 版本 >= 3.9

    按照文档操作 https://llonebot.github.io/zh-CN/guide/nonebot2

    Memos转发机器人的实现

    在nonebot 项目中

    新建 bot.py 内容为

    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 内容为

    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 中加入

    plugins = []
    plugin_dirs = ["memos/plugins"]

    运行机器人

    nb run

    使用机器人

    在聊天界面 使用命令

    /start token

    token 为 Memos 后台获取的对应的token
    绑定成功后如下图效果
    2024-09-12T00:44:49.png

    然后直接发送消息

    此时发送消息即可转发到 memos. 且默认为公开的 memos.

    如需默认为其他状态 需修改 qq_to_memos.py 中 "visibility" 的 值 .

    结尾

    如果你想尝试一下此功能可以添加 QQ 机器人

    153985848
    需添加为好友且 在 https://memos.ee 注册获取 token 便可以使用.

    老孙
    资深网民孙先生