我想在将单个大文件从一个文件路径复制到另一个文件路径时使用 tqdm 显示进度条。
这与复制多个小文件时显示进度条不同。
策略:
首先确定文件的长度,以便您可以告诉 tqdm 它是什么。
通常情况下,您会使用 shutil.copyfile (或 shutil.copyfileobj)来实际复制文件,但这两个函数都没有提供在复制过程中接收更新的方法,我们需要将其转发给 tqdm:
copyfileobj
的稍微修改版本,它支持传递在复制过程中定期调用的更新函数。update
方法作为更新函数传递给新的 copyfileobj
,以便在复制过程中定期调用它。代码:
import os
def copy_with_progress(source_filepath, target_filepath):
with open(source_filepath, 'rb') as source_file:
with open(target_filepath, 'wb') as target_file:
source_file.seek(0, os.SEEK_END)
file_length = source_file.tell()
print(f'Copying: {source_filepath} ({file_length:n} bytes)')
source_file.seek(0, os.SEEK_SET)
with tqdm(total=file_length, unit='B', unit_scale=True) as progress_bar:
copyfileobj_with_progress(
source_file, target_file,
progress_func=progress_bar.update)
# Based on shutil.copyfileobj
_WINDOWS = os.name == 'nt'
COPY_BUFSIZE = 1024 * 1024 if _WINDOWS else 64 * 1024
def copyfileobj_with_progress(fsrc, fdst, length=0, progress_func=None):
if not length:
length = COPY_BUFSIZE
if progress_func is None:
progress_func = lambda offset: None
# Localize variable access to minimize overhead.
fsrc_read = fsrc.read
fdst_write = fdst.write
while True:
buf = fsrc_read(length)
if not buf:
break
fdst_write(buf)
progress_func(len(buf)) # delta offset