企业微信应用类

快来打我* 2022-08-29 13:46 209阅读 0赞

> 用于发送消息，图片和文件

#!/usr/bin/python
    # -*- coding: UTF-8 -*-
    
    # python3
    import _locale
    _locale._getdefaultlocale = (lambda *args: ['zh_CN', 'utf8'])
    
    # python2
    # import sys
    # reload(sys)
    # sys.setdefaultencoding('utf8')
    
    import json
    import time
    import requests
    
    
    class WeChat():
        def __init__(self):
            """ 配置初始信息 """
            self.CORPID = "xxxxxxxxxxx"  # 企业ID
            self.CORPSECRET = "xxxxxxxxxxxxxxxxxxxxxxxxx"  # 应用Secret
            self.AGENTID = "xxxxxxxxxxxx"  # 应用Agentid
            # self.TOUSER = "xxx1|xxx2|xxx3" # 接收消息的userid
            self.TOUSER = "@all" # 接收消息的userid
            self.ACCESS_TOKEN_PATH = "access_token.conf" # 存放access_token的路径
    
        def _get_access_token(self):
            """ 调用接口返回登录信息access_token """
            # url = f"https://qyapi.weixin.qq.com/cgi-bin/gettoken?corpid={self.CORPID}&corpsecret={self.CORPSECRET}"
            url = "https://qyapi.weixin.qq.com/cgi-bin/gettoken?corpid={}&corpsecret={}".format(self.CORPID, self.CORPSECRET)
            res = requests.get(url=url)
            return json.loads(res.text)['access_token']
    
        def _save_access_token(self, cur_time):
            """ 将获取到的access_token保存到本地 """
            with open(self.ACCESS_TOKEN_PATH, "w")as f:
                access_token = self._get_access_token()
                # 保存获取时间以及access_token
                f.write("\t".join([str(cur_time), access_token]))
            return access_token
    
        def get_access_token(self):
            cur_time = time.time()
            try:
                with open(self.ACCESS_TOKEN_PATH, "r")as f:
                    t, access_token = f.read().split()
                    # 判断access_token是否有效
                    if 0 < cur_time-float(t) < 7200:
                        return access_token
                    else:
                        return self._save_access_token(cur_time)
            except:
                return self._save_access_token(cur_time)
    
        def send_message(self, message):
            """ 发送文本消息 """
            # url = f"https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token={self.get_access_token()}"
            url = "https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token={}".format(self.get_access_token())
            send_values = { 
                "touser": self.TOUSER,
                "msgtype": "text",
                "agentid": self.AGENTID,
                "text": { 
                    "content": message
                },
            }
            send_message = json.dumps(send_values)
            res = requests.post(url, send_message)
            return res.json()['errmsg']
    
        def _upload_file(self, file):
            """ 先将文件上传到临时媒体库 """
            # url = f"https://qyapi.weixin.qq.com/cgi-bin/media/upload?access_token={self.get_access_token()}&type=file"
            url = "https://qyapi.weixin.qq.com/cgi-bin/media/upload?access_token={}&type=file".format(self.get_access_token())
            data = { "file": open(file, "rb")}
            res = requests.post(url, files=data)
            return res.json()['media_id']
    
        def send_file(self, file):
            """ 发送文件 """
            media_id = self._upload_file(file) # 先将文件上传至临时媒体库
            # url = f"https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token={self.get_access_token()}"
            url = "https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token={}".format(self.get_access_token())
            send_values = { 
                "touser": self.TOUSER,
                "msgtype": "file",
                "agentid": self.AGENTID,
                "file": { 
                    "media_id": media_id
                },
            }
            send_message = (bytes(json.dumps(send_values), 'utf-8'))
            res = requests.post(url, send_message)
            return res.json()['errmsg']
    
        def _upload_image(self, image):
            url = "https://qyapi.weixin.qq.com/cgi-bin/media/upload?access_token={}&type=image".format(self.get_access_token())
            data = { "media": open(image, 'rb')}
            res = requests.post(url=url, files=data)
            return res.json()['media_id']
    
        def send_image(self, image):
            image = self._upload_image(image)
            url = "https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token={}".format(self.get_access_token())
            send_values = { 
                "touser": self.TOUSER,
                "agentid": self.AGENTID,
                "msgtype": "image",
                "image": { 
                    "media_id": image
                },
                    "safe": "0"
            }
            send_message = (bytes(json.dumps(send_values), 'utf-8'))
            res = requests.post(url, send_message)
            return res.json()['errmsg']
    
    
    
    wx = WeChat()
    image = "test.jpg"
    print(wx.send_image(image))