import requests
import os
base = "https://api.zyjj.cc/api/v1"
sk = "sk-xxx" # Replace with your secret key
path = "test.mp4"
# 1. Initialize multipart upload
file_size = os.path.getsize(path)
init_res = requests.post(
f"{base}/open/file/multipart/init",
json={"file_name": os.path.basename(path), "file_size": file_size},
headers={"Authorization": sk}
).json()
if init_res["code"] != 0:
raise Exception(init_res["msg"])
upload_id = init_res["data"]
# 2. Upload chunks
with open(path, "rb") as f:
part_list = []
chunk_size = 1024 * 1024 # 1MB chunks
part_num = 1
while True:
part_data = f.read(chunk_size)
if not part_data:
break
# Get authorization for this specific part
auth_res = requests.post(
f"{base}/open/file/multipart/part",
json={"upload_id": upload_id, "part_num": part_num},
headers={"Authorization": sk}
).json()
if auth_res["code"] != 0:
raise Exception(auth_res["msg"])
upload_info = auth_res["data"]
# Upload actual data to the provided URL
res = requests.put(
upload_info["url"],
data=part_data,
headers=upload_info["header"]
)
res.raise_for_status()
# Store part number and etag for completion
part_list.append({"part_num": part_num, "etag": res.headers.get("etag")})
part_num += 1
# 3. Complete the upload
complete_res = requests.post(
f"{base}/open/file/multipart/complete",
json={"upload_id": upload_id, "part_list": part_list},
headers={"Authorization": sk}
).json()
if complete_res["code"] != 0:
raise Exception(complete_res["msg"])
print(complete_res["data"])