Object Storage Service：Konfigurasi upload callbacks menggunakan OSS SDK untuk Python 1.0

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

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

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

# Tentukan wilayah tempat bucket berada. Contoh: cn-hangzhou. Parameter ini diperlukan jika Anda menggunakan algoritma tanda tangan V4.
region = "cn-hangzhou"

# Tentukan nama bucket.
bucket = oss2.Bucket(auth, endpoint, "yourBucketName", region=region)

# Tentukan fungsi yang digunakan untuk mengkodekan parameter callback dalam 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)))

# Tentukan parameter upload callback.
callback_params = {}
# Tentukan alamat server callback yang menerima permintaan callback. Contoh: http://oss-demo.aliyuncs.com:23450.
callback_params['callbackUrl'] = 'http://oss-demo.aliyuncs.com:23450'
# (Opsional) Tentukan bidang Host yang termasuk dalam header permintaan callback.
#callback_params['callbackHost'] = 'yourCallbackHost'
# Tentukan bidang body yang termasuk dalam permintaan callback. Gunakan placeholder untuk melewatkan informasi objek.
callback_params['callbackBody'] = 'bucket=${bucket}&object=${object}&size=${size}&my_var_1=${x:my_var1}&my_var_2=${x:my_var2}'
# Tentukan tipe konten permintaan callback.
callback_params['callbackBodyType'] = 'application/x-www-form-urlencoded'
encoded_callback = encode_callback(callback_params)
# Tentukan parameter kustom untuk permintaan callback. Setiap parameter kustom terdiri dari kunci dan nilai. Kunci harus dimulai dengan x:.
callback_var_params = {'x:my_var1': 'my_val1', 'x:my_var2': 'my_val2'}
encoded_callback_var = encode_callback(callback_var_params)

# Tentukan parameter upload callback.
params = {'x-oss-callback': encoded_callback, 'x-oss-callback-var': encoded_callback_var}
# Tentukan jalur penuh objek dan string yang ingin Anda unggah. Jangan sertakan nama bucket dalam jalur penuh.
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 kode contoh, pastikan bahwa variabel lingkungan OSS_ACCESS_KEY_ID dan OSS_ACCESS_KEY_SECRET telah dikonfigurasi.
auth = oss2.ProviderAuthV4(EnvironmentVariableCredentialsProvider())

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

# Tentukan wilayah tempat bucket berada. Contoh: cn-hangzhou. Parameter ini diperlukan jika Anda menggunakan algoritma tanda tangan V4.
region = "cn-hangzhou"

# Tentukan alamat server callback yang menerima permintaan callback. Contoh: http://oss-demo.aliyuncs.com:23450.
callback_url = 'http://oss-demo.aliyuncs.com:23450'

# Tentukan nama bucket.
bucket = oss2.Bucket(auth, endpoint, "yourBucketName", region=region)


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

# (Opsional) Tentukan bidang Host yang termasuk dalam header permintaan callback.
# callback_dict['callbackHost'] = 'oss-cn-hangzhou.aliyuncs.com'

## Tentukan bidang body yang termasuk dalam permintaan callback. Gunakan placeholder untuk melewatkan informasi objek.
callback_dict['callbackBody'] = 'bucket=${bucket}&object=${object}&size=${size}&mimeType=${mimeType}&my_var_1=${x:my_var1}&my_var_2=${x:my_var2}'
# Tentukan tipe konten 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)

# Tentukan parameter callback dalam format JSON dan kodekan parameter callback dalam Base64.
callback_param = json.dumps(callback_dict).strip()
base64_callback_body = oss2.utils.b64encode_as_string(callback_param)
# Tambahkan parameter callback yang telah dikodekan ke akhir header dan lewatkan parameter callback ke OSS.
headers = {'x-oss-callback': base64_callback_body, 'x-oss-callback-var': encoded_callback_var}


"""
Konfigurasikan upload callback untuk permintaan unggah multi-bagian
"""

# Konfigurasikan upload callback untuk permintaan unggah multi-bagian.
# Inisialisasi tugas unggah multi-bagian.
parts = []
upload_id = bucket.init_multipart_upload(key).upload_id
# Unggah 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 tugas unggah multi-bagian dan kirimkan upload callback.
result = bucket.complete_multipart_upload(key, upload_id, parts, headers)

# Jika tugas unggah multi-bagian dan permintaan upload callback berhasil, kode status HTTP 200 dikembalikan. Jika tugas unggah multi-bagian selesai tetapi permintaan upload callback gagal, kode status HTTP 203 dikembalikan.
if result.status == 200:
    print("Objek diunggah, dan permintaan callback berhasil (HTTP 200)")
elif result.status == 203:
    print("Objek diunggah, tetapi permintaan callback gagal (HTTP 203)")
else:
    print(f "Pengecualian unggah, kode status HTTP: {result.status}")


# Periksa apakah objek diunggah.
result = bucket.head_object(key)
assert 'x-oss-hash-crc64ecma' in result.headers

Untuk informasi lebih lanjut, lihat PostObject.

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


# Kode contoh berikut memberikan contoh cara Mengonfigurasi upload callback untuk permintaan PostObject. Implementasi ini tidak bergantung pada OSS SDK untuk Python.

# Inisialisasi informasi OSS, seperti ID AccessKey, rahasia AccessKey, dan titik akhir.
# Dapatkan informasi dari variabel lingkungan atau ganti variabel seperti <yourAccessKeyId> dengan nilai aktual.
access_key_id=os.getenv ('OSS_TEST_ACCESS_KEY_ID', '<yourAccessKeyId>')
access_key_secret=os.getenv ('OSS_TEST_ACCESS_KEY_SECRET, '<yourAccessKeySecret>')
bucket_name=os.getenv ('OSS_TEST_BUCKET, '<you bucket>')
endpoint=os.getenv ('OSS_TEST_ENDPOINT', '<yourendpoint>')
# Dalam contoh ini, oss-demo.aliyuncs.com:23450 digunakan.
call_back_url = "http://oss-demo.aliyuncs.com:23450"
# Tentukan wilayah tempat bucket berada. Dalam contoh ini, cn-hangzhou digunakan.
region = "<yourregion>"


# Pastikan parameter sebelumnya ditentukan dengan benar.
for param in (access_key_id, access_key_secret, bucket_name, endpoint):
    assert '<'not in param, ' Silakan tentukan parameter:' + param

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

def calculate_crc64(data):
    """ Hitung hash MD5 file lokal.
    : param data: data
    : Hash MD5 file lokal dikembalikan.
    """
    _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 string waktu GMT yang menentukan waktu kedaluwarsa permintaan.
    : param int expire_time: Periode timeout permintaan. Unit: detik.
    :return str: String waktu GMT yang menentukan waktu kedaluwarsa permintaan.
    """
    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: Periode timeout kebijakan. Unit: detik.
    :param list condition_list: Kondisi kebijakan.
    """
    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 tanda tangan.
    :param str access_key_secret: rahasia kunci akses
    :return str: Permintaan untuk 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()

    return signature

def bulid_callback(cb_url, cb_body, cb_body_type=None, cb_host=None):
    """ Hasilkan string callback.
    :param str cb_url: Tentukan URL server callback. OSS mengirimkan permintaan callback ke URL setelah objek diunggah.
    :param str cb_body: Tentukan tipe konten permintaan callback. Nilai default: application/x-www-form-urlencoded.
    :param str cb_body_type: Tentukan badan permintaan permintaan callback.
    :param str cb_host: Tentukan nilai header Host dalam permintaan callback.
    :return str: String callback yang telah dikodekan.
    """
    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 PostObject.
    :param str endpoint: titik akhir
    :param str bucket_name: nama bucket
    :return str: URL permintaan PostObject.
    """
    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):
    """ Tentukan badan permintaan PostObject.
    :param dict field_dict: Tentukan bidang formulir permintaan PostObject.
    :param str boundary: Tentukan string batas bidang formulir.
    :return str: Badan permintaan PostObject.
    """
    post_body = ''

    # Kodekan 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)

    # Tentukan konten objek yang ingin Anda unggah. Ini harus menjadi bidang terakhir dalam formulir.
    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 terminator ke bidang formulir.
    post_body += '\r\n--{0}--\r\n'.format(boundary)

    return post_body.encode('utf-8')

def build_post_headers(body_len, boundary, headers=None):
    """ Tentukan header dalam permintaan PostObject.
    :param str body_len: Tentukan panjang badan permintaan PostObject.
    :param str boundary: Tentukan string batas bidang formulir.
    # headers=dict(), # Tentukan 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()

# Tentukan bidang formulir dalam permintaan PostObject. Bidang formulir peka huruf besar-kecil.
field_dict = {}
# Tentukan nama objek.
field_dict['key'] = '0303/post.txt'
# access key id
field_dict['OSSAccessKeyId'] = access_key_id

product = "oss"


utc_time = datetime.datetime.utcnow()
# Atur periode validitas menjadi 3600 detik.
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')}"
# Minta tanda tangan.
field_dict['x-oss-signature'] = build_signature(access_key_secret, date)



# Tentukan token keamanan untuk pengguna sementara. Parameter ini diperlukan ketika pengguna sementara menggunakan kredensial sementara. Anda dapat membiarkan parameter ini kosong atau Anda tidak perlu menentukan parameter ini untuk pengguna non-sementara.
# field_dict['x-oss-security-token'] = ''
# Content-Disposition
field_dict['Content-Disposition'] = 'attachment;filename=download.txt'
# Tentukan metadata pengguna.
field_dict['x-oss-meta-uuid'] = 'uuid-xxx'
# Tentukan parameter callback. Jika Anda tidak memerlukan callback, jangan tentukan parameter ini.
field_dict['callback'] = bulid_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')
# Tentukan variabel kustom dalam callback. Jika Anda tidak memerlukan callback, jangan tentukan parameter ini.
field_dict['x:my_var1'] = 'value1'
field_dict['x:my_var2'] = 'value2'

# Unggah objek.
# with open("", r) as f:
#     content = f.read()
# field_dict['content'] = content

# Unggah konten objek.
field_dict['content'] = 'a'*64
# Tentukan tipe objek yang diunggah.
field_dict['content-type'] = 'text/plain'

# Tentukan string batas yang dihasilkan secara acak oleh formulir.
boundary = '9431149156168'

# Kirim permintaan PostObject.
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)

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