添加企业微信代理服务器

pt_site/wechat_push.py

@@ -0,0 +1,110 @@
+import time
+
+import requests
+
+
+class WechatPush:
+    def __init__(self, corp_id, secret, agent_id, server='https://qyapi.weixin.qq.com/'):
+        self.corp_id = corp_id
+        self.agent_id = agent_id
+        self.secret = secret
+        self.token = ''
+        self.expire = 0
+        self.server = server
+
+    def refresh_token(self):
+        r = requests.get(f"{self.server}cgi-bin/gettoken", params={
+            "corpid": self.corp_id,
+            "corpsecret": self.secret
+        }).json()
+        access_token = r['access_token']
+        if access_token and len(access_token) > 0:
+            self.token = access_token
+            self.expire = r['expires_in'] + time.time() - 10
+
+    def refresh_token_if_expire(self):
+        if time.time() > self.expire:
+            self.refresh_token()
+
+    def send_text(self, text, to_uid='@all'):
+        self.refresh_token_if_expire()
+        r = requests.post(f"{self.server}cgi-bin/message/send", params={
+            'access_token': self.token
+        }, json={
+            "touser": to_uid,
+            "agentid": self.agent_id,
+            "msgtype": "text",
+            "text": {
+                "content": text
+            },
+            "duplicate_check_interval": 600
+        }).json()
+        return r
+
+    def send_image(self, image_file, to_uid='@all'):
+        self.refresh_token_if_expire()
+        r = requests.post(f"{self.server}cgi-bin/media/upload", params={
+            "access_token": self.token,
+            "type": "image"
+        }, files={
+            "picture": image_file
+        }).json()
+        if "media_id" in r:
+            media_id = r['media_id']
+        else:
+            raise Exception("failed to upload: " + str(r))
+
+        r = requests.post(f"{self.server}cgi-bin/message/send", params={
+            "access_token": self.token
+        }, json={
+            "touser": to_uid,
+            "agentid": self.agent_id,
+            "msgtype": "image",
+            "image": {
+                "media_id": media_id
+            },
+            "duplicate_check_interval": 600
+        }).json()
+        return r
+
+    def send_markdown(self, markdown, to_uid='@all'):
+        self.refresh_token_if_expire()
+        r = requests.post(f"{self.server}cgi-bin/message/send", params={
+            "access_token": self.token
+        }, json={
+            "touser": to_uid,
+            "agentid": self.agent_id,
+            "msgtype": "markdown",
+            "markdown": {
+                "content": markdown
+            },
+            "duplicate_check_interval": 600
+        }).json()
+        return r
+
+    def send_file(self, file, filename="文件", to_uid='@all'):
+        self.refresh_token_if_expire()
+        r = requests.post(f"{self.server}cgi-bin/media/upload", params={
+            "access_token": self.token,
+            "type": "file"
+        }, files={
+            "media": (filename, file)
+        }).json()
+
+        if "media_id" in r:
+            media_id = r['media_id']
+        else:
+            raise Exception("failed to upload: " + str(r))
+
+        r = requests.post(f"{self.server}cgi-bin/message/send", params={
+            "access_token": self.token
+        }, json={
+            "touser": to_uid,
+            "agentid": self.agent_id,
+            "msgtype": "file",
+            "file": {
+                "media_id": media_id
+            },
+            "duplicate_check_interval": 600
+        }).json()
+        return r