首先提供一版可用的Python版本的OpenAPI调用示例,使用前需要先替换使用自己的AK/SK,并把ctecs-global.ctapi.ctyun.cn前面加上https和冒号以及双斜线(发表文章禁止携带):
# -*- coding: utf8 -*-
import urllib
import datetime
import requests
import json
import hashlib
import base64
import hmac
import datetime
import uuid
import os
import sys
if sys.version_info.major == 2:
from urllib import quote
reload(sys)
sys.setdefaultencoding('utf8')
VERSION = '2'
else:
VERSION = '3'
from urllib.parse import quote
from importlib import reload
from urllib3 import encode_multipart_formdata
from collections import OrderedDict
reload(sys)
METHOD_GET = 'GET'
METHOD_POST = 'POST'
file = False
# 官网accessKey
ak = '替换成您自己的AK'
# 官网securityKey
sk = '替换成您自己的SK'
def hmac_sha256(secret, data):
if type(secret) == bytes:
secret = bytearray(secret)
else:
secret = bytearray(secret, 'utf8')
data = bytearray(data, 'utf8')
return hmac.new(secret, data, digestmod=hashlib.sha256).digest()
def base64_of_hmac(data):
return base64.b64encode(data)
def get_request_uuid():
return str(uuid.uuid1())
def get_sorted_str(data, method):
"""
鉴权用的参数整理
:param data: dict 需要整理的参数
:return: str
"""
sorted_data = sorted(data.items(), key=lambda item: item[0])
str_list = map(lambda x_y: '%s=%s' % (x_y[0], x_y[1]), sorted_data)
return '&'.join(str_list)
def build_sign(query_params, body_params, eop_date, request_uuid, method):
"""
计算鉴权字段
:param query_params: dict get请求中的参数
:param body_params: dict post请求中的参数
:param eop_date: str 请求时间,格式为:'%Y%m%dT%H%M%SZ'
:return: str
"""
body_str = ""
if not file:
body_str = json.dumps(body_params) if body_params else ''
if method == METHOD_POST:
if file:
body_digest = hashlib.sha256(body_params).hexdigest()
else:
if isinstance(body_params, dict):
body_digest = hashlib.sha256(json.dumps(body_params).encode('utf-8')).hexdigest()
else:
body_digest = hashlib.sha256(body_params.encode('utf-8')).hexdigest()
else:
body_digest = hashlib.sha256(body_str.encode('utf-8')).hexdigest()
# 请求头中必要的两个参数
header_str = 'ctyun-eop-request-id:%s\neop-date:%s\n' % (request_uuid, eop_date)
# url中的参数,或get参数
query_str = encodeQueryStr(get_sorted_str(query_params, method))
signature_str = '%s\n%s\n%s' % (header_str, query_str, body_digest)
print_log(repr('signature_str is: %s' % signature_str))
sign_date = eop_date.split('T')[0]
# 计算鉴权密钥
k_time = hmac_sha256(sk, eop_date)
k_ak = hmac_sha256(k_time, ak)
k_date = hmac_sha256(k_ak, sign_date)
signature_base64 = base64_of_hmac(hmac_sha256(k_date, signature_str))
# 构建请求头的鉴权字段值
sign_header = '%s Headers=ctyun-eop-request-id;eop-date Signature=%s' % (ak, signature_base64.decode('utf8'))
print_log("sign_header :" + sign_header)
return sign_header.encode('utf8')
def get_sign_headers(query_params, body, method, content_type):
"""
获取鉴权用的请求头参数
:param query_params: dict get请求中的参数
:param body: dict post请求中的参数
:return:
"""
now = datetime.datetime.now()
eop_date = datetime.datetime.strftime(now, '%Y%m%dT%H%M%SZ')
request_uuid = get_request_uuid()
headers = { # 三个鉴权用的参数
'Content-type': content_type,
'ctyun-eop-request-id': request_uuid,
'Eop-Authorization': build_sign(query_params=query_params, body_params=body, eop_date=eop_date,
request_uuid=request_uuid, method=method),
'Eop-date': eop_date,
}
return headers
def get(url, query="", params=None, header_params=None):
return execute(url, query, method=METHOD_GET, params=params, header_params=header_params)
def post(url, query="", params=None, header_params=None, content_type='application/json;charset=UTF-8'):
return execute(url, query, method=METHOD_POST, params=params, header_params=header_params,
content_type=content_type)
def execute(url, query, method, params=None, header_params=None, content_type='application/json;charset=UTF-8'):
if 'application/x-www-form-urlencoded' in content_type:
if VERSION == '2':
params = urllib.urlencode(params)
else:
params = urllib.parse.urlencode(params)
params = params or {}
header_params = header_params or {}
if method == METHOD_GET:
query_params, body = (params, {})
else:
query_params = {}
if len(query) > 0:
for q in query.split('&'):
query_params[q.split("=")[0]] = q.split("=")[1]
body = params
headers = get_sign_headers(query_params, body, method, content_type)
headers.update(header_params)
headers.update(
{"User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10.15; rv:109.0) Gecko/20100101 Firefox/110.0"}
)
url = url + "?" + encodeQueryStr(query)
print_log('url: %s' % url)
print_log('请求方式: %s' % method)
print_log('请求头:\n %s' % headers)
print_log('请求参数:\n %s' % params)
print_log('请求参数类型:\n %s' % type(params))
requests.packages.urllib3.disable_warnings()
if method == METHOD_GET:
res = requests.get(url, params=params, headers=headers, verify=False)
else:
if 'application/x-www-form-urlencoded' in content_type:
res = requests.post(url, data=params, headers=headers, verify=False)
elif 'multipart/form-data' in content_type:
res = requests.post(url, data=params, headers=headers, verify=False)
else:
res = requests.post(url, json=params, headers=headers, verify=False)
print_log('返回状态码: %s' % res.status_code)
print_log('返回: %s' % res.json())
return res
def print_log(log_info):
return
now = datetime.datetime.now()
log_info = '[%s]: %s' % (str(now), log_info)
print(log_info)
def encodeQueryStr(query):
afterQuery = ""
if (len(query) != 0):
param = query.split("&")
param.sort()
for str in param:
if (len(afterQuery) < 1):
s = str.split("=")
if (len(s) <= 2):
encodeStr = quote(s[1])
str = s[0] + "=" + encodeStr
afterQuery = afterQuery + str
else:
encodeStr = ""
str = s[0] + "=" + encodeStr
afterQuery = afterQuery + str
else:
s = str.split("=")
if (len(s) >= 2):
encodeStr = quote(s[1])
str = s[0] + "=" + encodeStr
afterQuery = afterQuery + "&" + str
else:
encodeStr = ""
str = s[0] + "=" + encodeStr
afterQuery = afterQuery + "&" + str
return afterQuery
def generate_body(file_list, boundary, params):
lastbody = bytearray()
for file in file_list:
file_name_key = list(file.keys())[0]
file_path = list(file.values())[0]
file_name = os.path.basename(file_path)
body1_array = bytearray(
'--' + boundary + "\r\n" + "Content-Disposition: form-data; name=\"" + file_name_key + "\"; filename=\"" + file_name + "\r\n" + "Content-Type: application/octet-stream" + "\r\n" + "\r\n",
'utf8')
if os.path.exists(file_path):
with open(file_path, 'rb') as f:
body2_array = bytearray(f.read())
body3_array = bytearray("\r\n", 'utf8')
lastbody.extend(body1_array)
lastbody.extend(body2_array)
lastbody.extend(body3_array)
for param in params:
body1_array = bytearray(
'--' + boundary + "\r\n" + "Content-Disposition: form-data; name=\"" + list(param.keys())[
0] + "\r\n" + "\r\n" + list(param.values())[0] + "\r\n", 'utf8')
body3_array = bytearray("\r\n")
lastbody.extend(body1_array)
lastbody.extend(body3_array)
body4_array = bytearray('--' + boundary + "--\r\n", 'utf8')
lastbody.extend(body4_array)
return lastbody
list_regions_resp = get("ctecs-global.ctapi.ctyun.cn/v4/region/list-regions")
list_regions_resp_json = json.loads(list_regions_resp.text)
region_info_array = list_regions_resp_json['returnObj']['regionList']
for region_info in region_info_array:
params = {
"regionID": region_info["regionID"]
}
region_summary_resp = get("ctecs-global.ctapi.ctyun.cn/v4/region/get-summary", params=params)
region_summary = json.loads(region_summary_resp.text)['returnObj']
print("%s,%s,%s,%s" % (region_info['regionName'], region_summary['regionVersion'], region_summary['isMultiZones'], region_summary['openapiAvailable']))
使用这一版代码,可以获取到如下的细节信息:
Region | 版本 | 是否支持多可用区 | 是否支持OpenAPI |
南京2 | v3.0 | FALSE | TRUE |
长沙3 | v3.0 | FALSE | TRUE |
武汉3 | v3.0 | FALSE | TRUE |
九江 | v3.0 | FALSE | FALSE |
香港1 | v3.0 | FALSE | TRUE |
郴州2 | v3.0 | FALSE | FALSE |
西安3 | v3.0 | FALSE | TRUE |
福州3 | v3.0 | FALSE | TRUE |
重庆2 | v3.0 | FALSE | FALSE |
芜湖2 | v3.0 | FALSE | FALSE |
拉萨3 | v3.0 | FALSE | FALSE |
贵州3 | v3.0 | FALSE | FALSE |
杭州2 | v3.0 | FALSE | FALSE |
北京5 | v3.0 | FALSE | FALSE |
哈尔滨2 | v3.0 | FALSE | FALSE |
郑州2 | v3.0 | FALSE | FALSE |
成都4 | v3.0 | FALSE | FALSE |
南京3 | v3.0 | FALSE | FALSE |
海口2 | v3.0 | FALSE | FALSE |
晋中 | v3.0 | FALSE | FALSE |
武汉4 | v3.0 | FALSE | FALSE |
昆明2 | v3.0 | FALSE | FALSE |
天津2 | v3.0 | FALSE | FALSE |
西安4 | v3.0 | FALSE | FALSE |
福州4 | v3.0 | FALSE | FALSE |
南宁2 | v3.0 | FALSE | FALSE |
佛山3 | v3.0 | FALSE | FALSE |
石家庄2 | v3.0 | FALSE | FALSE |
内蒙6 | v3.0 | FALSE | FALSE |
上海7 | v3.0 | FALSE | FALSE |
乌鲁木齐2 | v3.0 | FALSE | FALSE |
南京4 | v3.0 | FALSE | TRUE |
沈阳4 | v3.0 | FALSE | FALSE |
兰州2 | v3.0 | FALSE | FALSE |
青岛3 | v3.0 | FALSE | FALSE |
兰州3 | v3.0 | FALSE | FALSE |
石家庄3 | v3.0 | FALSE | FALSE |
上海8 | v3.0 | FALSE | FALSE |
南京5 | v3.0 | FALSE | TRUE |
西安5 | v3.0 | FALSE | TRUE |
太原2 | v3.0 | FALSE | FALSE |
成都5 | v3.0 | FALSE | FALSE |
合肥 | v3.0 | FALSE | FALSE |
合肥2 | v3.0 | FALSE | TRUE |
广州6 | v3.0 | FALSE | TRUE |
华东1 | v4.0 | TRUE | TRUE |
广州5 | v3.0 | FALSE | FALSE |
福州5 | v3.0 | FALSE | FALSE |
南宁23 | v4.0 | TRUE | TRUE |
上海36 | v4.0 | TRUE | TRUE |
天津3 | v3.0 | FALSE | FALSE |
青岛4 | v3.0 | FALSE | FALSE |
上海9 | v3.0 | FALSE | FALSE |
石家庄20 | v3.0 | FALSE | TRUE |
辽阳2 | v3.0 | FALSE | FALSE |
辽阳1 | v3.0 | FALSE | TRUE |
南宁24 | v3.0 | FALSE | FALSE |
青岛20 | v4.0 | TRUE | TRUE |
新加坡3 | v3.0 | FALSE | TRUE |
法兰克福1 | v3.0 | FALSE | TRUE |
武汉41 | v4.0 | TRUE | TRUE |
福州25 | v3.0 | FALSE | TRUE |
圣保罗1 | v3.0 | FALSE | TRUE |
迪拜1 | v3.0 | FALSE | TRUE |
南昌18 | v3.0 | FALSE | FALSE |
乌鲁木齐27 | v3.0 | FALSE | TRUE |
华北2 | v4.0 | TRUE | TRUE |
长沙42 | v4.0 | TRUE | TRUE |
西南1 | v4.0 | TRUE | TRUE |
南昌5 | v4.0 | TRUE | TRUE |
浙江金华6 | v3.0 | FALSE | FALSE |
从这个结果我们可以分析出,天翼云的Region其实有不同的版本,分v3.0和v4.0,所有的v4.0的Region都支持多可用区,虽然比如南昌5只有一个可用区,但它是支持多可用区的,版本是v4.0。
另外,所有的v4.0版本Region都支持OpenAPI,但不是所有的v3.0 Region都支持OpenAPI。
感兴趣的同学可以使用上面的Python示例代码,去试试调用下其他的OpenAPI接口,肯定会有更多发现哟。