如何为Python remotezip下载文件添加进度条

remotezip的核心逻辑是通过RemoteFetcher发起HTTP Range请求获取远程ZIP的分段数据，再通过RemoteIO适配为标准文件对象供zipfile.ZipFile处理。要实现真实进度追踪，我们可以从拦截数据获取过程或自定义读取逻辑入手，下面提供两种可行方案：

方案一：分块手动读取+进度条更新 #

无需修改remotezip源码，直接在读取ZIP内文件时分块处理，结合tqdm显示进度：

from remotezip import RemoteZip
from tqdm import tqdm

def read_remote_file_with_progress(remote_zip, file_path, chunk_size=1024*1024):
    # 获取目标文件的元信息
    file_info = remote_zip.getinfo(file_path)
    # 实际需要下载的压缩数据大小（真实网络传输量）
    total_download_size = file_info.compress_size
    # 初始化进度条
    progress_bar = tqdm(
        total=total_download_size,
        unit="B",
        unit_scale=True,
        desc=file_path,
        leave=True
    )

    # 定位到文件在远程ZIP中的起始位置
    remote_zip.fp.seek(file_info.header_offset + len(file_info.FileHeader()))
    remaining = total_download_size
    content = b""

    while remaining > 0:
        read_chunk = remote_zip.fp.read(min(chunk_size, remaining))
        if not read_chunk:
            break
        content += read_chunk
        progress_bar.update(len(read_chunk))
        remaining -= len(read_chunk)
    
    progress_bar.close()
    # 如需解压后内容，可直接返回，ZipFile会自动处理解压
    return content

# 使用示例
with RemoteZip("https://example.com/your_remote_file.zip") as rz:
    file_content = read_remote_file_with_progress(rz, "subdir/target_file.txt")

方案二：自定义带进度追踪的RemoteFetcher #

通过继承RemoteFetcher类，拦截数据获取过程，自动统计下载进度，适配原有的read()方法：

from remotezip import RemoteZip, RemoteFetcher
from tqdm import tqdm

class ProgressFetcher(RemoteFetcher):
    def __init__(self, url, session=None, **kwargs):
        super().__init__(url, session, **kwargs)
        self._progress_bar = None
        self._total_size = 0
        self._downloaded = 0

    def set_progress_target(self, total_size, desc="Downloading"):
        """设置当前下载任务的总大小和进度条描述"""
        self._total_size = total_size
        self._downloaded = 0
        self._progress_bar = tqdm(
            total=total_size,
            unit="B",
            unit_scale=True,
            desc=desc
        )

    def fetch(self, start, end):
        """重写fetch方法，统计下载字节并更新进度"""
        data = super().fetch(start, end)
        self._downloaded += len(data)
        if self._progress_bar:
            self._progress_bar.update(len(data))
            # 任务完成后关闭进度条
            if self._downloaded >= self._total_size:
                self._progress_bar.close()
                self._progress_bar = None
        return data

class ProgressRemoteZip(RemoteZip):
    def __init__(self, url, session=None, **kwargs):
        # 使用自定义的ProgressFetcher
        super().__init__(url, session=session, fetcher=ProgressFetcher, **kwargs)

    def read_with_progress(self, file_path, desc=None):
        """带进度条的文件读取方法"""
        file_info = self.getinfo(file_path)
        # 设置真实下载的总大小（压缩后的数据量）
        self.fetcher.set_progress_target(file_info.compress_size, desc=desc or file_path)
        # 复用原有的read逻辑，自动触发进度更新
        return super().read(file_path)

# 使用示例
with ProgressRemoteZip("https://example.com/your_remote_file.zip") as rz:
    file_content = rz.read_with_progress("subdir/target_file.txt")

关键说明 #

• 如果需要显示解压后文件的读取进度（而非网络下载进度），只需将代码中的file_info.compress_size替换为file_info.file_size即可，但这无法反映真实的网络传输量。
• 两种方案均依赖tqdm库，可通过pip install tqdm安装。

内容的提问来源于stack exchange，提问作者klg