在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
頭,我們可以請求文件的特定部分,從而實現斷點續傳。我們還設置了重試策略,以便在網絡不穩定時自動重試下載。