用python写一个QQ机器人


写在前面

最近发现很多QQ群都有一个机器人,用来管理群、娱乐、聊天、查询信息的都有。看着挺好玩的,就自己弄了一个。
基于酷Q,使用python编写了一个解析抖音无水印视频的机器人,群成员在群里@机器人发送抖音短视频的分享链接,机器人自动发送包括无水印视频地址的相关视频信息。
这里看一下效果:

将视频链接复制到浏览器打开,播放正常,还可以下载

下面介绍一下实现流程。

下载酷Q

去酷Q官网:https://cqp.cc/t/23253 下载软件,有免费的Air版,也有收费的Pro版,Pro版的没用过,功能肯定也强大一些。这里选择免费的Air版本下载,下载图灵版

下载后是一个压缩包,加压之后点进去,有一个CQA.exe可执行文件,运行这个软件,就会弹出一个登陆框,这里就用你准备来做机器人的QQ来登陆,推荐使用小号,避免不测。

配置 CQHTTP 插件

这里还需要使用一个插件CQHTTP,详情请查看CQHTTP插件文档,将这个插件下载下来:

GitHub下载地址:https://github.com/richardchien/coolq-http-api/releases
备用下载地址:https://cloud.ztongyang.cn/#/s/A1FB

下载完成之后,将其复制到之前解压的酷Q目录\CQA-tuling\酷Q Air\app

接下来重新运行一下CQA.exe,登录机器人QQ,登陆成功后桌面出现一个带QQ头像的悬浮窗,右击,依次点击 应用->应用管理,打开应用管理窗口启用刚才下载的那个插件。

此时 酷Q 的 data/app/io.github.richardchien.coolqhttpapi/config/ 目录中应该已经有了一个名为 <user-id>.json 的文件(<user-id> 为你登录的 QQ 账号)。修改这个文件,修改如下配置项(如果不存在相应字段则添加):

{
    "ws_reverse_url": "ws://127.0.0.1:8080/ws/",
    "use_ws_reverse": true,
    "enable_heartbeat": true
}

这里的 127.0.0.1:8080 对应 nonebot.run() 中传入的 host 和 port,如果在 nonebot.run() 中传入的 host 是 0.0.0.0,则插件的配置中需使用任意一个能够访问到 NoneBot 所在环境的 IP,不要直接填 0.0.0.0。特别地,如果你的 酷Q 运行在 Docker 容器中,NoneBot 运行在宿主机中,则默认情况下这里需使用 172.17.0.1(即宿主机在 Docker 默认网桥上的 IP,不同机器有可能不同,如果是 Linux 系统,可以使用命令 ip addr show docker0 | grep -Po 'inet K[d.]+'来获取需要填入的ip;如果是 macOS 系统或者 Windows 系统,可以考虑使用 host.docker.internal,具体解释详见 Docker 文档的 Use cases and workarounds 的「I WANT TO CONNECT FROM A CONTAINER TO A SERVICE ON THE HOST」小标题)。

配置python环境

首先确保电脑上装有python环境(Python 3.7+),如果没有请自行安装。
然后编写QQ机器人需要使用一个第三方库nonebot,使用以下命令安装即可:

pip install nonebot

这样就可以开始操作了,先来一个小实例:

首先创建一个文件夹,用来管理之后的一些机器人运行和配置文件,在这个文件夹中创建一个python文件bot.py(名称随意),内容如下:

import nonebot

if __name__ == '__main__':
    nonebot.init()
    nonebot.load_builtin_plugins()
    nonebot.run(host='127.0.0.1', port=8080)

运行这段代码,如果输出以下日志:

[2020-03-16 15:50:26,435] 127.0.0.1:56363 GET /ws/ 1.1 101 - 7982
[2020-03-16 15:50:26,438] 127.0.0.1:56364 GET /ws/ 1.1 101 - 8977

那么这将是历史性的一次对话,这表示 CQHTTP 插件已经成功地连接上了 NoneBot,与此同时,CQHTTP 的日志控制台(和日志文件)中也会输出反向 WebSocket 连接成功的日志。

现在,尝试向你的 QQ 机器人账号发送内容:/echo 你好,世界
到这里如果一切 OK,你应该会收到机器人给你回复了 你好,世界。这一历史性的对话标志着你已经成功地运行了一个 NoneBot 的最小实例,开始了编写更强大的 QQ 机器人的创意之旅!

开始创建机器人

项目结构

根据以下目录结构,创建相应的文件夹和py文件

awesome-bot
├── awesome
│   └── plugins
│       └── dybot.py
├── bot.py
└── config.py

bot.py是程序入口,config.py是配置文件,dybot.py是自行编写的机器人插件,用来实现QQ机器人的具体功能。

接下来依次往里面填充内容。

bot.py

bot.py里面写入以下内容

from os import path

import nonebot

import config

if __name__ == '__main__':
    nonebot.init(config)
    nonebot.load_plugins(
        path.join(path.dirname(__file__), 'awesome', 'plugins'),
        'awesome.plugins'
    )
    nonebot.run()

这里解释一下,import config用来引入机器人的配置信息,后面再详细说明。nonebot.load_plugins() 函数的两个参数。第一个参数是插件目录的路径,这里根据 bot.py 的所在路径和相对路径拼接得到;第二个参数是导入插件模块时使用的模块名前缀,这个前缀要求必须是一个当前 Python 解释器可以导入的模块前缀,NoneBot 会在它后面加上插件的模块名共同组成完整的模块名来让解释器导入,因此这里传入 awesome.plugins,当运行 bot.py 的时候,Python 解释器就能够正确导入 awesome.plugins.dybot 这个插件模块了。

config.py

config.py里面填入以下内容:

from nonebot.default_config import *

HOST = '127.0.0.1'
PORT = 8080
SUPERUSERS = {2251513837}
COMMAND_START = {'', '/', '!', '/', '!'}

HOSTPORT是监听的IP和端口,SUPERUSERS是配置的超级用户,里面写入你想设置为超级用户的 QQ,这里可以不用配置这个。COMMAND_START是命令的起始字符。

dybot.py

终于轮到重头戏,在dybot.py里将实现机器人的逻辑和具体功能。NoneBot 是完全基于 asyncio 的,到这里需要对 Python 的 asyncio 编程有所了解。
首先导入所需的模块,缺少的模块就自己安装就行。

import nonebot,requests,re,json
from nonebot import on_command, CommandSession
from nonebot import on_natural_language, NLPSession, IntentCommand
from datetime import datetime

想要实现一个机器人的功能,我们就需要一个命令和参数,命令用来唤醒机器人,参数用来传递给机器人,然后机器人将我们传入的参数进行相关处理之后,给我们返回我们想要的信息。前面说到,这次实现的是一个解析抖音无水印视频的机器人,给机器人传入抖音分享链接,然后机器人给我们返回视频的无水印播放地址。
首先我们就随便命一个命令,就用抖音来作为命令好了,然后需要传入的参数就是抖音里的分享链接。

先贴上完整代码:

import nonebot,requests,re,json
from nonebot import on_command, CommandSession
from nonebot import on_natural_language, NLPSession, IntentCommand
from datetime import datetime

@on_command('抖音',aliases=('抖音解析'))
async def douyin(session: CommandSession):
    data = await parse_data(session.state['share_url'])
    await session.send(data)

@douyin.args_parser
async def parse_args(session: CommandSession):
    # 去掉消息首尾的空白符
    stripped_arg = session.current_arg_text.strip()
    if stripped_arg:
            session.state['share_url'] = stripped_arg
    elif not stripped_arg:
        session.finish('没有检测到分享链接,请重新输入')

async def parse_data(share_url):
    try:
        share_url = share_url.strip()
        pat = '(https://v.douyin.com/.*?/)'
        url = re.compile(pat).findall(share_url)[0]  #正则匹配分享链接
        headers = {'user-agent': 'Android',}
        res = requests.get(url, headers=headers, allow_redirects=False)        
        item_ids = res.text.split('/')[5]
        api = f'https://www.iesdouyin.com/web/api/v2/aweme/iteminfo/?item_ids={item_ids}'
        js_data = json.loads(requests.get(api,headers = headers).text)
        data = {}
        author_nickname = js_data['item_list'][0]['author']['nickname']
        douyin_desc = js_data['item_list'][0]['desc']
        douyin_create_time = datetime.fromtimestamp(js_data['item_list'][0]['create_time'])
        play_addr = js_data['item_list'][0]['video']['play_addr']['url_list'][0].replace('playwm','play')
        v_length = int(js_data['item_list'][0]['video']['duration'])
        data['author_nickname'] = author_nickname
        data['douyin_desc'] = douyin_desc
        data['play_addr'] = play_addr
        data['v_length'] = str(format(float(v_length/1000),".2f") + "s")
        data['douyin_create_time'] = douyin_create_time
        s = "作者:" + data['author_nickname'] + '\n' + "描述:" + data['douyin_desc'] + '\n' + "地址:" + data['play_addr'] + '\n' + '时长:' + data['v_length'] + '\n' + '发布时间:' + str(data['douyin_create_time'])
        return str(s)
    except:
        return "哭辽~~o(>_<)o ~~,解析失败!"

在下面的代码中

@on_command('抖音',aliases=('抖音解析'))
async def douyin(session: CommandSession):
    data = await parse_data(session.state['share_url'])
    await session.send(data)

on_command 装饰器将函数声明为一个命令处理器,抖音为命令的名字,同时允许使用别名「抖音解析」.data就是我们最终想要的数据,这个数据是由parse_data()这个函数处理后传递过来的,具体功能就是这个函数 实现的。session.send(data)实现了将处理结果返回给我们。

下面这个函数:

@douyin.args_parser
async def parse_args(session: CommandSession):
    # 去掉消息首尾的空白符
    stripped_arg = session.current_arg_text.strip()
    if stripped_arg:
            session.state['share_url'] = stripped_arg
    elif not stripped_arg:
        session.finish('没有检测到分享链接,请重新输入')

主要用来处理传入的参数,如果传入了参数(即向机器人发送了抖音短视频的分享链接),那么就将参数存入session.state这个字典里面,将其传入功能函数parse_data中进行处理。

# 打印session.state
{'share_url':'有一种美好,叫宫崎骏的夏天。🧡 https://v.douyin.com/JLWsEEJ/ 复制此链接,打开【抖音短视频】,直接观看视频!'}

如果检测到没有传入参数,机器人向用户返回 "没有检测到分享链接,请重新输入",并结束当前会话。

最后的parse_data()函数,主要就是从分享连接中去除多余字符提取真正的链接,然后请求使用request进行请求,抓取所需参数,访问抖音视频接口,从中提取所需信息。
这里我简单提取了下面几个信息

  • 视频作者
  • 视频描述
  • 视频地址
  • 视频时长
  • 发布时间

当然还有更多的信息,比如作者签名,作者头像,视频大小,视频配乐等都可以提取出来,这里就不一一介绍了。

最后

如果有服务器的话可以将其部署到服务器上,手里刚好有一台闲置低配服务器,部署上后使用正常,24小时不间断运行,推荐服务器用win系统,linux的话需要使用docker,没测试过,不知道能不能正常运行。
所有代码扔github上了,需要自取:https://github.com/yang2210670/qqBot
结束!

仅有一条评论

    周易测算网 游客

    23 天前 Win10  回复

    谢谢分享,日常打卡~ 滴滴~

https://file.ztongyang.cn/yang/picttures/QQqr.jpg https://file.ztongyang.cn/yang/picttures/wechatqr.png https://metu.ztongyang.cn/a/avatar.jpg avatar

南玖

生命不息,折腾不止

  网站咨询
  •   当前在线1人
  •   加载耗时37 ms
  •   文章数目40篇
  •   分类总数10个
  •   评论总数58条
  •   站点字数3.84 W
  •   运行时间90天
  访问信息
  •   
  •   2020-10-23 14:03:56 Fri
  •   ?浏览器
  •   操作系统
  热门文章