# ===========================================================================================================
# 文件操作工具类
# ===========================================================================================================
class FileOperateUtil(object):
"""
文件操作类
"""
@staticmethod
def file_iterator(down_file):
"""根据文件路径获取待下载文件流"""
content = open(down_file, 'rb+').read()
return content
@staticmethod
def extract_file_name(src_file_name):
"""
从完整路径名称中提取文件扩展名
:param src_file_name:
:return: 文件扩展名元组（文件路径，文件扩展名） eg:C:/tmp/test.ext --> ('C:/tmp/', '.txt')
"""
ext_meta = os.path.splitext(src_file_name)
return ext_meta
@staticmethod
def validate_folder_exists(folder, is_create=False):
"""
判断文件目录是否存在，如果不存在则创建
:param folder: 待判断目录
:param is_create: 是否创建
:return:
"""
if not os.path.exists(folder):
if is_create:
os.makedirs(folder)
else:
raise Exception("该目录不存在，请先创建该目录")
return True
@staticmethod
def validate_file_size(src_file_path, limit_size):
"""
判断文件大小，不能小于指定文件大小，更不能大于最大文件限制
:param src_file_path:源文件
:param limit_size: 限制大小
:return:
"""
if isinstance(src_file_path, object): # 文件对象
file_size = src_file_path.size
else:
file_size = os.path.getsize(src_file_path)
# 判断文件大小，是否小于最大文件限制
if file_size >= settings.UPLOAD_MAX_SIZE:
raise Exception("上传失败,文件过大，超过10M")
else:
if file_size >= limit_size:
raise Exception("上传失败，文件过大")
return True
@staticmethod
def copy_file_chunks(src_file_obj, dst_file_path):
"""
文件复制到指定目录【因为chunks貌似只有从request拿到的文件流才有，所以此方法只用于文件上传】
:param src_file_obj: 待复制文件流对象
:param dst_file_path: 存放目标文件路径，带文件名的完整路径 如:c:/tmp/test.txt
:return:
"""
# 创建存放文件流对象
destination = open(dst_file_path, 'wb+')
try:
# 循环源文件块对象
for chunk in src_file_obj.chunks():
# 将文件流写入
destination.write(chunk)
except Exception as e:
log.debug("复制文件失败,%s" % e)
os.remove(dst_file_path)
raise Exception("复制文件失败,%s" % e)
finally:
log.debug("关闭文件destination")
destination.close()
return True
@staticmethod
def copy_file(src_file_path, dst_file_path):
"""
文件复制到指定目录【任意文件】
:param src_file_path: 待复制文件完整路径
:param dst_file_path: 存放目标文件路径，带文件名的完整路径 如:c:/tmp/test.txt
:return:
"""
# 判断源文件是否是文件
if not os.path.isfile(src_file_path):
raise Exception("上传失败，不是一个文件!")
# 创建存放文件流对象
destination = open(dst_file_path, 'wb+')
try:
src_file = open(src_file_path, "rb+")
# 循环源文件块对象
for line in src_file:
# 将文件流写入
destination.write(line)
except Exception as e:
log.debug("复制文件失败,%s" % e)
os.remove(dst_file_path)
raise Exception("复制文件失败,%s" % e)
finally:
destination.close()
return True |
