最近在对接管家婆，文档上只有 java php .net 的例子，写了一个python的例子，里面部分数据按需填写。
加解密代码借鉴于知乎
python里面json对象转字符串，分号和逗号默认会有空格，会导致加密和签名不通过
2020-12-16 增加部分接口调用代码
import base64
import copy
import hashlib
import requests
import time
import json
from Crypto.Cipher import AES
class GuanjiapoOpenApiRequest:
class GuanjiapoOpenApiException(Exception):
pass
DEFAULT_APP_KEY = {
'app_key': '',
}
DEFAULT_SIGN_KEY = {
'sign_key': '',
}
DEFAULT_KEY = {
'key': '',
}
DEFAULT_IV = {
'iv': '',
}
GUANJIAPO_OPEN_API_URL = {
'get_authcode': 'http://apigateway.wsgjp.com.cn/api/login',
'get_token': 'http://apigateway.wsgjp.com.cn/api/token',
}
def __init__(self, key, iv):
self.key = key.encode('utf-8')
self.iv = iv.encode('utf-8')
def pkcs7padding(self, text):
"""明文使用PKCS7填充 """
bs = 16
length = len(text)
bytes_length = len(text.encode('utf-8'))
padding_size = length if (bytes_length == length) else bytes_length
padding = bs - padding_size % bs
padding_text = chr(padding) * padding
return text + padding_text
def aes_encrypt(self, content):
""" AES加密 """
cipher = AES.new(self.key, AES.MODE_CBC, self.iv)
# 处理明文
content_padding = self.pkcs7padding(content)
# 加密
encrypt_bytes = cipher.encrypt(content_padding.encode('utf-8'))
# 重新编码
result = str(base64.b64encode(encrypt_bytes), encoding='utf-8')
return result
def aes_decrypt(self, content):
"""AES解密 """
cipher = AES.new(self.key, AES.MODE_CBC, self.iv)
content = base64.b64decode(content)
text = cipher.decrypt(content).decode('utf-8')
return text
@classmethod
def str_md5(cls, data):
"""md5加密"""
m = hashlib.md5(data.encode('utf-8'))
return (m.hexdigest().upper())
@classmethod
def get_authcode(cls):
"""通过接口获取授权认证码"""
url = cls.GUANJIAPO_OPEN_API_URL['get_authcode']
app_key = GuanjiapoOpenApiRequest.DEFAULT_APP_KEY['app_key']
key = GuanjiapoOpenApiRequest.DEFAULT_KEY['key']
iv = GuanjiapoOpenApiRequest.DEFAULT_IV['iv']
ts = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
p_json = {
'CompanyName': '公司名',
'UserId': '用户名',
'Password': '密码',
'TimeStamp': ts,
}
aes_data = GuanjiapoOpenApiRequest(key=key, iv=iv)
p = aes_data.aes_encrypt(json.dumps(p_json, separators=(',', ':'), ensure_ascii=False))
sign_key = cls.DEFAULT_SIGN_KEY['sign_key']
sign_json = {
"appkey": app_key,
"p": p,
"signkey": sign_key,
}
sha256 = hashlib.sha256()
sha256.update(json.dumps(sign_json, separators=(',', ':'), ensure_ascii=False).encode('utf-8'))
sign_key = sha256.hexdigest()
post_data = {
'appkey': app_key,
'p': p,
'sign': sign_key,
}
res = requests.post(url, data=post_data)
return res
@classmethod
def get_token(cls, p_json=None):
"""根据 authcode 获取token"""
url = GuanjiapoOpenApiRequest.GUANJIAPO_OPEN_API_URL['get_token']
app_key = GuanjiapoOpenApiRequest.DEFAULT_APP_KEY['app_key']
key = GuanjiapoOpenApiRequest.DEFAULT_KEY['key']
iv = GuanjiapoOpenApiRequest.DEFAULT_IV['iv']
ts = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
res = GuanjiapoOpenApiRequest.get_authcode()
res_json = res.json()
auth_code = res_json['response']['authcode']
p_json = p_json or {
'TimeStamp': ts,
'GrantType': 'auth_token',
'AuthParam': auth_code,
}
aes_data = GuanjiapoOpenApiRequest(key=key, iv=iv)
p = aes_data.aes_encrypt(json.dumps(p_json, separators=(',', ':'), ensure_ascii=False))
sign_key = cls.DEFAULT_SIGN_KEY['sign_key']
sign_json = {
"appkey": app_key,
"p": p,
"signkey": sign_key,
}
sha256 = hashlib.sha256()
sha256.update(json.dumps(sign_json, separators=(',', ':'), ensure_ascii=False).encode('utf-8'))
sign_key = sha256.hexdigest()
post_data = {
'appkey': app_key,
'p': p,
'sign': sign_key,
}
res = requests.post(url, data=post_data)
res_json = res.json()
response = res_json['response']['response']
aes_data = GuanjiapoOpenApiRequest(key=key, iv=iv)
decrypt_str = aes_data.aes_decrypt(response)
decrypt_str = re.compile('[\\x00-\\x08\\x0b-\\x0c\\x0e-\\x1f\n\r\t]').sub('', decrypt_str)
res = json.loads(decrypt_str)
# return res['auth_token'] # 因为 refresh_token() 还需要调用，所以不能直接返回 token
return res
@classmethod
def refresh_token(cls):
"""根据 refresh_token 刷新token"""
res = GuanjiapoOpenApiRequest.get_token()
refresh_token = res['refresh_token']
ts = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
p_json = {
'TimeStamp': ts,
'GrantType': 'refresh_token',
'AuthParam': refresh_token,
}
ress = GuanjiapoOpenApiRequest.get_token(p_json)
return ress
@classmethod
def request_wrapper(cls, method, data) -> requests.Response:
"""请求封装"""
url = cls.GUANJIAPO_OPEN_API_URL['baseUrl']
app_key = cls.DEFAULT_APP_KEY['app_key']
sign_key = cls.DEFAULT_SIGN_KEY['sign_key']
ts = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
token = GuanjiapoOpenApiRequest.get_token()['auth_token']
add_list = []
sign_json = copy.deepcopy(data)
sign_json.update({
'method': method,
'appkey': app_key,
'timestamp': ts,
'token': token,
})
key_sort_list = [key for key in sign_json]
key_sort_list.sort()
for key in key_sort_list:
value = sign_json[key]
if value or value == 0: # 去掉空字符串、[]、{}
add_list.append(key + str(value))
sign_str = ''.join(add_list) + sign_key
sign = GuanjiapoOpenApiRequest.str_md5(sign_str)
json_data = copy.deepcopy(data)
json_data.update({
'appkey': app_key,
'method': method,
'timestamp': ts,
'token': token,
'sign': sign,
})
res = requests.post(url=url, data=json_data, verify=False)
return res
@classmethod
def beefun_selfbuiltmall_queryproductinfo(cls):
"""商品详情查询"""
method = 'beefun.selfbuiltmall.queryproductinfo'
# method = 'zyx.selfbuiltmall.queryproductinfo'
data = {
'pagesize': 50,
'pageno': 1,
'ptypeids': [],
'pricetype': 0,
'begintime': '2020-01-01 00:00:00',
'endtime': '2021-01-01 00:00:00',
}
res = GuanjiapoOpenApiRequest.request_wrapper(method=method, data=data)
return res
@classmethod
def zyx_selfbuiltmall_addsaleorder(cls):
method = 'zyx.selfbuiltmall.addsaleorder'
ts = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
# ts = "2020-10-10 16:57:22"
data = {
"shopkey": "3d37a7aa-ea05-4340-9291-1c09a960127c",
"tradeid": "DD00000000002",
"payno": "1234567890",
"tradestatus": "0",
"buyermessage": "",
"sellermemo": "bbeizhu",
"tradecreatetime": ts,
"tradepaiedtime": ts,
"tradefinishtime": ts,
"trademodifiedtime": ts,
"steptradedeliverytime": ts,
"tradetype": "0",
"shippingtype": "1",
"steptradestatus": "0",
"refundstatus": "0",
"invoicetitle": "lansemomoda",
"freightcode": "SF",
"freightbillno": "DH001",
"sellerflag": "0",
"tradetotal": "176",
"total": "100",
"paiedtotal": "0",
"preferentialtotal": "77",
"mallfee": "0",
"customerfreightfee": "9",
"buyerpayment": "0",
"expressagencyfee": "0",
"taxtotal": "0",
"taxamount": "0",
"taxfee": "0",
"orderdetails": '[{"BarCode":"barcode","comboId":0,"IsGift":false,"Oid":"DD00000000001001","OutId":"SJBM_111","PicUrl":null,"PlatformPropertiesName":"cccccccc","preferentialtotal":50.0,"Price":70.889,"ProductName":"测试商品","PtypeId":"110890-9001","Qty":2.0,"refundQty":0.0,"RefundStatus":0,"refundTotal":0.0,"SkuId":"110890-9001","TradeOriginalPrice":120.0},'
'{"BarCode":"barcode2","comboId":0,"IsGift":false,"Oid":"DD00000000001002","OutId":"SJBM_222","PicUrl":null,"PlatformPropertiesName":"qqqqq","preferentialtotal":28.0,"Price":30.0,"ProductName":"测试商品","PtypeId":"987654321001","Qty":2.0,"refundQty":0.0,"RefundStatus":0,"refundTotal":0.0,"SkuId":"1111111111111111111111111111111112","TradeOriginalPrice":58.0}]',
"eshopbuyer": '{"CustomerAllAddress":null,"CustomerEmail":null,"CustomerPayAccount":null,"CustomerReceiver":"苍茫北望","CustomerReceiverAddress":"ruanjianyuan","CustomerReceiverCity":"chengdu","CustomerReceiverCountry":"china","CustomerReceiverDistrict":"gaoxin","CustomerReceiverMobile":"12345678911","CustomerReceiverPhone":"12345678","CustomerReceiverProvince":"sichuan","CustomerReceiverZipcode":"123456","CustomerShopAccount":"zzzz","IdCard":null,"Profileid":0}',
"method": "zyx.selfbuiltmall.addsaleorder",
"appkey": "68943923115886070418838901844741",
"timestamp": ts,
"token": "nF1WXGcp27iiTJwC2P2GuysuJuGmtCE2FGBH7K1E"
}
res = GuanjiapoOpenApiRequest.request_wrapper(method=method, data=data)
return res
if __name__ == '__main__':
r = GuanjiapoOpenApiRequest.zyx_selfbuiltmall_addsaleorder()
res_json = r.json()
print(json.dumps(res_json, ensure_ascii=False, indent=4))
版权声明：本文为yt_xy原创文章，遵循CC 4.0 BY-SA版权协议，转载请附上原文出处链接和本声明。