1. AI技术聚合首页
  2. Python

制作属于自己的QQ机器人，超详细

青葱年少 • • Python • 阅读 103

目录

前期准备

1.QQ机器人框架的下载和配置

⑴首先需要一个QQ机器人的框架，这里我使用的是基于 Mirai 以及 MiraiG开发的go-cqhttp

下载地址

Windows推荐下载这个

 Mac推荐下这个

下载后得到三个文件

⑵双击第一个文件，输入0后回车，然后关闭

 

此时Windows系统的会在原来的目录中生成配置文件config.yml，而Macos系统的可以在访达中搜索找到

 ⑶双击打开config.yml文件，输入账号和密码(建议使用无登录保护的QQ，要不然会很麻烦)

 也可以不输入账号和密码，直接运行bat文件会出现二维码扫码登录，这样更加安全

最后在转到后两行，把#去掉，使其生效（很重要！！！）

⑷配置完成后直接双击bat文件，等待一段时间后会出现如下窗口

这两个端口都应该出现（很重要！！！）

如果运行bat文件后出现以下情况，不要慌，属于正常现象：

窗口会一直上报收到的信息，不要关闭

更多问题请查看官方文档

到这里，go-cqhttp的配置算是完成了

具体实现

1.发送信息

包括发送私聊信息和群聊信息，下面函数已经写好了，只需要传入参数即可

def send_msg(resp_dict):
    client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

    ip = '127.0.0.1'
    client.connect((ip, 5700))

    msg_type = resp_dict['msg_type']  # 回复类型（群聊/私聊）
    number = resp_dict['number']  # 回复账号（群号/好友号）
    msg = resp_dict['msg']  # 要回复的消息

    # 将字符中的特殊字符进行url编码
    msg = msg.replace(" ", "%20")
    msg = msg.replace("\n", "%0a")

    if msg_type == 'group':
        payload = "GET /send_group_msg?group_id=" + str(
            number) + "&message=" + msg + " HTTP/1.1\r\nHost:" + ip + ":5700\r\nConnection: close\r\n\r\n"
    elif msg_type == 'private':
        payload = "GET /send_private_msg?user_id=" + str(
            number) + "&message=" + msg + " HTTP/1.1\r\nHost:" + ip + ":5700\r\nConnection: close\r\n\r\n"
    print("发送" + payload)
    client.send(payload.encode("utf-8"))
    client.close()
    return 0

⑴发送私聊消息

resp_dict={'msg_type':'private','number':QQ号,'msg':'在吗'}
send_msg(resp_dict)

⑵发送群聊信息

resp_dict={'msg_type':'group','number':QQ号,'msg':'有人吗'}
send_msg(resp_dict)

⑶go-cqhttp框架不仅支持发送文本信息，并且还支持发送图片、音乐、表情包，甚至还可以@某人

例如：@某人

send_msg({'msg_type': ' provide', 'number': QQ号, 'msg': '[CQ:at]'})

发送表情

send_msg({'msg_type': 'group', 'number': QQ群号, 'msg': '[CQ:表情id]'})

更多功能请查看官方文档

2.接收上报的事件

新建一个py文件把名字改为receive，之后方便调用（不要运行这个代码，否则会引起端口冲突）。这个返回的是事件的上报信息，包括接收到的信息、加好友请求等等。

import socket
import json



ListenSocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
ListenSocket.bind(('127.0.0.1', 5701))
ListenSocket.listen(100)

HttpResponseHeader = '''HTTP/1.1 200 OK
Content-Type: text/html
'''

def request_to_json(msg):
    for i in range(len(msg)):
        if msg[i]=="{" and msg[-1]=="\n":
            return json.loads(msg[i:])
    return None

# 需要循环执行，返回值为json格式
def rev_msg():# json or None
    Client, Address = ListenSocket.accept()
    Request = Client.recv(1024).decode(encoding='utf-8')
    rev_json=request_to_json(Request)
    Client.sendall((HttpResponseHeader).encode(encoding='utf-8'))
    Client.close()
    return rev_json

3.实现简单的自动回复

需要写一个while循环一直获取rev，然后根据字段里面的信息做出回应。废话不多说，下面直接上代码：

from receive import rev_msg
import socket


# 发送消息函数
def send_msg(resp_dict):
    client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

    ip = '127.0.0.1'
    client.connect((ip, 5700))

    msg_type = resp_dict['msg_type']  # 回复类型（群聊/私聊）
    number = resp_dict['number']  # 回复账号（群号/好友号）
    msg = resp_dict['msg']  # 要回复的消息

    # 将字符中的特殊字符进行url编码
    msg = msg.replace(" ", "%20")
    msg = msg.replace("\n", "%0a")

    if msg_type == 'group':
        payload = "GET /send_group_msg?group_id=" + str(
            number) + "&message=" + msg + " HTTP/1.1\r\nHost:" + ip + ":5700\r\nConnection: close\r\n\r\n"
    elif msg_type == 'private':
        payload = "GET /send_private_msg?user_id=" + str(
            number) + "&message=" + msg + " HTTP/1.1\r\nHost:" + ip + ":5700\r\nConnection: close\r\n\r\n"
    print("发送" + payload)
    client.send(payload.encode("utf-8"))
    client.close()
    return 0




while True:
    try:
        rev = rev_msg()
        print(rev)
        if rev == None:
            continue
    except:
        continue

    if rev["post_type"] == "message":
        if rev["message_type"] == "private":  # 私聊
            if rev['raw_message'] == '在吗':
                qq = rev['sender']['user_id']
                send_msg({'msg_type': 'private', 'number': qq, 'msg': '在的'})  #
        elif rev["message_type"] == "group":  # 群聊
            group = rev['group_id']
            # if "[CQ:at,qq=机器人的QQ号]" in rev["raw_message"]:
            if rev['raw_message'] == '有人吗':
                qq = rev['sender']['user_id']
                send_msg({'msg_type': 'group', 'number': group, 'msg': '当然有'.format(qq)})
        else:
            continue
    else:
        continue

​ 

可以将上面的代码命名为main.py，之后的代码都是在这里添加的。需要一直运行，不能关闭

4.实现智能聊天功能

这里就要用到API接口了，目前我找到的智能回复的机器人是思知机器人，接口地址为：

https://api.ownthink.com/bot

首先先要去官网上注册账号，然后创建机器人，查看机器人的appid

如果有什么不懂得可以查看官方文档

请求参数：

返回参数： 

​ 

下面直接上代码

import requests
from receive import rev_msg
import socket


# 发送消息函数
def send_msg(resp_dict):
    client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

    ip = '127.0.0.1'
    client.connect((ip, 5700))

    msg_type = resp_dict['msg_type']  # 回复类型（群聊/私聊）
    number = resp_dict['number']  # 回复账号（群号/好友号）
    msg = resp_dict['msg']  # 要回复的消息

    # 将字符中的特殊字符进行url编码
    msg = msg.replace(" ", "%20")
    msg = msg.replace("\n", "%0a")

    if msg_type == 'group':
        payload = "GET /send_group_msg?group_id=" + str(
            number) + "&message=" + msg + " HTTP/1.1\r\nHost:" + ip + ":5700\r\nConnection: close\r\n\r\n"
    elif msg_type == 'private':
        payload = "GET /send_private_msg?user_id=" + str(
            number) + "&message=" + msg + " HTTP/1.1\r\nHost:" + ip + ":5700\r\nConnection: close\r\n\r\n"
    print("发送" + payload)
    client.send(payload.encode("utf-8"))
    client.close()
    return 0

# 控制整个机器人运行
def main02():
    while True:
        try:
            rev = rev_msg()
            print(rev)
            if rev == None:
                continue
        except:
            continue

        if rev["post_type"] == "message":
            message = rev['raw_message']
            data = {
                "appid": "自己机器人的appid",
                "userid": "自己的id",
                "spoken": message,
            }
            url = 'https://api.ownthink.com/bot'
            headers = {'Content-Type':'application/json'}
            response = requests.post(url=url, data=data, headers=headers)
            print(response.status_code)
            response.encoding = 'utf-8'
            result = response.json()
            answer = result['data']['info']['text']
            if rev['message_type'] == 'private':  #
                qq = rev['sender']['user_id']
                send_msg({'msg_type': 'private', 'number': qq, 'msg': answer})
            elif rev['message_type'] == 'group':
                qq01 = rev['group_id']
                send_msg({'msg_type': 'group', 'number': qq01, 'msg': answer})

if __name__ == '__main__':
    main02()

5.解决多次回复问题

接收本地端口的数据时会有重复的报文（大概5次），就像这样：

出现这种问题只需要把receive.py中的HttpResponseHeader修改一下就OK了

 

更多功能敬请期待，感谢大家对本文的支持的支持！！

文章出处登录后可见！

已经登录？立即刷新

共计人评分，平均分

到目前为止还没有投票！成为第一位评论此文章。

赞 (0)
青葱年少的头像青葱年少普通用户
0
上一篇 2023年6月26日
下一篇 2023年6月26日

相关推荐