-
Notifications
You must be signed in to change notification settings - Fork 1
/
wxpush.py
55 lines (49 loc) · 1.76 KB
/
wxpush.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
import json
import xmltodict
import config
import requests
from nonebot.log import logger
def qywx(msg, config):
get_token_url = f"https://qyapi.weixin.qq.com/cgi-bin/gettoken?corpid={config['wx_cid']}&corpsecret={config['wx_secret']}"
response = requests.get(get_token_url).content
if json.loads(response).get('errmsg') != 'ok':
return False
access_token = json.loads(response).get('access_token')
if access_token and len(access_token) > 0:
send_msg_url = f'https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token={access_token}'
data = {
"touser": '@all',
"agentid": config['wx_aid'],
"msgtype": "text",
"text": {
"content": msg
},
"duplicate_check_interval": 600
}
r = requests.post(send_msg_url, data=json.dumps(data))
result = json.loads(r.text)
if result['errcode'] == 0:
logger.info('企业微信消息发送成功')
else:
logger.info('企业微信消息发送失败' + str(result))
def pushplus(msg, token):
pushplus_url = 'http://pushplus.hxtrip.com/send'
data = {
'token': token,
'title': 'QQ消息',
'content': msg,
'template': 'html'
}
r = requests.post(url=pushplus_url, data=data)
result = xmltodict.parse(r.text)
if result['ResultT']['code'] == '200':
logger.info('pushplus消息发送成功')
else:
logger.info('pushplus消息发送失败' + str(result))
def main(msg):
pushplus_token = config.push_way['pushplus_token']
qywx_config = config.push_way['qywx_config']
if pushplus_token != '':
pushplus(msg, pushplus_token)
if qywx_config != '':
qywx(msg, qywx_config)