在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
头,我们可以请求文件的特定部分,从而实现断点续传。我们还设置了重试策略,以便在网络不稳定时自动重试下载。