​

python3 实现企微发送应用消息

想要使用企微作为提醒 必须要通过认证企微(忘记怎么认证了…) 需要获取企微主体id, 创建应用, 应用id, 密钥

效果图

获取企微id, 应用id, 密钥流程

点击跳转企微管理页面

  • 1. 获取企微主体id

获取企微主体id

  • 2. 创建应用(自建应用)

创建应用
自建应用

  • 3. 获取应用id, 密钥

获取应用id, 密钥

点击跳转企微文档

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
#!/usr/bin/env python
# -*- coding: utf-8 -*-
import requests
import json

from xwebhook.config import *
from xwebhook.redis_utils import redis_conn


class WeChat:
    def __init__(self):
        self.CORPID = WeChatCorpID  # 企业ID，在管理后台获取
        self.CORPSECRET = WeChatSecret  # 自建应用的Secret，每个自建应用里都有单独的secret
        self.AGENTID = WeChatAgentID  # 应用ID，在后台应用中获取
        self.TOUSER = WeChatToUser  # 接收者用户名,多个用户用|分割
        self.pipe = redis_conn()

    def _get_access_token(self):
        """
        功能介绍: 获取access_token
        :return:
        """
        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)
        self.pipe.set("access_token", data.get("access_token"), 60 * 60 * 2)
        self.pipe.execute()
        return data["access_token"]

    def get_access_token(self):
        """
        功能介绍: 判断token在两个小时内不用重复去请求
        :return:
        """
        try:
            access_token = self.pipe.execute(self.pipe.get("access_token"))[0].decode()
        except:
            access_token = self._get_access_token()
        return access_token

    def send_data(self, message):
        """
        功能介绍: 发送消息
        :param message:
        :return:
        """
        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": "0"
        }
        send_msges = (bytes(json.dumps(send_values), 'utf-8'))
        res = requests.post(send_url, send_msges)
        resp = res.json()  # 当返回的数据是json串的时候直接用.json即可将res转换成字典
        return resp["errmsg"]


# if __name__ == '__main__':
#     wx = WeChat()
#     state = wx.send_data("hey 你好吗")