在Python中,要实现类似于curl
命令的断点续传功能,可以使用requests
库结合requests_toolbelt
中的Retry
模块。以下是一个示例代码:
import requests from requests_toolbelt.adapters import HTTPAdapter from requests.packages.urllib3.util.retry import Retry url = "https://example.com/largefile.zip" local_filename = "largefile.zip" chunk_size = 1024 * 1024 # 1MB # 设置重试策略 retry_strategy = Retry( total=3, # 总共重试的次数 status_forcelist=[429, 500, 502, 503, 504], # 针对这些状态码进行重试 allowed_methods=["HEAD", "GET"], # 只允许HEAD和GET请求 backoff_factor=1 # 重试间隔的因子 ) # 创建一个HTTP适配器并应用重试策略 adapter = HTTPAdapter(max_retries=retry_strategy) # 使用Session对象来发送带有重试策略的请求 with requests.Session() as session: session.mount("https://", adapter) session.mount("http://", adapter) # 打开本地文件并设置写入模式 with open(local_filename, "ab") as file: # 发送HEAD请求以获取文件大小 response = session.head(url) file_size = int(response.headers["Content-Length"]) # 初始化文件指针位置 file.seek(0, os.SEEK_END) # 发送GET请求以分块下载文件 for start in range(0, file_size, chunk_size): end = min(start + chunk_size - 1, file_size - 1) headers = {"Range": f"bytes={start}-{end}"} response = session.get(url, headers=headers, stream=True) # 检查响应状态码是否为206(Partial Content) if response.status_code == 206: # 将数据写入本地文件 for chunk in response.iter_content(chunk_size=chunk_size): if chunk: file.write(chunk) else: print(f"Download failed with status code {response.status_code}") break
在这个示例中,我们首先使用HEAD
请求获取文件的大小,然后使用GET
请求分块下载文件。通过设置Range
头,我们可以请求文件的特定部分,从而实现断点续传。我们还设置了重试策略,以便在网络不稳定时自动重试下载。