Object Storage Service：Unggahan formulir menggunakan OSS SDK untuk Python 2.0

Perhatian

• Kode contoh dalam topik ini menggunakan ID wilayah China (Hangzhou), cn-hangzhou, sebagai contoh. Titik akhir publik digunakan secara default. Jika Anda ingin mengakses OSS dari produk Alibaba Cloud lainnya di wilayah yang sama, gunakan titik akhir internal. Untuk informasi lebih lanjut tentang pemetaan antara wilayah dan titik akhir yang didukung oleh OSS, lihat Wilayah dan titik akhir OSS.

• Ukuran objek yang diunggah menggunakan unggahan formulir tidak boleh melebihi 5 GB.

Kode contoh

Berikut adalah kode contoh yang menunjukkan proses lengkap unggahan formulir. Langkah-langkah utama adalah sebagai berikut:

1. Buat sebuah Kebijakan Post: Tentukan periode validitas dan kondisi untuk permintaan unggahan. Kondisi mencakup nama bucket, versi tanda tangan, informasi kredensial, tanggal permintaan, dan rentang panjang badan permintaan.

2. Serialize dan enkode Kebijakan: Serialize Kebijakan menjadi string JSON, lalu enkode menggunakan Base64.

3. Hasilkan kunci tanda tangan: Gunakan algoritma HMAC-SHA256 untuk menghasilkan kunci tanda tangan. Kunci tanda tangan mencakup tanggal, wilayah, produk, dan jenis permintaan.

4. Hitung tanda tangan: Gunakan kunci yang dihasilkan untuk menandatangani string Kebijakan yang di-enkode Base64, lalu ubah hasil tanda tangan menjadi string heksadesimal.

5. Bangun badan permintaan: Tambahkan kunci objek, kebijakan, versi tanda tangan, informasi kredensial, tanggal permintaan, dan tanda tangan ke formulir. Kemudian, tulis data yang akan diunggah ke formulir.

6. Buat dan jalankan permintaan: Buat permintaan HTTP POST, atur header permintaan, dan kirim permintaan. Periksa kode status tanggapan untuk memastikan bahwa permintaan berhasil.

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

# Membuat parser argumen baris perintah untuk sampel pengunggahan objek POST.
parser = argparse.ArgumentParser(description="post object sample")

# Menambahkan argumen baris perintah --region, yang menentukan wilayah tempat bucket berada. Parameter ini wajib.
parser.add_argument('--region', help='Wilayah tempat bucket berada.', required=True)

# Menambahkan argumen baris perintah --bucket, yang menentukan nama bucket. Parameter ini wajib.
parser.add_argument('--bucket', help='Nama bucket.', required=True)

# Menambahkan argumen baris perintah --endpoint, yang menentukan nama domain yang dapat digunakan layanan lain untuk mengakses OSS. Parameter ini opsional.
parser.add_argument('--endpoint', help='Nama domain yang dapat digunakan layanan lain untuk mengakses OSS')

# Menambahkan argumen baris perintah --key, yang menentukan nama objek. Parameter ini wajib.
parser.add_argument('--key', help='Nama objek.', required=True)


def main():
    # Tentukan konten yang akan diunggah.
    content = "hi oss"
    product = "oss"  # Pengenal produk, yaitu OSS.

    # Mengurai argumen baris perintah.
    args = parser.parse_args()
    region = args.region  # Informasi wilayah.
    bucket_name = args.bucket  # Nama bucket.
    object_name = args.key  # Nama objek.

    # Muat informasi kredensial dari variabel lingkungan untuk verifikasi identitas.
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()
    credential = credentials_provider.get_credentials()
    access_key_id = credential.access_key_id  # ID AccessKey.
    access_key_secret = credential.access_key_secret  # Rahasia AccessKey.

    # Dapatkan waktu UTC saat ini dan formatnya.
    utc_time = datetime.utcnow()
    date = utc_time.strftime("%Y%m%d")

    # Atur waktu kedaluwarsa satu jam kemudian dan buat Pemetaan Kebijakan.
    expiration = utc_time + timedelta(hours=1)
    policy_map = {
        "expiration": expiration.strftime("%Y-%m-%dT%H:%M:%S.000Z"),  # Waktu kedaluwarsa kebijakan.
        "conditions": [
            {"bucket": bucket_name},  # Tentukan bucket.
            {"x-oss-signature-version": "OSS4-HMAC-SHA256"},  # Tentukan versi tanda tangan.
            {"x-oss-credential": f"{access_key_id}/{date}/{region}/{product}/aliyun_v4_request"},  # Informasi kredensial.
            {"x-oss-date": utc_time.strftime("%Y%m%dT%H%M%SZ")},  # Tanggal permintaan.
            ["content-length-range", 1, 1024]  # Batas rentang panjang konten.
        ]
    }

    # Ubah kebijakan menjadi string JSON dan lakukan enkode Base64.
    policy = json.dumps(policy_map)
    string_to_sign = base64.b64encode(policy.encode()).decode()

    def build_post_body(field_dict, boundary):
        """
        Bangun badan permintaan POST dan enkode bidang formulir ke dalam format multipart/form-data.
        :param field_dict: Kamus bidang formulir.
        :param boundary: String pemisah.
        :return: Badan permintaan POST yang di-enkode.
        """
        post_body = ''

        # Enkode bidang formulir, kecuali untuk konten file dan tipe konten.
        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)

        # Konten file harus menjadi bidang formulir terakhir.
        post_body += '''--{0}\r\nContent-Disposition: form-data; name=\"file\"\r\n\r\n{1}'''.format(
            boundary, field_dict['content'])

        # Tambahkan terminator bidang formulir.
        post_body += '\r\n--{0}--\r\n'.format(boundary)

        return post_body.encode('utf-8')  # Kembalikan badan permintaan POST yang di-enkode dalam UTF-8.

    # Konstruksi kunci tanda tangan dan gunakan algoritma HMAC-SHA256 untuk menghasilkan tanda tangan.
    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()  # Ubah hasil tanda tangan menjadi string heksadesimal.

    # Bangun kamus bidang formulir yang diperlukan untuk permintaan POST.
    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

    # Hasilkan string acak sebagai pemisah formulir.
    boundary = ''.join(random.choice('0123456789') for _ in range(11))

    # Gunakan fungsi build_post_body untuk membangun badan permintaan POST.
    body = build_post_body(field_dict, boundary)

    # Konstruksi URL tujuan untuk permintaan POST.
    url = f"http://{bucket_name}.oss-{region}.aliyuncs.com"

    # Atur header HTTP, tentukan Content-Type sebagai multipart/form-data, dan sertakan string pemisah.
    headers = {
        "Content-Type": f"multipart/form-data; boundary={boundary}",
    }

    # Kirim permintaan POST ke OSS.
    response = requests.post(url, data=body, headers=headers)

    # Tentukan apakah pengunggahan berhasil berdasarkan kode status tanggapan.
    if response.status_code // 100 != 2:
        print(f"Pengunggahan Objek Gagal, kode status: {response.status_code}, alasan: {response.reason}")
    else:
        print(f"pengunggahan objek selesai, kode status: {response.status_code}, id permintaan: {response.headers.get('X-Oss-Request-Id')}")


if __name__ == "__main__":
    main()  # Titik masuk skrip. Fungsi utama dipanggil ketika file dijalankan langsung.

Skenario umum

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

# Membuat parser argumen baris perintah untuk menerima parameter input pengguna.
parser = argparse.ArgumentParser(description="post object sample")
parser.add_argument('--region', help='Wilayah tempat bucket berada.', required=True)
parser.add_argument('--bucket', help='Nama bucket.', required=True)
parser.add_argument('--endpoint', help='Nama domain yang dapat digunakan layanan lain untuk mengakses OSS')
parser.add_argument('--key', help='Nama objek.', required=True)
parser.add_argument('--callback_url', help='Alamat server callback.', required=True)

def main():
    # Tentukan konten yang akan diunggah.
    content = "hi oss"
    product = "oss"  # Nama produk.

    # Mengurai argumen baris perintah.
    args = parser.parse_args()
    region = args.region  # Informasi wilayah.
    bucket_name = args.bucket  # Nama bucket.
    object_name = args.key  # Nama objek (nama file setelah diunggah).

    # Buat penyedia kredensial menggunakan kredensial dalam variabel lingkungan.
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()
    credential = credentials_provider.get_credentials()  # Dapatkan kredensial.
    access_key_id = credential.access_key_id  # Dapatkan ID AccessKey.
    access_key_secret = credential.access_key_secret  # Dapatkan Rahasia AccessKey.

    # Dapatkan waktu UTC saat ini dan formatnya.
    utc_time = datetime.utcnow()
    date = utc_time.strftime("%Y%m%d")  # Format tanggal.
    expiration = utc_time + timedelta(hours=1)  # Atur waktu kedaluwarsa satu jam kemudian.

    # Bangun dokumen kebijakan untuk menentukan kondisi unggahan.
    policy_map = {
        "expiration": expiration.strftime("%Y-%m-%dT%H:%M:%S.000Z"),  # Waktu kedaluwarsa kebijakan.
        "conditions": [
            {"bucket": bucket_name},  # Tentukan nama bucket.
            {"x-oss-signature-version": "OSS4-HMAC-SHA256"},  # Versi tanda tangan.
            {"x-oss-credential": f"{access_key_id}/{date}/{region}/{product}/aliyun_v4_request"},  # Informasi kredensial.
            {"x-oss-date": utc_time.strftime("%Y%m%dT%H%M%SZ")},  # Waktu saat ini.
            ["content-length-range", 1, 1024]  # Batas rentang panjang konten.
        ]
    }
    # Ubah kebijakan menjadi string JSON.
    policy = json.dumps(policy_map)
    # Lakukan enkode Base64 pada kebijakan.
    string_to_sign = base64.b64encode(policy.encode()).decode()

    def build_post_body(field_dict, boundary):
        """
        Bangun badan permintaan POST.

        :param field_dict: Kamus bidang.
        :param boundary: String batas.
        :return: Badan permintaan yang di-enkode.
        """
        post_body = ''

        # Enkode bidang formulir.
        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)

        # Konten file yang diunggah harus menjadi bidang formulir terakhir.
        post_body += '''--{0}\r\nContent-Disposition: form-data; name=\"file\"\r\n\r\n{1}'''.format(
            boundary, field_dict['content'])

        # Tambahkan terminator bidang formulir.
        post_body += '\r\n--{0}--\r\n'.format(boundary)

        return post_body.encode('utf-8')

    # Bangun kunci tanda tangan.
    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()  # Hitung tanda tangan HMAC-SHA256.

    # Bangun kamus bidang formulir.
    field_dict = {}
    field_dict['key'] = object_name  # Nama objek.
    field_dict['policy'] = string_to_sign  # Kebijakan.
    field_dict['x-oss-signature-version'] = "OSS4-HMAC-SHA256"  # Versi tanda tangan.
    field_dict['x-oss-credential'] = f"{access_key_id}/{date}/{region}/{product}/aliyun_v4_request"  # Informasi kredensial.
    field_dict['x-oss-date'] = f"{utc_time.strftime('%Y%m%dT%H%M%SZ')}"  # Waktu saat ini.
    field_dict['x-oss-signature'] = signature  # Nilai tanda tangan.
    field_dict['content'] = content  # Konten file.

    def encode_callback(callback_params):
        """
        Lakukan enkode Base64 pada parameter callback.

        :param callback_params: Kamus parameter callback.
        :return: String yang di-enkode Base64.
        """
        cb_str = json.dumps(callback_params).strip()
        return base64.b64encode(cb_str.encode()).decode()

    # Tetapkan parameter callback unggahan.
    callback_params = {}
    callback_params['callbackUrl'] = args.callback_url  # Alamat server callback.
    callback_params['callbackBody'] = 'bucket=${bucket}&object=${object}&my_var_1=${x:my_var1}&my_var_2=${x:my_var2}'  # Badan permintaan callback.
    callback_params['callbackBodyType'] = 'application/x-www-form-urlencoded'  # Jenis badan permintaan callback.
    encoded_callback = encode_callback(callback_params)  # Enkode parameter callback.

    # Tambahkan bidang terkait callback ke data formulir.
    field_dict['callback'] = encoded_callback
    field_dict['x:my_var1'] = 'value1'
    field_dict['x:my_var2'] = 'value2'

    # Hasilkan string batas acak.
    boundary = ''.join(random.choice('0123456789') for _ in range(11))
    # Kirim permintaan POST.
    body = build_post_body(field_dict, boundary)

    # Bangun URL untuk layanan OSS.
    url = f"http://{bucket_name}.oss-{region}.aliyuncs.com"
    headers = {
        "Content-Type": f"multipart/form-data; boundary={boundary}",  # Atur header permintaan.
    }

    # Kirim permintaan POST.
    response = requests.post(url, data=body, headers=headers)

    # Proses tanggapan.
    if response.status_code // 100 != 2:
        print(f"Pengunggahan Objek Gagal, kode status: {response.status_code}, alasan: {response.reason}")
    else:
        print(f"pengunggahan objek selesai, kode status: {response.status_code}, id permintaan: {response.headers.get('X-Oss-Request-Id')}")

    # Cetak isi tanggapan.
    print(f"tanggapan: {response.text}")

if __name__ == "__main__":
    main()

Referensi

• Untuk contoh lengkap unggahan formulir, lihat post_object.py.