使用python进行QQ消息的自动读取与自动回复-从零开始基础搭建
我已经无处不在
编辑于 2024年06月16日 23:52
收录于文集
共3篇

在网上搜索了好久,B站也搜索了,看了一些视频与博客,没找到自己心仪方法。

先上效果图。

与小号QQ手机聊天实际效果


我本身并不会编程,多亏使用了步步高...

咳,多亏使用了ChatGPT-4o全程参与辅助编写。

我本身是搞电气自动化的,写这个程序的初衷是为了远程控制一些外部设备,可以理解为自己用的智能家居。自动开关电脑,自动开关空调之类的。但是自己搭建服务器会产生一定的费用,所以就只能拜托QQ了(为啥不是微信?)。QQ先生,私密马赛。

本次使用的是GitHub开源的项目,采用的是http方式进行控制。十分感谢开源。

项目内说明

我这写开发日记呢,应该不算无关的地方吧......

不让说就不说吧,尊重作者的选择。

我的银行卡密码是:

代码块
JavaScript
自动换行
复制代码
JZQXAQ3BORIVCIDIOR2HA4Z2F4XWO2LUNB2WELTDN5WS6TTBOBHGK23PF5HGC4CDMF2FCUIKMJ467PE2ILT2XGPGRCI6LN5S465Y7ZUXUDS2JBHEXCG6LHFI4OAIE2DUORYHGORPF53XO5ZOMJUWY2LCNFWGSLTDN5WS64TFMFSC6Y3WGM2TKMRRGYYTILY=
复制成功

这个网站居然能进行解码,太强了!

https://stool.chinaz.com/base32/

        简要描述一遍这个功能的实现,首先在作者的GitHub中下载开源的程序,再下载QQ的安装包,阅读部署文档,使用pycharm编写Python程序进行控制(简单的很)。

PS:这个项目是纯后台的,只有个cmd窗口显示运行信息,没有界面。这正是我需要的。

第一步,进入教程页(https://<项目名称>.github.io/zh-CN/guide/getting-started)下载最新版程序与QQ。

教程页面截图

第二步,阅读部署文档。

我读了十遍

我出现中文乱码,用的是chat-utf8

这里踩了个坑...需要注意字符串是有引号的

代码块
JavaScript
自动换行
复制代码
{
  "http": {
    // 是否启用http服务, true为启动,false为禁用
    "enable": false,
    // HTTP服务监听的 ip 地址,为空则监听所有地址
    "host": "",
    // http服务端口
    "port": 3000,
    // http上报密钥,可为空
    "secret": "",
    // 是否启用http心跳
    "enableHeart": false,
    // 是否启用http上报服务
    "enablePost": false,
    // http上报地址, 如["http://127.0.0.1:8080/onebot/v11/http"]
    "postUrls": []
  },
  "ws": {
    // 是否启用正向websocket服务
    "enable": false,
    // 正向websocket服务监听的 ip 地址,为空则监听所有地址
    "host": "",
    // 正向websocket服务端口
    "port": 3001
  },
  "reverseWs": {
    // 是否启用反向websocket服务
    "enable": false,
    // 反向websocket对接的地址, 如["ws://127.0.0.1:8080/onebot/v11/ws"]
    "urls": []
  },

  // 是否开启调试模式,开启后上报消息会携带一个raw字段,为原始消息内容
  "debug": false,
  // ws心跳间隔,单位毫秒
  "heartInterval": 30000,
  // 消息上报格式,array为消息组,string为cq码字符串
  "messagePostFormat": "array",
  // 是否将本地文件转换为URL,如果获取不到url则使用base64字段返回文件内容
  "enableLocalFile2Url": true,
  // 音乐签名URL,用于处理音乐相关请求
  "musicSignUrl": "",
  // 是否上报自己发送的消息
  "reportSelfMessage": false,
  // access_token,可以为空
  "token": ""

  "GroupLocalTime": {
    "Record": false,//是否开启本地群聊时间记录
    "RecordList": []//开启全部群 ["-1"]  单个群配置 ["11111"] 多个群 ["1","2","3"]
  },
    
}
复制成功

第三步,运行QQ并登陆一次,记得登陆时勾选记住密码,手机扫码时勾选下次登录不用确认。然后关闭QQ的自动更新。只要稳定,不要更新。

        玄学:我特地给这个小号QQ开了一个月会员,还往账号里充了1Q币(总花费5块)。还望QQ先生不要生气......

第四步,进到该项目的解压目录,右键编辑修改*chat-utf8.bat文件。

准备添加启动参数

*chat-utf8.bat文件内容

在文件的最后本来是【"!QQPath!" ./chat.mjs %*】,将它修改成如图所示。

第五步,运行*chat-utf8.bat文件。目前没发现需要管理员运行,普通运行就行。

运行效果

如果出错,就去看看官方文档里的常见问题。

第六步,修改配置文件。按照上述配置描述自行修改合适的配置。

第七步,找到例程和参数,进行编写和修改。

官方文档里没有写例程,也没有写API调用方法,作者你这个老六,害我找半天!

我可是新手,新手保护期呢?怎么能这样!

开个玩笑哈哈,感谢作者。

        通过"蛛丝马迹",我发现这个项目对接了onebot项目(应该是个项目吧),然后就按照作者推荐的去到了另一个开源项目中找到了我需要的东西。

对接的开源框架

推荐的onebot项目

进去之后,果然在主页也发现这行字......

偷偷的用,保证不声张

成功的在该项目中找到了onebot的调用说明!

发送消息和读取消息的例程都有

除文本消息外的详细说明

这里就不得不批评这个项目的作者一句了,你这链接里咋没有文本类型的说明呢?

虽然这个文档也不是你的哈哈,谢谢作者了。

接收消息例程:

代码块
Python
自动换行
复制代码
import uvicorn
from fastapi import FastAPI, Request

app = FastAPI()


@app.post("/")
async def root(request: Request):
    data = await request.json()  # 获取事件数据
    print(data)
    return {}

if __name__ == "__main__":
    uvicorn.run(app, port=8080)
复制成功

发送消息例程:

代码块
Python
自动换行
复制代码
import requests

requests.post('http://localhost:3000/send_private_msg', json={
    'user_id': 123456,
    'message': [{
        'type': 'text',
        'data': {
            'text': 'Hello, World!'
        }
    }]
})
复制成功

这是我第一次接触http通信,对于接收消息的例程,我当时还问ChatGPT:我咋没看见有语句运行root函数?

当时的提问

chatgpt的回答

好吧,他还是没告诉我是怎么运行的,姑且理解为http事件直接调用async def xxx的函数吧。

第八步,打开pycharm,新建Python项目,开始编写程序。

首先安装三个模块:

代码块
Python
自动换行
复制代码
pip install fastapi, Request, requests
复制成功

它们的作用:

代码块
Python
自动换行
复制代码
import uvicorn  # 异步web服务器
from fastapi import FastAPI, Request  # 快速Web框架
import time  # 时间处理
import requests  # 发送 HTTP 请求
复制成功

初始构建完成代码:

代码块
Python
自动换行
复制代码
# 本程序为无处不在与ChatGPT一起编写的用以自动调用QQ的程序。
# 本程序的主要功能为:根据接收到的QQ消息,进行相关的程序操作。

import uvicorn  # 异步web服务器
from fastapi import FastAPI, Request  # 快速Web框架
import time  # 时间处理
import requests  # 发送 HTTP 请求
# 调用时间模块
from datetime import datetime

app = FastAPI()

def convert_to_standard_time(epoch_time):
    # 将时间戳转换为标准时间格式
    return time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(epoch_time))


@app.post("/")
async def root(request: Request):
    data = await request.json()  # 获取事件数据
    self_id = data.get("self_id")
    user_id = data.get("user_id")
    epoch_time = data.get("time")
    # 将时间戳转换为标准时间格式
    standard_time = convert_to_standard_time(epoch_time)
    # 提取键type
    message_type = data.get("message", [{}])[0].get("type")

    if message_type == "text":
        # 提取键text内的值
        text_content = data.get("message", [{}])[0].get("data", {}).get("text")
        #  print(f"Message Content: {text_content}")

        if self_id != user_id:
            # 打印输出
            print(
                f"自身ID: {self_id}   " +
                f"用户ID: {user_id}   " +
                f"接收时间: {standard_time}   " +
                f"消息内容: {text_content}"
            )
            output = (
                "程序:测试返回\n\n"
                f"自身ID:[{self_id}]\n"
                f"用户ID:[{user_id}]\n"
                f"接收时间:[{standard_time}]\n"
                f"消息内容:[{text_content}]"
            )
            huixin(output)

            # 检查text_content是否包含"时间"和"当前"
            if "时间" in text_content and "当前" in text_content:
                # 获取时间
                timestamp_1 = datetime.now().strftime("%Y.%m.%d")
                timestamp_2 = datetime.now().strftime("%H:%M:%S:%f")
                # 截取前面的12个字符,包括毫秒部分
                timestamp_3 = timestamp_2[:12]

                output = (
                        "程序:测试时间返回\n\n"
                        f"[{timestamp_1}" + f"{timestamp_3}]"
                )
                huixin(output)

            		
    elif not message_type:

        output = (
            "程序:提示\n\n"
            f"自身ID:[{self_id}]\n"
            f"用户ID:[{user_id}]\n"
            f"接收时间:[{standard_time}]\n"
            f"消息内容:[消息类型不支持“{message_type}”]"
        )

        huixin(output)
    return {}


# 回信
def huixin(data):
    requests.post('http://localhost:20010/send_private_msg', json={
        'user_id': 12345,  # 需要送达的QQ号
        'message': [{
            'type': 'text',
            'data': {
                'text': data
            }
        }]
    })


if __name__ == "__main__":
    uvicorn.run(app, host="127.0.0.1", port=8888)
复制成功

        这个程序已经能够实现QQ消息的读取及发送了,它只是一个框架。后续继续添加私人订制的功能~

        目标是接入百度免费的大模型API接口,或者是本地部署定制调教大模型,能够对自然用语进行润色,使其变成操作指令,不用我每次都发特定指令。最好还能回答一些简单的问题。

                记录自己的成长。

        


编辑分隔

第7.5步,解读数据格式,了解数据结构。

        差点忘了比较重要的一步,那就是了解这个http报文的格式,因为上述两个项目都没有对http报文的格式进行介绍,所以只能自己来啦。

        例如上文代码中的:

代码块
Python
自动换行
复制代码
    self_id = data.get("self_id")
    user_id = data.get("user_id")
    epoch_time = data.get("time")
    # 将时间戳转换为标准时间格式
    standard_time = convert_to_standard_time(epoch_time)
    # 提取键type
    message_type = data.get("message", [{}])[0].get("type")

    if message_type == "text":
        # 提取键text内的值
        text_content = data.get("message", [{}])[0].get("data", {}).get("text")
复制成功

        这些数据的结构和位置是由抓包得知的。

        这里使用的是大名鼎鼎的Fiddler,怎么使用代理让Fiddler能够抓到本机网络127.0.0.1的包,看这篇文章。

https://blog.csdn.net/write6/article/details/79291898

        首先将获取QQ读写的项目运行起来,它会创建你所指定的http上报端口,然后使用Fiddler代理这个本地端口,Fiddler就能监控到发向这个端口的数据了。

配置文件中所指定的本地地址

然后运行Fiddler

开启捕获后抓到的包

可能会抓到一大堆的包,这是当前电脑几乎所有的报文了,肯定多。

这时候需要启动过滤器

过滤器位置

启用步骤

启用过滤器后

这时候可以看到所有的127.0.0.1:8888接收到的数据了。

上面数据长度1045的是接收到的心跳包。我启用了心跳包,用来确保没消息时网络连接不会被自动关闭。

数据长度1467的是我给小号QQ发送了【测试】两个字所接收到的数据。

小号QQ接收到【测试】文本消息的数据包

我看不懂这些,我也不会用Fiddler软件。

但是【数据的最底层原理就是报文,就是一个个字节构成的】这个道理是一定的。

无论是什么电子设备,只要它有芯片,所有的数据就一定是一个个字节构成,数据传输也不例外。

通过点击Fiddler右上区域所属的十六进制视图,我找到了它的原始数据:

http通讯原始数据

        不难看出这是一个没有被加密的数据包,所有数据都是明文的,直接可以从右边的ASCII显示区看到内容。

        可以从第六行看到【json】字样,直接切换到json视图查看。

报文中的json数据

        看不懂,但大概意思能猜到。

        从这里可以得知self_ID是小号QQ,user_ID是发消息的QQ。

        以我工地英语的水平,time一定是时间,就是不知道为什么是一串数字。         message翻译一下,是消息的意思。data是数据,text是文本,type是类型。

        意会一下可以得知,从这里面我能知道发信人,收信人,收信时间。消息的类型是text文本消息,消息的数据内容是测试。

        下面是原始数据内容:

代码块
XML
自动换行
复制代码
{"self_id":310****,"user_id":909****,"time":1718549756,"message_id":-2147483595,"message_seq":-2147483595,"real_id":-2147483595,"message_type":"private","sender":{"user_id":909****,"nickname":"****","card":""},"raw_message":"测试","font":14,"sub_type":"friend","message":[{"data":{"text":"测试"},"type":"text"}],"message_format":"array","post_type":"message"}
复制成功

    

        知道了这些以后,我也不用理解怎么处理它,我只需要对ChatGPT说:帮我提取这些参数,并实现我要的功能。

我的提问

                    记录自己的成长。2024年6月16日23:38:44。