# -*- coding: UTF-8 -*- import os import sys import json import oss2 import base64 import requests import time import logging import traceback from oss2 import SizedFileAdapter from oss2.models import PartInfo from aliyunsdkcore import client from aliyunsdkcore.utils import parameter_helper as helper from aliyunsdkvod.request.v20170321 import CreateUploadVideoRequest from aliyunsdkvod.request.v20170321 import RefreshUploadVideoRequest from aliyunsdkvod.request.v20170321 import CreateUploadImageRequest from aliyunsdkvod.request.v20170321 import CreateUploadAttachedMediaRequest from .utils import catch_error, VideoUtils from .exceptions import VideoException VOD_MAX_TITLE_LENGTH = 128 VOD_MAX_DESCRIPTION_LENGTH = 1024 logger = logging.getLogger(__name__) class VideoUploader: def __init__(self, accessKeyId, accessKeySecret, ecsRegionId=None): """ constructor for VodUpload :param accessKeyId: string, access key id :param accessKeySecret: string, access key secret :param ecsRegion: string, 部署迁移脚本的ECS所在的Region，详细参考：https://help.aliyun.com/document_detail/40654.html，如：cn-beijing :return """ self.__accessKeyId = accessKeyId self.__accessKeySecret = accessKeySecret self.__ecsRegion = ecsRegionId self.__vodApiRegion = None self.__connTimeout = 3 self.__bucketClient = None self.__maxRetryTimes = 3 self.__vodClient = None self.__EnableCrc = True # 分片上传参数 self.__multipartThreshold = 10 * 1024 * 1024 # 分片上传的阈值，超过此值开启分片上传 self.__multipartPartSize = 10 * 1024 * 1024 # 分片大小，单位byte self.__multipartThreadsNum = 3 # 分片上传时并行上传的线程数，暂时为串行上传，不支持并行，后续会支持。 self.setApiRegion('cn-shanghai') def setApiRegion(self, apiRegion): """ 设置VoD的接入地址，中国大陆为cn-shanghai，海外支持ap-southeast-1(新加坡)等区域，详情参考：https://help.aliyun.com/document_detail/98194.html :param apiRegion: 接入地址的Region英文表示 :return: """ self.__vodApiRegion = apiRegion self.__vodClient = self.__initVodClient() def setMultipartUpload( self, multipartThreshold=10 * 1024 * 1024, multipartPartSize=10 * 1024 * 1024, multipartThreadsNum=1): if multipartThreshold > 0: self.__multipartThreshold = multipartThreshold if multipartPartSize > 0: self.__multipartPartSize = multipartPartSize if multipartThreadsNum > 0: self.__multipartThreadsNum = multipartThreadsNum def setEnableCrc(self, isEnable=False): self.__EnableCrc = True if isEnable else False @catch_error def uploadLocalVideo(self, videoConfig, startUploadCallback=None): """ 上传本地视频或音频文件到点播，最大支持48.8TB的单个文件，暂不支持断点续传 :param videoConfig: UploadVideoRequest类的实例，注意filePath为本地文件的绝对路径 :param startUploadCallback为获取到上传地址和凭证(uploadInfo)后开始进行文件上传时的回调，可用于记录上传日志等；uploadId为设置的上传ID，可用于关联导入视频。 :return """ uploadInfo = self.__createUploadVideo(videoConfig) if startUploadCallback: startUploadCallback(videoConfig.uploadId, uploadInfo) headers = self.__getUploadHeaders(videoConfig) self.__uploadOssObjectWithRetry( videoConfig.filePath, uploadInfo['UploadAddress']['FileName'], uploadInfo, headers ) return uploadInfo['VideoId'] @catch_error def uploadImage(self, imageConfig, isLocalFile=True): """ 上传图片文件到点播，不支持断点续传；该接口可支持上传本地图片或网络图片 :param imageConfig: imageConfig，注意filePath为本地文件的绝对路径或网络文件的URL地址 :param isLocalFile: bool, 是否为本地文件。True：本地文件，False：网络文件 :return """ # 网络图片需要先下载到本地 if not isLocalFile: imageConfig = self.__downloadWebMedia(imageConfig) # 上传到点播 uploadInfo = self.__createUploadImage(imageConfig) self.__uploadOssObject( imageConfig.filePath, uploadInfo['UploadAddress']['FileName'], uploadInfo, None) # 删除本地临时文件 if not isLocalFile: os.remove(imageConfig.filePath) return uploadInfo['ImageId'], uploadInfo['ImageURL'] # 定义进度条回调函数；consumedBytes: 已经上传的数据量，totalBytes：总数据量 def uploadProgressCallback(self, consumedBytes, totalBytes): if totalBytes: rate = int(100 * (float(consumedBytes) / float(totalBytes))) else: rate = 0 print ("[%s]uploaded %s bytes, percent %s%s" % ( VideoUtils.getCurrentTimeStr(), consumedBytes, format(rate), '%')) sys.stdout.flush() def __initVodClient(self): return client.AcsClient( self.__accessKeyId, self.__accessKeySecret, self.__vodApiRegion, auto_retry=True, max_retry_time=self.__maxRetryTimes, timeout=self.__connTimeout) def __requestUploadInfo(self, request, mediaType): request.set_accept_format('JSON') result = json.loads(self.__vodClient.do_action_with_exception(request).decode('utf-8')) result['OriUploadAddress'] = result['UploadAddress'] result['OriUploadAuth'] = result['UploadAuth'] result['UploadAddress'] = json.loads(base64.b64decode(result['OriUploadAddress']).decode('utf-8')) result['UploadAuth'] = json.loads(base64.b64decode(result['OriUploadAuth']).decode('utf-8')) result['MediaType'] = mediaType if mediaType == 'video': result['MediaId'] = result['VideoId'] elif mediaType == 'image': result['MediaId'] = result['ImageId'] result['MediaURL'] = result['ImageURL'] return result # 获取视频上传地址和凭证 def __createUploadVideo(self, videoConfig): request = CreateUploadVideoRequest.CreateUploadVideoRequest() title = VideoUtils.subString(videoConfig.title, VOD_MAX_TITLE_LENGTH) request.set_Title(title) request.set_FileName(videoConfig.fileName) if videoConfig.description: description = VideoUtils.subString( videoConfig.description, VOD_MAX_DESCRIPTION_LENGTH) request.set_Description(description) if videoConfig.coverURL: request.set_CoverURL(videoConfig.coverURL) if videoConfig.tags: request.set_Tags(videoConfig.tags) if videoConfig.cateId: request.set_CateId(videoConfig.cateId) if videoConfig.templateGroupId: request.set_TemplateGroupId(videoConfig.templateGroupId) if videoConfig.storageLocation: request.set_StorageLocation(videoConfig.storageLocation) if videoConfig.userData: request.set_UserData(videoConfig.userData) if videoConfig.appId: request.set_AppId(videoConfig.appId) if videoConfig.workflowId: request.set_WorkflowId(videoConfig.workflowId) result = self.__requestUploadInfo(request, 'video') logger.info("CreateUploadVideo, FilePath: %s, VideoId: %s" % ( videoConfig.filePath, result['VideoId'])) return result # 刷新上传凭证 def __refresh_upload_video(self, videoId): request = RefreshUploadVideoRequest.RefreshUploadVideoRequest() request.set_VideoId(videoId) result = self.__requestUploadInfo(request, 'video') logger.info("RefreshUploadVideo, VideoId %s" % (result['VideoId'])) return result # 获取图片上传地址和凭证 def __createUploadImage(self, imageConfig): request = CreateUploadImageRequest.CreateUploadImageRequest() request.set_ImageType(imageConfig.imageType) request.set_ImageExt(imageConfig.imageExt) if imageConfig.title: title = VideoUtils.subString(imageConfig.title, VOD_MAX_TITLE_LENGTH) request.set_Title(title) if imageConfig.description: description = VideoUtils.subString( imageConfig.description, VOD_MAX_DESCRIPTION_LENGTH) request.set_Description(description) if imageConfig.tags: request.set_Tags(imageConfig.tags) if imageConfig.cateId: request.set_CateId(imageConfig.cateId) if imageConfig.storageLocation: request.set_StorageLocation(imageConfig.storageLocation) if imageConfig.userData: request.set_UserData(imageConfig.userData) if imageConfig.appId: request.set_AppId(imageConfig.appId) if imageConfig.workflowId: request.set_WorkflowId(imageConfig.workflowId) result = self.__requestUploadInfo(request, 'image') logger.info("CreateUploadImage, FilePath: %s, ImageId: %s, ImageUrl: %s" % ( imageConfig.filePath, result['ImageId'], result['ImageURL'])) return result def __createUploadAttachedMedia(self, uploadAttachedRequest): request = CreateUploadAttachedMediaRequest.CreateUploadAttachedMediaRequest() request.set_BusinessType(uploadAttachedRequest.businessType) request.set_MediaExt(uploadAttachedRequest.mediaExt) if uploadAttachedRequest.title: title = VideoUtils.subString(uploadAttachedRequest.title, VOD_MAX_TITLE_LENGTH) request.set_Title(title) if uploadAttachedRequest.description: description = VideoUtils.subString( uploadAttachedRequest.description, VOD_MAX_DESCRIPTION_LENGTH) request.set_Description(description) if uploadAttachedRequest.tags: request.set_Tags(uploadAttachedRequest.tags) if uploadAttachedRequest.cateId: request.set_CateId(uploadAttachedRequest.cateId) if uploadAttachedRequest.storageLocation: request.set_StorageLocation(uploadAttachedRequest.storageLocation) if uploadAttachedRequest.userData: request.set_UserData(uploadAttachedRequest.userData) if uploadAttachedRequest.appId: request.set_AppId(uploadAttachedRequest.appId) if uploadAttachedRequest.workflowId: request.set_WorkflowId(uploadAttachedRequest.workflowId) result = self.__requestUploadInfo(request, 'attached') logger.info("CreateUploadImage, FilePath: %s, MediaId: %s, MediaURL: %s" % ( uploadAttachedRequest.filePath, result['MediaId'], result['MediaURL'])) return result def __getUploadHeaders(self, videoConfig): if videoConfig.isShowWatermark is None: return None else: userData = "{\"Vod\":{\"UserData\":{\"IsShowWaterMark\": \"%s\"}}}" % ( videoConfig.isShowWatermark) return {'x-oss-notification': base64.b64encode(userData, 'utf-8')} # uploadType，可选：multipart, put, web def __uploadOssObjectWithRetry(self, filePath, object, uploadInfo, headers=None): retryTimes = 0 while retryTimes < self.__maxRetryTimes: try: return self.__uploadOssObject(filePath, object, uploadInfo, headers) except oss2.exceptions.OssError as e: # 上传凭证过期需要重新获取凭证 if e.code == 'SecurityTokenExpired' or e.code == 'InvalidAccessKeyId': uploadInfo = self.__refresh_upload_video(uploadInfo['MediaId']) except Exception as e: raise VideoException('UnkownError', repr(e), traceback.format_exc()) finally: retryTimes += 1 def __uploadOssObject(self, filePath, object, uploadInfo, headers=None): self.__createOssClient(uploadInfo['UploadAuth'], uploadInfo['UploadAddress']) """ p = os.path.dirname(os.path.realpath(__file__)) store = os.path.dirname(p) + '/osstmp' return oss2.resumable_upload(self.__bucketClient, object, filePath, store=oss2.ResumableStore(root=store), headers=headers, multipart_threshold=self.__multipartThreshold, part_size=self.__multipartPartSize, num_threads=self.__multipartThreadsNum, progress_callback=self.uploadProgressCallback) """ uploader = _VodResumableUploader(self.__bucketClient, filePath, object, uploadInfo, headers, self.uploadProgressCallback, self.__refreshUploadAuth) uploader.setMultipartInfo( self.__multipartThreshold, self.__multipartPartSize, self.__multipartThreadsNum) uploader.setClientId(self.__accessKeyId) res = uploader.upload() uploadAddress = uploadInfo['UploadAddress'] bucketHost = uploadAddress['Endpoint'].replace('://', '://' + uploadAddress['Bucket'] + ".") logger.info("UploadFile %s Finish, MediaId: %s, FilePath: %s, Destination: %s/%s" % ( uploadInfo['MediaType'], uploadInfo['MediaId'], filePath, bucketHost, object)) return res # 使用上传凭证和地址信息初始化OSS客户端（注意需要先Base64解码并Json Decode再传入） # 如果上传的ECS位于点播相同的存储区域（如上海），则可以指定internal为True，通过内网上传更快且免费 def __createOssClient(self, uploadAuth, uploadAddress): auth = oss2.StsAuth( uploadAuth['AccessKeyId'], uploadAuth['AccessKeySecret'], uploadAuth['SecurityToken']) endpoint = VideoUtils.convertOssInternal(uploadAddress['Endpoint'], self.__ecsRegion) self.__bucketClient = oss2.Bucket( auth, endpoint, uploadAddress['Bucket'], connect_timeout=self.__connTimeout, enable_crc=self.__EnableCrc) return self.__bucketClient def __refreshUploadAuth(self, videoId): uploadInfo = self.__refresh_upload_video(videoId) uploadAuth = uploadInfo['UploadAuth'] uploadAddress = uploadInfo['UploadAddress'] return self.__createOssClient(uploadAuth, uploadAddress) class _VodResumableUploader: def __init__( self, bucket, filePath, object, uploadInfo, headers, progressCallback, refreshAuthCallback): self.__bucket = bucket self.__filePath = filePath self.__object = object self.__uploadInfo = uploadInfo self.__totalSize = None self.__headers = headers self.__mtime = os.path.getmtime(filePath) self.__progressCallback = progressCallback self.__refreshAuthCallback = refreshAuthCallback self.__threshold = None self.__partSize = None self.__threadsNum = None self.__uploadId = 0 self.__record = {} self.__finishedSize = 0 self.__finishedParts = [] self.__filePartHash = None self.__clientId = None def setMultipartInfo(self, threshold, partSize, threadsNum): self.__threshold = threshold self.__partSize = partSize self.__threadsNum = threadsNum def setClientId(self, clientId): self.__clientId = clientId def upload(self): self.__totalSize = os.path.getsize(self.__filePath) if self.__threshold and self.__totalSize <= self.__threshold: return self.simpleUpload() else: return self.multipartUpload() def simpleUpload(self): with open(VideoUtils.toUnicode(self.__filePath), 'rb') as f: result = self.__bucket.put_object( self.__object, f, headers=self.__headers, progress_callback=None) if self.__uploadInfo['MediaType'] == 'video': self.__reportUploadProgress('put', 1, self.__totalSize) return result def multipartUpload(self): psize = oss2.determine_part_size(self.__totalSize, preferred_size=self.__partSize) # 初始化分片 self.__uploadId = self.__bucket.init_multipart_upload(self.__object).upload_id startTime = time.time() expireSeconds = 2500 # 上传凭证有效期3000秒，提前刷新 # 逐个上传分片 with open(VideoUtils.toUnicode(self.__filePath), 'rb') as fileObj: partNumber = 1 offset = 0 while offset < self.__totalSize: uploadSize = min(psize, self.__totalSize - offset) # logger.info("UploadPart, FilePath: %s, VideoId: %s, UploadId: %s, PartNumber: %s, PartSize: %s" % (self.__fileName, self.__videoId, self.__uploadId, partNumber, uploadSize)) result = self.__bucket.upload_part( self.__object, self.__uploadId, partNumber, SizedFileAdapter(fileObj, uploadSize) ) # print(result.request_id) self.__finishedParts.append(PartInfo(partNumber, result.etag)) offset += uploadSize partNumber += 1 # 上传进度回调 self.__progressCallback(offset, self.__totalSize) if self.__uploadInfo['MediaType'] == 'video': # 上报上传进度 self.__reportUploadProgress('multipart', partNumber - 1, offset) # 检测上传凭证是否过期 nowTime = time.time() if nowTime - startTime >= expireSeconds: self.__bucket = self.__refreshAuthCallback(self.__uploadInfo['MediaId']) startTime = nowTime # 完成分片上传 self.__bucket.complete_multipart_upload( self.__object, self.__uploadId, self.__finishedParts, headers=self.__headers) return result def __reportUploadProgress(self, uploadMethod, donePartsCount, doneBytes): reportHost = 'vod.cn-shanghai.aliyuncs.com' sdkVersion = '1.3.1' reportKey = 'HBL9nnSwhtU2$STX' uploadPoint = { 'upMethod': uploadMethod, 'partSize': self.__partSize, 'doneBytes': doneBytes } timestamp = int(time.time()) authInfo = VideoUtils.getStringMd5("%s|%s|%s" % (self.__clientId, reportKey, timestamp)) fields = { 'Action': 'ReportUploadProgress', 'Format': 'JSON', 'Version': '2017-03-21', 'Timestamp': helper.get_iso_8061_date(), 'SignatureNonce': helper.get_uuid(), 'VideoId': self.__uploadInfo['MediaId'], 'Source': 'PythonSDK', 'ClientId': self.__clientId, 'BusinessType': 'UploadVideo', 'TerminalType': 'PC', 'DeviceModel': 'Server', 'AppVersion': sdkVersion, 'AuthTimestamp': timestamp, 'AuthInfo': authInfo, 'FileName': self.__filePath, 'FileHash': self.__getFilePartHash(self.__clientId, self.__filePath, self.__totalSize), 'FileSize': self.__totalSize, 'FileCreateTime': timestamp, 'UploadRatio': 0, 'UploadId': self.__uploadId, 'DonePartsCount': donePartsCount, 'PartSize': self.__partSize, 'UploadPoint': json.dumps(uploadPoint), 'UploadAddress': self.__uploadInfo['OriUploadAddress'] } requests.post('http://' + reportHost, fields, timeout=1) def __getFilePartHash(self, clientId, filePath, fileSize): if self.__filePartHash: return self.__filePartHash length = 1 * 1024 * 1024 if fileSize < length: length = fileSize try: fp = open(VideoUtils.toUnicode(filePath), 'rb') strVal = fp.read(length) self.__filePartHash = VideoUtils.getStringMd5(strVal, False) fp.close() except Exception: self.__filePartHash = "%s|%s|%s" % (clientId, filePath, self.__mtime) return self.__filePartHash