Object Storage Service:Callback unggahan (Python SDK V1)

# -*- coding: utf-8 -*-
import json
import base64
import oss2
from oss2.credentials import EnvironmentVariableCredentialsProvider

# Dapatkan kredensial akses dari variabel lingkungan. Sebelum menjalankan contoh ini, pastikan Anda telah mengonfigurasi variabel lingkungan OSS_ACCESS_KEY_ID dan OSS_ACCESS_KEY_SECRET.
auth = oss2.ProviderAuthV4(EnvironmentVariableCredentialsProvider())

# Atur endpoint ke titik akhir wilayah tempat bucket Anda berada. Misalnya, jika bucket Anda berada di Wilayah China (Hangzhou), atur endpoint ke https://oss-cn-hangzhou.aliyuncs.com.
endpoint = "https://oss-cn-hangzhou.aliyuncs.com"

# Atur region ke wilayah yang sesuai dengan endpoint Anda, misalnya cn-hangzhou. Catatan: Parameter ini diperlukan untuk Signature V4.
region = "cn-hangzhou"

# Ganti yourBucketName dengan nama bucket Anda.
bucket = oss2.Bucket(auth, endpoint, "yourBucketName", region=region)

# Definisikan fungsi untuk meng-encode parameter callback ke Base64.
def encode_callback(callback_params):
    cb_str = json.dumps(callback_params).strip()
    return oss2.compat.to_string(base64.b64encode(oss2.compat.to_bytes(cb_str)))

# Atur parameter callback unggahan.
callback_params = {}
# Atur URL server callback Anda. Misalnya, http://oss-demo.aliyuncs.com:23450.
callback_params['callbackUrl'] = 'http://oss-demo.aliyuncs.com:23450'
# (Opsional) Atur nilai Host pada header permintaan callback. Nilai ini harus sesuai dengan Host yang dikonfigurasi di server Anda.
#callback_params['callbackHost'] = 'yourCallbackHost'
# Tentukan badan permintaan callback. Gunakan placeholder untuk meneruskan informasi objek secara dinamis.
callback_params['callbackBody'] = 'bucket=${bucket}&object=${object}&size=${size}&my_var_1=${x:my_var1}&my_var_2=${x:my_var2}'
# Tentukan Content-Type permintaan callback.
callback_params['callbackBodyType'] = 'application/x-www-form-urlencoded'
encoded_callback = encode_callback(callback_params)
# Atur parameter kustom untuk permintaan callback. Setiap parameter berupa pasangan kunci-nilai, dan kuncinya harus diawali dengan x:.
callback_var_params = {'x:my_var1': 'my_val1', 'x:my_var2': 'my_val2'}
encoded_callback_var = encode_callback(callback_var_params)

# Siapkan parameter untuk unggahan dengan callback.
params = {'x-oss-callback': encoded_callback, 'x-oss-callback-var': encoded_callback_var}
# Tentukan path lengkap dan konten objek. Jangan sertakan nama bucket dalam path lengkap.
result = bucket.put_object('examplefiles/exampleobject.txt', 'a'*1024*1024, params)

# -*- coding: utf-8 -*-

import json
from oss2.credentials import EnvironmentVariableCredentialsProvider
import oss2

key = 'exampleobject.txt'
content = "Anything you're good at contributes to happiness."

# Dapatkan kredensial akses dari variabel lingkungan. Sebelum menjalankan contoh ini, pastikan Anda telah mengonfigurasi variabel lingkungan OSS_ACCESS_KEY_ID dan OSS_ACCESS_KEY_SECRET.
auth = oss2.ProviderAuthV4(EnvironmentVariableCredentialsProvider())

# Atur endpoint ke titik akhir wilayah tempat bucket Anda berada. Misalnya, jika bucket Anda berada di Wilayah China (Hangzhou), atur endpoint ke https://oss-cn-hangzhou.aliyuncs.com.
endpoint = "https://oss-cn-hangzhou.aliyuncs.com"

# Atur region ke wilayah yang sesuai dengan endpoint Anda, misalnya cn-hangzhou. Catatan: Parameter ini diperlukan untuk Signature V4.
region = "cn-hangzhou"

# Atur URL server callback Anda. Misalnya, http://oss-demo.aliyuncs.com:23450.
callback_url = 'http://oss-demo.aliyuncs.com:23450'

# Ganti yourBucketName dengan nama bucket Anda.
bucket = oss2.Bucket(auth, endpoint, "yourBucketName", region=region)


# Siapkan parameter callback.
callback_dict = {}
callback_dict['callbackUrl'] = callback_url

# (Opsional) Atur nilai Host pada header permintaan callback. Nilai ini harus sesuai dengan Host yang dikonfigurasi di server Anda.
# callback_dict['callbackHost'] = 'oss-cn-hangzhou.aliyuncs.com'

# Tentukan badan permintaan callback. Gunakan placeholder untuk meneruskan informasi objek secara dinamis.
callback_dict['callbackBody'] = 'bucket=${bucket}&object=${object}&size=${size}&mimeType=${mimeType}&my_var_1=${x:my_var1}&my_var_2=${x:my_var2}'
# Tentukan Content-Type permintaan callback.
callback_dict['callbackBodyType'] = 'application/x-www-form-urlencoded'

callback_var_params = {'x:my_var1': 'my_val1', 'x:my_var2': 'my_val2'}
callback_var_param_json = json.dumps(callback_var_params).strip()
encoded_callback_var = oss2.utils.b64encode_as_string(callback_var_param_json)

# Parameter callback harus diformat dalam JSON dan di-encode Base64.
callback_param = json.dumps(callback_dict).strip()
base64_callback_body = oss2.utils.b64encode_as_string(callback_param)
# Tempatkan parameter callback yang telah di-encode dalam header permintaan.
headers = {'x-oss-callback': base64_callback_body, 'x-oss-callback-var': encoded_callback_var}


"""
Callback unggahan multi-bagian
"""

# Lakukan unggahan multi-bagian dengan callback.
# Inisialisasi tugas unggahan multi-bagian.
parts = []
upload_id = bucket.init_multipart_upload(key).upload_id
# Unggah satu bagian.
result = bucket.upload_part(key, upload_id, 1, content)
parts.append(oss2.models.PartInfo(1, result.etag, size = len(content), part_crc = result.crc))
# Selesaikan unggahan dan picu callback.
result = bucket.complete_multipart_upload(key, upload_id, parts, headers)

# Kode status 200 berarti unggahan dan callback berhasil. Kode status 203 berarti unggahan berhasil, tetapi callback gagal.
if result.status == 200:
    print("Unggahan berhasil, dan callback berhasil (HTTP 200)")
elif result.status == 203:
    print("Unggahan berhasil, tetapi callback gagal (HTTP 203)")
else:
    print(f"Unggahan gagal dengan kode status: {result.status}")


# Verifikasi bahwa file berhasil diunggah.
result = bucket.head_object(key)
assert 'x-oss-hash-crc64ecma' in result.headers

Untuk informasi selengkapnya tentang PostObject, lihat PostObject.

# -*- coding: utf-8 -*-
import os
import time
import datetime
import json
import base64
import hmac
import hashlib
import crcmod
import requests


# Kode berikut menunjukkan cara menggunakan PostObject. Implementasi ini tidak bergantung pada OSS SDK untuk Python.

# Pertama, inisialisasi parameter seperti ID AccessKey, AccessKey Secret, dan endpoint Anda.
# Dapatkan parameter ini dari variabel lingkungan atau ganti placeholder dengan nilai aktual Anda.
access_key_id = os.getenv('OSS_TEST_ACCESS_KEY_ID', '<your AccessKey ID>')
access_key_secret = os.getenv('OSS_TEST_ACCESS_KEY_SECRET', '<your AccessKey Secret>')
bucket_name = os.getenv('OSS_TEST_BUCKET', '<your bucket>')
endpoint = os.getenv('OSS_TEST_ENDPOINT', '<your endpoint>')
# Contoh ini menggunakan http://oss-demo.aliyuncs.com:23450.
call_back_url = "http://oss-demo.aliyuncs.com:23450"
# Misalnya, cn-hangzhou.
region = "<your region>"


# Pastikan parameter di atas telah dikonfigurasi dengan benar.
for param in (access_key_id, access_key_secret, bucket_name, endpoint):
    assert '<' not in param, 'Atur parameter berikut: ' + param

def convert_base64(input):
    return base64.b64encode(input.encode(encoding='utf-8')).decode('utf-8')

def calculate_crc64(data):
    """Hitung hash CRC64 dari data.
    :param data: Data.
    :return: Nilai CRC64 dari data.
    """
    _POLY = 0x142F0E1EBA9EA3693
    _XOROUT = 0XFFFFFFFFFFFFFFFF

    crc64 = crcmod.Crc(_POLY, initCrc=0, xorOut=_XOROUT)
    crc64.update(data.encode())

    return crc64.crcValue

def build_gmt_expired_time(expire_time):
    """Hasilkan waktu kedaluwarsa permintaan dalam format GMT.
    :param int expire_time: Periode timeout dalam detik.
    :return str: Waktu kedaluwarsa dalam format GMT.
    """
    now = int(time.time())
    expire_syncpoint  = now + expire_time

    expire_gmt = datetime.datetime.fromtimestamp(expire_syncpoint).isoformat()
    expire_gmt += 'Z'

    return expire_gmt

def build_encode_policy(expired_time, condition_list):
    """Hasilkan kebijakan.
    :param int expired_time: Waktu kedaluwarsa dalam detik.
    :param list condition_list: Daftar kondisi.
    """
    policy_dict = {}
    policy_dict['expiration'] = build_gmt_expired_time(expired_time)
    policy_dict['conditions'] = condition_list

    policy = json.dumps(policy_dict).strip()
    policy_encode = base64.b64encode(policy.encode())

    return policy_encode

def build_signature(access_key_secret, date):
    """Hasilkan signature.
    :param str access_key_secret: AccessKey Secret.
    :return str: Signature permintaan.
    """

    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()

    return signature

def build_callback(cb_url, cb_body, cb_body_type=None, cb_host=None):
    """Hasilkan string callback.
    :param str cb_url: URL server callback. OSS mengirimkan permintaan callback ke URL ini setelah file diunggah.
    :param str cb_body: Badan permintaan callback.
    :param str cb_body_type: Content-Type permintaan callback. Nilai default: application/x-www-form-urlencoded.
    :param str cb_host: Nilai header Host dalam permintaan callback.
    :return str: String callback yang telah di-encode.
    """
    callback_dict = {}

    callback_dict['callbackUrl'] = cb_url

    callback_dict['callbackBody'] = cb_body
    if cb_body_type is None:
        callback_dict['callbackBodyType'] = 'application/x-www-form-urlencoded'
    else:
        callback_dict['callbackBodyType'] = cb_body_type

    if cb_host is not None:
        callback_dict['callbackHost'] = cb_host

    callback_param = json.dumps(callback_dict).strip()
    base64_callback = base64.b64encode(callback_param.encode());

    return base64_callback.decode()

def build_post_url(endpoint, bucket_name):
    """Hasilkan URL untuk permintaan POST.
    :param str endpoint: Titik akhir.
    :param str bucket_name: Nama bucket.
    :return str: URL untuk permintaan POST.
    """
    if endpoint.startswith('http://'):
        return endpoint.replace('http://', 'http://{0}.'.format(bucket_name))
    elif endpoint.startswith('https://'):
        return endpoint.replace('https://', 'https://{0}.'.format(bucket_name))
    else:
        return 'http://{0}.{1}'.format(bucket_name, endpoint)

def build_post_body(field_dict, boundary):
    """Hasilkan badan untuk permintaan POST.
    :param dict field_dict: Bidang formulir untuk permintaan POST.
    :param str boundary: String pembatas untuk bidang formulir.
    :return str: Badan untuk permintaan POST.
    """
    post_body = ''

    # Encode 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 akan diunggah, yang harus menjadi bidang formulir terakhir.
    post_body += '''--{0}\r\nContent-Disposition: form-data; name=\"file\"; filename=\"{1}\"\r\nContent-Type: {2}\r\n\r\n{3}'''.format(
        boundary, field_dict['key'], field_dict['content-type'], field_dict['content'])

    # Tambahkan pembatas penutup.
    post_body += '\r\n--{0}--\r\n'.format(boundary)

    return post_body.encode('utf-8')

def build_post_headers(body_len, boundary, headers=None):
    """Hasilkan header untuk permintaan POST.
    :param str body_len: Panjang badan permintaan POST.
    :param str boundary: String pembatas bidang formulir.
    :param dict headers: Header permintaan.
    """
    headers = headers if headers else {}
    headers['Content-Length'] = str(body_len)
    headers['Content-Type'] = 'multipart/form-data; boundary={0}'.format(boundary)

    return headers

def encode_callback(callback_params):
    cb_str = json.dumps(callback_params).strip()
    return base64.b64encode(cb_str.encode()).decode()

# Bidang formulir untuk permintaan POST bersifat case-sensitive.
field_dict = {}
# Nama objek.
field_dict['key'] = '0303/post.txt'
# ID AccessKey.
field_dict['OSSAccessKeyId'] = access_key_id

product = "oss"


utc_time = datetime.datetime.utcnow()
# Atur waktu kedaluwarsa kebijakan.
expiration = '2120-01-01T12:00:00.000Z'
date = utc_time.strftime("%Y%m%d")
policy_map = {
    "expiration": expiration,
    "conditions": [
        {"bucket": bucket_name},
        {"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]
    ]
}
policy = json.dumps(policy_map)
print(policy)
string_to_sign = base64.b64encode(policy.encode()).decode()

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')}"
# Signature permintaan.
field_dict['x-oss-signature'] = build_signature(access_key_secret, date)



# Token STS. Diperlukan hanya saat menggunakan kredensial temporary. Jika tidak, biarkan kosong atau hapus.
# field_dict['x-oss-security-token'] = ''
# Content-Disposition
field_dict['Content-Disposition'] = 'attachment;filename=download.txt'
# Metadata pengguna kustom.
field_dict['x-oss-meta-uuid'] = 'uuid-xxx'
# Callback. Jangan sertakan bidang ini jika Anda tidak memerlukan callback.
field_dict['callback'] = build_callback(call_back_url,
                                        'bucket=${bucket}&object=${object}&size=${size}&mimeType=${mimeType}&my_var_1=${x:my_var1}&my_var_2=${x:my_var2}',
                                        'application/x-www-form-urlencoded')
# Variabel kustom untuk callback. Jangan sertakan bidang ini jika Anda tidak memerlukan callback.
field_dict['x:my_var1'] = 'value1'
field_dict['x:my_var2'] = 'value2'

# Untuk mengunggah file:
# with open("", r) as f:
#     content = f.read()
# field_dict['content'] = content

# Konten yang akan diunggah.
field_dict['content'] = 'a'*64
# Tipe konten file yang diunggah.
field_dict['content-type'] = 'text/plain'

# String pembatas untuk bidang formulir, yang biasanya berupa string acak.
boundary = '9431149156168'

# Kirim permintaan POST.
body = build_post_body(field_dict, boundary)
headers = build_post_headers(len(body), boundary)

resp = requests.post(build_post_url(endpoint, bucket_name),
                     data=body,
                     headers=headers)

# Periksa hasil permintaan.
print(resp.status_code)
assert resp.status_code == 200
assert resp.headers['x-oss-hash-crc64ecma'] == str(calculate_crc64(field_dict['content']))