python3调用阿里云openapi脚本 - 生产环境-Toy模板网

这篇具有很好参考价值的文章主要介绍了python3调用阿里云openapi脚本 - 生产环境。希望对大家有所帮助。如果存在错误或未考虑完全的地方，请大家不吝赐教，您也可以点击"举报违法"按钮提交疑问。

场景描述

下文内使用的脚本，主要应用场景为：

在日常使用阿里云服务时，获取数据时，一般都需要调用openapi，下面的脚本为调用阿里云直播接口的脚本，如大家并非使用直播产品的接口，需做以下准备：文章来源地址https://www.toymoban.com/news/detail-832460.html

查询接口版本，示例：‘Version’:‘2016-11-01’。

阿里云AK的RAM权限。

其他产品的地址，示例：http://live.aliyuncs.com为直播产品地址。

阿里云的openapi一般情况下同时支持GET和POST请求，但是也有例外的，需注意请求方式。

action_json 内的所有参数需要更改为openapi调用时的参数，此接口只有两个参数，如有其他参数，需增加一起输入。

脚本如下

# -*- coding: utf-8 -*-
import base64
import hmac
import json
import time
import sys
import requests
import urllib.parse

from pprint import pprint
from datetime import datetime,timedelta
from hashlib import sha1

class openapi_debug():
    def __init__(self,HttpMethod:str,action_json:dict):
        self.base_json = {
            'Format':'JSON',
            'Version':'2016-11-01',#接口版本
            'SignatureMethod':'HMAC-SHA1',
            'AccessKeyId':'xxxx',#AK
            'SignatureNonce': str(int(time.time()*10000)),
            'SignatureVersion': '1.0'
            }
        self.HttpMethod = HttpMethod
        self.access_key_secret = 'xxxx'#secret
        self.base_json.update(action_json)

    def get_utc_time(self):
        now_time = datetime.now()
        utc_time = now_time - timedelta(hours=8)  # UTC只是比北京时间提前了8个小时
        utc_time = utc_time.strftime("%Y-%m-%dT%H:%M:%SZ")
        return utc_time

    def percentEncode(self,str_v):
        res = urllib.parse.quote(str(str_v).encode('utf8').decode(sys.stdin.encoding), '')
        res = res.replace('+', '%20')
        res = res.replace('*', '%2A')
        res = res.replace('%7E', '~')
        return res

    def get_all_param(self):
        Timestamp = self.get_utc_time()
        self.base_json['Timestamp'] = Timestamp
        sortedD = sorted(self.base_json.items(), key=lambda x: x[0])
        canstring = ''
        for k,v in sortedD:
            canstring += '&' + self.percentEncode(k) + '=' + self.percentEncode(v)
        stringToSign = self.HttpMethod+ '&%2F&' + self.percentEncode(canstring[1:])
        h = hmac.new((self.access_key_secret+'&').encode('utf-8'), stringToSign.encode('utf-8'), sha1)
        signature = base64.encodebytes(h.digest()).strip()
        self.base_json['Signature'] = str(signature,encoding='utf-8')
        return self.base_json

class action_debug():
    def __init__(self,HttpMethod,base_json):
        self.base_json = base_json
        self.HttpMethod = HttpMethod

    def action_run(self):
        # 这里需要注意下下面的url内的地址，这里是直播调用，所以是live.aliyuncs.com，如果是其他产品，需要更改地址
        url = 'http://live.aliyuncs.com/?' + urllib.parse.urlencode(self.base_json)
        if self.HttpMethod == 'GET':
            res = requests.get(url=url, json=self.base_json)
            pprint(json.loads(res.text))
        elif self.HttpMethod == 'POST':
            res = requests.post(url=url,json=self.base_json)
            pprint(json.loads(res.text))

if __name__ == '__main__':
    # action_json 内的所有参数需要更改为openapi调用时的参数，此接口只有两个参数，如有其他参数，需增加一起输入。
    action_json = {
        'Action': 'xxxx', # 接口名
        'StartTime': str(int(time.time())),
        'EndTime': str(int(time.time() + 3600)),
    HttpMethod = 'POST' # GET/POST
    base_debug = openapi_debug(HttpMethod,action_json)
    base_json = base_debug.get_all_param()
    action_debug = action_debug(HttpMethod,base_json)
    action_debug.action_run()