从零搭建微信机器人(四)：封装消息发送接口

字数统计: 2.9k字   |   阅读时长: 11分

从零搭建微信机器人(一)：注册企业微信创建应用
从零搭建微信机器人(二)：发送文本消息
从零搭建微信机器人(三）：定时触发任务
从零搭建微信机器人(四)：封装消息发送接口

本项目的源码链接：hanfangyuan/wechat-robot，本文对应仓库tag为4.0
在上一篇从零搭建微信机器人(三）：定时触发任务中，我们学会了定时向微信发送消息，本篇将要把微信消息发送功能封装为一个类，便于后续发送其他更为复杂的消息。另外，还将介绍使用linux的screen命令保持程序的后台运行。

1. 封装WeChatAPI

python支持面向对象编程，在以前的博文中只是简单实现了消息发送的功能，并没有考虑保存token以及token过期这种情况的处理。把消息发送的功能封装成一个完整而单纯的类，只负责实现向微信发送消息，具体推送的消息内容在其他地方实现，这样代码结构更清晰，也便于代码的复用。

1.1 代码总览

按照往常的习惯，贴出完整的代码，该代码保存于wechat_api.py模块。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
# 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 代码说明

构造函数

1
2
3
4
5
# 接口类的构造函数，需要企业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请求函数

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
# 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更新函数

1
2
3
4
5
6
7
# 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。

文本消息发送函数

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
# 文本消息发送函数，传入要发送的文本消息标题，正文; 应用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的错误代码解释可以查看全局错误码说明。

实例化对象

1
2
# 实例化微信消息接口对象，供其他模块直接使用
wechat_api = WeChatAPI(CORP_ID, CORP_SECRET)

在该模块(wechat_api.py)中实例化处一个对象，主要是考虑到企业id以及应用secret全部都保存在了该文件中，其他模块要想使用发送消息的功能可以直接通过from wechat_api import wechat_api导入该对象，而不需要重复实例化该类，创建对象。

2. 发送文本消息

封装完毕后，接下来就要试试好不好用了。我们单独创建一个wechat_robot.py的文件，实现每天早上7:30向微信发送一条早安问候。在该文件中我们只需要关心发送消息的具体内容以及发送的时刻，而不用关心如何发送消息，只需要使用上文封装的微信消息接口即可。

代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
# 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命令即可开启发送消息的功能，可以在任务触发时间收到该消息。
在这里插入图片描述

3. 后台运行

因为代码要一直运行着，不可能开一个ssh窗口连接服务器一直运行着该命令，这样的话ssh连接挂掉，运行的命令也会挂掉。以我个人有限的水平解释，这是linux的进程启动、结束机制：一般进程都有父进程，如果父进程挂掉，后面的子进程、孙子进程通通也会结束。在ssh连接窗口运行python wechat_robot.py命令，相当于ssh连接会话的进程是上述命令的父进程，如果ssh会话由于网络不畅或主动关闭导致进程结束的话，运行该命令的进程也会挂掉。
一般有两种方式实现程序的后台运行，nohup以及screen。

3.1 nohup

nohup应该是linux系统自带的命令，它的全称是no hang up 顾名思义就是不打断，在待运行的命令前加上该命令，可以时命令的父进程变为linux的1号进程，1号进程是linux的初始化进程，辈分很高，不可以被暂停、终止，相当于该命令的进程认了个大佬当爹，被大佬罩着自然不会轻易被干掉。
使用nohup后台运行的命令为：
nohup python wechat_robot.py &
命令末尾的 & 符号表示在后台运行命令，不加的话会阻塞当前的ssh会话终端，不能在该终端输入其他命令。
要结束后台运行的进程先要使用命令
ps -ef | grep python wechat_robot.py
找到该进程的进程编号，然后使用命令
kill <进程编号>
结束该进程
补充
nohup python -u wechat_robot.py > wechat_robot.log 2>&1 &
-u python输出不缓存，立即写到wechat_robot.log中
标准输出流重定向到wechat_robot.log中
2>&1 错误重定向到标准输出流

3.2 screen

screen是一个会话管理命令，可以轻松实现使会话终端后台运行以及恢复正常的功能。
ubuntu可以通过如下命令安装。
sudo apt update && sudo apt install screen
使用screen实现后台运行以及关闭的步骤如下：

  1. screen -S wechat_robot 创建一个会话；
  2. 在该会话中使用命令python wechat_robot.py开启消息发送程序；
  3. 同时按ctrl + A释放该两个按键后，再按一下D键即可退出该会话并保持会话后台运行;
  4. screen -ls 查看会话列表；
  5. screen -r wechat_robot恢复进入后台的会话;
  6. 进入会话后ctrl + C 结束消息发送的程序，接着输入exit退出并终止该会话。

本篇博客介绍了封装微信消息接口、测试该接口的有效性以及后台运行微信机器人程序。至此，已经实现了较为完善的文本消息发送功能，下一篇将会在微信消息接口中添加图文消息发送接口，并利用该图文消息接口每日定时向微信推送知乎热榜。

赏

谢谢你请我喝杯可乐

支付宝
微信

扫一扫，分享到微信

微信分享二维码

tag:

    缺失模块。
    1、请确保node版本大于6.2
    2、在博客根目录（注意不是yilia根目录）执行以下命令：
    npm i hexo-generator-json-content --save

    3、在根目录_config.yml里添加配置： 

      jsonContent:
    meta: false
    pages: false
    posts:
      title: true
      date: true
      path: true
      text: false
      raw: false
      content: false
      slug: false
      updated: false
      comments: false
      link: false
      permalink: false
      excerpt: false
      categories: false
      tags: true

北科大计算机在读<br><br>爱好编程<br><br>热爱技术<br><br>喜欢折腾