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-CN,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-CN",
)
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-CN,zh;q=0.9,en;q=0.8",
"zh-TW,zh;q=0.9,en-US;q=0.8",
"en-US,en;q=0.9,zh-CN;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-CN",
timezone_id="Asia/Shanghai",
geolocation={"latitude": 39.9042, "longitude": 116.4074},
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-CN', '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#编程语言