Created
September 2, 2019 16:19
-
-
Save jie/e872fc095104e77ce28a70d538f2e2f9 to your computer and use it in GitHub Desktop.
upload qiniu by path
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import os | |
import sys | |
import fire | |
import json | |
from qiniu import Auth, put_file, etag, BucketManager, build_batch_stat | |
fail_list = [] | |
default_extname = 'jpg' | |
class QiniuCommandLine(object): | |
_access_key = '' | |
_secret_key = '' | |
_bucket_name = '' | |
_prefix = '' | |
def upload_by_qiniu(self, localName, key_expire_sec=60): | |
q = Auth(self._access_key, self._secret_key) | |
remoteKey = self.get_file_key(localName) | |
token = q.upload_token(self._bucket_name, remoteKey, key_expire_sec) | |
try: | |
ret, info = put_file(token, remoteKey, localName) | |
except Exception as e: | |
logger.exception(e) | |
fail_list.append({ | |
'localName': localName, | |
'error': 'qiniu_error' | |
}) | |
return | |
if not ret: | |
fail_list.append({ | |
'localName': localName, | |
'error': 'ret_error' | |
}) | |
return | |
if not ret.get('key') or ret['key'] != remoteKey: | |
fail_list.append({ | |
'localName': localName, | |
'error': 'key_error' | |
}) | |
return | |
elif not ret.get('key') or ret['hash'] != etag(localName): | |
fail_list.append({ | |
'localName': localName, | |
'error': 'etag_error' | |
}) | |
return | |
def get_file_key(self, localName): | |
filename = localName.split('/')[-1] | |
extname = filename.split('.')[-1].lower() | |
fname = filename.split('.')[0] | |
return self._prefix + filename + '.' + extname | |
def get_valid_files(self, path): | |
file_names = [] | |
local_files = os.listdir(path) | |
for item in local_files: | |
if not os.path.isfile(os.path.join(path, item)): | |
continue | |
if '.%s' % default_extname.lower() not in item.lower(): | |
continue | |
ext = item.split('/')[-1].split('.')[-1].lower() | |
if ext != default_extname: | |
continue | |
file_names.append(item) | |
return file_names | |
def upload(self, path, access_key, secret_key, bucket_name, prefix): | |
self._access_key = access_key | |
self._secret_key = secret_key | |
self._bucket_name = bucket_name | |
self._prefix = prefix | |
file_names = self.get_valid_files(path) | |
for index, item in enumerate(file_names): | |
self.upload_by_qiniu(os.path.join(path, item)) | |
print('%s / %s picture uploaded: %s' % (index + 1, len(file_names), item)) | |
for item in fail_list: | |
print('%s: %s' % (item['localName'], item['error'])) | |
print('<upload finished>') | |
print('<total: %s>' % len(file_names)) | |
print('<fail count: %s>' % len(fail_list)) | |
def check(self, path, access_key, secret_key, bucket_name, prefix): | |
self._access_key = access_key | |
self._secret_key = secret_key | |
self._bucket_name = bucket_name | |
self._prefix = prefix | |
q = Auth(access_key, secret_key) | |
bucket = BucketManager(q) | |
file_names = self.get_valid_files(path) | |
keys = [] | |
for index, item in enumerate(file_names): | |
localKey = self.get_file_key(item) | |
keys.append(localKey) | |
ops = build_batch_stat(bucket_name, keys) | |
ret, info = bucket.batch(ops) | |
if not info: | |
print('qiniu_info_error') | |
return | |
query_data = json.loads(info.text_body) | |
for index, item in enumerate(query_data): | |
if item.get('code') != 200: | |
print('[%s] %s' % (file_names[index], item['data']['error'] if item.get('data') and item['data']['error'] else item['code'])) | |
if __name__ == '__main__': | |
fire.Fire(QiniuCommandLine) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
python3 qiniu_upload.py upload--path=YOUR_IMAGES_PATH --access_key=YOUR_ACCESS_KEY --secret_key=YOUR_SECRET_KEY --bucket_name=YOUR_BUCKET_NAME --prefix=YOUR_FILE_PREFIX