Object Storage Service：表單上傳（Python SDK V2）

如果您希望在表單上傳後通知應用伺服器，可參考以下程式碼範例。

import argparse
import base64
import hashlib
import hmac
import json
import random
import requests
from datetime import datetime, timedelta
import alibabacloud_oss_v2 as oss

# 建立命令列參數解析器，用於接收使用者輸入的參數
parser = argparse.ArgumentParser(description="post object sample")
parser.add_argument('--region', help='The region in which the bucket is located.', required=True)
parser.add_argument('--bucket', help='The name of the bucket.', required=True)
parser.add_argument('--endpoint', help='The domain names that other services can use to access OSS')
parser.add_argument('--key', help='The name of the object.', required=True)
parser.add_argument('--callback_url', help='Callback server address.', required=True)

def main():
    # 定義要上傳的內容
    content = "hi oss"
    product = "oss"  # 產品名稱

    # 解析命令列參數
    args = parser.parse_args()
    region = args.region  # 地區資訊
    bucket_name = args.bucket  # 儲存空間名稱
    object_name = args.key  # 對象名稱（即上傳後的檔案名稱）

    # 使用環境變數中的憑證建立憑據提供者
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()
    credential = credentials_provider.get_credentials()  # 擷取憑證
    access_key_id = credential.access_key_id  # 擷取Access Key ID
    access_key_secret = credential.access_key_secret  # 擷取Access Key Secret

    # 擷取當前UTC時間並格式化
    utc_time = datetime.utcnow()
    date = utc_time.strftime("%Y%m%d")  # 格式化日期
    expiration = utc_time + timedelta(hours=1)  # 設定到期時間為1小時後

    # 構建策略文檔，定義上傳條件
    policy_map = {
        "expiration": expiration.strftime("%Y-%m-%dT%H:%M:%S.000Z"),  # 策略到期時間
        "conditions": [
            {"bucket": bucket_name},  # 指定Bucket名稱
            {"x-oss-signature-version": "OSS4-HMAC-SHA256"},  # 簽名版本
            {"x-oss-credential": f"{access_key_id}/{date}/{region}/{product}/aliyun_v4_request"},  # 憑證資訊
            {"x-oss-date": utc_time.strftime("%Y%m%dT%H%M%SZ")},  # 目前時間
            ["content-length-range", 1, 1024]  # 內容長度範圍限制
        ]
    }
    # 將策略轉換為JSON字串
    policy = json.dumps(policy_map)
    # 對策略進行Base64編碼
    string_to_sign = base64.b64encode(policy.encode()).decode()

    def build_post_body(field_dict, boundary):
        """
        構建POST請求體。

        :param field_dict: 欄位字典
        :param boundary: 邊界字串
        :return: 編碼後的請求體
        """
        post_body = ''

        # 編碼錶單欄位
        for k, v in field_dict.items():
            if k != 'content' and k != 'content-type':
                post_body += '''--{0}\r\nContent-Disposition: form-data; name=\"{1}\"\r\n\r\n{2}\r\n'''.format(boundary, k, v)

        # 上傳檔案的內容必須是最後一個表單欄位
        post_body += '''--{0}\r\nContent-Disposition: form-data; name=\"file\"\r\n\r\n{1}'''.format(
            boundary, field_dict['content'])

        # 添加表單欄位結束字元
        post_body += '\r\n--{0}--\r\n'.format(boundary)

        return post_body.encode('utf-8')

    # 構建簽名密鑰
    signing_key = "aliyun_v4" + access_key_secret
    h1 = hmac.new(signing_key.encode(), date.encode(), hashlib.sha256)
    h1_key = h1.digest()
    h2 = hmac.new(h1_key, region.encode(), hashlib.sha256)
    h2_key = h2.digest()
    h3 = hmac.new(h2_key, product.encode(), hashlib.sha256)
    h3_key = h3.digest()
    h4 = hmac.new(h3_key, "aliyun_v4_request".encode(), hashlib.sha256)
    h4_key = h4.digest()

    h = hmac.new(h4_key, string_to_sign.encode(), hashlib.sha256)
    signature = h.hexdigest()  # 計算HMAC-SHA256簽名

    # 構建表單欄位字典
    field_dict = {}
    field_dict['key'] = object_name  # 對象名稱
    field_dict['policy'] = string_to_sign  # 策略
    field_dict['x-oss-signature-version'] = "OSS4-HMAC-SHA256"  # 簽名版本
    field_dict['x-oss-credential'] = f"{access_key_id}/{date}/{region}/{product}/aliyun_v4_request"  # 憑證資訊
    field_dict['x-oss-date'] = f"{utc_time.strftime('%Y%m%dT%H%M%SZ')}"  # 目前時間
    field_dict['x-oss-signature'] = signature  # 簽名值
    field_dict['content'] = content  # 檔案內容

    def encode_callback(callback_params):
        """
        對回調參數進行Base64編碼。

        :param callback_params: 回調參數字典
        :return: Base64編碼後的字串
        """
        cb_str = json.dumps(callback_params).strip()
        return base64.b64encode(cb_str.encode()).decode()

    # 設定上傳回調參數
    callback_params = {}
    callback_params['callbackUrl'] = args.callback_url  # 回調伺服器位址
    callback_params['callbackBody'] = 'bucket=${bucket}&object=${object}&my_var_1=${x:my_var1}&my_var_2=${x:my_var2}'  # 回調請求體
    callback_params['callbackBodyType'] = 'application/x-www-form-urlencoded'  # 回調請求體類型
    encoded_callback = encode_callback(callback_params)  # 對回調參數進行編碼

    # 添加回調相關欄位到表單資料中
    field_dict['callback'] = encoded_callback
    field_dict['x:my_var1'] = 'value1'
    field_dict['x:my_var2'] = 'value2'

    # 產生隨機邊界字串
    boundary = ''.join(random.choice('0123456789') for _ in range(11))
    # 發送POST請求
    body = build_post_body(field_dict, boundary)

    # 構建OSS服務的URL
    url = f"http://{bucket_name}.oss-{region}.aliyuncs.com"
    headers = {
        "Content-Type": f"multipart/form-data; boundary={boundary}",  # 佈建要求頭
    }

    # 發送POST請求
    response = requests.post(url, data=body, headers=headers)

    # 處理響應結果
    if response.status_code // 100 != 2:
        print(f"Post Object Fail, status code: {response.status_code}, reason: {response.reason}")
    else:
        print(f"post object done, status code: {response.status_code}, request id: {response.headers.get('X-Oss-Request-Id')}")

    # 列印響應內容
    print(f"response: {response.text}")

if __name__ == "__main__":
    main()