快捷導(dǎo)航

go-cqhttp智能聊天功能的實現(xiàn)

更新時間：2022年09月13日 09:47:19 作者：A-L-Kun

這篇文章主要介紹了go-cqhttp智能聊天功能,本文通過實例代碼給大家介紹的非常詳細(xì)，對大家的學(xué)習(xí)或工作具有一定的參考借鑒價值，需要的朋友可以參考下

智能聊天

一、概述

我們將我們的qq聊天機(jī)器人的環(huán)境配置好后，其就可以開始接收消息啦！那么，我們除了可以接收特定的消息，是不是還需要接收那些不是我們指定的消息呢？我想是的！那么，我們要如何接入呢？

這里，我找了一個比較好用的聊天機(jī)器人的API接口。可以將其接入我們的程序中，做一個陪聊小助手。當(dāng)然，如果你機(jī)器學(xué)習(xí)學(xué)的比較好的話，你也可以自己訓(xùn)練一個模型來實現(xiàn)智能聊天的接口。

我們選擇的是青云客智能聊天

二、使用方法

同時，其還有一些拓展的功能！

三、接入程序

我們暫時只對私信消息進(jìn)行處理，接入我們的智能聊天接口

在private_script.py文件中，添加一個函數(shù)，同時修改處理私聊消息的接口：

#!/usr/bin/python3
# -*- coding: UTF-8 -*-
__author__ = "A.L.Kun"
__file__ = "private_script.py.py"
__time__ = "2022/9/9 22:04"

import httpx
from datetime import datetime
async def handle_private(uid, message):  # 處理私聊信息
    if message:  # 簡單的判斷，只是判斷其是否為空
        _ = await get_resp(message)
        ret = _.get("content", "獲取回復(fù)失敗")
        await send(uid, f"{ret}\n發(fā)送時間：{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
async def get_resp(message):  # 對接口發(fā)送請求，獲取響應(yīng)數(shù)據(jù)
    async with httpx.AsyncClient() as client:
        params = {
            "key": "free",
            "appid": 0,
            "msg": message,
        }
        resp = await client.get("http://api.qingyunke.com/api.php", params=params)
        return resp.json()
async def send(uid, message):
    """
    用于發(fā)送消息的函數(shù)
    :param uid: 用戶id
    :param message: 發(fā)送的消息
    :return: None
    """
    async with httpx.AsyncClient(base_url="http://127.0.0.1:5700") as client:
        # 如果發(fā)送的為私聊消息
        params = {
            "user_id": uid,
            "message": message,
        }
        await client.get("/send_private_msg", params=params)

這個文件負(fù)責(zé)發(fā)送私聊信息

四、智能群聊

我們這里開發(fā)一個智能水群的功能，其可以自動的根據(jù)群消息回復(fù)，同時增加了進(jìn)群歡迎的功能！

實現(xiàn)群聊艾特回復(fù)功能
實現(xiàn)戳一戳回復(fù)功能
實現(xiàn)新入群成員歡迎功能

在我們的main.py中，添加接口

#!/usr/bin/python3
# -*- coding: UTF-8 -*-
__author__ = "A.L.Kun"
__file__ = "main.py"
__time__ = "2022/9/9 22:03"

from flask import Flask, request
from flask_restful import Resource, Api
import private_script, group_script
import asyncio
app = Flask(__name__)

api = Api(app)
class AcceptMes(Resource):

    def post(self):
        # 這里對消息進(jìn)行分發(fā)，暫時先設(shè)置一個簡單的分發(fā)
        _ = request.json
        if _.get("message_type") == "private":  # 說明有好友發(fā)送信息過來
            uid = _["sender"]["user_id"]  # 獲取發(fā)信息的好友qq號
            message = _["raw_message"]  # 獲取發(fā)送過來的消息
            asyncio.run(private_script.handle_private(uid, message))
        elif _.get("message_type") == "group" and "[CQ:at,qq=2786631176]" in _["raw_message"]:  # 制作群聊消息
            message = _["raw_message"].replace("[CQ:at,qq=2786631176]", "")  # 獲取發(fā)送過來的消息
            gid = _["group_id"]  # 獲取發(fā)送消息的群號
            asyncio.run(group_script.handle_group(gid, message))、
        elif _.get("notice_type") == "group_increase":  # 有新成員加入
            uid = _["user_id"]  # 獲取加入者的qq
            gid = _["group_id"]  # 獲取群號
            asyncio.run(group_script.group_increase(uid, gid))  # 發(fā)送歡迎語
        elif _.get("sub_type") == "poke":  # 如果事件類型為戳一戳
            uid = _["user_id"]
            tid = _["target_id"]
            if str(tid) != "3500515050" and str(tid) != "2786631176":  # 判斷是否戳的是自己的賬號
                return
            try:
                gid = _["group_id"]
            except KeyError as e:
                gid = None
            asyncio.run(group_script.click_event(uid, gid))  # 傳入群號和qq號
api.add_resource(AcceptMes, "/", endpoint="index")
if __name__ == '__main__':
    app.run("0.0.0.0", 5701, debug=True)  # 注意，這里的端口要和配置文件中的保持一致

創(chuàng)建一個文件group_script.py，用于處理群聊消息

#!/usr/bin/python3
# -*- coding: UTF-8 -*-
__author__ = "A.L.Kun"
__file__ = "group_script.py"
__time__ = "2022/9/10 11:49"
import httpx
import private_script
import config
from random import choice
async def handle_group(gid, message):  # 處理群聊信息
    if str(gid) in config.allow_group_id:  # 簡單的判斷，只是判斷其是否為空
        if message.strip() == "":
            await send(gid, "艾特我干啥？又不發(fā)消息，一巴掌呼死你！[CQ:face,id=86][CQ:face,id=12]")
        else:
            _ = await private_script.get_resp(message)
            ret = _.get("content", "獲取回復(fù)失敗")
            await send(gid, ret)
async def group_increase(uid, gid):  # 處理有新成員加入的情況
    if str(gid) in config.allow_group_id:
        msg = config.welcome_group.get(str(gid), config.welcome_group["default"]) % uid  # welcome_group的鍵是qq群號，值是歡迎語
        await send(gid, msg)  # 發(fā)送信息


async def click_event(uid, gid):
    info = choice(config.click_info)  # 獲取戳一戳的信息
    try:
        info = info % uid
    except TypeError:
        if gid:  # 說明其為群戳戳
            info = f"[CQ:at,qq={uid}]" + info
    if gid:
        await send(gid, info)
    else:
        await private_script.send(uid, info)
async def send(gid, message):
    """
    用于發(fā)送消息的函數(shù)
    :param gid: 群號
    :param message: 發(fā)送的消息
    :return: None
    """
    async with httpx.AsyncClient(base_url="http://127.0.0.1:5700") as client:
        # 如果發(fā)送的為私聊消息
        params = {
            "group_id": gid,
            "message": message.replace("{br}", "\n").replace("菲菲", "坤坤"),
        }
        await client.get("/send_group_msg", params=params)

設(shè)置配置文件config.py：

#!/usr/bin/python3
# -*- coding: UTF-8 -*-
__author__ = "A.L.Kun"
__file__ = "config.py"
__time__ = "2022/9/10 11:57"
allow_group_id = [
]

welcome_group = {  # 新成員進(jìn)入回復(fù)信息
    "default": f"[CQ:at,qq=%d] 歡迎加入本群，來了就別想走哦！[CQ:face,id={43}]",
}
click_info = [  # 戳一戳的回復(fù)信息
    "？有事嗎？沒事我走了！[CQ:face,id=125]，goodbye",
    "沒問題么？〒沒問題的話，我就溜了哦！",
    "睡覺去，困死了！",
    "戳我干啥！本人在敘利亞做兼職呢？沒事別煩我！",
    "你好呀！我在躺平呢！請問有啥事呀？",
    "hello",
    "[CQ:poke,qq=%d]",
]