class
DeJian:
def
__init__(
self
, phone):
self
.phone
=
phone
self
.logger
=
self
.init_log()
def
init_log(
self
):
logger
=
logging.getLogger(
'DeJian'
)
logger.setLevel(logging.DEBUG)
ch
=
logging.StreamHandler()
ch.setLevel(logging.DEBUG)
formatter
=
logging.Formatter(
'%(message)s'
)
ch.setFormatter(formatter)
logger.addHandler(ch)
return
logger
def
req(
self
, method, url,
*
*
kwargs):
if
kwargs.get(
"headers"
):
kwargs[
"headers"
].update
=
{
"content-type"
:
"application/x-www-form-urlencoded"
,
"Host"
:
"dj.palmestore.com"
,
"user-agent"
:
"Dalvik/2.1.0 (Linux; U; Android 8.1.0; Nexus 5X Build/OPM1.171019.011)"
}
else
:
kwargs[
"headers"
]
=
{
"content-type"
:
"application/x-www-form-urlencoded"
,
"Host"
:
"dj.palmestore.com"
,
"user-agent"
:
"Dalvik/2.1.0 (Linux; U; Android 8.1.0; Nexus 5X Build/OPM1.171019.011)"
}
for
k, v
in
kwargs[
"data"
].items():
kwargs[
"data"
][k]
=
self
.url_encode(v)
kwargs[
"data"
]
=
self
.get_sorted_param_str(kwargs[
"data"
])
self
.logger.debug(f
"请求的参数为{method},url为{url}, 其他参数为{kwargs}"
)
r
=
requests.request(method
=
method, url
=
url,
*
*
kwargs)
self
.logger.info(f
"响应内容为{r.json()}"
)
return
r.json()
def
url_encode(
self
, s):
r
=
[
'+'
,
'/'
,
'='
]
res
=
s
for
i
in
r:
res
=
res.replace(i,
'%'
+
hex
(
ord
(i))[
2
:].upper())
return
res
def
sign(
self
, content):
private_key
=
b""
pri_key
=
RSA.importKey(private_key)
signer
=
PKCS1_v1_5.new(pri_key)
hash_obj
=
SHA1.new(content.encode())
sig1
=
signer.sign(hash_obj)
signature
=
base64.b64encode(sig1).decode()
res
=
self
.url_encode(signature)
return
res
def
get_sorted_param_str(
self
, dic):
content
=
'
'.join([f'
{k}
=
{dic[k]}&'
for
k
in
sorted
(dic)])[:
-
1
]
return
content
def
sendSms(
self
):
url
=
''
dic
=
{
"versionId"
:
"20005056"
,
"device"
:
"Nexus 5X"
,
"flag"
:
"1"
,
"imei"
: "",
"phone"
:
self
.phone,
"times"
:
"1"
,
"sendType"
:
"0"
,
"channelId"
:
"1240202"
,
"timestamp"
:
"1669439016670"
,
}
dic[
"sign"
]
=
self
.sign(
self
.get_sorted_param_str(dic))
r
=
self
.req(
'post'
, url, data
=
dic)