Python 调用示例

此脚本应能支持 Python 2 和 3。运行前请确保您已安装依赖包 requests，并且已修改 if __name__ == "__main__": 之后的内容。

# -*- coding: utf8 -*-
import base64
import datetime
import hashlib
import hmac
import json
import os
import sys
import uuid
from typing import Optional, Union

import requests

if sys.version_info.major == 2:
    from urllib import quote, urlencode

    reload(sys)  # type: ignore
    sys.setdefaultencoding("utf8")
else:
    from urllib.parse import quote, urlencode

METHOD_GET = "GET"
METHOD_POST = "POST"

file = False


def hmac_sha256(secret, data):  # type: (Union[bytes, str], str) -> bytes
    return hmac.new(
        key=bytearray(secret)
        if type(secret) == bytes
        else bytearray(secret, encoding="utf8"),
        msg=bytearray(data, encoding="utf8"),
        digestmod=hashlib.sha256,
    ).digest()


def base64_of_hmac(data):  # type: (bytes) -> bytes
    return base64.b64encode(s=data)


def get_request_uuid():  # type: () -> str
    return str(uuid.uuid1())


def get_sorted_str(data):  # type: (dict) -> str
    return "&".join(
        map(
            lambda x_y: "%s=%s" % (x_y[0], x_y[1]),
            sorted(data.items(), key=lambda item: item[0]),
        )
    )


def build_sign(
    query_params, body_params, eop_date, request_uuid, method
):  # type: (dict, Union[bytearray, dict], str, str, str) -> str
    body_str = json.dumps(body_params) if not file and body_params else ""
    body = (
        (
            body_params
            if file
            else (
                json.dumps(body_params).encode(encoding="utf-8")
                if isinstance(body_params, dict)
                else body_params.encode(encoding="utf-8")
            )
        )
        if method == METHOD_POST
        else body_str.encode(encoding="utf-8")
    )
    header_str = "ctyun-eop-request-id:%s\neop-date:%s\n" % (
        request_uuid,
        eop_date,  # 格式为：%Y%m%dT%H%M%SZ。
    )  # 请求头中必要的两个参数。
    query_str = encode_query_str(query=get_sorted_str(data=query_params))
    sign_date = eop_date.split("T")[0]

    # 计算鉴权密钥。
    k_time = hmac_sha256(data=eop_date, secret=sk)
    k_ak = hmac_sha256(data=ak, secret=k_time)
    k_date = hmac_sha256(data=sign_date, secret=k_ak)

    # 构建请求头的鉴权字段值。
    signature_str = "%s\n%s\n%s" % (
        header_str,
        query_str,
        hashlib.sha256(body).hexdigest(),
    )
    sign_header = "%s Headers=ctyun-eop-request-id;eop-date Signature=%s" % (
        ak,
        base64_of_hmac(data=hmac_sha256(data=signature_str, secret=k_date)).decode(
            encoding="utf8"
        ),
    )

    # format: off
    print_log(log_info="signature_str: %s" % repr(signature_str))
    print_log(log_info="sign_header: %s" % sign_header)
    # format: on

    return sign_header.encode(encoding="utf8")


def get_sign_headers(
    query_params, body, method, content_type
):  # type: (dict, dict, str, str) -> dict
    eop_date = datetime.datetime.strftime(datetime.datetime.now(), "%Y%m%dT%H%M%SZ")
    request_uuid = get_request_uuid()
    return {
        "Content-type": content_type,
        "ctyun-eop-request-id": request_uuid,
        "Eop-Authorization": build_sign(
            body_params=body,
            eop_date=eop_date,
            method=method,
            query_params=query_params,
            request_uuid=request_uuid,
        ),
        "Eop-date": eop_date,
    }


def get(
    url, query="", params=None, header_params=None
):  # type: (str, str, Optional[dict], Optional[dict]) -> requests.Response
    """发送 GET 请求。

    Parameters
    ----------
    url : str
        请求的 URL。

    query : str, optional
        请求的查询字符串（默认值：空字符串）。

    params : dict, optional
        请求的参数（默认值：空）。

    header_params : dict, optional
        请求的头部参数（默认值：空）。

    Returns
    -------
    requests.Response
        请求的响应。
    """
    return execute(
        header_params=header_params,
        method=METHOD_GET,
        params=params,
        query=query,
        url=url,
    )


def post(
    url,
    query="",
    params=None,
    header_params=None,
    content_type="application/json;charset=UTF-8",
):
    """发送 POST 请求。

    Parameters
    ----------
    url : str
        请求的 URL。

    query : str, optional
        请求的查询字符串（默认值：空字符串）。

    params : dict, optional
        请求的参数（默认值：空）。

    header_params : dict, optional
        请求的头部参数（默认值，空）。

    content_type : str, optional
        请求的内容类型（默认值：application/json;charset=UTF-8）。

    Returns
    -------
    requests.Response
        请求的响应。
    """
    return execute(
        content_type=content_type,
        header_params=header_params,
        method=METHOD_POST,
        params=params,
        query=query,
        url=url,
    )


def execute(
    url,
    query,
    method,
    params=None,
    header_params=None,
    content_type="application/json;charset=UTF-8",
):  # type: (str, str, str, Optional[dict], Optional[dict], str) -> requests.Response
    if "application/x-www-form-urlencoded" in content_type:
        params = urlencode(params)

    header_params = header_params or {}
    params = params or {}

    if method == METHOD_GET:
        body, query_params = ({}, params)
    else:
        body, query_params = (params, {})

        if len(query) > 0:
            for q in query.split("&"):
                query_params[q.split("=")[0]] = q.split("=")[1]

    print("\n" + "-" * 10)
    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"
        }
    )

    if query:
        url = url + "?" + encode_query_str(query)

    # fmt: off
    print_log(u"URL: %s" % url)
    print_log(u"请求方式: %s" % method)
    print_log(u"请求头: %s" % headers)
    print_log(u"请求参数: %s" % params)
    print_log(u"请求参数类型: %s" % type(params))
    # fmt: on

    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(data=params, headers=headers, url=url, verify=False)
        elif "multipart/form-data" in content_type:
            res = requests.post(data=params, headers=headers, url=url, verify=False)
        else:
            res = requests.post(headers=headers, json=params, url=url, verify=False)

    # fmt: off
    print_log(u"返回状态码: %s" % res.status_code)
    print_log(u"返回: %s" % res.text)
    # fmt: on

    return res


def print_log(log_info):  # type: (str) -> None
    print("[%s]: %s" % (str(object=datetime.datetime.now()), log_info))


def encode_query_str(query):  # type: (str) -> str
    after_query = ""

    if len(query) != 0:
        param = query.split("&")
        param.sort()

        for str in param:
            if len(after_query) < 1:
                s = str.split("=")

                if len(s) <= 2:
                    encode_str = quote(s[1])
                    str = s[0] + "=" + encode_str
                    after_query = after_query + str
                else:
                    encode_str = ""
                    str = s[0] + "=" + encode_str
                    after_query = after_query + str
            else:
                s = str.split("=")

                if len(s) >= 2:
                    encode_str = quote(s[1])
                    str = s[0] + "=" + encode_str
                    after_query = after_query + "&" + str
                else:
                    encode_str = ""
                    str = s[0] + "=" + encode_str
                    after_query = after_query + "&" + str

    return after_query


def generate_body(file_list, boundary, params):  # type: (list, str, dict) -> bytearray
    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",
            encoding="utf8",
        )

        if os.path.exists(file_path):
            with open(file_path, "rb") as f:
                body2_array = bytearray(f.read())

        body3_array = bytearray("\r\n", encoding="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",
            encoding="utf8",
        )
        body3_array = bytearray("\r\n")
        lastbody.extend(body1_array)
        lastbody.extend(body3_array)

    body4_array = bytearray("--" + boundary + "--\r\n", encoding="utf8")
    lastbody.extend(body4_array)
    return lastbody


if __name__ == "__main__":
    ak = "xxxxxx"  # NOTE: 请替换为您的 AccessKey。
    sk = "xxxxxx"  # NOTE：请替换为您的 SecurityKey。

    # NOTE: 以下为调用示例，请按实际情况做出修改。
    get(
        params={"regionID": "xxxxxx"},
        url="https://ctimage-global.ctapi.ctyun.cn/v4/image/list",
    )
    post(
        params={"regionID": "xxxxxx", "instanceID": "xxxxxx", "imageName": "xxxxxx"},
        url="https://ctimage-global.ctapi.ctyun.cn/v4/image/create",
    )