【Python小小白】京东云无线宝收入推送（server酱）源代码，寻大佬完善
本帖最后由 eoven8 于 2020-5-15 11:50 编辑1.先声明：本人Python纯小小白，只是闲每天看收入太麻烦，试着组装的这次脚本，参数都为写死代码中，请看注释修改即可
2.代码中登录部分是某大佬的作品，如有不合适请告知
3。第一次发帖，请见谅，第一次也不知道怎么发源码试试吧，大佬无聊可以完善下
import argparse
import os
import pickle
import random
import sys
import time
import json
import requests
import re
import logging
from bs4 import BeautifulSoup
from apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler.triggers.interval import IntervalTrigger
logging.basicConfig(
format="%(asctime)s - %(threadName)s - %(levelname)s - %(lineno)d - %(funcName)s - %(message)s",
level=logging.INFO
)
class JDSpider:
def __init__(self):
# init url related
self.home = 'https://passport.jd.com/new/login.aspx'
self.login = 'https://passport.jd.com/uc/loginService'
self.imag = 'https://authcode.jd.com/verify/image'
self.auth = 'https://passport.jd.com/uc/showAuthCode'
self.sess = requests.Session()
self.headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/75.0.3770.142 Safari/537.36',
'ContentType': 'text/html; charset=utf-8',
'Accept-Encoding': 'gzip, deflate, sdch',
'Accept-Language': 'zh-CN,zh;q=0.8',
'Connection': 'keep-alive',
}
self.cookies = {
}
self.eid = 'DHPVSQRUFPP6GFJ7WPOFDKYQUGQSREWJLJ5QJPDOSJ2BYF55IZHP5XX3K2BKW36H5IU3S4R6GPU7X3YOGRJGW7XCF4'
self.fp = 'b450f02af7d98727ef061e8806361c67'
#self.sb为京东云的MAC地址，多个自己存储，没做输入写死的 请修改为自己的
self.sb = ['DCD********','DCD********']
self.kw = {'source':'1','pageSize':1,'currentPage':1}
def checkLogin(self):
# 恢复之前保存的cookie
checkUrl = 'https://passport.jd.com/uc/qrCodeTicketValidation'
try:
print('+++++++++++++++++++++++++++++++++++++++++++++++++++++++')
print(f'{time.ctime()} > 自动登录中... ')
with open('cookie', 'rb') as f:
cookies = requests.utils.cookiejar_from_dict(pickle.load(f))
response = requests.get(checkUrl, cookies=cookies)
print(cookies)
print(response)
if response.status_code != requests.codes.OK:
print('登录过期, 请重新登录!')
return False
else:
print('登录成功!')
self.cookies.update(dict(cookies))
return True
except Exception as e:
logging.error(e)
return False
def login_by_QR(self):
# jd login by QR code
try:
print('+++++++++++++++++++++++++++++++++++++++++++++++++++++++')
print(f'{time.ctime()} > 请打开京东手机客户端，准备扫码登录:')
urls = (
'https://passport.jd.com/new/login.aspx',
'https://qr.m.jd.com/show',
'https://qr.m.jd.com/check',
'https://passport.jd.com/uc/qrCodeTicketValidation'
)
# step 1: open login page
response = self.sess.get(
urls,
headers=self.headers
)
if response.status_code != requests.codes.OK:
print(f"获取登录页失败:{response.status_code}")
return False
# update cookies
self.cookies.update(response.cookies)
# step 2: get QR image
response = self.sess.get(
urls,
headers=self.headers,
cookies=self.cookies,
params={
'appid': 133,
'size': 147,
't': int(time.time() * 1000),
}
)
if response.status_code != requests.codes.OK:
print(f"获取二维码失败:{response.status_code}")
return False
# update cookies
self.cookies.update(response.cookies)
# save QR code
image_file = 'qr.png'
with open(image_file, 'wb') as f:
for chunk in response.iter_content(chunk_size=1024):
f.write(chunk)
# scan QR code with phone
if os.name == "nt":
# for windows
os.system('start ' + image_file)
else:
if os.uname() == "Linux":
# for linux platform
os.system("eog " + image_file)
else:
# for Mac platform
os.system("open " + image_file)
# step 3: check scan result 京东上也是不断去发送check请求来判断是否扫码的
self.headers['Host'] = 'qr.m.jd.com'
self.headers['Referer'] = 'https://passport.jd.com/new/login.aspx'
# check if QR code scanned
qr_ticket = None
retry_times = 100# 尝试100次
while retry_times:
retry_times -= 1
response = self.sess.get(
urls,
headers=self.headers,
cookies=self.cookies,
params={
'callback': 'jQuery%d' % random.randint(1000000, 9999999),
'appid': 133,
'token': self.cookies['wlfstk_smdl'],
'_': int(time.time() * 1000)
}
)
if response.status_code != requests.codes.OK:
continue
rs = json.loads(re.search(r'{.*?}', response.text, re.S).group())
if rs['code'] == 200:
print(f"{rs['code']} : {rs['ticket']}")
qr_ticket = rs['ticket']
break
else:
print(f"{rs['code']} : {rs['msg']}")
time.sleep(3)
if not qr_ticket:
print("二维码登录失败")
return False
# step 4: validate scan result
# must have
self.headers['Host'] = 'passport.jd.com'
self.headers['Referer'] = 'https://passport.jd.com/new/login.aspx'
response = requests.get(
urls,
headers=self.headers,
cookies=self.cookies,
params={'t': qr_ticket},
)
if response.status_code != requests.codes.OK:
print(f"二维码登录校验失败:{response.status_code}")
return False
# 京东有时候会认为当前登录有危险，需要手动验证
# url: https://safe.jd.com/dangerousVerify/index.action?username=...
res = json.loads(response.text)
if not response.headers.get('p3p'):
if 'url' in res:
print(f"需要手动安全验证: {res['url']}")
return False
else:
print(res)
print('登录失败!!')
return False
# login succeed
print(response)
print(response.cookies)
self.headers['P3P'] = response.headers.get('P3P')
self.cookies.update(response.cookies)
# 保存cookie
with open('cookie', 'wb') as f:
pickle.dump(self.cookies, f)
print("登录成功")
return True
except Exception as e:
print(e)
raise
#-----------------分割线以上登录代码抄自大佬并删减不需要，地址：https://github.com/zstu-lly/JD_MASK_Robot------------------------------------------
def runReq(self):
for mac in self.sb:
self.kw['mac'] = mac
print(self.kw)
# params 接收一个字典或者字符串的查询参数，字典类型自动转换为url编码，不需要urlencode()
response = requests.get(
"https://router-app-api.jdcloud.com/v1/regions/cn-north-1/pointOperateRecords:show?",
headers = self.headers,
cookies=self.cookies,
params = self.kw
)
# 查看响应内容，response.text 返回的是Unicode格式的数据
jsonbk = json.loads(response.text)
print(jsonbk)
if jsonbk['code']!= 200:
print ("登录已失效")
else:
title = '（'+self.kw['mac']+'）今日收入:'
print (title)
score =str(jsonbk['result']['pointRecords']['pointAmount'])
print (score)
#以下请求为server酱的微信推送请求Key 自己去申请才能推送给自己wx
requests.get("https://sc.ftqq.com/SCU********************************************.send?text="+title+score+'&desp='+score)
print ("推送完毕")
# self.dojob()
def dojob(self):
#创建调度器：BlockingScheduler
scheduler = BlockingScheduler()
# 添加任务并设置触发方式为2小时执行一次
trigger = IntervalTrigger(seconds=7200)
scheduler.add_job(self.checkLogin, trigger)
# 每天6:00执行
scheduler.add_job(self.runReq, 'cron', hour=7, minute=45)
#添加任务,时间间隔5S
scheduler.start()
if __name__ == '__main__':
# help message
parser = argparse.ArgumentParser(description='Simulate to login Jing Dong, and buy sepecified good')
parser.add_argument('-a', '--area',
help='Area string, like: 1_72_2799_0 for Beijing', default='')
parser.add_argument('-g', '--good',
help='Jing Dong good ID', default='')
parser.add_argument('-c', '--count', type=int,
help='The count to buy', default=1)
parser.add_argument('-w', '--wait',
type=int, default=1000,
help='Flush time interval, unit MS')
parser.add_argument('-f', '--flush',
action='store_true',
help='Continue flash if good out of stock',
default=True)
parser.add_argument('-s', '--submit',
action='store_true',
help='Submit the order to Jing Dong',
default=True)
parser.add_argument('-t', '--timer',
type=str,
help='Set time to start monitoring. e.g. \"23:59:58\" , if no setting, start immediately',
default="")
options = parser.parse_args()
# for test
options.count = 1
options.good = '100010617232'
options.area = '15_1290_22049_22142'
options.timer = ""
spider = JDSpider()
if not spider.checkLogin():
if not spider.login_by_QR():
sys.exit(-1)
# spider.runReq()
spider.dojob() 用模拟器抓包，获取到wskey即可。
下面代码是我推送到钉钉的，有需求的朋友更换成server酱
# 京东云无线路由宝推送
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和sercret
token=""
secret=""
data = {
"msgtype": "markdown",
"markdown": {
"title":"京东云路由宝日报",
"text": msg
},
}
sendinfo_ding(token,secret,data)
# 抓包获取wskey
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)
Alex.Merceryj 发表于 2021-2-22 10:11
将 sendinfo_ding(token,secret,data) 注释，替换一个server酱的推送代码即可
还有 大佬 您的代码 如果直接用DD推送 会有个报错
Traceback (most recent call last):
File "C:\Users\QQQ\Desktop\JDCPointPush.py", line 81, in <module>
send_jd_router(wskey)
File "C:\Users\QQQQ\Desktop\JDCPointPush.py", line 59, in send_jd_router
dic=get_jd_total_avail_point(wskey)
NameError: name 'get_jd_total_avail_point' is not defined
很奇怪 完全看不懂的路过 看不懂，但是看起来很高级的样子 纯干货，支持一下。 支持一下!!!!!!!! 看不懂！！！ 这个 干哈的？ 这东西收益到底怎么样啊？ 利用保存的cookie直接调用runReq方法，加入云函数任务定时触发，可以省下服务器持续运行的资源消耗 Conastin 发表于 2020-7-26 20:26
利用保存的cookie直接调用runReq方法，加入云函数任务定时触发，可以省下服务器持续运行的资源消耗
纯小白一枚，还有这个cookie会失效保持不住咋处理