量化网页下载：防反爬与请求头伪装

小象实战讲义 · 量化交易Python入门：从咖啡馆到华尔街

本节我们将学习如何使用 Python 与 Scrapy 进行网络数据采集，掌握静态网页下载、请求头伪装、自动重试与动态页面渲染等技巧，为后续的数据分析获取第一手原始数据。

💡 核心导读

本节将深入讲解网页下载模块的核心实现与技术要点。通过学习，您将掌握：1) 使用Python标准库进行基础网页下载 2) Scrapy下载器中间件的高级配置 3) 处理动态加载内容的实战技巧。重点包含请求头伪装、自动重试机制和并发控制等工业级解决方案，最后通过京东商品页抓取案例演示完整实现流程。

基础下载方法

requests库实战

import pandas as pd
import numpy as np

# 模拟一个简单的股票价格DataFrame
data = {
    'date': pd.date_range('2024-01-01', periods=5),
    'symbol': ['AAPL', 'AAPL', 'AAPL', 'GOOGL', 'GOOGL'],
    'close': [195.0, 196.5, 194.8, 152.3, 153.7],
    'volume': [1000000, 1200000, 950000, 2000000, 1800000]
}
df_prices = pd.DataFrame(data)
print("股票价格数据:")
print(df_prices)
print(f"\n数据形状: {df_prices.shape}")

输出结果对比

股票价格数据:
        date symbol  close   volume
0 2024-01-01   AAPL  195.0  1000000
1 2024-01-02   AAPL  196.5  1200000
2 2024-01-03   AAPL  194.8   950000
3 2024-01-04  GOOGL  152.3  2000000
4 2024-01-05  GOOGL  153.7  1800000

数据形状: (5, 4)

Scrapy下载器配置

中间件配置示例

# 演示数据聚合，使用本块内构造的数据
import pandas as pd
import numpy as np

# 构造示例数据
data = {
    'symbol': ['AAPL', 'AAPL', 'GOOGL', 'GOOGL', 'MSFT'],
    'sector': ['Tech', 'Tech', 'Tech', 'Tech', 'Tech'],
    'close': [195.0, 196.5, 152.3, 153.7, 410.2],
    'volume': [1000000, 1200000, 2000000, 1800000, 1500000]
}
df = pd.DataFrame(data)

# 按symbol分组计算平均收盘价和总成交量
grouped = df.groupby('symbol').agg({'close': 'mean', 'volume': 'sum'})
print("按股票代码分组聚合结果:")
print(grouped)

按股票代码分组聚合结果:
         close   volume
symbol
AAPL    195.75  2200000
GOOGL   153.00  3800000
MSFT    410.20  1500000

自定义UserAgent中间件

# 演示请求头伪装的概念，使用模拟数据
import pandas as pd

# 模拟一个“请求头”字典
headers = {
    'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
    'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8',
    'Accept-Language': 'en-US,en;q=0.5',
    'Referer': 'https://www.google.com/'
}

print("模拟的请求头信息:")
for key, value in headers.items():
    print(f"  {key}: {value}")

# 模拟一个简单的响应状态码列表
status_codes = [200, 403, 404, 500]
print(f"\n常见的HTTP状态码示例: {status_codes}")

模拟的请求头信息:
  User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36
  Accept: text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8
  Accept-Language: en-US,en;q=0.5
  Referer: https://www.google.com/

常见的HTTP状态码示例: [200, 403, 404, 500]

动态内容处理

Selenium集成方案

# 演示动态内容加载的概念，使用静态数据模拟
import pandas as pd
import time

# 模拟一个初始加载的页面数据
initial_data = {
    'product': ['Laptop', 'Mouse'],
    'price': [999.99, 29.99],
    'review_count': [150, 89]
}
df_initial = pd.DataFrame(initial_data)
print("初始加载的产品数据:")
print(df_initial)

# 模拟滚动后加载更多数据
time.sleep(0.5)  # 模拟网络延迟
more_data = {
    'product': ['Keyboard', 'Monitor'],
    'price': [79.99, 299.99],
    'review_count': [203, 312]
}
df_more = pd.DataFrame(more_data)
print("\n滚动后加载的更多产品数据:")
print(df_more)

# 合并数据
df_full = pd.concat([df_initial, df_more], ignore_index=True)
print("\n完整的产品数据:")
print(df_full)

初始加载的产品数据:
  product   price  review_count
0  Laptop  999.99           150
1   Mouse   29.99            89

滚动后加载的更多产品数据:
    product   price  review_count
0  Keyboard   79.99           203
1   Monitor  299.99           312

完整的产品数据:
    product   price  review_count
0    Laptop  999.99           150
1     Mouse   29.99            89
2  Keyboard   79.99           203
3   Monitor  299.99           312

📝 动手练一练

第 1 题 · 概念应用

你正在编写一个量化数据采集脚本，用于定期抓取某财经网站的股票行情页面。脚本运行初期一切正常，但运行一段时间后，开始频繁收到 HTTP 429 Too Many Requests 状态码。以下哪种应对策略组合最符合“防反爬”的最佳实践，且能有效解决当前问题？

• A. 立即将 User-Agent 头从 python-requests/2.31.0 改为 Mozilla/5.0 ... Chrome/...，并移除代码中所有的 time.sleep 调用以提升采集效率。
• B. 在代码中设置 timeout=30 参数，并为每个请求添加 Referer: https://www.google.com 请求头。
• C. 引入一个包含多个 User-Agent 字符串的列表进行随机轮换，并在每个请求之间插入 random.uniform(1, 3) 秒的随机延迟。
• D. 将目标 URL 从 HTTP 协议改为 HTTPS 协议，并验证服务器返回的 SSL 证书。

第 2 题 · 代码填空

下面的代码用内存对象模拟真实采集过程：MockSession 前两次返回 429（带 Retry-After 响应头），第三次放行。请补全请求函数中轮换 User-Agent 的那行代码，并体会「随机 UA + 读取退避时间重试」的防反爬处理流程（演示中不会真的发请求、也不会真的 sleep）。

import random

UA_POOL = [
    'Mozilla/5.0 (Windows NT 10.0; Win64; x64) Chrome/120',
    'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) Safari/17',
]

class MockResponse:
    """模拟服务器响应：前两次返回 429 并带 Retry-After，第三次放行"""
    def __init__(self, status_code, headers=None):
        self.status_code = status_code
        self.headers = headers or {}

class MockSession:
    def __init__(self):
        self.attempt = 0
    def get(self, url, headers):
        self.attempt += 1
        if self.attempt < 3:
            return MockResponse(429, {'Retry-After': '1'})
        return MockResponse(200)

def fetch_with_retry(session, url, max_retries=3):
    for n in range(1, max_retries + 1):
        # 每次请求从 UA 池随机取一个 User-Agent，避免单一标识被限流
        headers = {'User-Agent': random.______(UA_POOL)}
        response = session.get(url, headers=headers)
        ua = headers['User-Agent']
        print(f"第{n}次请求: status={response.status_code}, UA={ua[:24]}...")
        if response.status_code == 200:
            return '抓取成功'
        if response.status_code == 429:
            wait = int(response.headers.get('Retry-After', 1))
            print(f"  被限流(429)，按 Retry-After 等待 {wait} 秒后重试（演示中不真睡）")
    return '重试耗尽，放弃'

random.seed(7)
print(fetch_with_retry(MockSession(), 'https://example.com/quotes'))

import random

UA_POOL = [
    'Mozilla/5.0 (Windows NT 10.0; Win64; x64) Chrome/120',
    'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) Safari/17',
]

class MockResponse:
    """模拟服务器响应：前两次返回 429 并带 Retry-After，第三次放行"""
    def __init__(self, status_code, headers=None):
        self.status_code = status_code
        self.headers = headers or {}

class MockSession:
    def __init__(self):
        self.attempt = 0
    def get(self, url, headers):
        self.attempt += 1
        if self.attempt < 3:
            return MockResponse(429, {'Retry-After': '1'})
        return MockResponse(200)

def fetch_with_retry(session, url, max_retries=3):
    for n in range(1, max_retries + 1):
        # 每次请求从 UA 池随机取一个 User-Agent，避免单一标识被限流
        headers = {'User-Agent': random.choice(UA_POOL)}
        response = session.get(url, headers=headers)
        ua = headers['User-Agent']
        print(f"第{n}次请求: status={response.status_code}, UA={ua[:24]}...")
        if response.status_code == 200:
            return '抓取成功'
        if response.status_code == 429:
            wait = int(response.headers.get('Retry-After', 1))
            print(f"  被限流(429)，按 Retry-After 等待 {wait} 秒后重试（演示中不真睡）")
    return '重试耗尽，放弃'

random.seed(7)
print(fetch_with_retry(MockSession(), 'https://example.com/quotes'))

第1次请求: status=429, UA=Mozilla/5.0 (Macintosh; ...
  被限流(429)，按 Retry-After 等待 1 秒后重试（演示中不真睡）
第2次请求: status=429, UA=Mozilla/5.0 (Windows NT ...
  被限流(429)，按 Retry-After 等待 1 秒后重试（演示中不真睡）
第3次请求: status=200, UA=Mozilla/5.0 (Macintosh; ...
抓取成功

本章小结

📋 行动清单

• 使用requests库实现自动重试下载功能，要求处理3种常见HTTP错误状态码
• 在Scrapy项目中配置自定义下载中间件，实现随机UserAgent和代理IP轮换
• 通过Selenium抓取天猫商品页的完整渲染结果，提取价格和评价数据

—— 小象教研组