搜尋引擎結果頁面(SERP)抓取涉及反爬蟲應對、動態參數、分頁處理、結果解析等進階技巧。本文完整實作百度搜尋抓取:User-Agent 旋轉、分頁參數、正則提取標題連結、反爬應對、結果去重與儲存。注意:請遵守 robots.txt 與服務條款,控制請求頻率,僅供技術學習參考。
import urllib.request
import urllib.parse
import urllib.error
import re
import random
import time
from typing import List, Tuple, Optional
class BaiduScraper:
def __init__(self, delay_range=(1, 3), timeout=10):
self.base_url = "https://www.baidu.com/s"
self.delay_range = delay_range
self.timeout = timeout
self.user_agents = [
'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36',
'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36',
# 更多現代 UA...
]
def _build_request(self, params: dict) -> urllib.request.Request:
url = self.base_url + "?" + urllib.parse.urlencode(params)
headers = {
'User-Agent': random.choice(self.user_agents),
'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8',
'Accept-Language': 'zh-TW,zh;q=0.9,en;q=0.8',
'Connection': 'keep-alive',
}
return urllib.request.Request(url, headers=headers)
def _fetch(self, request: urllib.request.Request) -> Optional[str]:
try:
with urllib.request.urlopen(request, timeout=self.timeout) as resp:
raw = resp.read()
# 編碼偵測
encoding = resp.headers.get_content_charset() or 'utf-8'
return raw.decode(encoding, errors='ignore')
except urllib.error.HTTPError as e:
if e.code == 429:
print("觸發頻率限制,需增加延遲或更換 IP")
return None
except Exception as e:
print(f"請求失敗: {e}")
return None
def _parse_results(self, html: str) -> List[Tuple[str, str]]:
"""解析搜尋結果,回傳 [(標題, 連結), ...]"""
results = []
# 百度結果容器正則(需根據實際 HTML 調整)
# 這裡使用簡化版,實際需分析當前頁面結構
pattern = re.compile(
r']*class="t"[^>]*>\s*]*href="([^"]*)"[^>]*>(.*?)',
re.DOTALL | re.IGNORECASE
)
for match in pattern.finditer(html):
url = match.group(1)
title = re.sub(r'<[^>]+>', '', match.group(2)).strip()
if url and title:
results.append((title, url))
return results
def search(self, keyword: str, max_pages: int = 5) -> List[Tuple[str, str]]:
all_results = []
seen_urls = set()
for page in range(max_pages):
pn = page * 10 # 百度分頁參數
params = {
'wd': keyword,
'pn': pn,
'rn': 10, # 每頁結果數
}
print(f"抓取第 {page+1} 頁: pn={pn}")
request = self._build_request(params)
html = self._fetch(request)
if not html:
print("獲取失敗,停止")
break
results = self._parse_results(html)
if not results:
print("無結果,可能被反爬或結構變更")
break
new_count = 0
for title, url in results:
# 去重
if url not in seen_urls:
seen_urls.add(url)
all_results.append((title, url))
new_count += 1
print(f" 新增 {new_count} 筆結果")
if new_count == 0:
print(" 無新結果,結束")
break
# 禮貌延遲
time.sleep(random.uniform(*self.delay_range))
return all_results
# 使用範例
if __name__ == "__main__":
scraper = BaiduScraper(delay_range=(2, 5))
results = scraper.search("Python 教學", max_pages=3)
for i, (title, url) in enumerate(results, 1):
print(f"{i}. {title} -> {url}")
pn:起始結果索引(0, 10, 20...)rn:每頁結果數(預設 10,可設 50)wd:搜尋關鍵字(URL 編碼)
# 維護 UA 池,定期更新
USER_AGENTS = [
# Chrome Windows
'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36',
# Chrome Mac
'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36',
# Firefox
'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:121.0) Gecko/20100101 Firefox/121.0',
# Safari
'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.2 Safari/605.1.15',
# Edge
'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36 Edg/120.0.0.0',
]
# 隨機選擇
import random
headers = {'User-Agent': random.choice(USER_AGENTS)}
selenium/playwright 或分析 API
# 使用 BeautifulSoup 較正則穩健
from bs4 import BeautifulSoup
def parse_with_bs4(html: str):
soup = BeautifulSoup(html, 'html.parser')
results = []
# 百度結果選擇器(需根據實際 DOM 調整)
for item in soup.select('.result .t a'):
title = item.get_text(strip=True)
url = item.get('href')
if title and url:
results.append((title, url))
return results
def resolve_real_url(baidu_url: str, headers: dict) -> str:
"""追蹤百度跳轉連結到真實目標"""
try:
req = urllib.request.Request(baidu_url, headers=headers)
with urllib.request.urlopen(req, timeout=10) as resp:
return resp.geturl() # 最終真實 URL
except Exception:
return baidu_url # 失敗回傳原連結
# 批次解析(可併發加速)
from concurrent.futures import ThreadPoolExecutor
def resolve_batch(urls: List[str], max_workers=5):
with ThreadPoolExecutor(max_workers=max_workers) as executor:
futures = {executor.submit(resolve_real_url, url, headers): url for url in urls}
for future in concurrent.futures.as_completed(futures):
url = futures[future]
try:
real = future.result()
print(f"{url} -> {real}")
except Exception as e:
print(f"{url} 解析失敗: {e}")
import json
import csv
from datetime import datetime
def save_results(results: List[Tuple[str, str]], keyword: str):
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
base = f"baidu_{keyword}_{timestamp}"
# JSON
with open(f"{base}.json", 'w', encoding='utf-8') as f:
json.dump([
{"rank": i, "title": t, "url": u}
for i, (t, u) in enumerate(results, 1)
], f, ensure_ascii=False, indent=2)
# CSV
with open(f"{base}.csv", 'w', encoding='utf-8', newline='') as f:
writer = csv.writer(f)
writer.writerow(['Rank', 'Title', 'URL'])
for i, (t, u) in enumerate(results, 1):
writer.writerow([i, t, u])
print(f"已儲存 {len(results)} 筆結果至 {base}.json/.csv")
https://www.baidu.com/robots.txt百度搜尋抓取涉及請求構建、反爬應對、HTML 解析、重導向追蹤、資料儲存等完整爬蟲流程。核心挑戰在於**反爬蟲機制**(IP、驗證碼、JS 渲染)與**頁面結構變動**。生產環境強烈建議:使用 requests+BeautifulSoup、代理 IP 池、selenium/playwright 處理 JS、申請官方 API。本文代碼為教學示範,實際部署需大幅增強穩健性、合規性與可維護性。