從 HTTP/HTTPS 下載檔案並解析回應標頭是爬蟲、下載器、API 客戶端的基礎技能。本文示範 urllib.request(標準庫)、requests(推薦)、串流下載大檔案、Cookie 處理、斷點續傳等實戰技巧。
import urllib.request
import os
def download_with_urllib(url, filepath, headers=None, cookies=None):
"""使用 urllib 下載,回傳 (檔案大小, Content-Type)"""
req = urllib.request.Request(url, headers=headers or {})
# Cookie 處理
if cookies:
cookie_header = '; '.join(f'{k}={v}' for k, v in cookies.items())
req.add_header('Cookie', cookie_header)
with urllib.request.urlopen(req) as resp:
# 解析標頭
meta = resp.info()
content_length = meta.get('Content-Length')
content_type = meta.get('Content-Type', '').split(';')[0]
file_size = int(content_length) if content_length else None
# 寫入檔案
with open(filepath, 'wb') as f:
f.write(resp.read())
return file_size, content_type
# 使用
size, ctype = download_with_urllib(
'https://example.com/file.pdf',
'downloaded.pdf',
headers={'User-Agent': 'MyDownloader/1.0'},
cookies={'session_id': 'abc123'}
)
print(f'Size: {size} bytes, Type: {ctype}')
import requests
def download_with_requests(url, filepath, headers=None, cookies=None, stream=True):
"""requests 下載,支援串流、進度條、重試"""
resp = requests.get(url, headers=headers, cookies=cookies, stream=stream, timeout=30)
resp.raise_for_status() # 非 2xx 拋出異常
content_type = resp.headers.get('Content-Type', '').split(';')[0]
content_length = resp.headers.get('Content-Length')
file_size = int(content_length) if content_length else None
if stream:
# 串流寫入,避免大檔案佔滿記憶體
with open(filepath, 'wb') as f:
for chunk in resp.iter_content(chunk_size=8192):
if chunk:
f.write(chunk)
else:
with open(filepath, 'wb') as f:
f.write(resp.content)
return file_size, content_type
from tqdm import tqdm # pip install tqdm
def download_with_progress(url, filepath):
resp = requests.get(url, stream=True, timeout=30)
resp.raise_for_status()
total = int(resp.headers.get('Content-Length', 0))
with open(filepath, 'wb') as f, tqdm(
total=total, unit='B', unit_scale=True, unit_divisor=1024,
desc=os.path.basename(filepath)
) as pbar:
for chunk in resp.iter_content(chunk_size=8192):
if chunk:
f.write(chunk)
pbar.update(len(chunk))
def download_resumable(url, filepath, chunk_size=8192):
"""支援斷點續傳的下載器"""
# 取得檔案總大小
head = requests.head(url, allow_redirects=True)
head.raise_for_status()
total_size = int(head.headers.get('Content-Length', 0))
# 已下載大小
downloaded = os.path.getsize(filepath) if os.path.exists(filepath) else 0
if downloaded >= total_size:
return total_size, head.headers.get('Content-Type')
# Range 請求
headers = {'Range': f'bytes={downloaded}-'}
resp = requests.get(url, headers=headers, stream=True, timeout=30)
if resp.status_code == 206: # Partial Content
mode = 'ab' # 追加模式
elif resp.status_code == 200: # 伺服器不支援 Range
downloaded = 0
mode = 'wb'
else:
resp.raise_for_status()
with open(filepath, mode) as f:
for chunk in resp.iter_content(chunk_size=chunk_size):
if chunk:
f.write(chunk)
return total_size, resp.headers.get('Content-Type')
import mimetypes
def guess_extension(content_type: str) -> str:
"""從 Content-Type 推斷副檔名"""
# 移除 charset 等參數
mime = content_type.split(';')[0].strip()
ext = mimetypes.guess_extension(mime)
return ext or '.bin'
# 常見對照
mimetypes.add_type('application/pdf', '.pdf')
mimetypes.add_type('image/webp', '.webp')
import mmap
def download_mmap(url, filepath):
"""使用記憶體映射寫入超大檔案"""
head = requests.head(url)
total_size = int(head.headers.get('Content-Length', 0))
# 預分配檔案空間
with open(filepath, 'wb') as f:
f.truncate(total_size)
resp = requests.get(url, stream=True)
with open(filepath, 'r+b') as f:
mm = mmap.mmap(f.fileno(), 0)
offset = 0
for chunk in resp.iter_content(chunk_size=65536):
mm[offset:offset+len(chunk)] = chunk
offset += len(chunk)
mm.flush()
from urllib3.util.retry import Retry
from requests.adapters import HTTPAdapter
def create_resilient_session():
session = requests.Session()
retry = Retry(
total=3,
backoff_factor=1,
status_forcelist=[429, 500, 502, 503, 504],
allowed_methods=['HEAD', 'GET', 'OPTIONS']
)
adapter = HTTPAdapter(max_retries=retry)
session.mount('https://', adapter)
session.mount('https://', adapter)
return session
# 使用
session = create_resilient_session()
session.get('https://example.com/file.zip', stream=True)
| 需求 | 推薦方案 |
|---|---|
| 零依賴、簡單下載 | urllib.request |
| 生產環境、豐富功能 | requests + Session |
| 超大檔案(>1GB) | 串流 + mmap / 斷點續傳 |
| 需進度條 | requests + tqdm |
| 高可靠、重試 | HTTPAdapter + Retry |
| 並發下載 | aiohttp (async) / ThreadPoolExecutor |