配置文件
注释里面有写,懒得用配置文件了
代码
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
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
|
# 京东云无线路由宝推送
import requests
import json
import time
import hmac
import hashlib
import base64
import urllib.parse
from datetime import datetime
# 发送钉钉消息
def sendinfo_ding(token,secret,data):
dic = get_timestamp_and_sign_by_secret(secret)
timestamp = dic[ 'timestamp' ]
sign = dic[ 'sign' ]
url = 'https://oapi.dingtalk.com/robot/send?access_token=%s×tamp=%s&sign=%s' % (token,timestamp,sign) #你的机器人webhook地址
headers = { 'Content-Type' : 'application/json' }
f = requests.post(url, data = json.dumps(data), headers = headers)
# 获取密签
def get_timestamp_and_sign_by_secret(secret):
timestamp = str ( round (time.time() * 1000 ))
secret_enc = secret.encode( 'utf-8' )
string_to_sign = '{}\n{}' . format (timestamp, secret)
string_to_sign_enc = string_to_sign.encode( 'utf-8' )
hmac_code = hmac.new(secret_enc, string_to_sign_enc, digestmod = hashlib.sha256).digest()
sign = urllib.parse.quote_plus(base64.b64encode(hmac_code))
return { "timestamp" :timestamp, "sign" :sign}
# 获取可用积分
def get_jd_total_avail_point(wskey):
url = 'https://router-app-api.jdcloud.com/v1/regions/cn-north-1/pinTotalAvailPoint'
headers = { 'Content-Type' : 'application/json' , 'wskey' :wskey}
rsp = requests.get(url,headers = headers)
data = (json.loads(rsp.text)[ 'result' ][ 'totalAvailPoint' ])
return { 'totalAvailPoint' :data}
# 获取积分详情
def get_jd_detail(wskey):
url = 'https://router-app-api.jdcloud.com/v1/regions/cn-north-1/todayPointDetail?sortField=today_point&sortDirection=DESC&pageSize=15¤tPage=1'
headers = { 'Content-Type' : 'application/json' , 'wskey' :wskey}
rsp = requests.get(url,headers = headers)
data = (json.loads(rsp.text)[ 'result' ])
items = data[ 'pointInfos' ]
total_today_point = 0
total_all_point = 0
dic = {}
dic[ 'todayDate' ] = data[ 'todayDate' ]
dic[ 'items' ] = []
for item in items:
mac = item[ 'mac' ]
today_point = item[ 'todayPointIncome' ]
all_point = item[ 'allPointIncome' ]
total_today_point + = int (today_point)
total_all_point + = int (all_point)
dic[ 'items' ].append(item)
dic[ 'total_today_point' ] = total_today_point
dic[ 'total_all_point' ] = total_all_point
return dic
# 发送京东路由宝日报
def send_jd_router(wskey):
dic = get_jd_detail(wskey)
msg = '# 京东路由宝日报\n'
msg + = '## %s \n' % (dic[ 'todayDate' ])
msg + = ( '> 今日获取总积分为**%d**分,对应金钱为 **%.2f**元\n' % (dic[ 'total_today_point' ], float (dic[ 'total_today_point' ]) / 100 ))
for item in dic[ 'items' ]:
msg + = ( '>> 设备**%s** \n' % (item[ 'mac' ]))
msg + = ( '>>> 今日获取积分为**%s**分,对应金钱为 **%.2f**元 \n\n' % (item[ 'todayPointIncome' ], float (item[ 'todayPointIncome' ]) / 100 ))
msg + = ( '> 累计总积分为 **%s** 分,对应金钱 **%.2f** 元 \n\n' % (dic[ 'total_all_point' ], float (dic[ 'total_all_point' ]) / 100 ))
dic = get_jd_total_avail_point(wskey)
msg + = ( '> 目前可用积分为 **%s** 分,对应金钱 **%.2f** 元' % (dic[ 'totalAvailPoint' ], float (dic[ 'totalAvailPoint' ]) / 100 ))
# token=""
# secret=""
# data = {
# "msgtype": "markdown",
# "markdown": {
# "title":"京东云路由宝日报",
# "text": msg
# },
# }
# sendinfo_ding(token,secret,data)
# 更改server酱发送
server_send_wechat( '京东路由宝日报' ,msg)
#Server酱发送微信
def server_send_wechat(text,desp = ''):
# Server酱 Key
key = ''
url = ( 'https://sc.ftqq.com/%s.send' % key)
params = {
'text' :text,
'desp' :desp
}
requests.get(url,params)
# 更改为配置文件获取 todo
wskey = ''
while True :
now = datetime.now()
print ( "任务执行的时间为:" + datetime.strftime(now, "%Y-%m-%d:%H:%M:%S" ))
if now.hour = = 8 and now.minute = = 0 :
send_jd_router(wskey)
time.sleep( 60 )
|
知识点扩展:Python编程之微信推送模板消息功能示例
本文实例讲述了Python微信推送模板消息功能。分享给大家供大家参考,具体如下:
具体代码如下:
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
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
|
#!/usr/bin/env python
#-*- coding: utf-8 -*-
import httplib
import json
import MySQLdb
#从数据库中获取access_token
access_token = ""
try :
conn = MySQLdb.connect(host = '192.168.1.1' ,user = 'root' ,passwd = 'root' ,db = 'db_weixin' ,port = 3306 )
cur = conn.cursor()
cur.execute( 'select access_token from weixin_public' )
result = cur.fetchone()
#print result
#print result[0]
access_token = result[ 0 ]
cur.close()
conn.close()
except MySQLdb.Error,e:
print "Mysql Error %d: %s" % (e.args[ 0 ], e.args[ 1 ])
#根据接口推送消息
if not access_token is None :
conn = httplib.HTTPConnection( "api.weixin.qq.com:80" ) #微信接口链接
headers = { "Content-type" : "application/json" } #application/x-www-form-urlencoded
params = ({ 'touser' : "oEGZ4johnKOtayJbnEVeuaZr6zQ0" , #用户openid
'template_id' : 'AtFuydv8k_15UGZuFntaBzJRCsHCkjNm1dcWD3A-11Y' , #模板消息ID
"topcolor" : "#667F00" ,#颜色
"data" : { #模板内容
"first" : { "value" : "尊敬的710.so : 您的网站http://www.zzvips.com (192.168.1.1) 有异常访问" , "color" : "#173177" },
"keyword1" : { "value" : "访问时间 2015-04-05 15:30:59 访问IP 192.168.1.2" , "color" : "#173177" },
"remark" : { "value" : "访问频率 10/s" , "color" : "#173177" }
}
}
)
conn.request( "POST" , "/cgi-bin/message/template/send?access_token=" + access_token, json.JSONEncoder().encode(params), headers) #推送消息请求
response = conn.getresponse()
data = response.read() #推送返回数据
if response.status = = 200 :
print 'success'
print data
else :
print 'fail'
conn.close()
|
希望本文所述对大家Python程序设计有所帮助。
以上就是Python 京东云无线宝消息推送的详细内容,更多关于Python 京东云无线宝推送的资料请关注服务器之家其它相关文章!
原文链接:https://www.cnblogs.com/Alex-Mercer/p/14429647.html