Python 爬虫技术全栈指南

Python 爬虫技术全栈指南

面向读者：有一定 Python 基础的数据开发工程师 版本说明：融入 2025-2026 年最新技术动态，覆盖 curl_cffi、Playwright 新版、AI 辅助逆向等前沿内容

目录

1. 爬虫全景概述

1.1 什么是爬虫

网络爬虫（Web Crawler / Spider）是一种按照一定规则自动抓取万维网信息的程序或脚本。其核心工作流程可以概括为：

┌─────────────────────────────────────────────────────────────┐
│ 爬虫工作流程全景图 │
├─────────────────────────────────────────────────────────────┤
│ │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌────────┐│
│ │ 种子URL │───>│ 发送请求 │───>│ 接收响应 │───>│ 解析提 ││
│ │ 队列 │ │ (HTTP) │ │ (HTML) │ │ 取数据 ││
│ └──────────┘ └──────────┘ └──────────┘ └───┬────┘│
│ ^ │ │
│ │ ┌──────────┐ ┌──────────┐ │ │
│ └─────────│ 调度控制 │<───│ 数据存储 │<────────┘ │
│ │ 引擎 │ │ Pipeline │ │
│ └──────────┘ └──────────┘ │
│ │
└─────────────────────────────────────────────────────────────┘

1.2 爬虫发展历史

1.3 爬虫分类

1.4 爬虫 vs RPA vs API调用

面试加分提示：在技术选型时，优先评估目标系统是否提供官方API。API调用是最稳定、合规、高效的方式。只有当API无法满足需求（如数据不完整、频率受限）时，才考虑爬虫方案。

1.5 应用场景全景

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}, # URL查询参数
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, # SSL证书验证
)

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 常见状态码

生产踩坑提醒：429 状态码是最明显的"你被封了"信号。生产环境必须实现指数退避重试策略，遇到 429 立即停止请求至少 60 秒。

2.4 Cookie/Session 机制

import requests

# 方式1：使用Session自动管理Cookie
session = requests.Session()

# 登录
login_resp = session.post("https://httpbin.org/post", data={
"username": "test_user",
"password": "test_pass_123"
})

# Session会自动携带登录后的Cookie
profile_resp = session.get("https://httpbin.org/cookies")
print(f"Cookies: {session.cookies.get_dict()}")

# 方式2：手动设置Cookie
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']}")

# asyncio.run(ws_crawler())

2.6 HTTP/2 特性

面试加分提示：2025 年起，越来越多网站强制要求 HTTP/2 支持。Python 的 httpx 和 curl_cffi 都原生支持 HTTP/2，而传统 requests 库只支持到 HTTP/1.1。

3. Python爬虫基础库

3.1 四大HTTP客户端对比

3.2 requests 进阶用法

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

# 创建带重试机制的Session
def create_robust_session():
session = requests.Session()

# 配置重试策略
retry_strategy = Retry(
total=3, # 最大重试次数
backoff_factor=1, # 退避因子：1s, 2s, 4s
status_forcelist=[429, 500, 502, 503, 504],
allowed_methods=["GET", "POST"], # 允许重试的方法
raise_on_status=False, # 不抛异常，返回Response
)

# 配置连接池
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, # 启用HTTP/2
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}")

# asyncio.run(async_crawler())

3.4 curl_cffi：TLS指纹伪装的利器

from curl_cffi import requests as cffi_requests

# 模拟Chrome浏览器的TLS指纹
session = cffi_requests.Session(impersonate="chrome136")

# 一次impersonate参数自动处理：
# – TLS指纹 (JA3/JA4)
# – HTTP/2 SETTINGS
# – 伪头部顺序 (masp for Chrome)
# – User-Agent、Sec-Ch-Ua 等Headers
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')}")

# 支持的浏览器目标（2025年curl_cffi 0.14版本）
# chrome99 ~ chrome142, safari153 ~ safari2601, firefox133/135/144
# edge99/101, tor145
# 简写: chrome, safari, firefox 自动指向最新版本

生产踩坑提醒：curl_cffi 的 JA3 哈希每次运行都会变化，这是正常行为。因为 GREASE 机制会在 ClientHello 中插入随机值，真实 Chrome 也是如此。关键是握手的"形状"（密码套件、扩展的组成和顺序）要匹配目标浏览器。

4. HTML解析与数据提取

4.1 六大解析方案对比

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>
"""

# ========== 方案1: BeautifulSoup ==========
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}")

# ========== 方案2: lxml + XPath ==========
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}")

# ========== 方案3: 正则表达式 ==========
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}")

# ========== 方案4: pyquery ==========
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 核心组件

5.3 Scrapy 实战示例

# items.py – 定义数据结构
import scrapy

class BookItem(scrapy.Item):
title = scrapy.Field()
price = scrapy.Field()
rating = scrapy.Field()
url = scrapy.Field()

# spiders/books.py – 爬虫逻辑
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)

# pipelines.py – 数据存储管道
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 框架横向对比

生产踩坑提醒：2025年 Scrapy 默认 download_delay 从 0 秒改为 1 秒，concurrent_requests_per_domain 从 8 降为 1。这是为了降低对目标网站的压力、降低法律风险。如需高并发，需手动调整配置。

6. 动态页面爬取

6.1 三大浏览器自动化工具对比

6.2 Playwright 实战

from playwright.sync_api import sync_playwright
import json

def crawl_spa_page():
"""使用Playwright爬取SPA动态页面"""
with sync_playwright() as p:
# 启动浏览器（Chromium，无头模式）
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()

# 拦截XHR/Fetch请求，提取API数据
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

# crawl_spa_page()

6.3 JS渲染等待策略

# 高级等待策略示例
# 等待直到某个元素的文本不再包含"加载中"
page.wait_for_function(
"() => !document.querySelector('.status').textContent.includes('loading')"
)

# 等待API请求完成
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 各层反爬详解

7.3 蜜罐陷阱检测

# 蜜罐检测示例 — 网站在HTML中放置隐藏链接
# 正常用户不可见，但爬虫会跟随

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()

# 检查display:none
if "display:none" in style or "display: none" in style:
return False
# 检查position移出视口
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）

# proxy_pool.py – 轻量级代理池服务
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连接
redis_client = redis.Redis(host="localhost", port=6379, db=0, decode_responses=True)

PROXY_KEY = "proxy_pool" # 代理池Hash

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识别器
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

# 对比：requests vs curl_cffi
# requests.get(url) -> 403 Forbidden (JA3被标记为Python)
# curl_cffi.get(url, impersonate="chrome136") -> 200 OK (JA3匹配Chrome)

8.4 请求伪装最佳实践

import random
import time

# 真实浏览器Header模板（2025-2026版本）
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 核心面板

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 常见加密算法识别

9.4 Hook 技术实战

// 在Chrome Console中注入的Hook脚本

// 1. Hook XMLHttpRequest – 拦截所有XHR请求
(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);
};
})();

// 2. Hook fetch – 拦截Fetch请求
(function() {
const originalFetch = window.fetch;
window.fetch = function(…args) {
console.log('[Fetch]', args[0], args[1]);
console.trace();
return originalFetch.apply(this, args);
};
})();

// 3. Hook setRequestHeader – 捕获加密Header
(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);
};
})();

// 4. Object.defineProperty Hook – 追踪属性读取
(function() {
// 监控 navigator.webdriver 何时被读取（常见反爬检测点）
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逆向获取的密钥
"""
# 参数按key排序，拼接为字符串
sorted_str = "&".join(
f"{k}={params[k]}" for k in sorted(params.keys())
)
# 拼接密钥后计算MD5
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:
# Base64解码
encrypted = base64.b64decode(token)

# AES-CBC解密
cipher = AES.new(
key.encode("utf-8")[:16], # 128位密钥
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()

# 构建 Unicode -> 字形名 的映射
mapping = {}
for code, name in cmap.items():
mapping[chr(code)] = name

return mapping

# 使用示例
# resp = requests.get("https://example.com/font.woff")
# mapping = parse_woff_font(resp.content)
# print(f"字符映射: {mapping}")

生产踩坑提醒：字体反爬的映射关系通常是动态生成的，每次请求都可能不同。你需要在每次请求时同步下载字体文件并重新解析映射表。

10.4 JS混淆对抗

11. APP爬虫与抓包

11.1 抓包工具对比

11.2 mitmproxy 脚本编写

# mitm_proxy.py – mitmproxy爬虫抓包脚本
# 启动命令: mitmweb -s mitm_proxy.py -p 8080

from mitmproxy import http
import json
import os

# 配置要抓取的API路径关键词
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 配置

# settings.py – Scrapy-Redis 分布式配置

# 使用Redis调度器
SCHEDULER = "scrapy_redis.scheduler.Scheduler"

# 使用Redis去重
DUPEFILTER_CLASS = "scrapy_redis.dupefilter.RFPDupeFilter"

# Redis连接配置
REDIS_URL = "redis://:your_password@redis-master:6379/0"

# 持久化队列（断点续爬）
SCHEDULER_PERSIST = True

# 队列类型
# SCHEDULER_QUEUE_CLASS = "scrapy_redis.queue.FifoQueue" # 先进先出
# SCHEDULER_QUEUE_CLASS = "scrapy_redis.queue.LifoQueue" # 后进先出(深度优先)
# SCHEDULER_QUEUE_CLASS = "scrapy_redis.queue.PriorityQueue" # 优先级队列

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) # True
print("page/3存在:", "https://example.com/page/3" in bf) # False

12.4 分布式调度方案对比

13. 数据存储与清洗

13.1 存储方案对比

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 中国相关法律法规

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 典型案例分析

生产踩坑提醒：2025年执法趋势明显收紧。即使抓取的是"公开"数据，如果涉及个人信息（手机号、身份证等），即使网站未做访问限制，也可能触犯《个人信息保护法》。建议在爬虫项目中引入法务审核环节。

15. 验证码专项突破

15.1 验证码类型与识别方案

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}")
# 预期: 100个请求约2-3秒完成

16.2 生产环境部署

# docker-compose.yml – 爬虫集群部署
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):
# 1. 初始化Session（TLS指纹伪装）
self.session = cffi_requests.Session(
impersonate="chrome136",
verify=True,
)

# 2. MongoDB存储
self.mongo = pymongo.MongoClient("mongodb://localhost:27017")
self.db = self.mongo["ecommerce"]
self.collection = self.db["products"]

# 3. 代理池
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,
)

# 使用示例
# crawler = EcommerceCrawler()

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

# 工作流程示例:
# 1. 通过DevTools定位到加密函数
# 2. 复制混淆后的函数代码
# 3. 构造prompt发送给LLM
# 4. 获取LLM返回的Python实现
# 5. 本地验证并调试

18.2 无浏览器爬虫新技术

18.3 tls-client：Go底层TLS模拟

# pip install tls-client
import tls_client

# 模拟浏览器TLS指纹
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驱动的爬虫

# pip install crawl4ai
# crawl4ai 是2025年新兴的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)}")
# 可进一步使用LLM进行结构化提取

# asyncio.run(ai_crawl())

18.5 技术趋势总结

┌──────────────────────────────────────────────────────────────────┐
│ 2025-2026 爬虫技术演进路线 │
├──────────────────────────────────────────────────────────────────┤
│ │
│ 浏览器自动化演进: │
│ Selenium –> Playwright主流 –> AI Agent爬虫 │
│ │
│ 反检测技术演进: │
│ 手动逆向JS –> curl_cffi流行 –> AI辅助逆向 │
│ │
│ 架构演进: │
│ 单机部署 –> 分布式集群 –> 边缘计算对抗 │
│ │
│ 关键变化: │
│ – TLS指纹检测成为主流反爬手段 │
│ – LLM被用于JS逆向分析和智能内容提取 │
│ – 无头浏览器检测与反检测的博弈持续升级 │
│ – 浏览器指纹检测维度扩展到WebGL/Audio/Fonts │
│ – 合规要求日益严格，爬虫法律风险上升 │
│ │
└──────────────────────────────────────────────────────────────────┘

附录：爬虫技术选型速查表

免责声明：本指南中的所有技术和代码仅供学习和研究目的使用。使用爬虫技术时，请务必遵守相关法律法规、网站服务条款和 robots.txt 协议。任何因不当使用本指南内容而产生的法律问题，由使用者自行承担。