Scrapy实战(4)：防反爬中间件配置

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

本章我们将探寻如何使用 Scrapy 框架进行网络数据采集。作为量化分析师，你将编写爬虫从财经网站爬取最新的股票行情和社交媒体上的咖啡店舆情，为策略引擎输入第一手另类数据。

💡 核心导读

本节将深入Scrapy框架的核心组件与高级功能，重点掌握Item Pipeline的定制开发、中间件机制的原理与实现，以及分布式爬虫的部署方案。通过实践数据清洗管道、自定义下载中间件等案例，你将学会处理反爬策略、实现数据持久化存储，并了解如何通过Scrapy-Redis构建分布式爬虫系统。

数据管道（Item Pipeline）开发

管道工作原理

Scrapy的Item Pipeline是数据处理流水线，按settings.py中定义的优先级顺序执行。典型处理阶段包括：

1. 数据清洗/验证
2. 去重处理
3. 数据存储

import sqlite3

class MySQLPipeline:
    def __init__(self):
        # 使用 sqlite3 在内存中模拟数据库连接
        self.conn = sqlite3.connect(':memory:')
        self.cursor = self.conn.cursor()
        # 创建示例表结构
        self.cursor.execute("""
            CREATE TABLE IF NOT EXISTS articles (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                title TEXT,
                url TEXT,
                content TEXT
            )
        """)
        self.conn.commit()

    def process_item(self, item, spider):
        sql = """INSERT INTO articles 
                (title, url, content) 
                VALUES (?, ?, ?)"""
        self.cursor.execute(sql, (
            item['title'],
            item['url'],
            item['content']
        ))
        self.conn.commit()
        return item

    def close_spider(self, spider):
        self.conn.close()

# 演示使用
pipeline = MySQLPipeline()
sample_item = {
    'title': '量化交易入门',
    'url': 'https://xxketang.com/quant',
    'content': '这是课程内容示例。'
}
result = pipeline.process_item(sample_item, spider=None)

# 查询并打印插入的数据以验证
cursor = pipeline.conn.cursor()
cursor.execute("SELECT title, url, content FROM articles")
rows = cursor.fetchall()
print("插入的数据记录：")
for row in rows:
    print(f"标题: {row[0]}, URL: {row[1]}, 内容: {row[2]}")

pipeline.close_spider(spider=None)

插入的数据记录：
标题: 量化交易入门, URL: https://xxketang.com/quant, 内容: 这是课程内容示例。

管道激活配置

# 此块通常位于 settings.py 中，为演示独立运行，我们模拟其结构并打印配置
ITEM_PIPELINES = {
    'myproject.pipelines.DuplicatesPipeline': 100,  # 优先级数值越小越先执行
    'myproject.pipelines.MySQLPipeline': 300,
}

# 打印配置以展示结构
print("ITEM_PIPELINES 配置字典：")
for key, value in ITEM_PIPELINES.items():
    print(f"  '{key}': {value}")

ITEM_PIPELINES 配置字典：
  'myproject.pipelines.DuplicatesPipeline': 100
  'myproject.pipelines.MySQLPipeline': 300

中间件开发实战

下载中间件示例

实现随机User-Agent和代理IP轮询：

import random

class RandomUserAgentMiddleware:
    def __init__(self, user_agents):
        self.user_agents = user_agents

    @classmethod
    def from_crawler(cls, crawler):
        # 模拟从设置中获取 USER_AGENTS 列表
        return cls(crawler.settings.getlist('USER_AGENTS'))

    def process_request(self, request, spider):
        # 固定选择第一个 User-Agent 以确保输出确定性
        request.headers['User-Agent'] = self.user_agents[0]

# 模拟的 settings 配置
USER_AGENTS = [
    'Mozilla/5.0 (Windows NT 10.0; Win64; x64)',
    'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7)',
    'Opera/9.80 (X11; Linux i686)'
]
DOWNLOADER_MIDDLEWARES = {
    'myproject.middlewares.RandomUserAgentMiddleware': 400,
}

# 演示中间件工作流程
# 模拟一个请求对象
class MockRequest:
    def __init__(self):
        self.headers = {}

# 模拟爬虫设置（简化）
class MockCrawler:
    class settings:
        @staticmethod
        def getlist(key):
            if key == 'USER_AGENTS':
                return USER_AGENTS
            return []

# 创建中间件实例
crawler = MockCrawler()
middleware = RandomUserAgentMiddleware.from_crawler(crawler)

# 处理请求
request = MockRequest()
print(f"处理前请求头: {request.headers}")
middleware.process_request(request, spider=None)
print(f"处理后请求头 User-Agent: {request.headers.get('User-Agent')}")

# 打印中间件配置
print(f"\nDOWNLOADER_MIDDLEWARES 配置: {DOWNLOADER_MIDDLEWARES}")

分布式爬虫部署

Scrapy-Redis配置

处理前请求头: {}
处理后请求头 User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64)

DOWNLOADER_MIDDLEWARES 配置: {'myproject.middlewares.RandomUserAgentMiddleware': 400}

# 此块通常位于 settings.py 中，为演示独立运行，我们打印配置信息
SCHEDULER = "scrapy_redis.scheduler.Scheduler"
DUPEFILTER_CLASS = "scrapy_redis.dupefilter.RFPDupeFilter"
REDIS_URL = 'redis://:password@192.168.1.100:6379'

print("分布式爬虫 Redis 配置：")
print(f"SCHEDULER = \"{SCHEDULER}\"")
print(f"DUPEFILTER_CLASS = \"{DUPEFILTER_CLASS}\"")
print(f"REDIS_URL = '{REDIS_URL}'")

分布式爬虫 Redis 配置：
SCHEDULER = "scrapy_redis.scheduler.Scheduler"
DUPEFILTER_CLASS = "scrapy_redis.dupefilter.RFPDupeFilter"
REDIS_URL = 'redis://:password@192.168.1.100:6379'

scrapy crawl myspider -s REDIS_URL=redis://:password@master_ip:6379

📝 动手练一练

第 1 题 · 概念应用

你正在编写一个爬虫，用于从多个财经新闻网站采集关于某只股票的舆情数据。为了规避网站基于 IP 地址的访问频率限制，你配置了一个下载器中间件，在 process_request 方法中动态地为每个请求设置不同的代理服务器地址。同时，你还在 process_response 方法中检查响应状态码，如果遇到 403 状态码，则将该请求的代理地址标记为“失效”，并从代理池中移除。

请问，当 Scrapy 引擎处理一个请求时，上述 process_request 和 process_response 方法分别会在哪个核心组件被调用前后执行？

• A. process_request 在 Spider 生成请求后、Engine 调度前执行；process_response 在 Downloader 下载完成后、Spider 解析前执行。
• B. process_request 在 Engine 将请求交给 Downloader 前执行；process_response 在 Downloader 返回响应给 Engine 后执行。
• C. process_request 在 Item Pipeline 处理数据前执行；process_response 在数据存储到数据库后执行。
• D. process_request 在 Spider 解析响应生成 Item 时执行；process_response 在 Item Pipeline 清洗数据时执行。

第 2 题 · 动手题（代码填空）

你正在开发一个数据清洗管道（Item Pipeline），用于处理爬取到的股票新闻。每条新闻数据是一个字典（Item），包含 title（标题）、publish_date（发布日期，字符串格式如 "2023-10-26"）和 content（正文）。你需要实现 process_item 方法，完成以下任务：

1. 验证 publish_date 字段是否存在且不为空。
2. 将 publish_date 的字符串转换为 datetime.date 对象，以便后续进行日期比较和分析。
3. 清洗 content 字段，移除其中所有的 HTML 标签（例如 <p>, <br/> 等），只保留纯文本。

请补全下面管道类中的代码，使其能完成上述功能。

import re
from datetime import datetime

class NewsCleaningPipeline:
    def process_item(self, item, spider):
        # 1. 验证 publish_date 字段
        if not item.get('publish_date'):
            raise ValueError("Item 缺少 'publish_date' 字段")

        # 2. 转换日期字符串为 date 对象
        # 提示：使用 datetime.strptime 并指定格式字符串
        date_str = item['publish_date']
        date_obj = datetime.strptime(date_str, ______).date()
        item['publish_date'] = date_obj

        # 3. 清洗 content，移除 HTML 标签
        # 提示：使用正则表达式 re.sub 替换掉所有尖括号 <> 及其包围的内容
        if 'content' in item and item['content']:
            html_pattern = re.compile(r'<[^>]+>')
            item['content'] = html_pattern.sub('', item['content'])

        return item

# 演示代码
pipeline = NewsCleaningPipeline()
sample_item = {
    'title': '央行发布最新货币政策报告',
    'publish_date': '2023-10-26',
    'content': '<p>报告指出，当前经济<strong>平稳运行</strong>。</p><br/>市场反应积极。'
}
try:
    cleaned_item = pipeline.process_item(sample_item, spider=None)
    print("清洗后的 Item:")
    print(f"  标题: {cleaned_item['title']}")
    print(f"  发布日期 (类型: {type(cleaned_item['publish_date'])}): {cleaned_item['publish_date']}")
    print(f"  正文: {cleaned_item['content']}")
except ValueError as e:
    print(f"处理失败: {e}")

import re
from datetime import datetime

class NewsCleaningPipeline:
    def process_item(self, item, spider):
        # 1. 验证 publish_date 字段
        if not item.get('publish_date'):
            raise ValueError("Item 缺少 'publish_date' 字段")

        # 2. 转换日期字符串为 date 对象
        date_str = item['publish_date']
        date_obj = datetime.strptime(date_str, "%Y-%m-%d").date()
        item['publish_date'] = date_obj

        # 3. 清洗 content，移除 HTML 标签
        if 'content' in item and item['content']:
            html_pattern = re.compile(r'<[^>]+>')
            item['content'] = html_pattern.sub("", item['content'])

        return item

# 演示代码
pipeline = NewsCleaningPipeline()
sample_item = {
    'title': '央行发布最新货币政策报告',
    'publish_date': '2023-10-26',
    'content': '<p>报告指出，当前经济<strong>平稳运行</strong>。</p><br/>市场反应积极。'
}
try:
    cleaned_item = pipeline.process_item(sample_item, spider=None)
    print("清洗后的 Item:")
    print(f"  标题: {cleaned_item['title']}")
    print(f"  发布日期 (类型: {type(cleaned_item['publish_date'])}): {cleaned_item['publish_date']}")
    print(f"  正文: {cleaned_item['content']}")
except ValueError as e:
    print(f"处理失败: {e}")

清洗后的 Item:
  标题: 央行发布最新货币政策报告
  发布日期 (类型: <class 'datetime.date'>): 2023-10-26
  正文: 报告指出，当前经济平稳运行。市场反应积极。

本章小结

📋 行动清单

• 实现一个数据清洗管道，对采集的文本内容进行HTML标签过滤和空白字符清理
• 开发自定义下载中间件，集成代理IP池服务实现自动切换
• 使用Scrapy-Redis部署两个爬虫Worker节点，观察Redis中的任务队列变化

—— 小象教研组