First commit
Former-commit-id: 424079609133edcc501ae185509836ee1181a02c
This commit is contained in:
0
change_font/util/__init__.py
Normal file
0
change_font/util/__init__.py
Normal file
36
change_font/util/get_ws_sign.py
Normal file
36
change_font/util/get_ws_sign.py
Normal file
@ -0,0 +1,36 @@
|
||||
from util.sign_client import get_signature
|
||||
from util.url import url_format,encode
|
||||
|
||||
|
||||
def get_sign(
|
||||
access_key_id,
|
||||
access_key_secret,
|
||||
timestamp,
|
||||
url,
|
||||
url_params):
|
||||
if access_key_id is None or len(access_key_id) == 0:
|
||||
raise RuntimeError('参数access_key_id不能为空')
|
||||
if access_key_secret is None or len(access_key_secret) == 0:
|
||||
raise RuntimeError('参数access_key_secret不能为空')
|
||||
if timestamp is None or len(timestamp) == 0:
|
||||
raise RuntimeError('参数timestamp不能为空')
|
||||
if url is None or len(url) == 0:
|
||||
raise RuntimeError('参数url不能为空')
|
||||
if url_params is None:
|
||||
raise RuntimeError('参数url_params不能为空')
|
||||
|
||||
url_params['access_key_id'] = access_key_id
|
||||
url_params['timestamp'] = timestamp
|
||||
|
||||
signature, signature_nonce = get_signature(
|
||||
url_params,
|
||||
None,
|
||||
'GET',
|
||||
'application/application_json',
|
||||
access_key_secret)
|
||||
|
||||
url_params["signature_nonce"] = signature_nonce
|
||||
url_params["signature"] = encode(signature)
|
||||
|
||||
url = url + '?' + url_format(url_params)
|
||||
return url
|
63
change_font/util/http.py
Normal file
63
change_font/util/http.py
Normal file
@ -0,0 +1,63 @@
|
||||
import requests
|
||||
|
||||
application_json = 'application/json'
|
||||
application_x_www_form_urlencoded = 'application/x-www-form-urlencoded'
|
||||
multipart_formdata = 'multipart/form-data'
|
||||
multipart_encoder = 'multipart_encoder'
|
||||
binary = 'binary'
|
||||
|
||||
|
||||
def send_delete(url, headers):
|
||||
response = requests.delete(url, headers=headers)
|
||||
return response.content.decode("utf-8")
|
||||
|
||||
|
||||
def send_get(url, params, headers):
|
||||
response = requests.get(url=url, params=params, headers=headers)
|
||||
return response.content.decode("utf-8")
|
||||
|
||||
|
||||
def send_post(url, request_body, headers):
|
||||
content_type = headers['content-type']
|
||||
if application_json == content_type:
|
||||
response = requests.post(url, json=request_body, headers=headers)
|
||||
elif multipart_formdata == content_type:
|
||||
data = get_multipart_data(request_body,headers)
|
||||
response = requests.post(url, data=data, headers=headers)
|
||||
else:
|
||||
response = requests.post(url, data=request_body, headers=headers)
|
||||
|
||||
return response.content.decode("utf-8")
|
||||
|
||||
|
||||
def send_put(url, request_body, headers):
|
||||
content_type = headers['content-type']
|
||||
if application_json == content_type:
|
||||
response = requests.put(url, json=request_body, headers=headers)
|
||||
elif multipart_formdata == content_type:
|
||||
data = get_multipart_data(request_body,headers)
|
||||
response = requests.put(url, data=data, headers=headers)
|
||||
else:
|
||||
response = requests.put(url, data=request_body, headers=headers)
|
||||
|
||||
return response.content.decode("utf-8")
|
||||
|
||||
|
||||
def send_patch(url, request_body, headers):
|
||||
content_type = headers['content-type']
|
||||
if application_json == content_type:
|
||||
response = requests.patch(url, json=request_body, headers=headers)
|
||||
elif multipart_formdata == content_type:
|
||||
data = get_multipart_data(request_body,headers)
|
||||
response = requests.patch(url, data=data, headers=headers)
|
||||
else:
|
||||
response = requests.patch(url, data=request_body, headers=headers)
|
||||
|
||||
return response.content.decode("utf-8")
|
||||
|
||||
|
||||
def get_multipart_data(request_body,headers):
|
||||
data = request_body['multipart_encoder']
|
||||
headers['content-type'] = data.content_type
|
||||
|
||||
return data
|
66
change_font/util/send_sign_http.py
Normal file
66
change_font/util/send_sign_http.py
Normal file
@ -0,0 +1,66 @@
|
||||
from util.http import send_patch,send_post,send_delete,send_get,send_put
|
||||
from util.sign_client import get_signature
|
||||
from util.url import encode,url_format
|
||||
|
||||
|
||||
def send_request(
|
||||
access_key_id,
|
||||
access_key_secret,
|
||||
timestamp,
|
||||
url,
|
||||
url_params,
|
||||
body_params,
|
||||
request_method,
|
||||
content_type
|
||||
) -> str:
|
||||
if access_key_id is None or len(access_key_id) == 0:
|
||||
raise RuntimeError('参数access_key_id不能为空')
|
||||
if access_key_secret is None or len(access_key_secret) == 0:
|
||||
raise RuntimeError('参数access_key_secret不能为空')
|
||||
if timestamp is None or len(timestamp) == 0:
|
||||
raise RuntimeError('参数timestamp不能为空')
|
||||
if url is None or len(url) == 0:
|
||||
raise RuntimeError('参数url不能为空')
|
||||
if url_params is None:
|
||||
raise RuntimeError('参数url_params不能为空')
|
||||
if body_params is None:
|
||||
raise RuntimeError('参数body_params不能为空')
|
||||
if request_method is None or len(request_method) == 0:
|
||||
raise RuntimeError('参数request_method不能为空')
|
||||
if content_type is None or len(content_type) == 0:
|
||||
raise RuntimeError('参数content_type不能为空')
|
||||
|
||||
url_params['access_key_id'] = access_key_id
|
||||
url_params['timestamp'] = timestamp
|
||||
|
||||
signature, signature_nonce = get_signature(
|
||||
url_params,
|
||||
body_params,
|
||||
request_method,
|
||||
content_type,
|
||||
access_key_secret)
|
||||
|
||||
url_params['signature'] = encode(signature)
|
||||
url_params['signature_nonce'] = signature_nonce
|
||||
url_params['timestamp'] = timestamp
|
||||
|
||||
url = url + '?' + url_format(url_params)
|
||||
|
||||
headers = {
|
||||
'content-type': content_type
|
||||
}
|
||||
|
||||
result = None
|
||||
if request_method == 'POST':
|
||||
result = send_post(url, request_body=body_params, headers=headers)
|
||||
elif request_method == 'PATCH':
|
||||
result = send_patch(url, request_body=body_params, headers=headers)
|
||||
elif request_method == 'PUT':
|
||||
result = send_put(url, request_body=body_params, headers=headers)
|
||||
elif request_method == 'GET':
|
||||
result = send_get(url, params=None, headers=headers)
|
||||
elif request_method == 'DELETE':
|
||||
result = send_delete(url, headers=headers)
|
||||
else:
|
||||
raise RuntimeError('支持[GET、POST、PUT、PATCH、DELETE]请求方式')
|
||||
return result
|
53
change_font/util/sign_client.py
Normal file
53
change_font/util/sign_client.py
Normal file
@ -0,0 +1,53 @@
|
||||
#!/usr/bin/python
|
||||
# -*- coding:utf-8 -*-
|
||||
|
||||
import base64
|
||||
import hmac
|
||||
from hashlib import sha1
|
||||
import uuid
|
||||
from urllib import parse
|
||||
import json
|
||||
from util.url import url_format_list
|
||||
from util.http import application_x_www_form_urlencoded,application_json
|
||||
|
||||
__request_body = "request_body"
|
||||
|
||||
|
||||
def __generate_signature(parameters, access_key_secret):
|
||||
sorted_parameters = sorted(parameters.items(), key=lambda parameters : parameters[0])
|
||||
string_to_sign = url_format_list(sorted_parameters)
|
||||
secret = access_key_secret + "&"
|
||||
|
||||
# print(secret)
|
||||
h = hmac.new(secret.encode('utf-8'), string_to_sign.encode('utf-8'), sha1)
|
||||
signature = base64.b64encode(h.digest()).strip()
|
||||
signature = str(signature, encoding="utf8")
|
||||
# signature = bytes(signature, encoding="utf8")
|
||||
return signature
|
||||
|
||||
|
||||
def get_signature(url_params, body_params, request_method, content_type, access_key_secret):
|
||||
signature_nonce = str(uuid.uuid1())
|
||||
|
||||
sign_param = {
|
||||
'signature_nonce': signature_nonce
|
||||
}
|
||||
|
||||
if (content_type == application_x_www_form_urlencoded or content_type == application_json) \
|
||||
and (request_method == 'POST' or request_method == 'PATCH' or request_method == 'PUT')\
|
||||
and body_params is not None and len(body_params) != 0:
|
||||
if content_type == application_x_www_form_urlencoded:
|
||||
sign_param[__request_body] = parse.urlencode(body_params)
|
||||
else:
|
||||
sign_param[__request_body] = json.dumps(body_params)
|
||||
# 生成签名使用Python,发送签名使用其他语言时使用separators
|
||||
# sign_param[__request_body] = json.dumps(body_params,separators=(',',':'))
|
||||
|
||||
for key in url_params.keys():
|
||||
sign_param[key] = url_params[key]
|
||||
|
||||
signature = __generate_signature(sign_param, access_key_secret)
|
||||
return signature, signature_nonce
|
||||
|
||||
|
||||
|
24
change_font/util/url.py
Normal file
24
change_font/util/url.py
Normal file
@ -0,0 +1,24 @@
|
||||
from urllib.parse import quote
|
||||
|
||||
|
||||
def encode(val):
|
||||
val = quote(val, 'utf-8')
|
||||
return val
|
||||
|
||||
|
||||
def url_format_list(parameters):
|
||||
param_list = []
|
||||
for (k, v) in parameters:
|
||||
param_str = '{}={}'.format(k, v)
|
||||
param_list.append(param_str)
|
||||
string_to_sign = '&'.join(param_list)
|
||||
return string_to_sign
|
||||
|
||||
|
||||
def url_format(parameters):
|
||||
param_list = []
|
||||
for key, value in parameters.items():
|
||||
param_str = '{}={}'.format(key, value)
|
||||
param_list.append(param_str)
|
||||
string_to_sign = '&'.join(param_list)
|
||||
return string_to_sign
|
Reference in New Issue
Block a user