47 lines
1.7 KiB
Python
47 lines
1.7 KiB
Python
# https://github.com/ucloud/ufile-sdk-python
|
||
import logging
|
||
|
||
from ufile import filemanager
|
||
|
||
from config.ucloud import PUBLIC_KEY, PRIVATE_KEY, UPLOAD_SUFFIX, DOWNLOAD_SUFFIX, BUCKET
|
||
|
||
|
||
def get_private_url(key):
|
||
get_ufile_handler = filemanager.FileManager(PUBLIC_KEY, PRIVATE_KEY, UPLOAD_SUFFIX, DOWNLOAD_SUFFIX)
|
||
|
||
# 判断文件是否存在
|
||
_, resp = get_ufile_handler.head_file(BUCKET, key)
|
||
if resp.status_code != 200:
|
||
logging.warning(f"uCloud中未找到({key})! status: {resp.status_code} error: {resp.error}")
|
||
return None
|
||
|
||
# 获取公有空间下载url
|
||
# url = get_ufile_handler.public_download_url(bucket, key)
|
||
|
||
# 获取私有空间下载url, expires为下载链接有效期，单位为秒
|
||
url = get_ufile_handler.private_download_url(BUCKET, key, expires=3600)
|
||
return url
|
||
|
||
|
||
def copy_file(source_bucket, source_key, target_bucket, target_key):
|
||
copy_ufile_handler = filemanager.FileManager(PUBLIC_KEY, PRIVATE_KEY, UPLOAD_SUFFIX, DOWNLOAD_SUFFIX)
|
||
|
||
# 拷贝文件
|
||
ret, resp = copy_ufile_handler.copy(target_bucket, target_key, source_bucket, source_key)
|
||
if resp.status_code != 200:
|
||
logging.warning(
|
||
f"将({source_key})从({source_bucket})拷贝到({target_bucket})失败! status: {resp.status_code} error: {resp.error}")
|
||
return False
|
||
return True
|
||
|
||
|
||
def upload_file(key, file_path):
|
||
putufile_handler = filemanager.FileManager(PUBLIC_KEY, PRIVATE_KEY, UPLOAD_SUFFIX, DOWNLOAD_SUFFIX)
|
||
|
||
# 普通上传文件至空间
|
||
ret, resp = putufile_handler.putfile(BUCKET, key, file_path, header=None)
|
||
if resp.status_code != 200:
|
||
logging.warning(f"上传({key})失败! status: {resp.status_code} error: {resp.error}")
|
||
return False
|
||
return True
|