Python 網路爬蟲 2026：反爬蟲繞過與生產級爬蟲實戰指南
编程语言
2026 年網頁爬蟲的挑戰
2026 年的 Web 早已不是簡單的靜態頁面時代。現代網站普遍部署了多層反爬機制，讓資料採集變得愈發困難：
- 反 Bot 偵測：Cloudflare Turnstile、Akamai Bot Manager、PerimeterX 等商業方案透過行為分析、TLS 指紋、Canvas 指紋等手段精準識別自動化請求
- CAPTCHA 升級：從簡單的圖片驗證碼進化到 hCaptcha、reCAPTCHA v3 無感驗證、GeeTest 滑塊驗證，甚至需要 AI 輔助破解
- 動態渲染：React/Vue/Svelte 建構的 SPA 應用，頁面內容由 JavaScript 動態產生，傳統 requests + BeautifulSoup 方案完全失效
- 請求頻率限制：基於 IP + User-Agent + Cookie 的多維限流，簡單的時間間隔控制已無法繞過
- 資料加密：關鍵資料透過 Webpack 打包混淆、介面簽名校驗、回應資料加密等手段保護
面對這些挑戰，我們需要一套系統化的爬蟲方案。本文將從核心概念出發，逐步建構生產級爬蟲系統。
核心概念速查
| 概念 | 說明 | 典型工具 |
|---|---|---|
| 靜態爬取 | 直接請求 HTML，解析 DOM | requests + BeautifulSoup |
| 動態渲染 | 模擬瀏覽器執行 JS | Playwright、Selenium |
| 反爬繞過 | 偽裝請求特徵，規避偵測 | 代理池、指紋偽裝、CAPTCHA 服務 |
| 分散式爬取 | 多節點並發抓取 | Scrapy + Redis、Celery |
| 資料管線 | 清洗、轉換、儲存抓取結果 | Pandas、SQLAlchemy、Item Pipeline |
| 增量爬取 | 只抓取新增/變更內容 | URL 去重、內容雜湊比對 |
| 限流控制 | 控制請求頻率，避免封禁 | asyncio.Semaphore、Scrapy AutoThrottle |
五大核心挑戰分析
挑戰一：TLS 指紋偵測
現代反爬系統透過 TLS 握手特徵（JA3/JA4 指紋）識別請求來源。Python 的 requests 和 aiohttp 預設 TLS 指紋與瀏覽器差異巨大，一擊即中。
挑戰二：瀏覽器指紋追蹤
Canvas、WebGL、AudioContext、字型列表等瀏覽器 API 回傳值構成唯一指紋，Headless 瀏覽器與真實瀏覽器存在可偵測差異。
挑戰三：JavaScript 動態渲染
SPA 應用的核心資料透過 AJAX/Fetch 非同步載入，初始 HTML 中不包含目標資料，必須等待 JS 執行完成後才能取得。
挑戰四：驗證碼攔截
登入、搜尋、翻頁等關鍵操作觸發驗證碼，傳統 OCR 方案對現代驗證碼效果極差，需要借助第三方打碼平台或 AI 模型。
挑戰五：IP 封禁與限流
高頻請求觸發 IP 封禁，單一代理容易失效，需要建構高可用代理池並實現智慧輪換策略。
方案一：Scrapy 框架大規模爬取
Scrapy 是 Python 最成熟的爬蟲框架，適合結構化、大規模的資料採集任務。
專案初始化
pip install scrapy scrapy-playwright
scrapy startproject news_crawler
cd news_crawler
scrapy genspider tech_news example.com
完整 Spider 範例
# news_crawler/spiders/tech_news.py
import scrapy
from scrapy.linkextractors import LinkExtractor
from scrapy.spiders import CrawlSpider, Rule
from ..items import NewsItem
class TechNewsSpider(CrawlSpider):
name = "tech_news"
allowed_domains = ["example.com"]
start_urls = ["https://example.com/tech"]
custom_settings = {
"CONCURRENT_REQUESTS": 16,
"DOWNLOAD_DELAY": 1.5,
"AUTOTHROTTLE_ENABLED": True,
"AUTOTHROTTLE_TARGET_CONCURRENCY": 8,
"AUTOTHROTTLE_MAX_DELAY": 10,
"ROBOTSTXT_OBEY": True,
"USER_AGENT": (
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
"AppleWebKit/537.36 (KHTML, like Gecko) "
"Chrome/126.0.0.0 Safari/537.36"
),
"DEFAULT_REQUEST_HEADERS": {
"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",
"Accept-Encoding": "gzip, deflate, br",
},
"FEEDS": {
"output/tech_news.json": {
"format": "json",
"encoding": "utf-8",
"indent": 2,
},
},
"ITEM_PIPELINES": {
"news_crawler.pipelines.CleanHtmlPipeline": 100,
"news_crawler.pipelines.DeduplicatePipeline": 200,
"news_crawler.pipelines.SqlitePipeline": 300,
},
}
rules = (
Rule(
LinkExtractor(allow=r"/tech/article/\d+"),
callback="parse_article",
follow=True,
),
Rule(
LinkExtractor(allow=r"/tech/page/\d+"),
follow=True,
),
)
def parse_article(self, response):
item = NewsItem()
item["title"] = response.css("h1.article-title::text").get("").strip()
item["author"] = response.css(".author-name::text").get("").strip()
item["publish_time"] = response.css(
"time.article-time::attr(datetime)"
).get("")
item["content"] = response.css(".article-content").get("")
item["tags"] = response.css(".tag-item::text").getall()
item["url"] = response.url
yield item
Item 定義
# news_crawler/items.py
import scrapy
from itemloaders.processors import TakeFirst, MapCompose, Join
def strip_text(value):
return value.strip() if value else ""
class NewsItem(scrapy.Item):
title = scrapy.Field(
input_processor=MapCompose(strip_text),
output_processor=TakeFirst(),
)
author = scrapy.Field(
input_processor=MapCompose(strip_text),
output_processor=TakeFirst(),
)
publish_time = scrapy.Field(output_processor=TakeFirst())
content = scrapy.Field(
input_processor=MapCompose(strip_text),
output_processor=Join(""),
)
tags = scrapy.Field()
url = scrapy.Field(output_processor=TakeFirst())
Pipeline 資料處理
# news_crawler/pipelines.py
import hashlib
import sqlite3
from html.parser import HTMLParser
from itemadapter import ItemAdapter
class HtmlStripper(HTMLParser):
def __init__(self):
super().__init__()
self.reset()
self.fed = []
def handle_data(self, d):
self.fed.append(d)
def get_data(self):
return "".join(self.fed)
class CleanHtmlPipeline:
"""清洗 HTML 標籤"""
def process_item(self, item, spider):
adapter = ItemAdapter(item)
content = adapter.get("content", "")
if content:
stripper = HtmlStripper()
stripper.feed(content)
adapter["content"] = stripper.get_data().strip()
return item
class DeduplicatePipeline:
"""基於內容雜湊去重"""
def __init__(self):
self.seen_hashes = set()
def process_item(self, item, spider):
adapter = ItemAdapter(item)
content = adapter.get("content", "")
content_hash = hashlib.md5(content.encode()).hexdigest()
if content_hash in self.seen_hashes:
spider.logger.info(f"重複內容，跳過: {adapter.get('url')}")
raise scrapy.exceptions.DropItem("重複內容")
self.seen_hashes.add(content_hash)
return item
class SqlitePipeline:
"""儲存到 SQLite"""
def open_spider(self, spider):
self.conn = sqlite3.connect("news_data.db")
self.cursor = self.conn.cursor()
self.cursor.execute("""
CREATE TABLE IF NOT EXISTS articles (
id INTEGER PRIMARY KEY AUTOINCREMENT,
title TEXT,
author TEXT,
publish_time TEXT,
content TEXT,
tags TEXT,
url TEXT UNIQUE,
crawl_time TEXT
)
""")
self.conn.commit()
def close_spider(self, spider):
self.conn.close()
def process_item(self, item, spider):
adapter = ItemAdapter(item)
self.cursor.execute(
"""INSERT OR IGNORE INTO articles
(title, author, publish_time, content, tags, url)
VALUES (?, ?, ?, ?, ?, ?)""",
(
adapter.get("title"),
adapter.get("author"),
adapter.get("publish_time"),
adapter.get("content"),
",".join(adapter.get("tags", [])),
adapter.get("url"),
),
)
self.conn.commit()
return item
執行爬蟲
scrapy crawl tech_news -s LOG_LEVEL=INFO
方案二：Playwright 動態 SPA 抓取
對於 JavaScript 渲染的 SPA 應用，Playwright 是 2026 年的首選方案，比 Selenium 更快、更穩定。
安裝
pip install playwright
playwright install chromium
基礎動態頁面抓取
import asyncio
from playwright.async_api import async_playwright
async def scrape_spa():
async with async_playwright() as p:
browser = await p.chromium.launch(headless=True)
context = await browser.new_context(
viewport={"width": 1920, "height": 1080},
user_agent=(
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
"AppleWebKit/537.36 (KHTML, like Gecko) "
"Chrome/126.0.0.0 Safari/537.36"
),
locale="zh-TW",
)
page = await context.new_page()
await page.goto(
"https://spa-example.com/products",
wait_until="networkidle",
)
await page.wait_for_selector(".product-card", timeout=10000)
# 捲動載入更多內容
for _ in range(3):
await page.evaluate(
"window.scrollTo(0, document.body.scrollHeight)"
)
await page.wait_for_timeout(1500)
# 提取資料
products = await page.evaluate("""() => {
const cards = document.querySelectorAll('.product-card');
return Array.from(cards).map(card => ({
name: card.querySelector('.product-name')?.textContent?.trim() || '',
price: card.querySelector('.price')?.textContent?.trim() || '',
rating: card.querySelector('.rating')?.textContent?.trim() || '',
image: card.querySelector('img')?.src || '',
}));
}""")
print(f"抓取到 {len(products)} 個商品")
for product in products[:5]:
print(f" {product['name']} - {product['price']}")
await browser.close()
return products
asyncio.run(scrape_spa())
處理無限捲動 + 分頁
import asyncio
from playwright.async_api import async_playwright
async def scrape_infinite_scroll():
async with async_playwright() as p:
browser = await p.chromium.launch(headless=True)
page = await browser.new_page()
await page.goto(
"https://example.com/feed", wait_until="networkidle"
)
all_items = []
previous_count = 0
max_scrolls = 20
for scroll_round in range(max_scrolls):
await page.evaluate(
"window.scrollTo(0, document.body.scrollHeight)"
)
try:
await page.wait_for_function(
"document.querySelectorAll('.feed-item').length > arguments[0]",
previous_count,
timeout=5000,
)
except Exception:
print(f"第 {scroll_round + 1} 輪無新內容，停止捲動")
break
current_items = await page.evaluate("""() =>
Array.from(document.querySelectorAll('.feed-item')).map(
el => el.textContent.trim()
)
""")
all_items = current_items
previous_count = len(current_items)
print(f"第 {scroll_round + 1} 輪：共 {len(current_items)} 筆")
await page.wait_for_timeout(
int(asyncio.get_event_loop().time() % 2000 + 1000)
)
await browser.close()
return all_items
asyncio.run(scrape_infinite_scroll())
攔截網路請求提取 API 資料
import asyncio
import json
from playwright.async_api import async_playwright
async def intercept_api_data():
async with async_playwright() as p:
browser = await p.chromium.launch(headless=True)
page = await browser.new_page()
api_responses = []
async def handle_response(response):
if "/api/v2/products" in response.url and response.status == 200:
try:
data = await response.json()
api_responses.append(data)
print(f"攔截到 API 資料: {response.url}")
except Exception:
pass
page.on("response", handle_response)
await page.goto(
"https://example.com/shop", wait_until="networkidle"
)
await page.wait_for_timeout(3000)
all_products = []
for resp in api_responses:
items = resp.get("data", {}).get("items", [])
all_products.extend(items)
print(f"透過 API 攔截取得 {len(all_products)} 筆資料")
with open("api_data.json", "w", encoding="utf-8") as f:
json.dump(all_products, f, ensure_ascii=False, indent=2)
await browser.close()
return all_products
asyncio.run(intercept_api_data())
方案三：反爬蟲繞過實戰
請求標頭偽裝
import random
USER_AGENTS = [
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36",
"Mozilla/5.0 (Macintosh; Intel Mac OS X 14_5) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.5 Safari/605.1.15",
"Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36",
"Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:127.0) Gecko/20100101 Firefox/127.0",
]
ACCEPT_LANGUAGES = [
"zh-TW,zh;q=0.9,en;q=0.8",
"zh-CN,zh;q=0.9,en-US;q=0.8",
"en-US,en;q=0.9,zh-TW;q=0.8",
]
def get_random_headers():
"""產生隨機請求標頭"""
return {
"User-Agent": random.choice(USER_AGENTS),
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,*/*;q=0.8",
"Accept-Language": random.choice(ACCEPT_LANGUAGES),
"Accept-Encoding": "gzip, deflate, br",
"DNT": "1",
"Connection": "keep-alive",
"Upgrade-Insecure-Requests": "1",
"Sec-Fetch-Dest": "document",
"Sec-Fetch-Mode": "navigate",
"Sec-Fetch-Site": "none",
"Sec-Fetch-User": "?1",
"Cache-Control": "max-age=0",
}
代理池輪換
import asyncio
import aiohttp
import random
from dataclasses import dataclass, field
@dataclass
class ProxyPool:
"""高可用代理池"""
proxies: list[str] = field(default_factory=list)
failed_count: dict[str, int] = field(default_factory=dict)
max_failures: int = 3
def add_proxy(self, proxy: str):
self.proxies.append(proxy)
self.failed_count[proxy] = 0
def get_proxy(self) -> str | None:
available = [
p for p in self.proxies
if self.failed_count.get(p, 0) < self.max_failures
]
if not available:
return None
return random.choice(available)
def mark_success(self, proxy: str):
self.failed_count[proxy] = 0
def mark_failure(self, proxy: str):
self.failed_count[proxy] = self.failed_count.get(proxy, 0) + 1
if self.failed_count[proxy] >= self.max_failures:
if proxy in self.proxies:
self.proxies.remove(proxy)
print(f"代理 {proxy} 已移除（失敗次數過多）")
async def fetch_with_proxy(pool: ProxyPool, url: str, max_retries: int = 3):
"""使用代理池請求"""
for attempt in range(max_retries):
proxy = pool.get_proxy()
if not proxy:
print("代理池耗盡，使用直連")
proxy = None
try:
async with aiohttp.ClientSession() as session:
async with session.get(
url,
proxy=f"http://{proxy}" if proxy else None,
headers=get_random_headers(),
timeout=aiohttp.ClientTimeout(total=15),
ssl=False,
) as response:
if response.status == 200:
if proxy:
pool.mark_success(proxy)
return await response.text()
elif response.status in (403, 429):
if proxy:
pool.mark_failure(proxy)
print(f"狀態碼 {response.status}，切換代理（第 {attempt + 1} 次）")
await asyncio.sleep(random.uniform(2, 5))
except Exception as e:
if proxy:
pool.mark_failure(proxy)
print(f"請求異常: {e}（第 {attempt + 1} 次）")
await asyncio.sleep(random.uniform(1, 3))
return None
Playwright 指紋偽裝
import asyncio
from playwright.async_api import async_playwright
async def stealth_scrape():
async with async_playwright() as p:
browser = await p.chromium.launch(
headless=True,
args=[
"--disable-blink-features=AutomationControlled",
"--disable-features=IsolateOrigins,site-per-process",
"--disable-dev-shm-usage",
"--no-sandbox",
],
)
context = await browser.new_context(
viewport={"width": 1920, "height": 1080},
screen={"width": 1920, "height": 1080},
user_agent=(
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
"AppleWebKit/537.36 (KHTML, like Gecko) "
"Chrome/126.0.0.0 Safari/537.36"
),
locale="zh-TW",
timezone_id="Asia/Taipei",
geolocation={"latitude": 25.0330, "longitude": 121.5654},
permissions=["geolocation"],
color_scheme="light",
)
page = await context.new_page()
# 注入反偵測腳本
await page.add_init_script("""
Object.defineProperty(navigator, 'webdriver', {
get: () => undefined
});
window.chrome = {
runtime: {}, loadTimes: function() {},
csi: function() {}, app: {}
};
const originalQuery = window.navigator.permissions.query;
window.navigator.permissions.query = (parameters) => (
parameters.name === 'notifications' ?
Promise.resolve({ state: Notification.permission }) :
originalQuery(parameters)
);
Object.defineProperty(navigator, 'plugins', {
get: () => [1, 2, 3, 4, 5]
});
Object.defineProperty(navigator, 'languages', {
get: () => ['zh-TW', 'zh', 'en-US', 'en']
});
""")
await page.goto("https://example.com", wait_until="networkidle")
content = await page.content()
print(f"頁面長度: {len(content)}")
await browser.close()
return content
asyncio.run(stealth_scrape())
方案四：aiohttp + BeautifulSoup 非同步爬蟲
對於不需要 JS 渲染的場景，非同步爬蟲是效率最高的選擇。
import asyncio
import aiohttp
from bs4 import BeautifulSoup
from dataclasses import dataclass, field
import json
import time
@dataclass
class AsyncScraper:
"""非同步爬蟲核心類別"""
base_url: str
max_concurrency: int = 10
request_delay: float = 0.5
timeout: int = 15
results: list = field(default_factory=list)
visited: set = field(default_factory=set)
async def fetch_page(self, session: aiohttp.ClientSession, url: str) -> str | None:
"""請求單一頁面"""
try:
async with session.get(
url,
headers=get_random_headers(),
timeout=aiohttp.ClientTimeout(total=self.timeout),
ssl=False,
) as response:
if response.status == 200:
return await response.text()
print(f"[{response.status}] {url}")
return None
except asyncio.TimeoutError:
print(f"[逾時] {url}")
return None
except Exception as e:
print(f"[異常] {url}: {e}")
return None
def parse_page(self, html: str, url: str) -> dict | None:
"""解析頁面內容"""
soup = BeautifulSoup(html, "lxml")
title = soup.select_one("h1.article-title")
if not title:
return None
content_paragraphs = soup.select(".article-content p")
content = "\n".join(p.get_text(strip=True) for p in content_paragraphs)
return {
"title": title.get_text(strip=True),
"content": content,
"url": url,
}
async def scrape_urls(self, urls: list[str]):
"""並發爬取多個 URL"""
semaphore = asyncio.Semaphore(self.max_concurrency)
async def limited_fetch(session, url):
async with semaphore:
result = await self.fetch_page(session, url)
await asyncio.sleep(self.request_delay)
return url, result
async with aiohttp.ClientSession() as session:
tasks = [
limited_fetch(session, url)
for url in urls if url not in self.visited
]
responses = await asyncio.gather(*tasks, return_exceptions=True)
for resp in responses:
if isinstance(resp, Exception):
continue
url, html = resp
if html:
self.visited.add(url)
parsed = self.parse_page(html, url)
if parsed:
self.results.append(parsed)
return self.results
def save_to_json(self, filename: str = "scraped_data.json"):
"""儲存結果到 JSON"""
with open(filename, "w", encoding="utf-8") as f:
json.dump(self.results, f, ensure_ascii=False, indent=2)
print(f"已儲存 {len(self.results)} 筆資料到 {filename}")
async def main():
scraper = AsyncScraper(
base_url="https://example.com", max_concurrency=8
)
urls = [f"https://example.com/article/{i}" for i in range(1, 51)]
start = time.perf_counter()
results = await scraper.scrape_urls(urls)
elapsed = time.perf_counter() - start
print(f"抓取完成: {len(results)} 筆，耗時 {elapsed:.2f}s")
scraper.save_to_json()
asyncio.run(main())
方案五：資料管線與儲存
CSV 匯出
import csv
from dataclasses import dataclass
@dataclass
class CsvExporter:
filename: str
fieldnames: list[str]
def export(self, data: list[dict]):
with open(self.filename, "w", newline="", encoding="utf-8-sig") as f:
writer = csv.DictWriter(f, fieldnames=self.fieldnames)
writer.writeheader()
writer.writerows(data)
print(f"已匯出 {len(data)} 筆到 {self.filename}")
exporter = CsvExporter("products.csv", ["name", "price", "rating", "url"])
exporter.export(products_data)
JSON 匯出
import json
from datetime import datetime
class JsonExporter:
def __init__(self, filename: str, indent: int = 2):
self.filename = filename
self.indent = indent
def export(self, data: list[dict]):
output = {
"metadata": {
"total": len(data),
"export_time": datetime.now().isoformat(),
"version": "1.0",
},
"data": data,
}
with open(self.filename, "w", encoding="utf-8") as f:
json.dump(output, f, ensure_ascii=False, indent=self.indent)
print(f"已匯出 {len(data)} 筆到 {self.filename}")
SQLite 儲存
import sqlite3
from contextlib import contextmanager
class SqliteStorage:
def __init__(self, db_path: str = "scraped.db"):
self.db_path = db_path
self._init_db()
@contextmanager
def get_connection(self):
conn = sqlite3.connect(self.db_path)
conn.row_factory = sqlite3.Row
try:
yield conn
finally:
conn.close()
def _init_db(self):
with self.get_connection() as conn:
conn.execute("""
CREATE TABLE IF NOT EXISTS scraped_data (
id INTEGER PRIMARY KEY AUTOINCREMENT,
title TEXT NOT NULL,
content TEXT,
url TEXT UNIQUE,
tags TEXT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
conn.execute("""
CREATE INDEX IF NOT EXISTS idx_url ON scraped_data(url)
""")
conn.commit()
def insert(self, item: dict):
with self.get_connection() as conn:
conn.execute(
"""INSERT OR IGNORE INTO scraped_data
(title, content, url, tags) VALUES (?, ?, ?, ?)""",
(
item.get("title"),
item.get("content"),
item.get("url"),
",".join(item.get("tags", [])),
),
)
conn.commit()
def query(self, keyword: str, limit: int = 50) -> list[dict]:
with self.get_connection() as conn:
rows = conn.execute(
"""SELECT * FROM scraped_data
WHERE title LIKE ? OR content LIKE ?
ORDER BY created_at DESC LIMIT ?""",
(f"%{keyword}%", f"%{keyword}%", limit),
).fetchall()
return [dict(row) for row in rows]
常見陷阱指南
陷阱一：忽略 robots.txt
❌ 直接暴力爬取，無視網站爬取規則
✅ 先檢查 robots.txt，尊重 Disallow 規則；Scrapy 預設開啟 ROBOTSTXT_OBEY
陷阱二：固定請求間隔
❌ 使用固定 time.sleep(2) 等待，模式過於規律
✅ 使用隨機延遲 await asyncio.sleep(random.uniform(1.5, 4.0))，模擬人類瀏覽行為
陷阱三：忽略異常重試
❌ 請求失敗直接跳過，資料大量遺失
✅ 實現指數退避重試機制，區分可重試錯誤（429、503）和不可重試錯誤（404、403）
陷阱四：記憶體中累積全部資料
❌ 將所有結果儲存在列表中，百萬級資料導致 OOM
✅ 使用串流寫入，每批 N 筆寫入檔案/資料庫後清空快取
陷阱五：硬編碼選擇器
❌ 直接硬編碼 CSS/XPath 選擇器，網站改版後全部失效
✅ 使用多種選擇器策略作為降級方案，或基於資料特徵（如 JSON-LD 結構化資料）提取
錯誤排錯表
| 錯誤現象 | 可能原因 | 解決方案 |
|---|---|---|
| HTTP 403 Forbidden | 請求標頭缺失或被識別為爬蟲 | 新增完整瀏覽器請求標頭，使用指紋偽裝 |
| HTTP 429 Too Many Requests | 請求頻率過高 | 降低並發數，增加隨機延遲，使用代理輪換 |
| 連線逾時 TimeoutError | 網路不穩定或目標伺服器限速 | 增加逾時時間，實現重試機制，切換代理 |
| SSL 驗證失敗 | 代理中間人憑證或 TLS 指紋偵測 | 設定 ssl=False（開發環境），或使用 curl_cffi |
| 頁面內容為空 | JS 動態渲染未完成 | 切換到 Playwright，等待元素載入 |
| CAPTCHA 驗證碼攔截 | 觸發反爬規則 | 降低請求頻率，接入打碼平台，使用 Cookie 池 |
| 資料重複抓取 | URL 去重邏輯缺失 | 使用 Bloom Filter 或 Redis Set 去重 |
| 記憶體溢位 OOM | 大量資料堆積在記憶體 | 串流寫入磁碟，分批處理，使用產生器 |
| 編碼錯誤 UnicodeDecodeError | 回應編碼與宣告不一致 | 使用 response.encoding = response.apparent_encoding |
| ElementNotInteractable | 元素被遮擋或尚未可互動 | 等待元素可見 wait_for_selector，捲動到元素位置 |
進階最佳化技巧
1. Bloom Filter 去重
from pybloom_live import ScalableBloomFilter
class UrlDeduplicator:
def __init__(self):
self.bloom = ScalableBloomFilter(
initial_capacity=100000, error_rate=0.001,
)
def is_duplicate(self, url: str) -> bool:
if url in self.bloom:
return True
self.bloom.add(url)
return False
2. 智慧限流 AutoThrottle
# Scrapy 設定
AUTOTHROTTLE_ENABLED = True
AUTOTHROTTLE_START_DELAY = 1.0
AUTOTHROTTLE_MAX_DELAY = 30.0
AUTOTHROTTLE_TARGET_CONCURRENCY = 8.0
AUTOTHROTTLE_DEBUG = True
3. 分散式爬取（Scrapy + Redis）
pip install scrapy-redis
# settings.py
SCHEDULER = "scrapy_redis.scheduler.Scheduler"
DUPEFILTER_CLASS = "scrapy_redis.dupefilter.RFPDupeFilter"
REDIS_URL = "redis://localhost:6379/0"
SCHEDULER_PERSIST = True
4. curl_cffi 繞過 TLS 指紋
from curl_cffi import requests as curl_requests
response = curl_requests.get(
"https://tls-protected.example.com",
impersonate="chrome126",
timeout=15,
)
print(response.status_code)
5. Cookie 池管理
import json
import random
from pathlib import Path
class CookiePool:
def __init__(self, cookie_file: str = "cookies.json"):
self.cookie_file = Path(cookie_file)
self.cookies: list[dict] = []
self._load()
def _load(self):
if self.cookie_file.exists():
self.cookies = json.loads(
self.cookie_file.read_text(encoding="utf-8")
)
def _save(self):
self.cookie_file.write_text(
json.dumps(self.cookies, ensure_ascii=False, indent=2),
encoding="utf-8",
)
def add_cookie(self, cookie: dict):
self.cookies.append(cookie)
self._save()
def get_random_cookie(self) -> dict | None:
if not self.cookies:
return None
return random.choice(self.cookies)
def remove_cookie(self, cookie: dict):
self.cookies = [c for c in self.cookies if c != cookie]
self._save()
框架對比選型
| 維度 | Scrapy | Playwright | Selenium | BeautifulSoup + requests |
|---|---|---|---|---|
| 適用場景 | 大規模結構化爬取 | 動態 SPA 頁面 | 舊系統相容 | 小規模靜態頁面 |
| JS 渲染 | 需 scrapy-playwright | 原生支援 | 原生支援 | 不支援 |
| 效能 | 高（非同步+Twisted） | 中（瀏覽器實例） | 低（WebDriver 開銷） | 高（輕量 HTTP） |
| 反爬能力 | 中（需額外設定） | 高（真實瀏覽器） | 中（易被偵測） | 低（裸請求） |
| 學習曲線 | 陡峭 | 中等 | 簡單 | 簡單 |
| 分散式 | 原生支援（scrapy-redis） | 需自建 | 需自建 | 需自建 |
| 資料管線 | 內建 Pipeline | 需自建 | 需自建 | 需自建 |
| 除錯工具 | Scrapy Shell | Playwright Inspector | Selenium IDE | 瀏覽器 DevTools |
| 資源佔用 | 低 | 高（瀏覽器程序） | 高（WebDriver） | 極低 |
| 推薦指數 | ⭐⭐⭐⭐⭐ | ⭐⭐⭐⭐ | ⭐⭐⭐ | ⭐⭐⭐ |
選型建議：
- 大規模結構化資料採集 → Scrapy
- 動態 SPA / 需要互動操作 → Playwright
- 簡單靜態頁面快速抓取 → BeautifulSoup + requests
- 已有 Selenium 測試體系 → Selenium（但建議遷移 Playwright）
相關工具推薦
在爬蟲開發實踐中，以下 工具庫 工具可以幫到你：
- JSON 格式化 — 格式化 API 回應和抓取結果，快速排查資料結構問題
- Hash 計算 — 產生內容雜湊用於去重，計算 URL 指紋
- cURL 轉程式碼 — 將瀏覽器 cURL 請求一鍵轉為 Python 程式碼，快速建構請求範本
爬蟲的本質是「與反爬系統的博弈」。沒有銀彈，只有根據目標網站特徵選擇合適的技術棧，在效率與隱蔽性之間找到平衡。遵守 robots.txt、控制請求頻率、尊重資料版權，是每個爬蟲開發者的底線。
本站提供瀏覽器本地工具，免註冊即可試用 →
#Python爬虫#反爬虫#Scrapy#Playwright爬虫#2026#编程语言