酷Q [CQHTTP Python Async SDK] 入坑指南

作者: zsh2517 分类: 未分类 发布时间: 2020-04-19 11:07

相关链接 见文末

酷Q [CQHTTP Python Async SDK] 入坑指南

介绍和文档

CoolQ HTTP API 插件
CQHTTP Python Async SDK

CQHTTP Python Async SDK 是 酷Q 的 CQHTTP 插件的 异步 Python SDK 版本，采用异步 I/O，封装了 web 服务器相关的代码，支持 CQHTTP 的 HTTP 和反向 WebSocket 两种通信方式，让使用 Python 的开发者能方便地开发插件。
本 SDK 要求使用 Python 3.7 或更高版本、CQHTTP v4.8 或更高版本。
—-CQHTTP Python Async SDK

为什么用这个？

  1. 我之前写的 python 的机器人框架用的这个
  2. 基于 quart 框架写的采用了异步的方式，因此效率更高。（flask 和 quart 的对比）
  3. 保留了 quart 的功能，可以额外添加网页访问等功能。
  4. 采用 websocket 可以实现酷Q程序和机器人主要代码分离。

前置知识

flask

一个 flask 程序大概是这样的

code 1_flask.py

from flask import Flask
app = Flask(__name__)
@app.route('/')
def index():
    return 'Index Page'


@app.route('/hello')
def hello():
    return 'Hello, World'


app.run(host="127.0.0.1", port="2333")

@app.route("xxx") 定义了一个路由，即一个特定的URL的位置。

之后当访问特定的URL时，会解析到该位置

例如

访问地址 路由 值
localhost:2333 / Index Page
localhost:2333/hello /hello Hello World

异步程序

Python 3.4 引入了 asyncio 模块，增加了异步编程，跟 JavaScript 的async/await 极为类似，大大方便了异步任务的处理。它受到了开发者的欢迎，成为从 Python 2 升级到 Python 3 的主要理由之一。
asyncio 模块最大特点就是，只存在一个线程，跟 JavaScript 一样。
由于只有一个线程，就不可能多个任务同时运行。asyncio 是”多任务合作”模式（cooperative multitasking），允许异步任务交出执行权给其他任务，等到其他任务完成，再收回执行权继续往下执行，这跟 JavaScript 也是一样的。
—-Python 异步编程入门@阮一峰的网络日志

简单来说

一个问题：
早晨起床家里就你一个人在家，要做下面的事情
烧水 20min
洗漱 10min
淘米 10min
煮饭 20min
整理房间 15min
你会怎么干？

你的任务 同时进行的 时长
洗漱 烧水 10min
淘米 烧水 10min
整理房间 煮饭 20min

简单来说，就是异步程序，一个时间还是只能执行一个程序，但是执行顺序并不是固定的，在等待的时候可以把控制权交出，执行别的程序，等到完成后再收回控制权。这样表现出来就是交叉运行了（但是实际上还是同一时刻只有一个进程在运行）

这里不过多展开，需要的话可以自行了解

quart

quart 是一个和 flask 非常类似的 web 框架，区别是底层采用了异步的架构，简单来说，从一个 flask 程序过渡到 quart，只需要改变全部的库（和 flask 有关的变成 quart 对应的模块），然后对应的加上 await / async 即可。

code 2_quart.py

from quart import Quart
app = Quart(__name__)
@app.route('/')
async def index():
    return 'Index Page'


@app.route('/hello')
async def hello():
    return 'Hello, World'

app.run(host="127.0.0.1", port="2333")

cqhttp

cqhttp 是酷Q端的插件。负责将酷Q的数据通过 websocket 发给后端。

配置

参考 配置 CQHTTP

首先下载酷Q和cqhttp插件io.github.richardchien.coolqhttpapi.cpk

将插件放到 酷Q下的app目录，之后打开酷Q

在 CQHTTP 配置文件 <酷Q>\data\app\io.github.richardchien.coolqhttpapi\config 中，填写 ws_reverse_url 值为 ws://127.0.0.1:8080/ws/ ，这里 127.0.0.1:8080 应根据情况改为 bot.py 中传给 bot.run 的 host 和 port 参数。
然后，如果有的话，删掉 ws_reverse_event_url 和 ws_reverse_api_url 这两个配置项。
接着设置 use_ws_reverse 为 true。
最后重启 CQHTTP。

python-aiocqhttp 介绍

模块内部内置了一个 quart 作为 web 服务器，因此语法和 quart/flask 非常类似。python-aiocqhttp 作为 web 服务器，与酷Q的插件 cqhttp 相结合进行通信。cqhttp 将酷Q的消息编码成 json 发送到 python-aiocqhttp，同时 python-aiocqhttp 对其进行解析和封装。

除了酷Q的各种事件注册成了路由，python-aiocqhttp 同时也支持添加 quart 的路由。详见 aicocqhttp 文档 #添加路由

实例

上手第一个程序

一个简单的人工 ZZ
看一段对话

> 在吗？
< 在！
> 你好
< 我好
> 去吃饭吗？
< 去吃饭！
> 你好吗？
< 我好！
> 一起去吃饭？
< 一起去吃饭！
> 好
< 好
> 上班了吗？
< 上班了！
> 好吧
< 好吧
> 再见
< 再见

实现只有两行

while True:
    print(input("").replace("你","我").replace("吗","").replace("？","！"))

展开来看核心代码

t = input()
t = t.replace("你","我")
t = t.replace("吗","")
t = t.replace("？","！")
print(t)

详见这段代码

code 3_aizz.py

from aiocqhttp import CQHttp, Event

bot = CQHttp()


@bot.on_message('private')
async def _(event: Event):
    msg = event.message
    msg = msg.replace("你", "我")
    msg = msg.replace("吗", "")
    msg = msg.replace("？", "！")
    # print(event, msg, "\n\n\n\n\n\n\n")
    await bot.send_private_msg(user_id=event.user_id, message=msg)


@bot.on_message('group')
async def _(event: Event):
    available = [1061750983]  # 群号
    if event.group_id in available:
        msg = event.message
        # 常用的比如 event.message, event.user_id, event.group_id
        msg = msg.replace("你", "我")
        msg = msg.replace("吗", "")
        msg = msg.replace("？", "！")
        await bot.send_group_msg(group_id=event.group_id, message=msg)
    # return {'reply': event.message}

bot.run(host='127.0.0.1', port=8080)

QQ点歌系统

酷Q + python

之前试过了，QQ音乐的数据最好弄到。所以就用QQ音乐了

1. 抓包获取API

以勾指起誓为例，搜索勾指起誓，推测服务器端返回一个json信息，在F12中一条条找，找到这个链接

https://c.y.qq.com/soso/fcgi-bin/client_search_cp?ct=24&qqmusic_ver=1298&new_json=1&remoteplace=txt.yqq.song&searchid=72043884567630163&t=0&aggr=1&cr=1&catZhida=1&lossless=0&flag_qc=0&p=1&n=10&w=%E5%8B%BE%E6%8C%87%E8%B5%B7%E8%AA%93&g_tk_new_20200303=5381&g_tk=5381&loginUin=0&hostUin=0&format=json&inCharset=utf8&outCharset=utf-8&notice=0&platform=yqq.json&needNewCode=0

对这段代码URL反编码可以看到，如图，参数列表有这么多。如果不想折腾可以直接替换里面的勾指起誓这四个字。但是作为尝试，为什么不折腾呢？
放到postman里面，一个参数一个参数尝试，发现至少需要保留 w=歌名 format=json 这两个。（如果不带format，得到的是 callback( jsondata ) 的形式。之后别的就没有影响了。记下来得到的最简化的URL https://c.y.qq.com/soso/fcgi-bin/client_search_cp?format=json&w={0}

https://c.y.qq.com/soso/fcgi-bin/client_search_cp?ct=24&qqmusic_ver=1298&new_json=1&remoteplace=txt.yqq.song&searchid=72043884567630163&t=0&aggr=1&cr=1&catZhida=1&lossless=0&flag_qc=0&p=1&n=10&w=勾指起誓&g_tk_new_20200303=5381&g_tk=5381&loginUin=0&hostUin=0&format=json&inCharset=utf8&outCharset=utf-8&notice=0&platform=yqq.json&needNewCode=0

https://c.y.qq.com/soso/fcgi-bin/client_search_cp
ct=24
qqmusic_ver=1298
new_json=1
remoteplace=txt.yqq.song
searchid=72043884567630163
t=0
aggr=1
cr=1
catZhida=1
lossless=0
flag_qc=0
p=1
n=10
w=勾指起誓
g_tk_new_20200303=5381
g_tk=5381
loginUin=0
hostUin=0
format=json
inCharset=utf8
outCharset=utf-8
notice=0
platform=yqq.json
needNewCode=0

2. 测试 API

先写一个requests请求测试一下。没有问题，然后分析json

获取到的数据可以点这查看 json

第一首歌是 泠鸢yousa 的，链接https://y.qq.com/n/yqq/song/001uRLnP2aCu1T.html

看起来这一串001uRLnP2aCu1T是歌曲的一个ID之类的。去 json 里面搜索这个 应该是 .data.song.list[0].songmid

3. demo

结合上面获取到的API，可以写出下面的一个程序，获得搜索列表并且生成第一首歌的地址。

code 4_qqmusic.py

import json
import requests
m = input("请输入歌名：")
data = requests.get("https://c.y.qq.com/soso/fcgi-bin/client_search_cp?format=json&w={0}".format(m))
j = json.loads(data.text)
f = open("test.json", "w", encoding="utf-8")
json.dump(j, f, ensure_ascii=False, indent=4)
url = "https://y.qq.com/n/yqq/song/{0}.html"
url = url.format(j["data"]["song"]["list"][0]["songmid"])
print(url)

4. 酷Q

这个时候和酷Q结合一下就是这样了。（假设通过 #点歌 指令响应）

code 5_qqmusic_2.py

import json
import requests
from aiocqhttp import CQHttp, Event

bot = CQHttp()


@bot.on_message('private')
async def _(event: Event):
    msg = event.message
    if len(msg) > 3 and msg[0:3] == "#点歌":
        msg = msg[3:]
        data = requests.get(
            "https://c.y.qq.com/soso/fcgi-bin/client_search_cp?format=json&w={0}".format(msg))
        j = json.loads(data.text)
        f = open("test.json", "w", encoding="utf-8")
        json.dump(j, f, ensure_ascii=False, indent=4)
        url = "https://y.qq.com/n/yqq/song/{0}.html"
        url = url.format(j["data"]["song"]["list"][0]["songmid"])
        print(url)
        await bot.send_private_msg(user_id=event.user_id, message=url)


@bot.on_message('group')
async def _(event: Event):
    msg = event.message
    if len(msg) > 3 and msg[0:3] == "#点歌":
        msg = msg[3:]
        data = requests.get(
            "https://c.y.qq.com/soso/fcgi-bin/client_search_cp?format=json&w={0}".format(msg))
        j = json.loads(data.text)
        f = open("test.json", "w", encoding="utf-8")
        json.dump(j, f, ensure_ascii=False, indent=4)
        url = "https://y.qq.com/n/yqq/song/{0}.html"
        url = url.format(j["data"]["song"]["list"][0]["songmid"])
        print(url)
        await bot.send_group_msg(group_id=event.group_id, message=("[CQ:at,qq={0}]" + url).format(event.user_id), auto_escape=False)


bot.run(host='127.0.0.1', port=8080)

这里的 auto_escape=False 是转义 酷Q 码。比如这里面用到了 [CQ:at,qq={0}] 有的时候可能并不希望酷Q输出各种奇奇怪怪的信息（比如利用这个特性发送卡片消息，大量@等等，可以进行转义（即不会解析成特殊消息））

进一步的，还能得到音频的歌手、简介等等

info = ret["data"]["song"]["list"][0]
url = "https://y.qq.com/n/yqq/song/%s.html" %(info["songmid"])
song_name = info["songname"]
singers = info["singer"]
singer_name = ""
for singer in singers:
    singer_name += singer["name"] + " / "
singer_name = singer_name[:-3]

完整版本（我目前用的）如下（还是整个项目的代码片段，项目开源）

code 6_qqmusic_3.py

import json
import requests
from aiocqhttp import CQHttp, Event
import aiohttp
import urllib
bot = CQHttp()


def text_share(url, title, content, image):
    text = "[CQ:share,url={0},title={1},content={2},image={3}]"
    text = text.format(url, title, content, image)
    return text


async def getqqmusic(key):
    async with aiohttp.ClientSession() as session:
        # data.append(key)
        apiurl = 'https://c.y.qq.com/soso/fcgi-bin/client_search_cp?format=json&w='
        url = apiurl + urllib.parse.quote(key.encode("utf-8"))
        resp = await session.get(url)
        ret = await resp.text()
        ret = json.loads(ret)
        f = open("music.json", "w")
        json.dump(ret, f, ensure_ascii=False, indent=4)
        f.close()
        info = ret["data"]["song"]["list"][0]
        url = "https://y.qq.com/n/yqq/song/%s.html" % (info["songmid"])
        song_name = info["songname"]
        singers = info["singer"]
        singer_name = ""
        for singer in singers:
            singer_name += singer["name"] + " / "
        singer_name = singer_name[:-3]
        img_url = "http://y.gtimg.cn/music/photo_new/T002R300x300M000%s.jpg" % (
            info["albummid"])
        text = text_share(url=url, title=song_name,
                          content=singer_name, image=img_url)
        print(text)
        # text 就是可以直接发过去的 酷Q码了
        return text


@bot.on_message('private')
async def _(event: Event):
    msg = event.message
    if len(msg) > 3 and msg[0:3] == "#点歌":
        msg = msg[3:]
        await bot.send_private_msg(user_id=event.user_id, message=await getqqmusic(msg))


@bot.on_message('group')
async def _(event: Event):
    msg = event.message
    if len(msg) > 3 and msg[0:3] == "#点歌":
        msg = msg[3:]
        await bot.send_group_msg(group_id=event.group_id, message=await getqqmusic(msg), auto_escape=True)


bot.run(host='127.0.0.1', port=8080)

在获取了这些信息之后，酷Q是有卡片消息分享的（需要酷Q pro 才能正常使用），详见这里，使用最后一个方式分享即可。（上面的那个音乐分享，因为没有音频链接这一项，所以没用）

5. 酷Q码

CQ 码@CQHTTP
CQ 码@酷Q官方

进群欢迎

code 7_welcome.py

import json
import requests
from aiocqhttp import CQHttp, Event

bot = CQHttp()


@bot.on_notice('group_increase')
async def _(event: Event):
    print(event.sub_type)
    # if event.sub_type == "approve":
    await bot.send_group_msg(group_id=event.group_id, message="欢迎[CQ:at,qq={0}]加入本群".format(event.user_id))


@bot.on_notice('group_decrease')
async def _(event: Event):
    print(event.sub_type)
    # if event.sub_type == "approve":
    await bot.send_group_msg(group_id=event.group_id, message="[CQ:at,qq={0}]离开了我们".format(event.user_id))


bot.run(host='127.0.0.1', port=8080)

更多功能

例如群成员增加、减少，撤回消息等等各种功能，看文档吧。

还有什么应用？

理论上，凡是python可以做到的东西，你都可以写成酷Q的插件（即将QQ作为输入输出）
除了常见的比如群管理、小游戏、使用工具等等QQ机器人之外
再比如我能想到的比较典型的比如这些

功能 思路
服务器状态监控 python定时获取/需要时获取服务器状态，处理后返回信息
物联网设备开关 在有API的情况下，可以通过python发送指令
和shell交互
比如mc的服务端等等		 重定向stdio
客服系统 机器人对于收到的消息进行汇总，然后分发给不同的后端客服

跨服务器

cqhttp 和 python-aiocqhttp 之间是通过 websocket 交互的，也就是二者不必像酷Q一样必须运行在同一个设备。这样可以比如酷Q放到Windows上面，然后插件主体放到Linux，或者酷Q本身运行在普通账户，插件主体运行在低权限账户上面。

其他资料链接

链接：

我目前正在做的一个，在 python-aiocqhttp 的基础上进行的一个适合于小游戏等等侧重交互性的插件的框架。
GitHub 地址
个人仓库地址(镜像, 备用)

本次分享的源码
下载
下载(备用)

备用站点为镜像/反代站点。速度相较于原站点较慢，建议仅在直接访问非大陆的服务器（原站点）更慢的时候再选择备用站点。

如果觉得我的文章对您有用，请随意打赏。您的支持将鼓励我继续创作！

发表评论

电子邮件地址不会被公开。 必填项已用*标注

更多阅读
标签云