配置文件
注释里面有写,懒得用配置文件了
代码
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