酷Q [CQHTTP Python Async SDK] 入坑指南
Contents
相关链接 见文末
酷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
为什么用这个?
- 我之前写的 python 的机器人框架用的这个
- 基于 quart 框架写的采用了异步的方式,因此效率更高。(flask 和 quart 的对比)
- 保留了 quart 的功能,可以额外添加网页访问等功能。
- 采用 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 发给后端。
配置
首先下载酷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¬ice=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¬ice=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码
进群欢迎
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 地址
个人仓库地址(镜像, 备用)
备用站点为镜像/反代站点。速度相较于原站点较慢,建议仅在直接访问非大陆的服务器(原站点)更慢的时候再选择备用站点。