代码示例

以下代码是基于普通API签名方式实现的签名验证Python3代码,可直接作为SDK使用

import datetime
import hashlib
import json
import time
import urllib
import urllib.parse
import urllib.request

import requests
import urllib3


urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

HEADERS = {
    "Accept": "*/*",
    "Connection": "keep-alive",
    "Accept-Encoding": "gzip, deflate",
    "Content-Type": "application/json",
    'Bbx-Dev': 'API'
}


def http_get_request(url, params, headers=None):
    param_data = urllib.parse.urlencode(params)

    response = None
    try:
        response = requests.get(url, param_data, headers=headers, timeout=5, verify=False)
        if response.status_code == 200:
            return response.json()
        else:
            return None
    except BaseException as e:
        if response:
            print("BBX http get failed, detail: %s,%s" % (response.text, e))
        else:
            print("BBX http get failed, detail: %s" % e)
        return None


def http_post_request(url, params, headers=None):
    post_data = json.dumps(params)

    response = None
    try:
        response = requests.post(url, post_data, headers=headers, timeout=10, verify=False)
        if response.status_code == 200:
            return response.json()
        else:
            return None
    except BaseException as e:
        if response:
            print("BBX http post failed, detail: %s,%s" % (response.text, e))
        else:
            print("BBX http post failed, detail: %s" % e)
        return None


def api_key_get(url, access_key, secret_key, params=''):
    param_data = urllib.parse.urlencode(params)

    _headers = HEADERS.copy()

    ts = str(int(time.time() * 1000 * 1000))    # 当前UTC时间的时间戳,单位微秒

    bbx_sign = secret_key + ts
    _headers["Bbx-Sign"] = get_md5_value(_str=bbx_sign)
    _headers['Bbx-Accesskey'] = access_key
    _headers['Bbx-Ts'] = ts

    response = None
    try:
        response = requests.get(url, param_data, headers=_headers, timeout=5, verify=False)
        if response.status_code == 200:
            return response.json()
        else:
            return None
    except BaseException as e:
        if response:
            print("BBX API Key http get failed, detail: %s,%s" % (response.text, e))
        else:
            print("BBX API Key http get failed, detail: %s" % e)
        return None


def api_key_post(url, access_key, secret_key, params):
    params['nonce'] = int(time.time())
    post_data = json.dumps(params)

    _headers = HEADERS.copy()

    ts = str(int(time.time() * 1000 * 1000))    # 当前UTC时间的时间戳,单位微秒

    bbx_sign = json.dumps(params) + secret_key + ts
    _headers["Bbx-Sign"] = get_md5_value(_str=bbx_sign)
    _headers['Bbx-Accesskey'] = access_key
    _headers['Bbx-Ts'] = ts

    response = None
    try:
        response = requests.post(url, post_data, headers=_headers, timeout=30, verify=False)
        if response.status_code == 200:
            return response.json()
        else:
            return None
    except BaseException as e:
        if response:
            print("BBX API Key http post failed, detail: %s,%s" % (response.text, e))
        else:
            print("BBX API Key http post failed, detail: %s" % e)
        return None


def get_md5_value(_str):
    file_md5 = hashlib.md5()
    file_md5.update(_str.encode(encoding="utf-8"))
    md5value = file_md5.hexdigest()
    return md5value


def fix_time(tm):
    if not tm:
        return ''
    split_parts = tm.split('.')
    k = split_parts[1]

    if len(k) > 7:
        d = k[:6] + 'Z'
        s = str.format('%s.%s' % (split_parts[0], d))
        return s
    else:
        return tm


def parse_time(str_tm: str) -> datetime.datetime:
    fixed_str_tm = fix_time(str_tm)
    return datetime.datetime.strptime(fixed_str_tm, '%Y-%m-%dT%H:%M:%S.%fZ').replace(tzinfo=datetime.timezone.utc)


def check_response(response) -> bool:
    if not response or not isinstance(response, dict):
        return False

    if response.get('errno', '').lower() != 'ok':
        return False
    if response.get('message', '').lower() != 'success':
        return False
    return True

results matching ""

    No results matching ""