Python 爬虫技术全栈指南
面向读者：有一定 Python 基础的数据开发工程师
版本说明：融入 2025-2026 年最新技术动态，覆盖 curl_cffi、Playwright 新版、AI 辅助逆向等前沿内容
目录
- 爬虫全景概述
- HTTP协议与网络基础
- Python爬虫基础库
- HTML解析与数据提取
- 主流爬虫框架详解
- 动态页面爬取
- 反爬虫技术全解
- 反反爬虫实战
- 逆向工程基础
- JS逆向深度实战
- APP爬虫与抓包
- 分布式爬虫架构
- 数据存储与清洗
- 法律合规与robots协议
- 验证码专项突破
- 高并发与性能优化
- 实战项目
- 2025-2026爬虫新技术趋势
1. 爬虫全景概述
1.1 什么是爬虫
网络爬虫（Web Crawler / Spider）是一种按照一定规则自动抓取万维网信息的程序或脚本。其核心工作流程可以概括为：
┌─────────────────────────────────────────────────────────────┐
│ 爬虫工作流程全景图 │
├─────────────────────────────────────────────────────────────┤
│ │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌────────┐│
│ │ 种子URL │───>│ 发送请求 │───>│ 接收响应 │───>│ 解析提 ││
│ │ 队列 │ │ (HTTP) │ │ (HTML) │ │ 取数据 ││
│ └──────────┘ └──────────┘ └──────────┘ └───┬────┘│
│ ^ │ │
│ │ ┌──────────┐ ┌──────────┐ │ │
│ └─────────│ 调度控制 │<───│ 数据存储 │<────────┘ │
│ │ 引擎 │ │ Pipeline │ │
│ └──────────┘ └──────────┘ │
│ │
└─────────────────────────────────────────────────────────────┘
1.2 爬虫发展历史
| 阶段 | 时间 | 特征 | 代表技术 |
|---|
| 萌芽期 | 1993-2000 | 搜索引擎爬虫诞生 | World Wide Web Wanderer、Googlebot |
| 成长期 | 2000-2010 | 开源框架涌现 | Scrapy(2008)、Nutch、Heritrix |
| 对抗期 | 2010-2020 | 反爬与反反爬博弈 | Selenium、PhantomJS、代理池 |
| 智能化 | 2020-2025 | AI辅助逆向、无头浏览器普及 | Playwright、curl_cffi、GPT逆向 |
| 新纪元 | 2025-2026 | TLS指纹伪装、LLM分析JS | curl_cffi 0.14+、AI Agent爬虫 |
1.3 爬虫分类
| 类型 | 说明 | 适用场景 | 示例 |
|---|
| 通用爬虫 | 大规模抓取全网数据 | 搜索引擎索引 | Googlebot、BaiduSpider |
| 聚焦爬虫 | 针对特定主题/网站 | 电商比价、舆情监控 | 商品价格采集器 |
| 增量爬虫 | 只抓取新增/变更内容 | 新闻监控、更新检测 | 新闻聚合系统 |
| 分布式爬虫 | 多机协同大规模采集 | 全网数据、PB级 | Scrapy-Redis集群 |
| 深度学习爬虫 | 结合AI进行智能提取 | 结构化信息抽取 | 基于BERT的网页解析 |
1.4 爬虫 vs RPA vs API调用
| 对比维度 | 网络爬虫 | RPA（机器人流程自动化） | API调用 |
|---|
| 实现方式 | 模拟HTTP请求 | 模拟人类UI操作 | 直接调用接口 |
| 开发难度 | 中等 | 低（可视化配置） | 低（有文档时） |
| 执行效率 | 高 | 低（需渲染UI） | 最高 |
| 稳定性 | 中（受反爬影响） | 低（UI变化敏感） | 高 |
| 数据完整性 | 可控 | 与人类操作一致 | 取决于API设计 |
| 法律风险 | 中-高 | 中 | 低 |
| 典型工具 | Scrapy/Playwright | UiPath/影刀 | requests/httpx |
面试加分提示：在技术选型时，优先评估目标系统是否提供官方API。API调用是最稳定、合规、高效的方式。只有当API无法满足需求（如数据不完整、频率受限）时，才考虑爬虫方案。
1.5 应用场景全景
| 领域 | 典型应用 | 数据量级 |
|---|
| 电商 | 商品比价、销量监控、评价分析 | 百万级SKU |
| 金融 | 股票行情、财报抓取、舆情预警 | 实时/日更 |
| 学术 | 论文检索、期刊下载、专利分析 | 十万级文档 |
| 社交媒体 | 话题追踪、KOL分析、趋势预测 | 亿级消息 |
| 招聘 | 职位聚合、薪资分析、技能需求 | 百万级岗位 |
| 房地产 | 房价监控、楼盘对比、租金指数 | 十万级房源 |
2. HTTP协议与网络基础
2.1 HTTP/HTTPS 原理
HTTP（HyperText Transfer Protocol）是 Web 数据通信的基础协议。HTTPS 在 HTTP 基础上增加了 TLS/SSL 加密层。
┌─────────────────────────────────────────────────────────┐
│ HTTP vs HTTPS 协议栈对比 │
├─────────────────────────────────────────────────────────┤
│ │
│ HTTP 协议栈: HTTPS 协议栈: │
│ ┌──────────────┐ ┌──────────────┐ │
│ │ HTTP/1.1 │ │ HTTP/2 │ │
│ ├──────────────┤ ├──────────────┤ │
│ │ │ │ TLS 1.3 │ │
│ │ │ ├──────────────┤ │
│ ├──────────────┤ │ │ │
│ │ TCP │ ├──────────────┤ │
│ ├──────────────┤ │ TCP │ │
│ │ IP │ ├──────────────┤ │
│ └──────────────┘ │ IP │ │
│ └──────────────┘ │
└─────────────────────────────────────────────────────────┘
HTTPS 握手过程（TLS 1.3 简化版）：
客户端 服务端
│ │
│──── ClientHello ──────────────────────>│ (支持的密码套件、扩展、SNI)
│ │
│<──── ServerHello ─────────────────────│ (选定的密码套件、证书)
│<──── Certificate ─────────────────────│ (服务器证书链)
│<──── EncryptedExtensions ─────────────│
│<──── Finished ────────────────────────│
│ │
│──── Finished ─────────────────────────>│ (握手完成，开始加密通信)
│ │
│<════ 加密的应用数据 ══════════════════>│
2.2 请求响应模型
import requests
response = requests.get(
url="https://httpbin.org/get",
params={"page": 1, "size": 20},
headers={
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
"AppleWebKit/537.36 (KHTML, like Gecko) "
"Chrome/131.0.0.0 Safari/537.36",
"Accept": "application/json",
"Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8",
"Accept-Encoding": "gzip, deflate, br",
"Connection": "keep-alive",
},
cookies={"session_id": "abc123def456"},
timeout=(5, 30),
allow_redirects=True,
verify=True,
)
print(f"状态码: {response.status_code}")
print(f"响应头: {response.headers['Content-Type']}")
print(f"编码: {response.encoding}")
print(f"JSON数据: {response.json()}")
print(f"请求耗时: {response.elapsed.total_seconds():.3f}s")
2.3 常见状态码
| 状态码 | 含义 | 爬虫应对策略 |
|---|
| 200 | 成功 | 正常解析 |
| 301 | 永久重定向 | 更新URL库 |
| 302 | 临时重定向 | 跟踪Location头 |
| 304 | 未修改（缓存） | 利用缓存策略 |
| 400 | 请求错误 | 检查请求参数 |
| 401 | 未授权 | 需要认证信息 |
| 403 | 禁止访问 | 可能触发反爬，检查UA/IP |
| 404 | 资源不存在 | 跳过该URL |
| 429 | 请求过多 | 必须降速，增加延迟 |
| 500 | 服务器错误 | 重试或跳过 |
| 503 | 服务不可用 | 可能是反爬（如Cloudflare），等待重试 |
生产踩坑提醒：429 状态码是最明显的"你被封了"信号。生产环境必须实现指数退避重试策略，遇到 429 立即停止请求至少 60 秒。
2.4 Cookie/Session 机制
import requests
session = requests.Session()
login_resp = session.post("https://httpbin.org/post", data={
"username": "test_user",
"password": "test_pass_123"
})
profile_resp = session.get("https://httpbin.org/cookies")
print(f"Cookies: {session.cookies.get_dict()}")
session.cookies.set("token", "eyJhbGciOiJIUzI1NiJ9.xxxx", domain=".example.com")
2.5 WebSocket 基础
import asyncio
import websockets
import json
async def ws_crawler():
"""WebSocket数据抓取示例"""
uri = "wss://stream.binance.com:9443/ws/btcusdt@trade"
async with websockets.connect(uri) as websocket:
for i in range(100):
message = await websocket.recv()
data = json.loads(message)
print(f"价格: {data['p']}, 数量: {data['q']}, 时间: {data['T']}")
2.6 HTTP/2 特性
| 特性 | HTTP/1.1 | HTTP/2 | 爬虫影响 |
|---|
| 多路复用 | 不支持，队头阻塞 | 支持，单连接多请求 | 减少连接数 |
| 头部压缩 | 不支持 | 支持 HPACK | 减少带宽 |
| 服务器推送 | 不支持 | 支持 | 可预取资源 |
| 二进制帧 | 文本 | 二进制 | 解析更高效 |
| 流优先级 | 不支持 | 支持 | 资源加载优化 |
面试加分提示：2025 年起，越来越多网站强制要求 HTTP/2 支持。Python 的 httpx 和 curl_cffi 都原生支持 HTTP/2，而传统 requests 库只支持到 HTTP/1.1。
3. Python爬虫基础库
3.1 四大HTTP客户端对比
| 特性 | requests | httpx | aiohttp | curl_cffi |
|---|
| 同步/异步 | 同步 | 同步+异步 | 异步 | 同步+异步 |
| HTTP/2 | 不支持 | 支持 | 支持 | 支持 |
| TLS指纹伪装 | 不支持 | 不支持 | 不支持 | 支持 |
| 浏览器模拟 | 不支持 | 不支持 | 不支持 | 支持 impersonate |
| 连接池 | 支持 | 支持 | 支持 | 支持 |
| WebSocket | 不支持 | 支持 | 支持 | 不支持 |
| 自动重试 | 不支持 | 支持 | 支持 | 支持 |
| 学习曲线 | 低 | 低 | 中 | 低 |
| 适用场景 | 简单脚本 | 现代替代 | 高并发 | 反爬绕过 |
| 维护状态 | 维护中 | 活跃 | 活跃 | 活跃(2025热门) |
3.2 requests 进阶用法
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
def create_robust_session():
session = requests.Session()
retry_strategy = Retry(
total=3,
backoff_factor=1,
status_forcelist=[429, 500, 502, 503, 504],
allowed_methods=["GET", "POST"],
raise_on_status=False,
)
adapter = HTTPAdapter(
max_retries=retry_strategy,
pool_connections=10,
pool_maxsize=20,
)
session.mount("http://", adapter)
session.mount("https://", adapter)
return session
proxies = {
"http": "http://user:pass@proxy.example.com:8080",
"https": "http://user:pass@proxy.example.com:8080",
}
session = create_robust_session()
resp = session.get("https://httpbin.org/get", proxies=proxies, timeout=10)
3.3 httpx 异步实战
import httpx
import asyncio
async def async_crawler():
"""httpx异步爬虫示例 - 高并发请求"""
urls = [
"https://httpbin.org/get",
"https://httpbin.org/headers",
"https://httpbin.org/cookies",
]
async with httpx.AsyncClient(
http2=True,
timeout=httpx.Timeout(10.0),
limits=httpx.Limits(
max_connections=100,
max_keepalive_connections=20,
),
follow_redirects=True,
) as client:
tasks = [client.get(url) for url in urls]
responses = await asyncio.gather(*tasks, return_exceptions=True)
for url, resp in zip(urls, responses):
if isinstance(resp, httpx.Response):
print(f"{url} -> {resp.status_code}")
else:
print(f"{url} -> ERROR: {resp}")
3.4 curl_cffi：TLS指纹伪装的利器
from curl_cffi import requests as cffi_requests
session = cffi_requests.Session(impersonate="chrome136")
response = session.get("https://tls.browserleaks.com/json")
data = response.json()
print(f"JA3 Hash: {data.get('ja3_hash', 'N/A')}")
print(f"HTTP Version: {data.get('http_version', 'N/A')}")
生产踩坑提醒：curl_cffi 的 JA3 哈希每次运行都会变化，这是正常行为。因为 GREASE 机制会在 ClientHello 中插入随机值，真实 Chrome 也是如此。关键是握手的"形状"（密码套件、扩展的组成和顺序）要匹配目标浏览器。
4. HTML解析与数据提取
4.1 六大解析方案对比
| 方案 | 速度 | 易用性 | 功能 | 依赖 | 推荐场景 |
|---|
| 正则表达式 | 极快 | 中等 | 基础 | 无 | 简单模式匹配 |
| BeautifulSoup | 较慢 | 极好 | 全面 | bs4 | 快速开发 |
| lxml/XPath | 极快 | 中等 | 强大 | lxml | 高性能解析 |
| CSS选择器 | 快 | 好 | 中等 | cssselect | 前端开发者 |
| pyquery | 快 | 好 | jQuery风格 | pyquery | jQuery熟悉者 |
| selectolax | 极快 | 中等 | 基础 | selectolax | 极致性能 |
4.2 实战代码对比
html = """
<html>
<body>
<div class="product-list">
<div class="item" data-id="1001">
<h3 class="title">Python编程从入门到实践</h3>
<span class="price">89.00</span>
<span class="stock">有货</span>
<a href="/book/1001" class="link">详情</a>
</div>
<div class="item" data-id="1002">
<h3 class="title">流畅的Python</h3>
<span class="price">128.00</span>
<span class="stock">缺货</span>
<a href="/book/1002" class="link">详情</a>
</div>
</div>
</body>
</html>
"""
from bs4 import BeautifulSoup
soup = BeautifulSoup(html, "lxml")
for item in soup.select(".item"):
title = item.select_one(".title").get_text()
price = item.select_one(".price").get_text()
book_id = item.get("data-id")
print(f"[BS4] ID:{book_id} {title} {price}")
from lxml import etree
tree = etree.HTML(html)
items = tree.xpath('//div[@class="item"]')
for item in items:
title = item.xpath('.//h3[@class="title"]/text()')[0]
price = item.xpath('.//span[@class="price"]/text()')[0]
book_id = item.get("data-id")
print(f"[XPath] ID:{book_id} {title} {price}")
import re
titles = re.findall(r'class="title">(.*?)</h3>', html)
prices = re.findall(r'class="price">(.*?)</span>', html)
ids = re.findall(r'data-id="(\d+)"', html)
for i, t, p in zip(ids, titles, prices):
print(f"[Regex] ID:{i} {t} {p}")
from pyquery import PyQuery as pq
doc = pq(html)
for item in doc(".item").items():
title = item(".title").text()
price = item(".price").text()
book_id = item.attr("data-id")
print(f"[PyQuery] ID:{book_id} {title} {price}")
4.3 性能基准测试参考
测试条件：10000次重复解析同一HTML页面
┌─────────────────┬──────────┬──────────┐
│ 方案 │ 耗时(s) │ 内存(MB) │
├─────────────────┼──────────┼──────────┤
│ selectolax │ 0.3 │ 2.1 │
│ lxml (XPath) │ 0.8 │ 5.3 │
│ parsel │ 1.2 │ 6.1 │
│ pyquery │ 1.5 │ 7.8 │
│ BS4 + lxml │ 3.2 │ 12.4 │
│ BS4+html.parser │ 8.5 │ 10.2 │
└─────────────────┴──────────┴──────────┘
面试加分提示：在高并发爬虫场景（如 Scrapy），推荐使用 parsel（Scrapy内置）或 lxml，它们基于 C 扩展，性能是 BeautifulSoup 的 3-10 倍。selectolax 基于 Modest/lexbor 引擎，是目前最快的 Python HTML 解析库。
5. 主流爬虫框架详解
5.1 Scrapy 架构全景
┌─────────────────────────────────────────────────────────────────┐
│ Scrapy 核心架构 │
├─────────────────────────────────────────────────────────────────┤
│ │
│ ┌──────────┐ │
│ │ Engine │ (核心引擎，驱动数据流) │
│ └────┬─────┘ │
│ │ │
│ ┌──────────────┼──────────────┐ │
│ │ │ │ │
│ ┌─────────▼──┐ ┌───────▼──────┐ ┌───▼──────────┐ │
│ │ Scheduler │ │ Downloader │ │ Spider │ │
│ │ (调度器) │ │ (下载器) │ │ (爬虫逻辑) │ │
│ └─────┬──────┘ └───────┬──────┘ └─────┬────────┘ │
│ │ │ │ │
│ ┌─────▼──────┐ ┌──────▼───────┐ ┌────▼─────────┐ │
│ │ 请求队列 │ │ 下载中间件 │ │ 解析回调 │ │
│ │(Redis/本地)│ │(代理/重试/UA)│ │ (parse) │ │
│ └────────────┘ └──────────────┘ └──────┬───────┘ │
│ │ │
│ ┌──────▼───────┐ │
│ │ Pipeline │ │
│ │(清洗/验证/存储)│ │
│ └──────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────┘
5.2 Scrapy 核心组件
| 组件 | 职责 | 关键配置 |
|---|
| Engine | 控制数据流，触发事件 | 自动管理 |
| Scheduler | 管理请求队列，调度优先级 | SCHEDULER |
| Downloader | 执行HTTP请求，获取网页 | DOWNLOAD_TIMEOUT |
| Spider | 解析响应，提取数据/新请求 | 自定义 parse() |
| Item | 定义数据结构 | scrapy.Item |
| Pipeline | 清洗、验证、存储数据 | ITEM_PIPELINES |
| Middleware | 拦截/修改请求和响应 | DOWNLOADER_MIDDLEWARES |
5.3 Scrapy 实战示例
import scrapy
class BookItem(scrapy.Item):
title = scrapy.Field()
price = scrapy.Field()
rating = scrapy.Field()
url = scrapy.Field()
import scrapy
from ..items import BookItem
class BooksSpider(scrapy.Spider):
name = "books"
allowed_domains = ["books.toscrape.com"]
start_urls = ["https://books.toscrape.com/"]
custom_settings = {
"DOWNLOAD_DELAY": 1,
"CONCURRENT_REQUESTS_PER_DOMAIN": 4,
}
def parse(self, response):
for book in response.css("article.product_pod"):
item = BookItem()
item["title"] = book.css("h3 a::attr(title)").get()
item["price"] = book.css(".price_color::text").get()
item["rating"] = book.css("p.star-rating::attr(class)").get()
item["url"] = response.urljoin(book.css("h3 a::attr(href)").get())
yield item
next_page = response.css("li.next a::attr(href)").get()
if next_page:
yield response.follow(next_page, self.parse)
import pymongo
class MongoDBPipeline:
def __init__(self, mongo_uri, db_name):
self.mongo_uri = mongo_uri
self.db_name = db_name
@classmethod
def from_crawler(cls, crawler):
return cls(
mongo_uri=crawler.settings.get("MONGO_URI"),
db_name=crawler.settings.get("MONGO_DB"),
)
def open_spider(self, spider):
self.client = pymongo.MongoClient(self.mongo_uri)
self.db = self.client[self.db_name]
def process_item(self, item, spider):
self.db["books"].update_one(
{"url": item["url"]},
{"$set": dict(item)},
upsert=True,
)
return item
def close_spider(self, spider):
self.client.close()
5.4 框架横向对比
| 框架 | 语言 | 异步 | 分布式 | 动态渲染 | 学习曲线 | 适用规模 |
|---|
| Scrapy | Python | 支持(asyncio) | 支持(scrapy-redis) | 需扩展 | 中 | 大型 |
| httpx+bs4 | Python | 支持(asyncio) | 需自建 | 需配合 | 低 | 小型 |
| Feapder | Python | 支持 | 支持 | 内置Playwright | 低 | 中型 |
| Crawley | Python | 不支持 | 不支持 | 不支持 | 低 | 小型 |
| PySpider | Python | 支持 | 支持 | 支持 | 中 | 中型(已停更) |
| Colly | Go | 支持 | 需自建 | 不支持 | 中 | 大型 |
生产踩坑提醒：2025年 Scrapy 默认 download_delay 从 0 秒改为 1 秒，concurrent_requests_per_domain 从 8 降为 1。这是为了降低对目标网站的压力、降低法律风险。如需高并发，需手动调整配置。
6. 动态页面爬取
6.1 三大浏览器自动化工具对比
| 特性 | Selenium | Playwright | Puppeteer |
|---|
| 开发商 | SeleniumHQ | Microsoft | Google |
| 浏览器支持 | Chrome/Firefox/Safari/Edge | Chromium/Firefox/WebKit | 仅Chromium |
| 语言支持 | Python/Java/C#/JS | Python/JS/Java/.NET | JS/Python(非官方) |
| 自动等待 | 不支持，需手动 | 支持智能等待 | 不支持，需手动 |
| 网络拦截 | 有限 | 强大 | 强大 |
| 执行速度 | 慢 | 快（快30%以上） | 快 |
| 无头模式 | 支持 | 支持（默认） | 支持 |
| 移动端模拟 | 有限 | 支持设备描述符 | 支持 |
| 截图/录屏 | 截图 | 截图+录屏 | 截图+录屏 |
| 社区活跃度 | 高（老牌） | 极高（新星） | 中 |
6.2 Playwright 实战
from playwright.sync_api import sync_playwright
import json
def crawl_spa_page():
"""使用Playwright爬取SPA动态页面"""
with sync_playwright() as p:
browser = p.chromium.launch(
headless=True,
args=["--disable-blink-features=AutomationControlled"]
)
context = 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/131.0.0.0 Safari/537.36",
locale="zh-CN",
timezone_id="Asia/Shanghai",
)
page = context.new_page()
api_responses = []
def handle_response(response):
if "/api/" in response.url:
try:
api_responses.append({
"url": response.url,
"status": response.status,
"body": response.json()
})
except Exception:
pass
page.on("response", handle_response)
page.goto("https://quotes.toscrape.com/js/")
page.wait_for_selector(".quote", state="visible", timeout=10000)
quotes = page.query_selector_all(".quote")
results = []
for quote in quotes:
text = quote.query_selector(".text").inner_text()
author = quote.query_selector(".author").inner_text()
results.append({"text": text, "author": author})
print(f"提取到 {len(results)} 条名言")
print(f"拦截到 {len(api_responses)} 个API响应")
browser.close()
return results
6.3 JS渲染等待策略
| 策略 | 说明 | 适用场景 | 代码示例 |
|---|
| 固定等待 | time.sleep(n) | 调试用，不推荐 | time.sleep(3) |
| 元素等待 | 等待特定元素出现 | 最常用 | wait_for_selector(".data") |
| 网络空闲 | 等待无新请求 | SPA加载 | wait_for_load_state("networkidle") |
| DOM加载 | DOM解析完成 | 静态内容 | wait_for_load_state("domcontentloaded") |
| 自定义条件 | 轮询检查 | 复杂场景 | page.wait_for_function(...) |
page.wait_for_function(
"() => !document.querySelector('.status').textContent.includes('loading')"
)
with page.expect_response("**/api/data*") as response_info:
page.click("#load-button")
response = response_info.value
data = response.json()
面试加分提示：Playwright 的 expect_response 是拦截 API 数据的利器。很多 SPA 页面的数据通过 JSON API 返回，直接拦截 API 响应比解析 DOM 更高效、更稳定。
7. 反爬虫技术全解
7.1 反爬技术分类全景
┌──────────────────────────────────────────────────────────────────┐
│ 反爬虫技术体系全景 │
├──────────────────────────────────────────────────────────────────┤
│ │
│ 请求层检测: │
│ +------------------------------------------------------------+ │
│ | User-Agent检测 | Referer校验 | Accept头校验 | │
│ | Cookie完整性 | 请求头顺序 | Accept-Language | │
│ +------------------------------------------------------------+ │
│ │
│ 网络层检测: │
│ +------------------------------------------------------------+ │
│ | IP频率限制 | IP信誉评分 | TLS指纹(JA3/JA4) | │
│ | HTTP/2指纹 | TCP指纹 | 地理位置检测 | │
│ +------------------------------------------------------------+ │
│ │
│ 行为层检测: │
│ +------------------------------------------------------------+ │
│ | 请求频率异常 | 鼠标轨迹分析 | 页面停留时间 | │
│ | 滚动行为模式 | 点击模式 | 浏览器指纹(Canvas/WebGL) | │
│ +------------------------------------------------------------+ │
│ │
│ 验证层挑战: │
│ +------------------------------------------------------------+ │
│ | 图片验证码 | 滑块验证码 | 点选验证码 | │
│ | 行为验证码 | reCAPTCHA v3 | hCaptcha | │
│ +------------------------------------------------------------+ │
│ │
│ 高级防护: │
│ +------------------------------------------------------------+ │
│ | JS混淆/JSVMP | 蜜罐陷阱 | 动态Token | │
│ | 字体反爬 | Cookie加密 | 瑞数/数美商业防护 | │
│ +------------------------------------------------------------+ │
│ │
└──────────────────────────────────────────────────────────────────┘
7.2 各层反爬详解
| 反爬类型 | 检测原理 | 难度 | 典型产品 |
|---|
| User-Agent检测 | 检查UA是否为已知爬虫 | 低 | 基础防护 |
| IP限制 | 单IP请求频率过高则封禁 | 中 | 大多数网站 |
| Cookie验证 | 首次访问设置加密Cookie | 高 | Cloudflare |
| JS挑战 | 执行JS计算生成Token | 高 | 瑞数信息 |
| TLS指纹 | 检测JA3/JA4指纹 | 很高 | Akamai/DataDome |
| 浏览器指纹 | Canvas/WebGL/Audio指纹 | 很高 | Cloudflare |
| 行为分析 | 鼠标轨迹、点击模式 | 很高 | 数美/极验 |
| JSVMP | JS虚拟化保护 | 极高 | 瑞数/数美 |
7.3 蜜罐陷阱检测
from lxml import etree
hidden_html = """
<div style="display:none; position:absolute; left:-9999px;">
<a href="/admin/secret" class="trap">管理入口</a>
<a href="/api/internal" class="trap">内部接口</a>
</div>
<a href="/real-page" class="real">正常链接</a>
"""
def is_visible(element):
"""检查元素是否可见（简化版）"""
style = element.get("style", "")
parent = element.getparent()
if "display:none" in style or "display: none" in style:
return False
if "left:-9999" in style or "top:-9999" in style:
return False
if parent is not None and parent.tag != "html":
return is_visible(parent)
return True
tree = etree.HTML(hidden_html)
for link in tree.xpath("//a"):
if is_visible(link):
print(f"可见链接: {link.get('href')}")
else:
print(f"蜜罐链接，跳过: {link.get('href')}")
生产踩坑提醒：很多网站使用蜜罐链接检测爬虫。如果你的爬虫无差别地跟随所有链接，很可能触发蜜罐而被封禁。建议在提取链接时，检查元素的可见性。
8. 反反爬虫实战
8.1 代理池搭建（Redis + FastAPI）
import redis
import time
import random
import asyncio
from fastapi import FastAPI
from typing import Optional
from datetime import datetime
app = FastAPI(title="Proxy Pool Service")
redis_client = redis.Redis(host="localhost", port=6379, db=0, decode_responses=True)
PROXY_KEY = "proxy_pool"
class ProxyManager:
"""代理池管理器"""
MAX_SCORE = 100
MIN_SCORE = 0
INIT_SCORE = 10
CHECK_INTERVAL = 300
@classmethod
async def is_valid_proxy(cls, proxy: str) -> bool:
"""验证代理是否可用"""
import httpx
try:
async with httpx.AsyncClient(
proxies={"all://": f"http://{proxy}"},
timeout=10
) as client:
resp = await client.get("https://httpbin.org/get")
return resp.status_code == 200
except Exception:
return False
@classmethod
async def add_proxy(cls, proxy: str):
"""添加代理到池中"""
if not redis_client.hexists(PROXY_KEY, proxy):
redis_client.hset(PROXY_KEY, proxy, cls.INIT_SCORE)
@classmethod
async def get_random_proxy(cls) -> Optional[str]:
"""随机获取一个高分代理"""
proxies = {}
all_proxies = redis_client.hgetall(PROXY_KEY)
for proxy, score in all_proxies.items():
if int(score) >= 50:
proxies[proxy] = int(score)
if not proxies:
return None
total = sum(proxies.values())
r = random.uniform(0, total)
cumulative = 0
for proxy, score in proxies.items():
cumulative += score
if r <= cumulative:
return proxy
return list(proxies.keys())[0]
@classmethod
async def increase_score(cls, proxy: str):
"""代理可用时加分"""
current = int(redis_client.hget(PROXY_KEY, proxy) or 0)
if current < cls.MAX_SCORE:
redis_client.hset(PROXY_KEY, proxy, current + 1)
@classmethod
async def decrease_score(cls, proxy: str):
"""代理不可用时减分"""
current = int(redis_client.hget(PROXY_KEY, proxy) or 0)
if current > cls.MIN_SCORE:
redis_client.hset(PROXY_KEY, proxy, current - 1)
else:
redis_client.hdel(PROXY_KEY, proxy)
proxy_mgr = ProxyManager()
@app.get("/proxy/random")
async def get_proxy():
"""获取随机代理"""
proxy = await proxy_mgr.get_random_proxy()
if proxy:
return {"proxy": proxy, "url": f"http://{proxy}"}
return {"error": "no available proxy"}
@app.post("/proxy/add")
async def add_proxy(proxy: str):
"""添加代理"""
await proxy_mgr.add_proxy(proxy)
return {"status": "added", "proxy": proxy}
8.2 验证码识别（ddddocr）
import ddddocr
import requests
ocr = ddddocr.DdddOcr(show_ad=False)
def recognize_captcha(image_url: str) -> str:
"""识别图片验证码"""
response = requests.get(image_url, timeout=10)
image_bytes = response.content
result = ocr.classification(image_bytes)
return result
captcha_text = recognize_captcha("https://example.com/captcha.jpg")
print(f"验证码识别结果: {captcha_text}")
det = ddddocr.DdddOcr(det=False, ocr=False, show_ad=False)
def find_slider_gap(bg_bytes: bytes, slider_bytes: bytes) -> int:
"""找到滑块缺口的x坐标"""
result = det.slide_match(slider_bytes, bg_bytes, simple_target=True)
return result.get("target", [0, 0])[0]
8.3 curl_cffi 实战绕过 TLS 检测
from curl_cffi import requests as cffi_requests
def bypass_cloudflare(url: str):
"""使用curl_cffi绕过Cloudflare的TLS指纹检测"""
session = cffi_requests.Session(
impersonate="chrome136",
verify=True,
)
response = session.get(url, timeout=30)
if "cf-chl-bypass" in response.text.lower():
print("Cloudflare挑战页面，可能需要额外处理")
elif response.status_code == 200:
print(f"成功绕过，状态码: {response.status_code}")
print(f"页面长度: {len(response.text)} 字符")
else:
print(f"请求失败: {response.status_code}")
return response
8.4 请求伪装最佳实践
import random
import time
CHROME_HEADERS_TEMPLATES = [
{
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
"(KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36",
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,"
"image/webp,image/apng,*/*;q=0.8",
"Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8",
"Accept-Encoding": "gzip, deflate, br, zstd",
"Sec-Ch-Ua": '"Chromium";v="131", "Not_A Brand";v="24"',
"Sec-Ch-Ua-Mobile": "?0",
"Sec-Ch-Ua-Platform": '"Windows"',
"Sec-Fetch-Dest": "document",
"Sec-Fetch-Mode": "navigate",
"Sec-Fetch-Site": "none",
"Sec-Fetch-User": "?1",
"Upgrade-Insecure-Requests": "1",
},
]
def random_delay(min_sec: float = 1.0, max_sec: float = 5.0):
"""随机延迟，模拟人类行为"""
delay = random.uniform(min_sec, max_sec)
time.sleep(delay)
def get_stealth_headers() -> dict:
"""获取伪装Headers"""
return random.choice(CHROME_HEADERS_TEMPLATES).copy()
9. 逆向工程基础
9.1 Chrome DevTools 核心面板
| 面板 | 用途 | 爬虫相关功能 |
|---|
| Network | 网络请求分析 | XHR/Fetch过滤、请求重放、参数查看 |
| Sources | 源码调试 | 断点、单步调试、Call Stack追踪 |
| Elements | DOM检查 | 元素审查、事件监听器查看 |
| Console | 控制台 | Hook注入、代码执行、调试输出 |
| Application | 存储查看 | Cookie/LocalStorage/SessionStorage |
| Performance | 性能分析 | 识别关键渲染路径 |
9.2 逆向核心流程
┌───────────────────────────────────────────────────────────────┐
│ JS逆向工程标准流程 │
├───────────────────────────────────────────────────────────────┤
│ │
│ Step 1: 抓包分析 Step 2: 定位入口 │
│ +------------------+ +------------------+ │
│ | Network | | XHR断点 | │
│ | 面板过滤XHR |--> | 全局搜索sign | │
│ | 分析请求参数 | | 搜索加密函数 | │
│ +------------------+ +------------------+ │
│ | │
│ Step 3: 断点调试 v │
│ +------------------+ │
│ | CallStack追踪 | │
│ | 逐步跟踪 | │
│ | 找到加密逻辑 | │
│ +------------------+ │
│ | │
│ Step 4: 算法识别 Step 5: 逻辑还原 Step 6: Python复现 │
│ +------------------+ +------------------+ +----------------+│
│ | MD5/SHA/AES | | AST还原控制流 | | requests本地 ││
│ | RSA/Base64 |->| 字符串表还原 |->| 验证签名 ││
│ | HMAC/自定义 | | 死代码消除 | | 对比结果 ││
│ +------------------+ +------------------+ +----------------+│
│ │
└───────────────────────────────────────────────────────────────┘
9.3 常见加密算法识别
| 代码特征 | 算法 | 输出特征 | 应对方案 |
|---|
CryptoJS.MD5() | MD5 | 32位hex | hashlib.md5 |
CryptoJS.SHA256() | SHA-256 | 64位hex | hashlib.sha256 |
CryptoJS.AES.encrypt() | AES | Base64或hex | pycryptodome |
JSEncrypt / encrypt | RSA | Base64 | rsa库 |
btoa() / atob() | Base64 | 结尾= | base64模块 |
encodeURIComponent | URL编码 | %XX格式 | urllib.parse |
9.4 Hook 技术实战
(function() {
const originalOpen = XMLHttpRequest.prototype.open;
const originalSend = XMLHttpRequest.prototype.send;
XMLHttpRequest.prototype.open = function(method, url) {
this._url = url;
this._method = method;
console.log('[XHR] ' + method + ' ' + url);
return originalOpen.apply(this, arguments);
};
XMLHttpRequest.prototype.send = function(body) {
if (this._url && this._url.includes('/api/')) {
console.log('[XHR Body]', body);
console.trace();
}
return originalSend.apply(this, arguments);
};
})();
(function() {
const originalFetch = window.fetch;
window.fetch = function(...args) {
console.log('[Fetch]', args[0], args[1]);
console.trace();
return originalFetch.apply(this, args);
};
})();
(function() {
const originalSetHeader = XMLHttpRequest.prototype.setRequestHeader;
XMLHttpRequest.prototype.setRequestHeader = function(name, value) {
if (/sign|token|auth|x-/i.test(name)) {
console.log('[Header] ' + name + ': ' + value);
console.trace();
}
return originalSetHeader.apply(this, arguments);
};
})();
(function() {
Object.defineProperty(navigator, 'webdriver', {
get: function() {
console.trace('[检测点] navigator.webdriver 被读取');
return false;
}
});
})();
面试加分提示：XHR断点+Call Stack追踪是JS逆向最基本也最有效的方法。在 DevTools Sources XHR/fetch Breakpoints 中添加接口关键词，触发请求后即可自动断在参数拼装位置，沿调用栈向上追溯即可找到加密逻辑。
10. JS逆向深度实战
10.1 签名参数（sign）破解流程
┌───────────────────────────────────────────────────────────────┐
│ 签名参数逆向完整流程 │
├───────────────────────────────────────────────────────────────┤
│ │
│ Step 1: 抓包发现 │
│ GET /api/products?timestamp=1720000000&sign=a1b2c3d4e5 │
│ ^^^^^^^^^^^^^^^ ^^^^^^^^^^^^^^^^ │
│ 时间戳参数 签名参数(32位hex=MD5?) │
│ │
│ Step 2: 搜索定位 │
│ DevTools -> Ctrl+Shift+F -> 搜索 "sign=" 或 "generateSign" │
│ │
│ Step 3: 断点分析 │
│ 找到关键函数: │
│ function generateSign(params, secret) { │
│ let sorted = Object.keys(params).sort() │
│ .map(k => k + '=' + params[k]).join('&'); │
│ return CryptoJS.MD5(secret + sorted).toString(); │
│ } │
│ │
│ Step 4: Python复现 │
│ sign = md5(secret + sorted_params).hexdigest() │
│ │
└───────────────────────────────────────────────────────────────┘
import hashlib
import time
import requests
def generate_sign(params: dict, secret: str = "x$12*_d#@") -> str:
"""
还原目标网站的签名算法
:param params: 请求参数字典
:param secret: 通过JS逆向获取的密钥
"""
sorted_str = "&".join(
f"{k}={params[k]}" for k in sorted(params.keys())
)
sign_str = secret + sorted_str + secret
return hashlib.md5(sign_str.encode("utf-8")).hexdigest().upper()
def crawl_with_sign():
"""带签名的请求"""
params = {
"page": 1,
"size": 20,
"category": "electronics",
"timestamp": int(time.time()),
}
params["sign"] = generate_sign(params)
response = requests.get(
"https://api.example.com/products",
params=params,
headers={
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
"AppleWebKit/537.36 Chrome/131.0.0.0 Safari/537.36",
"Referer": "https://www.example.com/",
},
)
return response.json()
10.2 Cookie 加密逆向
import re
import json
import base64
from Crypto.Cipher import AES
from Crypto.Util.Padding import unpad
def decrypt_cookie_token(token: str, key: str) -> dict:
"""
解密Cookie中的加密Token
常见于电商网站的加密Cookie
"""
try:
encrypted = base64.b64decode(token)
cipher = AES.new(
key.encode("utf-8")[:16],
AES.MODE_CBC,
iv=key.encode("utf-8")[:16],
)
decrypted = unpad(cipher.decrypt(encrypted), AES.block_size)
return json.loads(decrypted.decode("utf-8"))
except Exception as e:
print(f"解密失败: {e}")
return {}
10.3 字体反爬破解
import io
import fontTools
from fontTools.ttLib import TTFont
import requests
def parse_woff_font(font_bytes: bytes) -> dict:
"""
解析woff字体文件，提取字符映射关系
用于破解字体反爬（常见于电商价格、电话号码）
"""
font = TTFont(io.BytesIO(font_bytes))
cmap = font.getBestCmap()
mapping = {}
for code, name in cmap.items():
mapping[chr(code)] = name
return mapping
生产踩坑提醒：字体反爬的映射关系通常是动态生成的，每次请求都可能不同。你需要在每次请求时同步下载字体文件并重新解析映射表。
10.4 JS混淆对抗
| 混淆技术 | 特征 | 破解方法 |
|---|
| 变量名混淆 | _0x3f2a12 等无意义名称 | AST反混淆+语义分析 |
| 控制流平坦化 | 巨大的switch-case | AST分析+状态机还原 |
| 字符串加密 | 字符串数组+索引访问 | 解密函数定位+批量还原 |
| 死代码注入 | 未使用的函数/变量 | 代码覆盖率分析 |
| JSVMP | 字节码+解释器循环 | 插桩追踪+Hook出口 |
11. APP爬虫与抓包
11.1 抓包工具对比
| 工具 | 平台 | HTTPS解密 | 脚本能力 | 免费 | 推荐场景 |
|---|
| Charles | 全平台 | 支持 | Rewrite规则 | 试用30min | Mac用户首选 |
| Fiddler | Windows | 支持 | FiddlerScript | 免费 | Windows用户 |
| Burp Suite | 全平台 | 支持 | Extension扩展 | Community版 | 安全测试 |
| mitmproxy | 全平台 | 支持 | Python脚本 | 开源免费 | 程序员首选 |
| Wireshark | 全平台 | 需密钥 | 不支持 | 免费 | 底层协议分析 |
11.2 mitmproxy 脚本编写
from mitmproxy import http
import json
import os
API_KEYWORDS = ["/api/", "/graphql", "/v1/", "/v2/"]
class CrawlAddon:
"""mitmproxy爬虫插件"""
def __init__(self):
self.output_dir = "./captured_data"
os.makedirs(self.output_dir, exist_ok=True)
self.request_count = 0
def response(self, flow: http.HTTPFlow):
"""拦截响应"""
url = flow.request.url
if not any(kw in url for kw in API_KEYWORDS):
return
content_type = flow.response.headers.get("content-type", "")
if "json" not in content_type:
return
self.request_count += 1
try:
data = json.loads(flow.response.get_text())
filename = f"{self.output_dir}/req_{self.request_count:04d}.json"
with open(filename, "w", encoding="utf-8") as f:
json.dump({
"url": url,
"method": flow.request.method,
"request_headers": dict(flow.request.headers),
"request_body": flow.request.get_text(),
"status_code": flow.response.status_code,
"response_data": data,
}, f, ensure_ascii=False, indent=2)
print(f"[{self.request_count}] {flow.request.method} {url}")
except json.JSONDecodeError:
pass
addons = [CrawlAddon()]
11.3 APP抓包关键步骤
┌───────────────────────────────────────────────────────────────┐
│ APP抓包完整流程 │
├───────────────────────────────────────────────────────────────┤
│ │
│ [1.安装证书] [2.配置代理] [3.信任证书] │
│ 导出CA证书 --> WiFi代理 --> 系统设置 │
│ 到手机 指向电脑 安装证书 │
│ | │
│ [4.验证抓包] [5.分析协议] [6.模拟请求] │
│ 打开APP --> 筛选API --> Python复现 │
│ 触发操作 分析参数 请求 │
│ │
│ 常见问题: │
│ - Android 7+不信任用户证书 -> Root后安装到系统目录 │
│ - APP证书绑定(Certificate Pinning) -> Frida Hook │
│ - 请求参数加密 -> 逆向APP或Hook加密函数 │
│ │
└───────────────────────────────────────────────────────────────┘
面试加分提示：Android 7.0+ 默认不信任用户安装的 CA 证书。解决方案：(1) Root 后将证书安装到系统目录；(2) 使用 Frida 绕过 Certificate Pinning；(3) 使用已 Root 的模拟器配合 Xposed 的 JustTrustMe 模块。
12. 分布式爬虫架构
12.1 Scrapy-Redis 分布式原理
┌──────────────────────────────────────────────────────────────────┐
│ Scrapy-Redis 分布式架构 │
├──────────────────────────────────────────────────────────────────┤
│ │
│ ┌──────────────────────────────────────────────┐ │
│ │ Redis 中央存储 │ │
│ │ ┌────────────┐ ┌──────────┐ ┌────────────┐ │ │
│ │ │ 请求队列 │ │ 去重集合 │ │ 结果队列 │ │ │
│ │ │ (List/ZSET)│ │ (Set) │ │ (List) │ │ │
│ │ └─────┬──────┘ └────┬─────┘ └──────┬─────┘ │ │
│ └────────┼──────────────┼──────────────┼────────┘ │
│ │ │ │ │
│ ┌──────┼──────────────┼──────────────┼──────┐ │
│ │ │ │ │ │ │
│ [Worker 1] [Worker 2] [Worker N] │
│ +---------+ +----------+ +---------+ │
│ |Scrapy | | Scrapy | | Scrapy | │
│ |Engine | | Engine | | Engine | │
│ +----+----+ +----+-----+ +----+----+ │
│ | | | │
│ [Pipeline] [Pipeline] [Pipeline] │
│ (本地存储) (本地存储) (本地存储) │
│ │
└──────────────────────────────────────────────────────────────────┘
12.2 Scrapy-Redis 配置
SCHEDULER = "scrapy_redis.scheduler.Scheduler"
DUPEFILTER_CLASS = "scrapy_redis.dupefilter.RFPDupeFilter"
REDIS_URL = "redis://:your_password@redis-master:6379/0"
SCHEDULER_PERSIST = True
12.3 布隆过滤器去重
import math
import mmh3
from bitarray import bitarray
class BloomFilter:
"""
布隆过滤器 - 内存高效的URL去重方案
适用于海量URL去重，误判率可控
"""
def __init__(self, expected_count: int, error_rate: float = 0.001):
self.size = self._optimal_size(expected_count, error_rate)
self.hash_count = self._optimal_hash_count(self.size, expected_count)
self.bit_array = bitarray(self.size)
self.bit_array.setall(0)
self.count = 0
@staticmethod
def _optimal_size(n: int, p: float) -> int:
m = -(n * math.log(p)) / (math.log(2) ** 2)
return int(m)
@staticmethod
def _optimal_hash_count(m: int, n: int) -> int:
k = (m / n) * math.log(2)
return int(k)
def _hash_values(self, item: str):
for i in range(self.hash_count):
yield mmh3.hash(item, i) % self.size
def add(self, item: str):
for bit_pos in self._hash_values(item):
self.bit_array[bit_pos] = 1
self.count += 1
def __contains__(self, item: str) -> bool:
return all(
self.bit_array[bit_pos]
for bit_pos in self._hash_values(item)
)
bf = BloomFilter(expected_count=10_000_000, error_rate=0.001)
print(f"位数组大小: {bf.size / 1024 / 1024:.1f} MB")
print(f"哈希函数数: {bf.hash_count}")
bf.add("https://example.com/page/1")
bf.add("https://example.com/page/2")
print("page/1存在:", "https://example.com/page/1" in bf)
print("page/3存在:", "https://example.com/page/3" in bf)
12.4 分布式调度方案对比
| 方案 | 特点 | 适用规模 | 复杂度 |
|---|
| Scrapy-Redis | 简单，Redis队列 | 中小（10节点内） | 低 |
| Celery + Redis | 任务队列成熟 | 中等 | 中 |
| Kafka + Scrapy | 高吞吐，持久化 | 大型（100+节点） | 高 |
| RabbitMQ + Scrapy | 可靠消息传递 | 中大型 | 中 |
| 自研调度（Redis+FastAPI） | 完全可控 | 按需 | 高 |
13. 数据存储与清洗
13.1 存储方案对比
| 数据库 | 类型 | 写入速度 | 查询能力 | 适用场景 | 推荐规模 |
|---|
| MongoDB | 文档型 | 快 | 灵活 | 半结构化数据 | 百万-千万 |
| MySQL | 关系型 | 中 | SQL强大 | 结构化数据 | 十万-百万 |
| PostgreSQL | 关系型 | 中 | SQL+JSON | 复杂查询 | 百万-千万 |
| Elasticsearch | 搜索引擎 | 快 | 全文检索 | 日志/搜索 | 千万-亿 |
| CSV/JSON | 文件 | 极快 | 无 | 临时存储 | 万级 |
| Redis | 缓存 | 极快 | K-V | 队列/缓存 | 内存限制 |
13.2 MongoDB 存储实战
import pymongo
from datetime import datetime
class MongoStorage:
"""MongoDB数据存储管道"""
def __init__(self, uri: str = "mongodb://localhost:27017",
db_name: str = "crawler_data"):
self.client = pymongo.MongoClient(uri)
self.db = self.client[db_name]
def save_items(self, collection_name: str, items: list):
"""批量保存数据"""
collection = self.db[collection_name]
for item in items:
item["crawled_at"] = datetime.utcnow()
operations = []
for item in items:
operations.append(
pymongo.UpdateOne(
{"url": item.get("url")},
{"$set": item},
upsert=True,
)
)
if operations:
result = collection.bulk_write(operations)
print(f"插入: {result.upserted_count}, 更新: {result.modified_count}")
def create_indexes(self, collection_name: str):
"""创建索引加速查询"""
collection = self.db[collection_name]
collection.create_index([("url", pymongo.HASHED)])
collection.create_index([("crawled_at", pymongo.DESCENDING)])
collection.create_index([("title", pymongo.TEXT)])
13.3 数据清洗 Pipeline
import pandas as pd
import re
from typing import List, Dict
class DataCleaner:
"""数据清洗管道"""
@staticmethod
def clean_price(price_str: str) -> float:
"""清洗价格字段"""
if not price_str:
return 0.0
cleaned = re.sub(r'[^0-9.]', '', price_str)
try:
return float(cleaned)
except ValueError:
return 0.0
@staticmethod
def clean_text(text: str) -> str:
"""清洗文本字段"""
if not text:
return ""
text = re.sub(r'\s+', ' ', text).strip()
text = re.sub(r'[\x00-\x08\x0b\x0c\x0e-\x1f\x7f]', '', text)
return text
@staticmethod
def deduplicate(items: List[Dict], key_field: str) -> List[Dict]:
"""基于关键字段去重"""
seen = set()
unique_items = []
for item in items:
key = item.get(key_field)
if key and key not in seen:
seen.add(key)
unique_items.append(item)
return unique_items
@staticmethod
def to_dataframe(items: List[Dict]) -> pd.DataFrame:
"""转换为DataFrame进行分析"""
df = pd.DataFrame(items)
if 'price' in df.columns:
df['price'] = pd.to_numeric(df['price'], errors='coerce')
df.dropna(how='all', inplace=True)
return df
@staticmethod
def export_csv(df: pd.DataFrame, filepath: str):
"""导出为CSV"""
df.to_csv(filepath, index=False, encoding='utf-8-sig')
生产踩坑提醒：MongoDB 的 update_one + upsert 模式在高频写入时可能产生竞态条件。生产环境建议使用 bulk_write 批量操作，并在 url 字段上创建唯一索引。
14. 法律合规与robots协议
14.1 中国相关法律法规
| 法律法规 | 生效时间 | 与爬虫相关的核心条款 |
|---|
| 《网络安全法》 | 2017.06 | 第27条：禁止从事非法侵入网络、干扰网络正常功能 |
| 《数据安全法》 | 2021.09 | 第32条：任何组织和个人收集数据应当合法正当 |
| 《个人信息保护法》 | 2021.11 | 第13条：处理个人信息需取得同意或有法定事由 |
| 《反不正当竞争法》 | 2019.04 | 第12条：禁止利用技术手段妨碍网络产品正常运行 |
| 《刑法》第285条 | - | 非法获取计算机信息系统数据罪（最高7年） |
14.2 robots.txt 规范
from urllib.robotparser import RobotFileParser
def check_robots_txt(base_url: str, user_agent: str = "*", path: str = "/") -> bool:
"""检查robots.txt是否允许爬取"""
rp = RobotFileParser()
robots_url = f"{base_url}/robots.txt"
rp.set_url(robots_url)
try:
rp.read()
can_fetch = rp.can_fetch(user_agent, path)
crawl_delay = rp.crawl_delay(user_agent)
print(f"URL: {path}")
print(f"允许爬取: {can_fetch}")
print(f"请求间隔: {crawl_delay}秒")
return can_fetch
except Exception as e:
print(f"读取robots.txt失败: {e}")
return True
14.3 爬虫合规边界
┌───────────────────────────────────────────────────────────────┐
│ 爬虫行为合规光谱 │
├───────────────────────────────────────────────────────────────┤
│ │
│ [合规区域] [违法区域] │
│ +-------------------------+ +-------------------------+ │
│ | 遵守robots.txt | | 绕过登录认证 | │
│ | 合理请求频率 | | 批量抓取个人信息 | │
│ | 抓取公开数据 | | 破解加密接口 | │
│ | 标注数据来源 | | 干扰网站运行 | │
│ | 用于内部分析 | | 商业竞争牟利 | │
│ | 学术研究用途 | | 出售个人信息 | │
│ +-------------------------+ +-------------------------+ │
│ │
│ [灰色地带 - 需谨慎评估] │
│ +-------------------------------------------------------+ │
│ | 抓取非公开但可访问的数据（需登录但无权限控制） | │
│ | 大规模抓取用于商业分析 | │
│ | 突破反爬措施（技术对抗本身可能违法） | │
│ | 抓取内容涉及版权作品 | │
│ +-------------------------------------------------------+ │
│ │
└───────────────────────────────────────────────────────────────┘
14.4 典型案例分析
| 案例 | 年份 | 行为 | 判决 | 启示 |
|---|
| 微脉宝爬虫案 | 2019 | 批量爬取用户数据并出售 | 有期徒刑3年 | 个人信息不可随意抓取 |
| 某数据公司案 | 2020 | 破解反爬获取竞品数据 | 构成不正当竞争 | 商业竞争爬虫风险极高 |
| 新浪诉脉脉案 | 2019 | 超授权范围抓取用户信息 | 赔偿+停止侵权 | 开放平台不等于无限抓取 |
生产踩坑提醒：2025年执法趋势明显收紧。即使抓取的是"公开"数据，如果涉及个人信息（手机号、身份证等），即使网站未做访问限制，也可能触犯《个人信息保护法》。建议在爬虫项目中引入法务审核环节。
15. 验证码专项突破
15.1 验证码类型与识别方案
| 验证码类型 | 原理 | 难度 | 推荐方案 | 成功率 |
|---|
| 数字字母验证码 | 图片+噪点 | 中 | ddddocr/Tesseract | 85-95% |
| 滑块验证码 | 拼图缺口匹配 | 高 | OpenCV模板匹配 | 90-98% |
| 点选文字验证码 | OCR+坐标定位 | 很高 | ddddocr+坐标计算 | 70-85% |
| 点选图标验证码 | 图像识别+定位 | 很高 | CNN模型 | 60-80% |
| 行为验证码（极验） | 鼠标轨迹分析 | 极高 | 专业打码平台 | 90%以上 |
| reCAPTCHA v3 | 行为评分 | 极高 | 真人打码/养Cookie | 80%以上 |
| hCaptcha | 图像分类 | 很高 | 打码平台 | 75-90% |
15.2 滑块验证码识别
import cv2
import numpy as np
import requests
import random
import time
def find_slider_gap(bg_bytes: bytes, piece_bytes: bytes) -> int:
"""
使用OpenCV模板匹配找到滑块缺口位置
"""
bg_array = np.asarray(bytearray(bg_bytes), dtype=np.uint8)
piece_array = np.asarray(bytearray(piece_bytes), dtype=np.uint8)
bg_img = cv2.imdecode(bg_array, cv2.IMREAD_COLOR)
piece_img = cv2.imdecode(piece_array, cv2.IMREAD_COLOR)
bg_gray = cv2.cvtColor(bg_img, cv2.COLOR_BGR2GRAY)
piece_gray = cv2.cvtColor(piece_img, cv2.COLOR_BGR2GRAY)
bg_edge = cv2.Canny(bg_gray, 100, 200)
piece_edge = cv2.Canny(piece_gray, 100, 200)
result = cv2.matchTemplate(bg_edge, piece_edge, cv2.TM_CCOEFF_NORMED)
_, max_val, _, max_loc = cv2.minMaxLoc(result)
gap_x = max_loc[0]
print(f"缺口位置: x={gap_x}, 匹配置信度: {max_val:.3f}")
return gap_x
def generate_track(distance: int) -> list:
"""
生成模拟人类拖拽的轨迹（加速 -> 减速 -> 微调）
"""
track = []
current = 0
mid = distance * 4 / 5
t = 0.2
v = 0
while current < distance:
if current < mid:
a = 2 + random.uniform(-0.5, 0.5)
else:
a = -3 + random.uniform(-0.5, 0.5)
v0 = v
v = v0 + a * t
move = v0 * t + 0.5 * a * t * t
current += move
track.append(round(move))
track.append(-random.randint(1, 3))
track.append(random.randint(1, 2))
return track
15.3 打码平台集成
import requests
import base64
import time
class CaptchaSolver:
"""验证码识别服务 - 支持本地OCR和打码平台"""
def __init__(self, api_key: str = None, api_user: str = None):
self.api_key = api_key
self.api_user = api_user
def solve_local(self, image_bytes: bytes) -> str:
"""本地ddddocr识别"""
import ddddocr
ocr = ddddocr.DdddOcr(show_ad=False)
return ocr.classification(image_bytes)
def solve_remote(self, image_bytes: bytes, captcha_type: str = "1901") -> str:
"""远程打码平台识别"""
if not self.api_key:
raise ValueError("未配置打码平台API Key")
img_base64 = base64.b64encode(image_bytes).decode()
data = {
"user": self.api_user,
"pass2": self.api_key,
"softid": "your_software_id",
"codetype": captcha_type,
"file_base64": img_base64,
}
response = requests.post(
"http://upload.chaojiying.net/Upload/Processing.php",
data=data,
)
result = response.json()
if result.get("err_no") == 0:
return result.get("pic_str")
else:
raise Exception(f"识别失败: {result}")
16. 高并发与性能优化
16.1 asyncio 异步爬虫
import asyncio
import aiohttp
import time
from typing import List, Dict
class AsyncCrawler:
"""高性能异步爬虫"""
def __init__(self, concurrency: int = 50, timeout: int = 30):
self.concurrency = concurrency
self.timeout = timeout
self.semaphore = asyncio.Semaphore(concurrency)
self.results: List[Dict] = []
self.stats = {"success": 0, "failed": 0, "total_time": 0}
async def fetch(self, session: aiohttp.ClientSession, url: str) -> Dict:
"""单个请求"""
async with self.semaphore:
try:
async with session.get(
url,
timeout=aiohttp.ClientTimeout(total=self.timeout),
) as response:
html = await response.text()
self.stats["success"] += 1
return {
"url": url,
"status": response.status,
"length": len(html),
"html": html,
}
except Exception as e:
self.stats["failed"] += 1
return {"url": url, "error": str(e)}
async def crawl_all(self, urls: List[str]) -> List[Dict]:
"""批量爬取"""
start_time = time.time()
connector = aiohttp.TCPConnector(
limit=self.concurrency,
limit_per_host=10,
ttl_dns_cache=300,
enable_cleanup_closed=True,
)
async with aiohttp.ClientSession(
connector=connector,
headers={
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
"AppleWebKit/537.36 Chrome/131.0.0.0 Safari/537.36",
},
) as session:
tasks = [self.fetch(session, url) for url in urls]
self.results = await asyncio.gather(*tasks)
self.stats["total_time"] = time.time() - start_time
return self.results
async def benchmark():
urls = [f"https://httpbin.org/get?id={i}" for i in range(100)]
crawler = AsyncCrawler(concurrency=50)
results = await crawler.crawl_all(urls)
print(f"异步爬取统计: {crawler.stats}")
16.2 生产环境部署
version: '3.8'
services:
redis:
image: redis:7-alpine
ports:
- "6379:6379"
command: redis-server --requirepass crawler_pass
volumes:
- redis_data:/data
crawler-worker-1:
build: ./crawler
environment:
- REDIS_URL=redis://:crawler_pass@redis:6379/0
- SPIDER_NAME=books
- CONCURRENT_REQUESTS=32
depends_on:
- redis
deploy:
resources:
limits:
cpus: '2'
memory: 2G
crawler-worker-2:
build: ./crawler
environment:
- REDIS_URL=redis://:crawler_pass@redis:6379/0
- SPIDER_NAME=books
- CONCURRENT_REQUESTS=32
depends_on:
- redis
mongo:
image: mongo:7
ports:
- "27017:27017"
environment:
- MONGO_INITDB_ROOT_USERNAME=crawler
- MONGO_INITDB_ROOT_PASSWORD=mongo_pass_2025
volumes:
- mongo_data:/data/db
proxy-api:
build: ./proxy_pool
ports:
- "8080:8080"
depends_on:
- redis
volumes:
redis_data:
mongo_data:
; supervisord.conf - 进程管理配置
[program:crawler-worker]
command=scrapy crawl %(ENV_SPIDER_NAME)s
directory=/app
user=crawler
autostart=true
autorestart=true
startsecs=10
stopwaitsecs=60
redirect_stderr=true
stdout_logfile=/var/log/crawler/%(program_name)s.log
stdout_logfile_maxbytes=50MB
stdout_logfile_backups=5
environment=
PYTHONUNBUFFERED="1",
SCRAPY_SETTINGS_MODULE="crawler.settings"
[program:proxy-pool]
command=uvicorn proxy_pool:app --host 0.0.0.0 --port 8080
directory=/app
user=crawler
autostart=true
autorestart=true
redirect_stderr=true
stdout_logfile=/var/log/crawler/proxy_pool.log
面试加分提示：生产环境爬虫部署的三件套是 Docker（环境隔离）+ Supervisor（进程守护）+ Prometheus+Grafana（监控告警）。关键监控指标包括：请求成功率、响应时间P99、队列积压量、内存使用率。
17. 实战项目
17.1 电商商品数据采集（含反反爬）
"""
电商商品数据采集完整示例
技术栈: curl_cffi + lxml + MongoDB + 代理池
"""
import json
import time
import random
from curl_cffi import requests as cffi_requests
from lxml import etree
from urllib.parse import urljoin
import pymongo
class EcommerceCrawler:
"""电商爬虫 - 带完整反反爬策略"""
def __init__(self):
self.session = cffi_requests.Session(
impersonate="chrome136",
verify=True,
)
self.mongo = pymongo.MongoClient("mongodb://localhost:27017")
self.db = self.mongo["ecommerce"]
self.collection = self.db["products"]
self.proxies = [
"http://proxy1.example.com:8080",
"http://proxy2.example.com:8080",
]
def get_headers(self) -> dict:
"""生成随机Headers"""
return {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
"AppleWebKit/537.36 Chrome/131.0.0.0 Safari/537.36",
"Accept": "text/html,application/xhtml+xml,*/*;q=0.8",
"Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8",
"Referer": "https://www.example.com/",
}
def fetch_page(self, url: str, max_retries: int = 3) -> str:
"""带重试的页面获取"""
for attempt in range(max_retries):
try:
proxy = random.choice(self.proxies) if self.proxies else None
response = self.session.get(
url,
headers=self.get_headers(),
proxies={"http": proxy, "https": proxy} if proxy else None,
timeout=30,
)
if response.status_code == 200:
return response.text
elif response.status_code == 429:
wait_time = (attempt + 1) * 30
print(f"被限流，等待{wait_time}秒...")
time.sleep(wait_time)
except Exception as e:
print(f"请求失败(第{attempt+1}次): {e}")
time.sleep(5)
return None
def parse_product(self, html: str, url: str) -> dict:
"""解析商品页面"""
tree = etree.HTML(html)
product = {
"url": url,
"title": self._safe_text(tree, '//h1[@class="title"]/text()'),
"price": self._safe_text(tree, '//span[@class="price"]/text()'),
"sales": self._safe_text(tree, '//span[@class="sales-count"]/text()'),
"shop_name": self._safe_text(tree, '//a[@class="shop-name"]/text()'),
"crawled_at": time.strftime("%Y-%m-%d %H:%M:%S"),
}
return {k: v for k, v in product.items() if v}
@staticmethod
def _safe_text(tree, xpath: str) -> str:
result = tree.xpath(xpath)
return result[0].strip() if result else ""
def save_product(self, product: dict):
"""保存商品数据（去重更新）"""
self.collection.update_one(
{"url": product["url"]},
{"$set": product},
upsert=True,
)
17.2 新闻聚合爬虫
import asyncio
import httpx
from lxml import etree
from datetime import datetime
from typing import List, Dict
class NewsAggregator:
"""多源新闻聚合爬虫"""
SOURCES = {
"澎湃新闻": {
"url": "https://www.thepaper.cn/",
"title_xpath": '//h2/a/text()',
"link_xpath": '//h2/a/@href',
},
"36氪": {
"url": "https://36kr.com/newsflashes",
"title_xpath": '//a[@class="item-title"]/text()',
"link_xpath": '//a[@class="item-title"]/@href',
},
}
def __init__(self):
self.results: List[Dict] = []
async def crawl_source(self, client: httpx.AsyncClient,
name: str, config: dict) -> List[Dict]:
"""爬取单个新闻源"""
try:
response = await client.get(
config["url"], timeout=15, follow_redirects=True,
)
tree = etree.HTML(response.text)
items = []
titles = tree.xpath(config["title_xpath"])
links = tree.xpath(config["link_xpath"])
for title, link in zip(titles[:20], links[:20]):
items.append({
"source": name,
"title": title.strip(),
"url": link if link.startswith("http")
else f"{config['url'].rstrip('/')}{link}",
"crawled_at": datetime.now().isoformat(),
})
print(f"[{name}] 获取到 {len(items)} 条新闻")
return items
except Exception as e:
print(f"[{name}] 爬取失败: {e}")
return []
async def crawl_all(self) -> List[Dict]:
"""并发爬取所有源"""
async with httpx.AsyncClient(
headers={"User-Agent": "Mozilla/5.0 Chrome/131.0.0.0"},
http2=True,
) as client:
tasks = [
self.crawl_source(client, name, config)
for name, config in self.SOURCES.items()
]
results = await asyncio.gather(*tasks)
self.results = [item for sublist in results for item in sublist]
print(f"总计获取 {len(self.results)} 条新闻")
return self.results
17.3 学术论文批量下载
import requests
import os
import time
import hashlib
from lxml import etree
class PaperDownloader:
"""学术论文下载器（以ArXiv为例）"""
BASE_URL = "https://arxiv.org"
def __init__(self, download_dir: str = "./papers"):
self.download_dir = download_dir
os.makedirs(download_dir, exist_ok=True)
self.session = requests.Session()
self.session.headers.update({
"User-Agent": "Mozilla/5.0 Research Bot/1.0"
})
def search_papers(self, query: str, max_results: int = 50) -> list:
"""搜索论文"""
params = {"query": query, "searchtype": "all", "start": 0}
papers = []
while len(papers) < max_results:
response = self.session.get(
f"{self.BASE_URL}/search/", params=params
)
tree = etree.HTML(response.text)
items = tree.xpath('//li[@class="arxiv-result"]')
for item in items:
title = item.xpath('.//p[@class="title"]/text()')
pdf_link = item.xpath('.//a[contains(text(),"PDF")]/@href')
if title and pdf_link:
papers.append({
"title": title[0].strip(),
"pdf_url": pdf_link[0] if pdf_link else "",
})
next_page = tree.xpath('//a[@class="next"]/text()')
if not next_page or len(papers) >= max_results:
break
params["start"] += 25
time.sleep(3)
return papers[:max_results]
def download_pdf(self, pdf_url: str) -> str:
"""下载PDF文件"""
filename = hashlib.md5(pdf_url.encode()).hexdigest()[:12] + ".pdf"
filepath = os.path.join(self.download_dir, filename)
if os.path.exists(filepath):
return filepath
response = self.session.get(pdf_url, timeout=60, stream=True)
if response.status_code == 200:
with open(filepath, "wb") as f:
for chunk in response.iter_content(chunk_size=8192):
f.write(chunk)
print(f"已下载: {filepath}")
return filepath
def batch_download(self, query: str, max_papers: int = 20):
"""批量搜索并下载"""
print(f"搜索论文: {query}")
papers = self.search_papers(query, max_papers)
print(f"找到 {len(papers)} 篇论文，开始下载...")
for i, paper in enumerate(papers, 1):
print(f"[{i}/{len(papers)}] {paper['title'][:60]}...")
if paper["pdf_url"]:
self.download_pdf(paper["pdf_url"])
time.sleep(3)
18. 2025-2026爬虫新技术趋势
18.1 AI辅助逆向（LLM分析JS）
2025年出现了一个革命性的趋势：使用大语言模型辅助JS逆向分析。
"""
使用LLM辅助JS逆向分析的工作流
将混淆的JS代码片段发送给LLM，获取算法还原建议
"""
def prepare_js_for_analysis(js_code: str, max_tokens: int = 4000) -> str:
"""预处理JS代码，构建发送给LLM的prompt"""
prompt = f"""
你是一个JavaScript逆向工程专家。请分析以下混淆的JS代码，
还原其核心算法逻辑，并给出Python实现。
要求：
1. 识别使用的加密算法（MD5/SHA/AES/RSA等）
2. 找出关键参数和密钥
3. 给出等价的Python代码
混淆JS代码：
{js_code[:max_tokens]}
"""
return prompt
18.2 无浏览器爬虫新技术
| 工具 | 原理 | 优势 | 局限 |
|---|
| curl_cffi | TLS指纹模拟 | 无需浏览器，速度快 | 不执行JS |
| tls-client | Go底层TLS模拟 | 跨平台，高性能 | 不执行JS |
| nodriver | CDP协议精简版 | 比Selenium轻量 | 仍需Chromium |
| botasaurus | 高级爬虫框架 | 自动反反爬 | 相对较新 |
| crawl4ai | AI驱动爬虫 | 智能内容提取 | 需要GPU |
18.3 tls-client：Go底层TLS模拟
import tls_client
session = tls_client.Session(
client_identifier="chrome_131",
random_tls_extension_order=True,
)
response = session.get(
"https://target-website.com/api/data",
headers={"User-Agent": "Mozilla/5.0 Chrome/131.0.0.0"},
proxy_url="http://proxy:8080",
)
print(f"Status: {response.status_code}")
18.4 crawl4ai：AI驱动的爬虫
from crawl4ai import AsyncWebCrawler
from crawl4ai.extraction import LLMExtraction
import asyncio
async def ai_crawl():
"""使用AI进行智能内容提取"""
async with AsyncWebCrawler() as crawler:
result = await crawler.arun(
url="https://example.com/products",
wait_until="networkidle",
)
if result.success:
print(f"页面长度: {len(result.html)}")
18.5 技术趋势总结
┌──────────────────────────────────────────────────────────────────┐
│ 2025-2026 爬虫技术演进路线 │
├──────────────────────────────────────────────────────────────────┤
│ │
│ 浏览器自动化演进: │
│ Selenium --> Playwright主流 --> AI Agent爬虫 │
│ │
│ 反检测技术演进: │
│ 手动逆向JS --> curl_cffi流行 --> AI辅助逆向 │
│ │
│ 架构演进: │
│ 单机部署 --> 分布式集群 --> 边缘计算对抗 │
│ │
│ 关键变化: │
│ - TLS指纹检测成为主流反爬手段 │
│ - LLM被用于JS逆向分析和智能内容提取 │
│ - 无头浏览器检测与反检测的博弈持续升级 │
│ - 浏览器指纹检测维度扩展到WebGL/Audio/Fonts │
│ - 合规要求日益严格，爬虫法律风险上升 │
│ │
└──────────────────────────────────────────────────────────────────┘
附录：爬虫技术选型速查表
| 场景 | 推荐技术栈 | 理由 |
|---|
| 简单静态页面 | requests + lxml | 快速开发，性能够用 |
| 中等规模项目 | Scrapy + MongoDB | 框架成熟，扩展性好 |
| 反TLS检测 | curl_cffi | 一行代码模拟浏览器指纹 |
| SPA动态页面 | Playwright | 自动等待，API拦截 |
| 大规模分布式 | Scrapy-Redis + Kafka | 高吞吐，可扩展 |
| APP数据 | mitmproxy + Frida | 抓包+Hook |
| 验证码突破 | ddddocr + 打码平台 | 本地识别+远程兜底 |
| JS逆向 | DevTools + LLM辅助 | 传统方法+AI加速 |
免责声明：本指南中的所有技术和代码仅供学习和研究目的使用。使用爬虫技术时，请务必遵守相关法律法规、网站服务条款和 robots.txt 协议。任何因不当使用本指南内容而产生的法律问题，由使用者自行承担。
25
0
已为社区贡献1条内容
所有评论(0)