前幾天已經完成了不少爬蟲功能,從最基本的 Requests、BeautifulSoup,到自動翻頁、Session、User-Agent 和資料清理。
不過實際執行爬蟲時,還有一個很常遇到的問題:
網路不一定每次都成功。
例如:
網路突然不穩
網站暫時沒有回應
Request Timeout
Server 暫時發生錯誤
某一次 Request 剛好失敗
如果程式只要遇到一次錯誤就直接:
爬蟲結束
那前面抓了幾百筆資料可能就白費了。
所以今天想學的是 Retry(重試機制)。
一、為什麼需要 Retry?
假設我要抓 5 個網頁:
第1頁 → 成功
第2頁 → 成功
第3頁 → 失敗
第4頁 → 成功
第5頁 → 成功
如果程式寫成:
try:
response = requests.get(url)
except requests.RequestException:
break
當第 3 頁失敗時:
第1頁 ✓
第2頁 ✓
第3頁 ✗
↓
程式停止
後面的第 4、5 頁也不會繼續。
但有時候第 3 頁只是「暫時失敗」。
如果重新 Request 一次,可能就成功了。
所以可以設計:
Request
↓
失敗
↓
等待
↓
重新 Request
↓
成功
二、最簡單的 Retry
其實不用一開始就使用很複雜的套件。
可以先自己寫一個 Function。
import requests
import time
def get_page(url, retries=3):
for attempt in range(retries):
try:
response = requests.get(
url,
timeout=10
)
response.raise_for_status()
return response
except requests.RequestException as e:
print(
"Request 失敗",
attempt + 1,
"次:",
e
)
time.sleep(2)
return None
這個 Function 的概念很簡單。
如果:
retries=3
代表最多嘗試 3 次。
三、測試 Retry
例如:
url = "https://books.toscrape.com/"
response = get_page(url)
if response:
print("成功取得網頁")
print(response.status_code)
else:
print("多次嘗試後仍然失敗")
如果第一次成功:
Request
↓
成功
↓
直接繼續
如果第一次失敗:
第一次失敗
↓
等待 2 秒
↓
第二次
↓
成功
如果三次都失敗:
第一次 ✗
第二次 ✗
第三次 ✗
↓
回傳 None
四、為什麼不能一直 Retry?
看到 Retry 之後,可能會覺得:
「那我設定 100 次不就好了?」
其實不適合。
如果網站真的沒有回應,程式一直 Request:
Request
Request
Request
Request
Request
Request
...
反而可能增加網站負擔。
所以 Retry 通常會設定一個合理的次數,例如:
3 次
5 次
而且每次重試之間可以等待。
這也是爬蟲設計裡很重要的一個概念:
失敗時不要一直快速重送 Request。
五、加入逐漸增加的等待時間
前面的程式是:
time.sleep(2)
每次都等待 2 秒。
其實還可以讓等待時間逐漸增加。
例如:
第一次失敗
↓
等待 1 秒
第二次失敗
↓
等待 2 秒
第三次失敗
↓
等待 4 秒
這種方式叫做 Exponential Backoff(指數退避)。
可以寫成:
wait_time = 2 ** attempt
例如:
attempt = 0 → 1 秒
attempt = 1 → 2 秒
attempt = 2 → 4 秒
六、自己實作 Backoff
import requests
import time
def get_page(url, retries=3):
for attempt in range(retries):
try:
response = requests.get(
url,
timeout=10
)
response.raise_for_status()
return response
except requests.RequestException as e:
print(
f"第 {attempt + 1} 次 Request 失敗:",
e
)
if attempt < retries - 1:
wait_time = 2 ** attempt
print(
f"等待 {wait_time} 秒後重試..."
)
time.sleep(wait_time)
return None
這樣如果連續失敗:
第 1 次失敗
等待 1 秒
第 2 次失敗
等待 2 秒
第 3 次失敗
停止
七、把 Retry 放進爬蟲
接下來把這個功能和前幾天的爬蟲結合。
import requests
from bs4 import BeautifulSoup
from urllib.parse import urljoin
import time
session = requests.Session()
session.headers.update({
"User-Agent": "Mozilla/5.0"
})
def get_page(url, retries=3):
for attempt in range(retries):
try:
response = session.get(
url,
timeout=10
)
response.raise_for_status()
return response
except requests.RequestException as e:
print(
f"第 {attempt + 1} 次失敗:",
e
)
if attempt < retries - 1:
wait_time = 2 ** attempt
print(
f"等待 {wait_time} 秒後重試..."
)
time.sleep(wait_time)
return None
url = "https://books.toscrape.com/"
all_books = []
while url:
print("正在抓取:", url)
response = get_page(url)
if response is None:
print("這個頁面無法取得,停止爬蟲。")
break
soup = BeautifulSoup(
response.text,
"html.parser"
)
books = soup.select(
"article.product_pod"
)
for book in books:
title = book.h3.a["title"]
price = book.select_one(
".price_color"
).text.strip()
availability = book.select_one(
".availability"
).text.strip()
all_books.append({
"title": title,
"price": price,
"availability": availability
})
next_button = soup.select_one(
"li.next a"
)
if next_button:
next_url = next_button.get("href")
url = urljoin(
url,
next_url
)
else:
url = None
time.sleep(1)
print()
print("爬蟲完成")
print("總共取得:", len(all_books), "筆資料")
現在整個爬蟲的架構就變成:
Session
↓
User-Agent
↓
Request
↓
失敗?
├── 沒有 → 繼續
│
└── 有
↓
等待
↓
Retry
↓
成功 → 繼續
失敗 → 再 Retry
↓
多次失敗 → 停止
八、今天也學到「不要讓一筆資料害整個程式停止」
前幾天的資料清理可能會遇到另一個問題。
例如正常資料:
£51.77
£35.02
£17.46
但如果某筆資料突然變成:
N/A
這時候:
float("N/A")
就會發生錯誤。
所以資料清理也可以搭配 try...except。
def clean_price(price_text):
try:
price_text = price_text.replace(
"£", ""
)
return float(price_text)
except ValueError:
return None
測試:
print(clean_price("£51.77"))
print(clean_price("N/A"))
結果會類似:
51.77
None
這樣即使某一筆價格有問題,也不一定要讓整個爬蟲停止。
九、把錯誤資料另外記錄
如果直接:
return None
雖然程式可以繼續,但我們可能會不知道到底是哪筆資料有問題。
所以也可以另外建立:
invalid_books = []
例如:
invalid_books = []
for book in books:
title = book.h3.a["title"]
price_text = book.select_one(
".price_color"
).text.strip()
price = clean_price(price_text)
if price is None:
invalid_books.append({
"title": title,
"price": price_text
})
continue
這樣就可以把有問題的資料留下來。
最後可以檢查:
print("正常資料:", len(all_books))
print("異常資料:", len(invalid_books))
這對之後做資料清理非常有用。