最近项目上了活动，流量比较大，运营需要时刻监控流量来控制活动内容，本来想开发一个微信机器人，然而github上微信机器人的开源项目都对微信版本有要求，可移植性差（没跑起来），转而开发QQ机器人。
I.基于go-cqhttp 1. 配置go-cqhttp官方文档
使用版本：v1.0.0-rc1
根据官方文档修改config.yml，第一次运行go-cqhttp会自动创建。
做一个简单的自动回复消息的机器人只需要考虑如下地方：
account: # 账号相关 uin: 123456 # QQ账号 password: 123123 # 密码为空时使用扫码登录 servers: # HTTP 通信设置 - http: host: 127.0.0.1 # 服务端监听地址 port: 5700 # 服务端监听端口 …… post: # 反向HTTP POST地址列表 - url: http://127.0.0.1:5701/ # 地址
反向post地址和端口号按需修改。
2. 启动go-cqhttp:windows系统下直接运行go-cqhttp.bat即可。
linux系统下使用nohup ./go-cqhttp & 命令执行后台运行。
消息发送的host:port取决于config.yml中配置的服务端监听地址和服务端监听端口。
1.定时任务主动推送消息一个简单的示例：
# coding = utf-8
import os
import json
import Model_mysql as mySql
import httpx
def start():
project_array = load_config("db_config.json") # 自定义获取数据库列表的方法
contact_array = load_config("contact.json") # 自定义获取联系人列表的方法
data = load_data(project_array ) # 自定义从数据库获取数据的方法
for data_obj in data :
for contact_obj in contact_array:
host = "http://127.0.0.1:5700/send_private_msg?user_id=%s&message=%s" % (contact_obj, data_obj)
httpx.get(host) # 将每份数据发送给每个联系人
start()
2.自动回复（接受并处理指令）
go-cqhttp如果在config.yml中配置了反向HTTP POST地址，就会在收到消息时把消息post到指定地址，我们可以通过socket来接收消息并处理。下面是一个简单的例子:
import socket
import json
import threading
import requests
def do(conn):
try:
while True:
buf = conn.recv(1024).decode()
if len(buf) != 0:
# 接收数据，格式化http的报文
http_data = receive_data(buf)
# 返回响应,如果不返回响应 go-cqhttp会一直报错
# [WARNING]: 上报 Event 数据到 http://127.0.0.1:5701/ 失败: Post "http://127.0.0.1:5701/": dial tcp 127.0.0.1:5701: connectex: No connection could be made because the target machine actively refused it.
conn.sendall(bytes(response_success(), encoding="utf-8"))
# 处理数据
handle_data(http_data)
else:
break
except Exception as e:
print(e)
conn.close()
def response_success():
return "HTTP/1.1 200 OKrnContent-Type: application/jsonrnrn{result:success}rn"
def receive_data(raw):
context = str(raw).split("n")
result = dict()
if len(context) > 1:
for i in range(1, len(context)):
obj = context[i].strip()
if obj.find(":") > -1 and obj.startswith("{") is False:
result[obj.split(":")[0].strip()] = obj.split(":")[1].strip()
elif obj.startswith("{") and obj.endswith("}"):
result["message"] = obj
else:
pass
else:
result = None
return result
def handle_data(data):
if data:
if data["message"]:
message = json.loads(data["message"])
if message.get("meta_event_type") and message.get("meta_event_type") == "heartbeat":
# go-cqhttp心跳，忽略数据
pass
else:
{
"private": lambda x: private(x),# 私聊消息
"group": lambda x: group(x)# 群组消息
}[message.get("message_type")](message)
def private(message):
# do something
pass
def group(message):
# do something
pass
print("Server is starting")
port = 5701 #
max_connection = 5
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.bind(('127.0.0.1', port)) # 配置socket，绑定IP地址和端口号，同config.yml中配置的反向HTTP POST地址和端口
sock.listen(max_connection) # 设置最大允许连接数，各连接和server的通信遵循FIFO原则
print("Server is listening port %s , with max connection %s" % (port, max_connection))
while True:
connection, address = sock.accept()
# 获得一个连接，然后开始循环处理这个连接发送的信息
# 使用多线程处理
threading.Thread(target=do, args=[connection, ]).start()
运行这个python无限循环监听socket即可。
附：添加到系统计划任务由于需要每小时都推送一次消息，在python代码里面写一个休眠1小时执行1次的循环实在是不优雅，把定时循环的功能交给系统就行了。
1. windows右键此电脑-管理-任务计划程序-任务计划程序库，就在任务计划程序库中创建任务。
在任务计划程序库界面右边选择创建任务
名称和描述随便填写即可。
触发器里面选择新建，我这个任务需要每天每小时触发，所以设置里面选择每天，开始时间设置为下一次需要触发的时间，比如说现在时间是21:10，下次应该是22:00，那么开始时间就写22:00，我曾经试过开始时间在当前时间以前，然后勾选上如果过了计划开始时间，立即启动任务，然而任务并没有启动╮(╯▽╰)╭
操作里面也选择新建，设置中程序或者脚本要选择python的启动器，就是python安装目录下的那个python.exe文件，添加参数填入要运行的py文件绝对路径（如果你的Python脚本中使用了相对路径，那么需要把这个Python脚本所在目录填到起始于,不然运行python脚本时会找不到相对路径的文件。）。然后点确定保存退出即可。
使用crontab添加计划任务即可。
具体可以参考我的另一篇文章:服务器设置定时任务使用python爬取网页
