import oss2 import requests import urllib.parse OSS_ACCESS_KEY_ID = "LTAIP6x1l3DXfSxm" OSS_ACCESS_KEY_SECRET = "KbTaM9ars4OX3PMS6Xm7rtxGr1FLon" # OSS_BUCKET_ENDPOINT = "oss-cn-hangzhou-internal.aliyuncs.com" # 内网地址 OSS_BUCKET_ENDPOINT = "oss-cn-hangzhou.aliyuncs.com" # 外网地址 def upload_to_oss(video_id): """ Uploads video file to OSS :param video_id: :return: """ # print("start download video...") # await download_video(video_url, "temp.mp4") # print("video download successfully done") oss_object_key = f'single_video/{video_id}' # print(key) auth = oss2.Auth(OSS_ACCESS_KEY_ID, OSS_ACCESS_KEY_SECRET) bucket = oss2.Bucket(auth, OSS_BUCKET_ENDPOINT, "art-pubbucket") response = bucket.put_object_from_file(oss_object_key, "temp.mp4") print("Upload completed successfully.") if 'Content-Length' in response.headers: return { 'status': response.status, 'oss_object_key': oss_object_key} raise AssertionError(f'OSS上传失败,请求ID: \n{response.headers["x-oss-request-id"]}') upload_to_oss(video_id="123456789")