基于 OneBot API 的简易异步 QQ 群批量发消息脚本

HowieHz

2024-12-23

🌍简体中文编程语言-Python

本文介绍了一个基于 OneBot API 的简易异步 QQ 群批量发消息脚本。

亮点

高效异步执行
自带进度条
支持随机选择信息
易于使用
轻量 - 仅引用两个第三方库（一个用于异步发送请求，另一个用于显示进度条）

注意：为适配自动化框架，程序将在运行结束 30 秒后自动退出。

如何使用

部署 OneBot API 软件

使用这个脚本前要部署完成实现了 OneBot API 的软件

一种 onebot api 的实现软件：NapCat | NapCatQQ | 开始安装

运行脚本前请记得创建\开启 http 服务器，开启 0.0.0.0:3000 作为服务器地址（http://localhost:3000 地址是默认的地址，如果你开启的是 0.0.0.0:3000，那么下面示例配置文件的 onebot_api_http_server 这一项无需修改）

填写配置文件

需要在脚本同目录创建 config.json 作为配置文件

onebot_api_http_server：指的是 http 服务器地址
group_id_list：指的是你要发送信息的群
messages：指的是从其中随机选一个消息发送

（注意："messages": [ "" ]，将不会发出任何消息。如果设置为 "messages": [ "打卡", "" ] 那将会有 50% 的概率发送打卡消息）

（编辑配置文件的时候请注意在 JSON 中，尾随逗号（即在最后一个元素后添加的逗号）是不被允许的。建议在编辑完之后使用在线 JSON 检查工具检查正确性）

以下是一个示例配置文件，请将 onebot_api_http_server 和 group_id_list 改为实际的地址和实际的目标群号。

{
    "onebot_api_http_server": "http://localhost:3000",
    "group_id_list": [
        "123123125",
        "123123124",
        "123123123"
    ],
    "messages": [
        "冒泡睡觉🌙",
        "打卡睡觉🌙",
        "群友晚安🌙",
        "碎觉时间到🌙"
    ]
}

运行程序

从二进制文件运行（一键启动）

下载地址：软件发布页

点击 qqbot-auto-send-message-to-group.exe 下载即可

从源代码运行

直接从代码运行需要 Python 环境，之后安装以下第三方库。

pip install aiohttp tqdm

从镜像源下载

pip install aiohttp tqdm -i https://pypi.tuna.tsinghua.edu.cn/simple

最新版本代码发布地址：HowieHz/qqbot-auto-send-message-to-group
（目前无二次更新，以下代码本体即为最新代码）

代码本体

import asyncio
import json
import os
import random
import time

import aiohttp
from tqdm.asyncio import tqdm_asyncio


# 从配置文件读取
def load_config(config_path: str) -> dict[str, any]:
    if not os.path.exists(config_path):
        print(f"配置文件未找到：{config_path}")
        exit(1)
    with open(config_path, "r", encoding="utf-8") as file:
        return json.load(file)


config: dict[str, any] = load_config("config.json")
onebot_api: str = config.get("onebot_api_http_server", "http://localhost:3000")
group_id_list: list[int] = config.get("group_id_list", [])
# 默认消息为打卡
messages: list[str] = config.get("messages", ["打卡"])


def get_mag() -> str:
    """随机获取一个睡觉打卡消息

    Returns:
        str: 打卡消息
    """
    return random.choice(messages)


async def send_group_msg_rate_limited(group_id: int, msg: str) -> tuple[str, int]:
    """发送消息到指定群组，限制发送频率

    Args:
        group_id (int): 群号
        msg (str): 消息内容

    Returns:
        _type_: _description_
    """
    async with aiohttp.ClientSession() as session:
        async with session.post(
            f"{onebot_api}/send_group_msg_rate_limited",
            json={
                "group_id": group_id,
                "message": msg,
            },
        ) as response:
            await response.text()
            return msg, group_id


async def main():
    tasks = [
        send_group_msg_rate_limited(group_id, get_mag()) for group_id in group_id_list
    ]
    results = []
    for core in tqdm_asyncio.as_completed(tasks, desc="发送进度", total=len(tasks)):
        msg, group_id = await core
        results.append((msg, group_id))

    for msg, group_id in results:
        print(f"发送成功：{group_id} -> {msg}")
    print(f"已发送 {len(results)} 条消息")


if __name__ == "__main__":
    asyncio.run(main())
    print("30 秒后自动结束程序")
    time.sleep(30)

代码可能的改进点

代码中抛弃了 await response.text() 的返回值，因为在 NapCat 实现中，发现返回值为 {"status":"failed","retcode":200,"data":null,"message":"Timeout: NTEvent serviceAndMethod:NodeIKernelMsgService/sendMsg ListenerName:NodeIKernelMsgListener/onMsgInfoListUpdate EventRet:\n{\n \"result\": 0,\n \"errMsg\": \"\"\n}\n","wording":"Timeout: NTEvent serviceAndMethod:NodeIKernelMsgService/sendMsg ListenerName:NodeIKernelMsgListener/onMsgInfoListUpdate EventRet:\n{\n \"result\": 0,\n \"errMsg\": \"\"\n}\n","echo":null} 而不是类似 {"status":"ok","retcode":0,"data":{"message_id":409173648},"message":"","wording":"","echo":null} 但是消息依然是发送成功的，所以我选择忽略此返回值。
无法设置每日自动某时刻发送消息。此处我选择通过外部软件进行定时任务设置，如 Windows 的“任务计划程序”。