如何通过Python调用cURL实现每10%输出一次下载进度？

要实现每10%打印一次下载进度并实时输出日志，有两种可行方案，具体如下：

方案一：通过Python解析curl的机器可读进度输出 #

利用curl的--progress-meter参数输出机器可读的进度数据，再通过Python捕获并解析这些数据，按10%间隔打印进度。

实现代码 #

import subprocess
import sys

def track_curl_progress(url, output_path):
    # 构造curl命令，输出机器可读进度到stderr
    cmd = [
        'curl',
        '-o', output_path,
        '--progress-meter',  # curl 7.67+支持，旧版本可替换为--progress-bar
        url
    ]
    # 启动进程，实时读取stderr输出
    proc = subprocess.Popen(
        cmd,
        stderr=subprocess.PIPE,
        text=True,
        bufsize=1,  # 行缓冲模式，确保实时获取输出
        universal_newlines=True
    )

    last_printed_percent = -10  # 初始值保证首次10%进度能触发打印

    for line in proc.stderr:
        line = line.strip('\r\n')
        if not line:
            continue
        parts = line.split()
        if len(parts) < 2:
            continue
        try:
            downloaded = int(parts[0])
            total = int(parts[1])
        except ValueError:
            continue

        if total == 0:
            continue  # 总文件大小未知时跳过

        percent = int((downloaded / total) * 100)
        # 仅当进度达到10%整数倍且未打印过时输出
        if percent >= last_printed_percent + 10:
            print(min(percent, 100))
            sys.stdout.flush()  # 强制刷新输出，确保日志实时写入K8s
            last_printed_percent = (percent // 10) * 10

    proc.wait()
    # 下载完成后确保打印100%
    if last_printed_percent < 100:
        print(100)
        sys.stdout.flush()

if __name__ == '__main__':
    if len(sys.argv) != 3:
        print(f"用法: {sys.argv[0]} <下载URL> <输出路径>")
        sys.exit(1)
    track_curl_progress(sys.argv[1], sys.argv[2])

关键说明 #

• --progress-meter：让curl输出包含已下载字节、总字节的机器可读行，便于解析。若环境curl版本低于7.67，可替换为--progress-bar，此时需调整代码改为解析行中的百分比字符串（用正则匹配(\d+\.?\d+)%）。
• 捕获stderr：curl的进度信息默认输出到标准错误流，而非标准输出。
• 行缓冲与强制刷新：bufsize=1确保实时读取输出，sys.stdout.flush()避免输出缓存，保证K8s能实时采集到进度日志。

方案二：纯Python实现下载与进度跟踪 #

直接使用Python的requests库实现流式下载，自行计算并输出进度，无需依赖curl，更灵活可控。

实现代码 #

import requests
import sys

def download_with_progress(url, output_path):
    # 流式请求，避免一次性加载大文件到内存
    response = requests.get(url, stream=True)
    response.raise_for_status()  # 捕获请求错误

    total_size = int(response.headers.get('content-length', 0))
    downloaded_size = 0
    last_printed_percent = -10

    with open(output_path, 'wb') as f:
        # 按1MB块读取下载内容
        for chunk in response.iter_content(chunk_size=1024*1024):
            if chunk:
                f.write(chunk)
                downloaded_size += len(chunk)
                if total_size > 0:
                    percent = int((downloaded_size / total_size) * 100)
                    if percent >= last_printed_percent + 10:
                        print(min(percent, 100))
                        sys.stdout.flush()
                        last_printed_percent = (percent // 10) * 10

    # 确保最终打印100%
    if total_size > 0 and last_printed_percent < 100:
        print(100)
        sys.stdout.flush()

if __name__ == '__main__':
    if len(sys.argv) != 3:
        print(f"用法: {sys.argv[0]} <下载URL> <输出路径>")
        sys.exit(1)
    download_with_progress(sys.argv[1], sys.argv[2])

关键说明 #

• 流式下载：stream=True和iter_content避免大文件占用过多内存，适合K8s容器环境。
• 进度计算：通过content-length响应头获取总文件大小，逐块累加已下载大小计算百分比。
• 错误处理：response.raise_for_status()会在请求失败时抛出异常，便于容器捕获错误状态。

选择建议 #

• 若环境已依赖curl或需要利用curl的高级特性（如断点续传、代理配置），选择方案一。
• 若希望纯Python实现、减少外部依赖，选择方案二。

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