从零搭建微信机器人(四)：封装消息发送接口 |
您所在的位置：网站首页 › 微信群机器人发链接 › 从零搭建微信机器人(四)：封装消息发送接口 |
从零搭建微信机器人(四)：封装消息发送接口
|
从零搭建微信机器人(一)：注册企业微信创建应用 从零搭建微信机器人(二)：发送文本消息 从零搭建微信机器人(三）：定时触发任务 从零搭建微信机器人(四)：封装消息发送接口 文章目录 1. 封装WeChatAPI1.1 代码总览1.2 代码说明构造函数token请求函数token更新函数文本消息发送函数实例化对象 2. 发送文本消息代码 3. 后台运行3.1 nohup3.2 screen 本项目的源码链接： hanfangyuan/wechat-robot，本文对应仓库tag为4.0 在上一篇 从零搭建微信机器人(三）：定时触发任务中，我们学会了定时向微信发送消息，本篇将要把微信消息发送功能封装为一个类，便于后续发送其他更为复杂的消息。另外，还将介绍使用linux的screen命令保持程序的后台运行。 1. 封装WeChatAPIpython支持面向对象编程，在以前的博文中只是简单实现了消息发送的功能，并没有考虑保存token以及token过期这种情况的处理。把消息发送的功能封装成一个完整而单纯的类，只负责实现向微信发送消息，具体推送的消息内容在其他地方实现，这样代码结构更清晰，也便于代码的复用。 1.1 代码总览按照往常的习惯，贴出完整的代码，该代码保存于wechat_api.py模块。 # wechat_api.py import requests import json # 企业id、key CORP_ID = 'xxx' # 更换为你的企业id CORP_SECRET = 'xxx' # 更换为你的应用secret AGENT_ID = 1000002 #更换为你的应用id，注意是数字类型，不是字符串 # 微信消息接口类 class WeChatAPI(): # 接口类的构造函数，需要企业id以及应用的secret def __init__(self, corp_id, corp_secret): self.corp_id = corp_id self.corp_secret = corp_secret self.update_token() # token请求函数，根据企业id以及应用secret请求token def get_token(self): # token获取的api token_api = ( 'https://qyapi.weixin.qq.com/cgi-bin/gettoken?' + f'corpid={self.corp_id}&corpsecret={self.corp_secret}' ) # 设置10s的超时限制，无响应则打印错误信息 try: response = requests.get(token_api, timeout=10) except: print('请求超时，无法获取access_token') return 'error' return response.json()['access_token'] # token更新函数，返回的token其实是有有效期的，在token过期后要更新token， # 我们在得到token后要及时保存，避免频繁重复请求token，微信对token的请求频率也是有限制的 def update_token(self): self.access_token = self.get_token() # 获取新的token后，放入消息发送的api中 self.send_api = 'https://qyapi.weixin.qq.com/cgi-bin/message/send?' + f'access_token={self.access_token}' # 文本消息发送函数，传入要发送的文本消息标题，正文; 应用id以及推送的用户有默认值，需要时可以传入参数 def send_text_message(self, title, text, agent_id=AGENT_ID, touser="@all"): # 消息体字典转换为json字符串 data = json.dumps({ "touser" : touser, "msgtype" : "text", "agentid" : agent_id, "text" : { # 文本消息可以设置一个标题还有正文，这样阅读更清晰 "content" : title + '\n\n' + text }, "safe":0 }) # 设置10s的超时限制，无响应则打印错误信息 try: response = requests.post(self.send_api, data=data, timeout=10).json() except: print('请求超时，无法发送消息') return # 企业微信文档中介绍了42001，40014这两个错误代码都跟token有关，需要更新token，再次发送消息 # 更新token，重新发送消息 if response['errcode']==42001 or response['errcode']==40014: self.update_token() self.send_text_message(title, text, agent_id=agent_id, touser=touser) # 发送成功 elif response['errcode'] == 0: print(title, '发送成功') # 其他错误 else: print('发送失败') print('data: ', data) print('response: ', response) print('token: ', self.access_token) # 实例化微信消息接口对象，供其他模块直接使用 wechat_api = WeChatAPI(CORP_ID, CORP_SECRET) 1.2 代码说明 构造函数 # 接口类的构造函数，需要企业id以及应用的secret def __init__(self, corp_id, corp_secret): self.corp_id = corp_id self.corp_secret = corp_secret self.update_token()__init__(self, corp_id, corp_secret) 为python中类的构造函数，在实例化类创建对象的时候会调用构造函数。微信消息接口类的构造函数参数有corp_id以及corp_secret分别为企业的id以及应用secret，这两个参数用于后续请求token。构造函数的self参数类似C++或者java的this指针，是类成员函数必须的且放在第一个的参数。 token请求函数 # token请求函数，根据企业id以及应用secret请求token def get_token(self): # token获取的api token_api = ( 'https://qyapi.weixin.qq.com/cgi-bin/gettoken?' + f'corpid={self.corp_id}&corpsecret={self.corp_secret}' ) # 设置10s的超时限制，无响应则打印错误信息 try: response = requests.get(token_api, timeout=10) except: print('请求超时，无法获取access_token') return 'error' return response.json()['access_token']根据企业id以及应用secret请求token。 token更新函数 # token更新函数，返回的token其实是有有效期的，在token过期后要更新token， # 我们在得到token后要及时保存，避免频繁重复请求token，微信对token的请求频率也是有限制的 def update_token(self): self.access_token = self.get_token() # 获取新的token后，放入消息发送的api中 self.send_api = 'https://qyapi.weixin.qq.com/cgi-bin/message/send?' + f'access_token={self.access_token}'token一般会在两个小时后过期，使用过期后的token发送消息会返回表示token过期的错误代码，这时候就需要重新请求token，然后生成新的消息推送请求URL。 文本消息发送函数 # 文本消息发送函数，传入要发送的文本消息标题，正文; 应用id以及推送的用户有默认值，需要时可以传入参数 def send_text_message(self, title, text, agent_id=AGENT_ID, touser="@all"): # 消息体字典转换为json字符串 data = json.dumps({ "touser" : touser, "msgtype" : "text", "agentid" : agent_id, "text" : { # 文本消息可以设置一个标题还有正文，这样阅读更清晰 "content" : title + '\n\n' + text }, "safe":0 }) # 设置10s的超时限制，无响应则打印错误信息 try: response = requests.post(self.send_api, data=data, timeout=10).json() except: print('请求超时，无法发送消息') return # 企业微信文档中介绍了42001，40014这两个错误代码都跟token有关，需要更新token，再次发送消息 # 更新token，重新发送消息 if response['errcode']==42001 or response['errcode']==40014: self.update_token() self.send_message(title, text, agent_id=agent_id, touser=touser) # 发送成功 elif response['errcode'] == 0: print(title, '发送成功') # 其他错误 else: print('发送失败') print('data: ', data) print('response: ', response) print('token: ', self.access_token)使用该成员函数向微信推送文本消息，注释中已经解释得很明白了，不再过多解释。有关token的错误代码解释可以查看全局错误码说明。 实例化对象 # 实例化微信消息接口对象，供其他模块直接使用 wechat_api = WeChatAPI(CORP_ID, CORP_SECRET)在该模块(wechat_api.py)中实例化处一个对象，主要是考虑到企业id以及应用secret全部都保存在了该文件中，其他模块要想使用发送消息的功能可以直接通过from wechat_api import wechat_api导入该对象，而不需要重复实例化该类，创建对象。 2. 发送文本消息封装完毕后，接下来就要试试好不好用了。我们单独创建一个wechat_robot.py的文件，实现每天早上7:30向微信发送一条早安问候。在该文件中我们只需要关心发送消息的具体内容以及发送的时刻，而不用关心如何发送消息，只需要使用上文封装的微信消息接口即可。 代码 # wechat_robot.py from apscheduler.schedulers.blocking import BlockingScheduler from .wechat_api import wechat_api # 导入是实例化的微信消息接口对象 # 早安消息 def good_morning(): title = '早安问候' text = '早上起床，拥抱太阳，满满的正能量！' wechat_api.send_text_message(title, text) if __name__ == '__main__': # 创建scheduler，timezone时区信息可以不设置，默认为系统时区 scheduler = BlockingScheduler(timezone="Asia/Shanghai") # 每天七点半发送问候消息 scheduler.add_job(good_morning, 'cron', hour=7, minute=30) # 开启scheduler print('开启任务......') scheduler.start()可以看到，在进行封装后，代码变得一目了然，使用python wechat_robot.py命令即可开启发送消息的功能，可以在任务触发时间收到该消息。 因为代码要一直运行着，不可能开一个ssh窗口连接服务器一直运行着该命令，这样的话ssh连接挂掉，运行的命令也会挂掉。以我个人有限的水平解释，这是linux的进程启动、结束机制：一般进程都有父进程，如果父进程挂掉，后面的子进程、孙子进程通通也会结束。在ssh连接窗口运行python wechat_robot.py命令，相当于ssh连接会话的进程是上述命令的父进程，如果ssh会话由于网络不畅或主动关闭导致进程结束的话，运行该命令的进程也会挂掉。 一般有两种方式实现程序的后台运行，nohup以及screen。 3.1 nohupnohup应该是linux系统自带的命令，它的全称是no hang up 顾名思义就是不打断，在待运行的命令前加上该命令，可以时命令的父进程变为linux的1号进程，1号进程是linux的初始化进程，辈分很高，不可以被暂停、终止，相当于该命令的进程认了个大佬当爹，被大佬罩着自然不会轻易被干掉。 使用nohup后台运行的命令为： nohup python wechat_robot.py & 命令末尾的 & 符号表示在后台运行命令，不加的话会阻塞当前的ssh会话终端，不能在该终端输入其他命令。 要结束后台运行的进程先要使用命令 ps -ef | grep python wechat_robot.py 找到该进程的进程编号，然后使用命令 kill 结束该进程 批量删除命令 ps -ef | grep | grep -v grep | awk '{print $2}' | xargs kill -9 | 管道符，用来隔开两个命令，管道符左边命令的输出会作为管道符右边命令的输入。 ps 命令用来列出系统中当前运行的进程，ps -ef 显示所有进程信息，联通命令行。 grep 命令用于过滤/搜索特定字符，grep 在这里为搜索过滤所有含某关键字的进程 grep -v grep -v 显示不包含匹配文本的所有行，在这里为筛选出所有不包含 grep 名称的进程，对上一步的进程再做一次筛选(因为 ps -ef 列出了所有的命令，包括命令行) awk 在文件或字符串中基于指定规则浏览和抽取信息；把文件逐行读入，以空格为默认分隔符将每行切片，然后再进行后序处理。 awk ‘{print $2}’ 将上一步中过滤得到的进程进行打印，$2 表示打印第二个域(PID，进程号) $0 表示所有域, 1 表 示 第 一 个 域 ， 1 表示第一个域， 1表示第一个域，n 表示第 n 个域。 xargs 命令是给命令传递参数的过滤器，把标准数据数据转换成命令行参数。在这里则是将获取前一个命令的标准输出然后转换成命令行参数传递给后面的 kill 命令。 kill -9 强制关闭进程。 补充 nohup python -u wechat_robot.py > wechat_robot.log 2>&1 & -u python输出不缓存，立即写到wechat_robot.log中 标准输出流重定向到wechat_robot.log中 2>&1 错误重定向到标准输出流 3.2 screenscreen是一个会话管理命令，可以轻松实现使会话终端后台运行以及恢复正常的功能。 ubuntu可以通过如下命令安装。 sudo apt update && sudo apt install screen 使用screen实现后台运行以及关闭的步骤如下： screen -S wechat_robot 创建一个会话；在该会话中使用命令python wechat_robot.py开启消息发送程序；同时按ctrl + A释放该两个按键后，再按一下D键即可退出该会话并保持会话后台运行;screen -ls 查看会话列表；screen -r wechat_robot恢复进入后台的会话;进入会话后ctrl + C 结束消息发送的程序，接着输入exit退出并终止该会话。本篇博客介绍了封装微信消息接口、测试该接口的有效性以及后台运行微信机器人程序。至此，已经实现了较为完善的文本消息发送功能，下一篇将会在微信消息接口中添加图文消息发送接口，并利用该图文消息接口每日定时向微信推送知乎热榜。 参考链接： [1] “平安报”自动化折腾日记 |
【本文地址】