使用Python发送企业微信消息

2023-08-07 10:58:22

准备工作：

到企业微信官网，注册一个企业；登录企业微信后台，创建一个“自建”应用，获取企业ID、agentid、secret这3个必要的参数；在企业微信的通讯录中，创建多个测试账号；在手机端安装“企业微信”APP，使用测试账号登录到企业微信，准备接收消息。

程序代码：

企业微信提供API开发接口，通过HTTPS的GET、POST方法与企业微信后台进行交互，完成获取令牌、发送数据、获取数据的操作。

Python代码主要使用requests库，将企业微信API进行简单封装，模拟https的GET、POST操作，向指定的用户发送企业微信消息。

#!/usr/bin/env python

# -*- coding: utf-8 -*-

import time

import requests

import json

class WeChat:

    def __init__(self):

        self.CORPID = 'ww2e1234567895498f5498f'  #企业ID，在管理后台获取

        self.CORPSECRET = 'xy11234567898hk_ecJ123456789DhKy4_1y12345OI'#自建应用的Secret，每个自建应用里都有单独的secret

        self.AGENTID = ''  #应用ID，在后台应用中获取

        self.TOUSER = "maomao|dingding"  # 接收者用户名,多个用户用|分割

    def _get_access_token(self):

        url = 'https://qyapi.weixin.qq.com/cgi-bin/gettoken'

        values = {'corpid': self.CORPID,

                  'corpsecret': self.CORPSECRET,

                  }

        req = requests.post(url, params=values)

        data = json.loads(req.text)

        return data["access_token"]

    def get_access_token(self):

        try:

            with open('./tmp/access_token.conf', 'r') as f:

                t, access_token = f.read().split()

        except:

            with open('./tmp/access_token.conf', 'w') as f:

                access_token = self._get_access_token()

                cur_time = time.time()

                f.write('\t'.join([str(cur_time), access_token]))

                return access_token

        else:

            cur_time = time.time()

            if 0 < cur_time - float(t) < 7260:

                return access_token

            else:

                with open('./tmp/access_token.conf', 'w') as f:

                    access_token = self._get_access_token()

                    f.write('\t'.join([str(cur_time), access_token]))

                    return access_token

    def send_data(self, message):

        send_url = 'https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token=' + self.get_access_token()

        send_values = {

            "touser": self.TOUSER,

            "msgtype": "text",

            "agentid": self.AGENTID,

            "text": {

                "content": message

                },

            "safe": ""

            }

        send_msges=(bytes(json.dumps(send_values), 'utf-8'))

        respone = requests.post(send_url, send_msges)

        respone = respone.json()   #当返回的数据是json串的时候直接用.json即可将respone转换成字典

        return respone["errmsg"]

if __name__ == '__main__':

    wx = WeChat()

    wx.send_data("这是程序发送的第1条消息！\n Python程序调用企业微信API,从自建应用“告警测试应用”发送给管理员的消息！")

    wx.send_data("这是程序发送的第2条消息！")