import hashlib import json import os import aliyunsdkcore import oss2 as oss2 from aliyunsdkcore.client import AcsClient, DEFAULT_SDK_CONNECTION_TIMEOUT_IN_SECONDS from aliyunsdkcore.request import CommonRequest class UploadPhoto(AcsClient): def __init__(self, ak='LTAIeS8aBuPBZxV2', secret='hyPeTaDQBQs6jetYcqY0BUdpacXTH3', region_id="cn-hongkong", auto_retry=True, max_retry_time=3, user_agent=None, port=80, timeout=DEFAULT_SDK_CONNECTION_TIMEOUT_IN_SECONDS, public_key_id=None, private_key=None, session_period=3600, credential=None, debug=False): super().__init__(ak, secret, region_id, auto_retry, max_retry_time, user_agent, port, timeout, public_key_id, private_key, session_period, credential, debug) aliyunsdkcore.request.set_default_protocol_type("https") self.domain = 'cloudphoto.cn-shanghai.aliyuncs.com' self.version = '2017-07-11' def get_md5_01(self, file_path): md5 = None if os.path.isfile(file_path): f = open(file_path, 'rb') md5_obj = hashlib.md5() md5_obj.update(f.read()) hash_code = md5_obj.hexdigest() f.close() md5 = str(hash_code).lower() return md5 def assumeRole(self): request = CommonRequest(domain='sts.aliyuncs.com', version='2015-04-01', action_name='AssumeRole') RoleSessionName = 'pqh' request.add_query_param('RoleArn', 'acs:ram::1098806312754985:role/aliyuncloudphotodefaultrole') request.add_query_param('RoleSessionName', RoleSessionName) response = json.loads(self.do_action_with_exception(request).decode()) print(response['Credentials']['SecurityToken']) def createAction(self, action): return CommonRequest(domain=self.domain, version=self.version, action_name=action) def do_action_with_exception(self, acs_request): return json.loads(super().do_action_with_exception(acs_request).decode()) def showOption(self, options, key, desc): choose = [] for index, option in enumerate(options): choose.append(option[key]) print('{index}:{name}'.format(index=index + 1, name=option[key])) return choose[(int)(input('输入{desc}\n'.format(desc=desc))) - 1] if __name__ == '__main__': client = UploadPhoto() request = client.createAction('ListPhotoStores') response = client.do_action_with_exception(request) print('PhotoStores:') photoStores = response['PhotoStores'] StoreName = client.showOption(photoStores, 'Name', 'StoreName') request = client.createAction('GetPhotoStore') request.add_query_param('StoreName', StoreName) response = client.do_action_with_exception(request) print(response) request = client.createAction('FetchLibraries') request.add_query_param('StoreName', StoreName) response = client.do_action_with_exception(request) Libraries = response['Libraries'] LibraryId = client.showOption(Libraries, 'LibraryId', 'LibraryId') request = client.createAction('GetLibrary') request.add_body_params('StoreName', StoreName) request.add_body_params('LibraryId', LibraryId) response = client.do_action_with_exception(request) print(response) request = client.createAction('CreateTransaction') filePath = input('输入上传文件路径') fileName = filePath.split('/')[-1] request.add_query_param('Size', os.path.getsize(filePath)) request.add_query_param('Ext', fileName[-fileName[::-1].index('.'):]) request.add_query_param('Md5', client.get_md5_01(filePath)) request.add_query_param('StoreName', StoreName) # request.add_query_param('LibraryId', liberaryId) # response = json.loads(client.do_action_with_exception(request)) # print(response) # # Upload=response['Transaction']['Upload'] # FileId = Upload['FileId'] # SessionId = Upload['SessionId'] # Bucket=Upload['Bucket'] # OssEndpoint=Upload['OssEndpoint'] # ObjectKey=Upload['ObjectKey'] # # auth = oss2.Auth(client.get_access_key(), client.get_access_secret()) # bucket = oss2.Bucket(auth, OssEndpoint,Bucket) # with open(filePath, 'rb') as fileobj: # result=bucket.put_object(ObjectKey,fileobj) # print('文件上传状态{status}'.format(status=result.status)) # # request = client.createAction('CreatePhoto') # request.add_query_param('FileId', FileId) # request.add_query_param('PhotoTitle', fileName) # request.add_query_param('SessionId', SessionId) # request.add_query_param('StoreName', StoreName) # request.add_query_param('UploadType', 'manual') # request.add_query_param('LibraryId', liberaryId) # response = json.loads(client.do_action_with_exception(request)) # print(response)