使用Python下载大文件并显示进度条与耗时

传统缓存式下载为何容易导致内存飙升

初学者常使用requests.get(url).content或直接遍历响应体进行文件落盘。该写法会在底层缓冲区内暂存全部原始数据，随后才交由用户态代码处理。当目标资源达到数百兆或数吉字节级别时，单一内存池会被瞬间撑满，进而触发操作系统级别的OOM或页面交换延迟。

流式下载的核心设计思路

规避瞬时内存压力的关键在于切断“请求→全量缓冲→一次性写入”的链路，改为“按需拉取→分块累积→边传边写”。具体依赖于Response.iter_content()迭代器与stream=True标志位。配合服务端提供的Content-Length字段，即可建立已下载区间与总长度的映射关系，从而驱动进度算法。

构建带进度监控的完整下载流程

第1步：初始化环境与配置分块大小

目的：加载系统级工具模块，确立固定吞吐单元以避免频繁的系统调用开销。

• 入口：脚本全局命名空间
• 动作：导入requests、os与time标准库
• 重要参数：chunk_size设定为1024（字节），低带宽环境可适当调高以提升I/O效率
• 可观察状态：解释器成功加载依赖，路径管理器就绪

第2步：建立长连接通道并验证元数据

目的：维持单次TCP握手，确保后续分块能够无缝续传。

• 入口：requests.get()接口
• 动作：传入目标URL并强制启用stream=True模式
• 关键参数：提取response.headers.get("Content-Length")作为总容量基准
• 完成状态：HTTP响应码为200，长度值大于零，连接句柄保持存活

第3步：逐块持久化存储并驱动终端UI刷新

目的：同步磁盘写入进程，消除批量等待造成的交互停滞。

• 入口：response.iter_content(chunk_size)消费端
• 动作：以二进制模式打开目标文件，追加写入每一片数据并累加已读计数
• 核心逻辑：进度百分比 = (total_downloaded / content_length) × 100；填充条纹长度按固定比例缩放
• 完成状态：控制台使用\r转义序列覆盖上一帧文本，循环终止后自动计算差值并打印耗时

可直接复制运行的参考脚本

import requests
import os
import time

def download_with_progress(url, save_dir, file_name):
    # 检查并创建保存目录
    if not os.path.exists(save_dir):
        os.makedirs(save_dir)
    
    start_time = time.time()
    chunk_size = 1024  # 每次读取1KB
    total_downloaded = 0
    
    try:
        # 开启流式传输，避免全量加载至内存
        response = requests.get(url, stream=True)
        response.raise_for_status()
        
        # 从响应头获取文件总大小
        content_length = int(response.headers.get('Content-Length', 0))
        if content_length == 0:
            print("警告：服务端未返回 Content-Length，无法显示准确进度条")
            return
            
        file_path = os.path.join(save_dir, file_name)
        print(f"开始下载 | 预估大小: {content_length / 1024 / 1024:.2f} MB")
        
        with open(file_path, 'wb') as f:
            for chunk in response.iter_content(chunk_size=chunk_size):
                if chunk:
                    f.write(chunk)
                    total_downloaded += len(chunk)
                    
                    # 计算进度百分比与填充长度
                    percent = total_downloaded / content_length
                    filled_length = int(percent * 50)
                    bar = ">" * filled_length
                    
                    # \r 使光标回到行首实现原地刷新
                    print(f"\r[下载进度]: {bar} {percent*100:.2f}% ", end="")
        
        end_time = time.time()
        print(f"\n✅ 下载完成! 实际耗时: {end_time - start_time:.2f} 秒")
        
    except requests.exceptions.RequestException as e:
        print(f"\n❌ 下载失败: {e}")
    except Exception as e:
        print(f"\n❌ 未知错误: {e}")

# 使用示例（替换为真实资源链接）
if __name__ == "__main__":
    target_url = ""  # 填入有效下载链接
    download_with_progress(target_url, "./downloads", "large_file.bin")

异常排查与适配建议

若运行期间未出现进度数值变化但CPU占用持续升高，请核对URL是否指向重定向链或动态鉴权页面。stream=True仅对常规HTTP/HTTPS有效，对于Websocket或SSE等非标准协议不适用。此外，部分云存储节点默认采用分块传输编码（Chunked），此时Content-Length将缺失，属正常的服务端协商行为。若需接入图形化界面或断点续传能力，建议将底层的进度渲染逻辑替换为成熟的tqdm组件。

以上就是基于Python与requests库实现大文件下载进度条及耗时显示的详细内容，更多关于Python网络IO优化与自动化脚本编写的资料请关注本站其它相关文章！