python+go

#python+go| 来源: 网络整理| 查看: 265

最近项目上了活动，流量比较大，运营需要时刻监控流量来控制活动内容，本来想开发一个微信机器人，然而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 & 命令执行后台运行。

II.功能实现

消息发送的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: # socket持续监听，无需断开连接 # 低性能主机长时间保持长连接会导致CPU占用居高不下 # 比如我的J4125运行6小时以后CPU占用率会到90% # 解决办法：在config.yml中关闭心跳，socket.recv()方法默认阻塞，没有消息的时候死循环会阻塞减少CPU占用 while True: # 设置获取的字节串最大长度 b: bytes = conn.recv(1024 * 10) buf = b.decode("utf-8") 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")) # 处理数据 if data and data != "": handle_data(data) except Exception as e: print(e) conn.close() def response_success(): return "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\n\r\n{result:success}\r\n" def receive_data(raw: str): try: # 正则表达式匹配报文内容 data_re = re.search(r"{(.|\s)*}", raw, re.S) if data_re: data_str = str(data_re.group()) return data_str except Exception as e: print(e) print("错误位置:%s" % e.__traceback__.tb_lineno) def handle_data(data): if data_str.find("heartbeat") != -1: # 心跳 pass else: # 自定义机器人指令逻辑 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脚本时会找不到相对路径的文件。）。然后点确定保存退出即可。

2. Linux

使用crontab添加计划任务即可。具体可以参考我的另一篇文章:服务器设置定时任务使用python爬取网页

【本文地址】

公司简介

联系我们