查看全课大纲（第 24 / 52 节）

Scrapy实战(3)：启用数据管道清洗行情

约 10 分钟

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

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

💡 核心导读

本节将深入Scrapy框架的核心工作流程，重点讲解Item Pipeline的数据处理机制、中间件(Middleware)的定制方法以及项目优化技巧。通过完整的爬虫案例演示如何清洗存储数据、处理异常请求，并实现分布式爬取。掌握这些技能后，读者将能够构建工业级的数据采集系统。

数据处理管道(Item Pipeline)

管道工作机制

Scrapy的Item Pipeline是数据处理流水线，包含多个按序执行的处理器：

爬虫解析 → Pipeline1 → Pipeline2 → ... → 持久化存储

典型管道组件实现

# 演示一个简单的数据清洗管道（不使用外部依赖）
class PriceConverterPipeline:
    """将价格字段统一转换为美元"""
    exchange_rate = 7.2  # 人民币兑美元汇率

    def process_item(self, item, spider):
        # 模拟 ItemAdapter 的功能
        if 'price_cny' in item:
            item['price_usd'] = round(item['price_cny'] / self.exchange_rate, 2)
            del item['price_cny']
        return item

class SimpleStoragePipeline:
    """模拟存储管道，实际打印数据"""
    def process_item(self, item, spider):
        print(f"存储数据: {item}")
        return item

# 演示管道处理一个示例数据项
# 创建示例数据
sample_item = {'symbol': 'AAPL', 'price_cny': 720.0, 'volume': 1000}
print("原始数据项:", sample_item)

# 测试价格转换管道
price_pipe = PriceConverterPipeline()
processed_item = price_pipe.process_item(sample_item, spider=None)
print("转换后数据项:", processed_item)

# 测试存储管道
storage_pipe = SimpleStoragePipeline()
storage_pipe.process_item(processed_item, spider=None)

原始数据项: {'symbol': 'AAPL', 'price_cny': 720.0, 'volume': 1000}
转换后数据项: {'symbol': 'AAPL', 'volume': 1000, 'price_usd': 100.0}
存储数据: {'symbol': 'AAPL', 'volume': 1000, 'price_usd': 100.0}

管道启用配置

# 演示settings.py中的配置（仅打印配置）
ITEM_PIPELINES = {
    'project.pipelines.PriceConverterPipeline': 100,  # 数值越小优先级越高
    'project.pipelines.SimpleStoragePipeline': 200,
}
MONGO_URI = 'mongodb://localhost:27017'
MONGO_DB = 'scrapy_data'

# 打印配置以供核对
print("ITEM_PIPELINES 配置:", ITEM_PIPELINES)
print("MONGO_URI:", MONGO_URI)
print("MONGO_DB:", MONGO_DB)

ITEM_PIPELINES 配置: {'project.pipelines.PriceConverterPipeline': 100, 'project.pipelines.SimpleStoragePipeline': 200}
MONGO_URI: mongodb://localhost:27017
MONGO_DB: scrapy_data

中间件开发实战

下载中间件示例

import random

class RandomUserAgentMiddleware:
    """随机User-Agent中间件（简化版，不依赖Scrapy）"""
    user_agents = [
        'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36',
        'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/14.1.1 Safari/605.1.15',
        'Mozilla/5.0 (X11; Linux x86_64; rv:89.0) Gecko/20100101 Firefox/89.0'
    ]

    def process_request(self, request):
        # 固定选择第一个User-Agent以确保输出确定性
        chosen_ua = self.user_agents[0]
        request['headers']['User-Agent'] = chosen_ua
        # 为演示，打印出选择的User-Agent
        print(f"为请求 {request['url']} 设置 User-Agent: {chosen_ua[:50]}...")

# 演示中间件使用
# 模拟一个请求字典（代替Scrapy Request对象）
middleware = RandomUserAgentMiddleware()
demo_request = {
    'url': 'https://example.com/stock/AAPL',
    'headers': {}
}

# 调用中间件处理请求
middleware.process_request(demo_request)
print("请求头中的User-Agent:", demo_request['headers'].get('User-Agent'))

为请求 https://example.com/stock/AAPL 设置 User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWeb...
请求头中的User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36

中间件配置激活

# 演示DOWNLOADER_MIDDLEWARES配置（仅打印配置）
DOWNLOADER_MIDDLEWARES = {
    'project.middlewares.RandomUserAgentMiddleware': 543,
}

# 打印配置
print("DOWNLOADER_MIDDLEWARES 配置:", DOWNLOADER_MIDDLEWARES)

DOWNLOADER_MIDDLEWARES 配置: {'project.middlewares.RandomUserAgentMiddleware': 543}

分布式爬虫实现

Redis分布式配置

# 演示分布式爬虫配置（Scrapy-Redis）（仅打印配置）
SCHEDULER = "scrapy_redis.scheduler.Scheduler"
DUPEFILTER_CLASS = "scrapy_redis.dupefilter.RFPDupeFilter"
REDIS_URL = 'redis://:password@localhost:6379'

# 打印配置
print("SCHEDULER:", SCHEDULER)
print("DUPEFILTER_CLASS:", DUPEFILTER_CLASS)
print("REDIS_URL:", REDIS_URL)

数据去重对比

SCHEDULER: scrapy_redis.scheduler.Scheduler
DUPEFILTER_CLASS: scrapy_redis.dupefilter.RFPDupeFilter
REDIS_URL: redis://:password@localhost:6379

📝 动手练一练

第 1 题 · 概念应用

你正在构建一个量化数据采集系统，用于清洗从多个数据源爬取的股票行情。你设计了一个 DataValidationPipeline，其 process_item 方法会检查数据项的 price 字段，如果价格小于等于 0 或为 None，则抛出 DropItem 异常。同时，你还配置了一个 MongoDBStoragePipeline 用于将清洗后的数据存入数据库。

如果某个数据项 {‘symbol’: ‘XYZ’， ‘price’: -5.0， ‘volume’: 1000} 进入管道，且 DataValidationPipeline 的优先级为 100，MongoDBStoragePipeline 的优先级为 800，那么以下哪种情况会发生？

A. 数据项会被 DataValidationPipeline 丢弃，MongoDBStoragePipeline 不会收到它，数据库中将没有 XYZ 的这条记录。
B. 数据项会通过 DataValidationPipeline 的检查，MongoDBStoragePipeline 会收到它，但会因为价格异常而拒绝写入数据库。
C. DataValidationPipeline 会捕获 DropItem 异常，将价格修正为 0 后继续传递给 MongoDBStoragePipeline。
D. Scrapy 引擎会停止整个爬虫任务，并报告一个严重的数据错误。

👉 点击查看参考答案

答案：A（数据项会被 DataValidationPipeline 丢弃，MongoDBStoragePipeline 不会收到它，数据库中将没有 XYZ 的这条记录。）

推理过程：在 Scrapy 的 Item Pipeline 机制中，当某个管道的 process_item 方法抛出 DropItem 异常时，该数据项会立即从处理流水线中被丢弃，后续所有优先级更低的管道都不会再处理这个数据项。这是一种主动过滤无效数据的标准做法。因此，价格无效的 XYZ 数据项在优先级为 100 的 DataValidationPipeline 处被拦截后，优先级为 800 的 MongoDBStoragePipeline 根本不会“看到”它，自然也就不会尝试存储。

第 2 题 · 代码填空

以下代码模拟了一个简化的数据清洗管道 VolumeNormalizerPipeline，用于将爬取到的原始成交量字符串（如 “1.2K”、“5M”）统一转换为整数。请补全管道中处理 ‘1.2K’ 和 ‘5M’ 这类带后缀字符串的核心转换逻辑（空位 ______）。

class VolumeNormalizerPipeline:
    """将带K/M后缀的成交量字符串转换为整数"""
    
    def _normalize_volume(self, vol_str):
        """转换逻辑：K代表千，M代表百万"""
        if vol_str is None:
            return 0
        vol_str = vol_str.upper().strip()
        multiplier = 1
        if vol_str.endswith('K'):
            multiplier = 1000
            vol_str = vol_str[:-1]
        elif vol_str.endswith('M'):
            multiplier = ______  # 填空：M代表百万，乘数应为？
            vol_str = vol_str[:-1]
        try:
            # 处理可能的小数，如 '1.2K'
            base_value = float(vol_str)
            return int(base_value * multiplier)
        except ValueError:
            return 0
    
    def process_item(self, item, spider):
        if 'volume_str' in item:
            item['volume'] = self._normalize_volume(item['volume_str'])
            del item['volume_str']
        return item

# 测试管道
pipeline = VolumeNormalizerPipeline()
test_items = [
    {'symbol': 'A', 'volume_str': '1500'},
    {'symbol': 'B', 'volume_str': '1.2K'},
    {'symbol': 'C', 'volume_str': '5M'},
    {'symbol': 'D', 'volume_str': 'invalid'},
]

for item in test_items:
    result = pipeline.process_item(item.copy(), spider=None)
    print(f"{result['symbol']}: volume = {result.get('volume')}")

👉 点击查看参考答案

空位应填：1000000

补全后的完整可运行代码块：

class VolumeNormalizerPipeline:
    """将带K/M后缀的成交量字符串转换为整数"""
    
    def _normalize_volume(self, vol_str):
        """转换逻辑：K代表千，M代表百万"""
        if vol_str is None:
            return 0
        vol_str = vol_str.upper().strip()
        multiplier = 1
        if vol_str.endswith('K'):
            multiplier = 1000
            vol_str = vol_str[:-1]
        elif vol_str.endswith('M'):
            multiplier = 1000000  # M代表 Million (百万)
            vol_str = vol_str[:-1]
        try:
            # 处理可能的小数，如 '1.2K'
            base_value = float(vol_str)
            return int(base_value * multiplier)
        except ValueError:
            return 0
    
    def process_item(self, item, spider):
        if 'volume_str' in item:
            item['volume'] = self._normalize_volume(item['volume_str'])
            del item['volume_str']
        return item

# 测试管道
pipeline = VolumeNormalizerPipeline()
test_items = [
    {'symbol': 'A', 'volume_str': '1500'},
    {'symbol': 'B', 'volume_str': '1.2K'},
    {'symbol': 'C', 'volume_str': '5M'},
    {'symbol': 'D', 'volume_str': 'invalid'},
]

for item in test_items:
    result = pipeline.process_item(item.copy(), spider=None)
    print(f"{result['symbol']}: volume = {result.get('volume')}")

A: volume = 1500
B: volume = 1200
C: volume = 5000000
D: volume = 0

逐行解释：

对于 ‘1500’：无后缀，multiplier=1，直接转换为整数 1500。
对于 ‘1.2K’：识别后缀 ‘K’，multiplier=1000，去掉后缀后 base_value=1.2，计算 1.2 * 1000 = 1200。
对于 ‘5M’：识别后缀 ‘M’，multiplier=1000000（即一百万），去掉后缀后 base_value=5，计算 5 * 1000000 = 5000000。
对于 ‘invalid’：尝试转换为 float 时触发 ValueError 异常，返回 0。

本章小结

📋 行动清单

实现一个自定义Pipeline，将采集的日期字段统一转换为ISO8601格式
开发下载中间件，自动处理遇到403状态码时切换代理IP
配置Redis分布式环境，验证多节点协同爬取效果

—— 小象教研组