网页解析(2)：使用CSS选择器与ItemLoader

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

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

💡 核心导读

本节将深入探讨Scrapy框架中的网页解析进阶技巧，重点讲解XPath与CSS选择器的混合使用策略、动态数据提取方法以及数据清洗管道设计。通过本章学习，你将掌握：使用scrapy.Selector处理嵌套结构数据、应对反爬机制的解析技巧、以及通过ItemLoader实现标准化数据流。最后通过一个完整的电商网站数据采集案例，演示如何构建高可靠性的解析模块。

混合选择器实战策略

XPath与CSS选择器组合

import re

# 构造一个模拟的HTML响应，用于演示解析，不发起真实网络请求
html_content = '''
<html>
<body>
<div class="product-item" data-sku="SKU123">
    <h2 data-name="高性能笔记本">高性能笔记本</h2>
    <span class="price">$1299.99</span>
</div>
<div class="product-item" data-alt-sku="SKU456">
    <h2 data-name="无线鼠标">无线鼠标</h2>
    <span class="price">$29.95</span>
</div>
</body>
</html>
'''

# 模拟爬虫的parse方法逻辑，使用正则表达式模拟CSS选择器和XPath
def parse_demo(html_str):
    results = []
    # 使用正则查找所有product-item div
    product_pattern = r'<div class="product-item"[^>]*>.*?</div>'
    products = re.findall(product_pattern, html_str, re.DOTALL)
    
    for product in products:
        # 提取data-name属性
        name_match = re.search(r'data-name="([^"]+)"', product)
        name = name_match.group(1) if name_match else None
        
        # 提取价格文本并匹配数字
        price_match = re.search(r'<span class="price">[^<]*?(\d+\.\d{2})', product)
        price = price_match.group(1) if price_match else None
        
        # 提取sku属性，优先data-sku，其次data-alt-sku
        sku_match = re.search(r'data-sku="([^"]+)"', product)
        if not sku_match:
            sku_match = re.search(r'data-alt-sku="([^"]+)"', product)
        sku = sku_match.group(1) if sku_match else None
        
        if name and price and sku:
            results.append({
                'name': name,
                'price': price,
                'sku': sku
            })
    return results

# 执行解析并打印结果
parsed_items = parse_demo(html_content)
for item in parsed_items:
    print(item)

{'name': '高性能笔记本', 'price': '1299.99', 'sku': 'SKU123'}
{'name': '无线鼠标', 'price': '29.95', 'sku': 'SKU456'}

反爬应对方案

import time
import hashlib

# 模拟一个HTML片段，包含价格信息
html_fragment = '''
<span class="price">599.99</span>
<meta itemprop="price" content="599.99">
'''

# 模拟解析过程，不依赖外部库
def parse_dynamic_content_demo(html_str):
    # 简单模拟XPath提取逻辑
    # 查找包含价格的span标签
    if '<span class="price">' in html_str:
        start = html_str.find('<span class="price">') + len('<span class="price">')
        end = html_str.find('</span>', start)
        real_price = html_str[start:end]
    else:
        # 备用方案：查找meta标签
        if 'meta itemprop="price"' in html_str:
            start = html_str.find('content="', html_str.find('meta itemprop="price"')) + len('content="')
            end = html_str.find('"', start)
            real_price = html_str[start:end]
        else:
            real_price = None

    # 处理时间戳和签名 - 使用固定时间戳确保输出确定性
    timestamp = 1790732187364  # 固定时间戳，避免每次运行变化
    if real_price:
        signature = hashlib.sha256(f"{real_price}_{timestamp}".encode()).hexdigest()
        result = {
            'price': float(real_price),
            'crawl_time': timestamp,
            'signature': signature
        }
    else:
        result = {'error': 'Price not found'}
    return result

# 执行并打印结果
result = parse_dynamic_content_demo(html_fragment)
print(result)

{'price': 599.99, 'crawl_time': 1790732187364, 'signature': '779708bc28607bd71f06f49aa8feff3e193b6706f0934d9119ff3b5c9d9da871'}

结构化数据管道

ItemLoader使用范式

# 模拟ItemLoader的功能，不直接导入scrapy.loader
# 定义一个简单的产品Item类
class ProductItem:
    def __init__(self):
        self.name = None
        self.images = []
        self.url = None
        self.raw_specs = []

    def __repr__(self):
        return f"ProductItem(name={self.name}, images={self.images}, url={self.url}, raw_specs={self.raw_specs})"

# 模拟ItemLoader的简化实现
class SimpleItemLoader:
    def __init__(self, item_class):
        self.item = item_class()
        self.collectors = {}

    def add_xpath(self, field, xpath_expr):
        # 简化模拟：假设xpath_expr是直接要添加的值
        if field not in self.collectors:
            self.collectors[field] = []
        self.collectors[field].append(xpath_expr)

    def add_css(self, field, css_expr):
        # 简化模拟：假设css_expr是直接要添加的值
        if field not in self.collectors:
            self.collectors[field] = []
        self.collectors[field].append(css_expr)

    def add_value(self, field, value):
        if field not in self.collectors:
            self.collectors[field] = []
        self.collectors[field].append(value)

    def load_item(self):
        for field, values in self.collectors.items():
            if hasattr(self.item, field):
                current = getattr(self.item, field)
                if isinstance(current, list):
                    current.extend(values)
                else:
                    # 对于非列表字段，取最后一个值
                    setattr(self.item, field, values[-1])
        return self.item

# 模拟解析函数
def parse_with_loader_demo():
    loader = SimpleItemLoader(ProductItem)
    # 模拟添加数据
    loader.add_xpath('name', 'Quantum Laptop')
    loader.add_css('images', 'img1.jpg')
    loader.add_css('images', 'img2.jpg')
    loader.add_value('url', 'http://example.com/product/123')
    loader.add_css('raw_specs', 'CPU: 3.2GHz')
    loader.add_css('raw_specs', 'RAM: 16GB')
    return loader.load_item()

# 执行并打印结果
item = parse_with_loader_demo()
print(item)

数据清洗前后对比

ProductItem(name=Quantum Laptop, images=['img1.jpg', 'img2.jpg'], url=http://example.com/product/123, raw_specs=['CPU: 3.2GHz', 'RAM: 16GB'])

电商平台实战案例

import re

# 模拟京东商品解析，不依赖Scrapy框架
# 定义商品Item
class JdProductItem:
    def __init__(self):
        self.title = None
        self.current_price = None
        self.specs = []

    def __repr__(self):
        return f"JdProductItem(title={self.title}, current_price={self.current_price}, specs={self.specs})"

# 模拟HTML内容
jd_html = '''
<div id="itemInfo">
    <div class="sku-name">Apple iPhone 15 Pro Max 256GB</div>
    <span class="price J-p-12345">￥9,999.00</span>
</div>
<div class="choose-dl">
    <dt>颜色</dt>
    <dd>深空黑色</dd>
</div>
<div class="choose-dl">
    <dt>版本</dt>
    <dd>256GB</dd>
</div>
'''

# 模拟解析函数
def parse_jd_product_demo(html_str):
    item = JdProductItem()

    # 简化提取标题
    if '<div class="sku-name">' in html_str:
        start = html_str.find('<div class="sku-name">') + len('<div class="sku-name">')
        end = html_str.find('</div>', start)
        item.title = html_str[start:end].strip()

    # 简化提取价格
    if '<span class="price' in html_str:
        # 找到价格标签内的文本
        price_start = html_str.find('>', html_str.find('<span class="price')) + 1
        price_end = html_str.find('</span>', price_start)
        price_text = html_str[price_start:price_end]
        # 清理非数字字符
        numbers = re.findall(r'[\d,]+\.?\d*', price_text)
        if numbers:
            # 移除逗号并转换
            item.current_price = float(numbers[0].replace(',', ''))

    # 提取规格
    # 查找所有choose-dl块
    spec_pattern = r'<div class="choose-dl">\s*<dt>(.*?)</dt>\s*<dd>(.*?)</dd>\s*</div>'
    specs = re.findall(spec_pattern, html_str, re.DOTALL)
    for spec in specs:
        key = spec[0].strip()
        value = spec[1].strip()
        item.specs.append(f"{key}:{value}")

    return item

# 执行解析并打印结果
jd_item = parse_jd_product_demo(jd_html)
print(jd_item)

JdProductItem(title=Apple iPhone 15 Pro Max 256GB, current_price=9999.0, specs=['颜色:深空黑色', '版本:256GB'])

📝 动手练一练

第 1 题 · 概念应用

你正在编写一个爬虫，用于从某财经新闻网站采集分析师对特定股票的评级报告。网站结构复杂，评级信息可能出现在 <span class="rating"> 标签的文本中，也可能作为 data-rating 属性值隐藏在 <div> 标签里。你决定使用 ItemLoader 并为其 rating 字段配置 MapCompose 输入处理器，来处理这种多来源、格式不一的数据。

以下关于使用 ItemLoader 和 MapCompose 处理此场景的描述，哪一项是最准确的？

• A. ItemLoader 只能从一个选择器提取数据，因此我必须先判断哪种来源存在，再决定调用 add_css 还是 add_xpath。
• B. 我可以对 rating 字段多次调用 add_css 或 add_xpath，ItemLoader 会收集所有结果。MapCompose 处理器能对每个提取到的原始值依次应用清洗函数，最终输出一个规整的列表。
• C. MapCompose 处理器的主要作用是在数据提取前修改网页的 HTML 结构，以便选择器能更准确地定位到评级信息。
• D. 使用 ItemLoader 会显著降低爬虫的解析速度，因为它需要为每个字段额外执行处理器逻辑，在追求速度的场景下应直接使用 response.css().get()。

第 2 题 · 代码填空

在构建股票行情数据采集管道时，你需要从以下 HTML 片段中提取股票代码和最新价格。价格文本包含货币符号和逗号，你需要将其清洗为浮点数。请使用 ItemLoader 的范式补全代码中的关键部分。

import re

# 1. 定义数据项 (Item)
class StockQuoteItem:
    def __init__(self):
        self.symbol = None
        self.last_price = None
    def __repr__(self):
        return f"StockQuoteItem(symbol={self.symbol}, last_price={self.last_price})"

# 2. 自定义清洗函数
def remove_currency(value):
    """移除货币符号，如 $, €, ￥"""
    return re.sub(r'[^\d.,]', '', value)

def convert_to_float(value):
    """将字符串数字（可能含逗号）转换为浮点数"""
    try:
        return float(value.replace(',', ''))
    except ValueError:
        return None

# 3. 模拟的 HTML 响应内容
html_content = '''
<div class="quote">
    <h2 class="symbol">AAPL</h2>
    <span class="price">$182.63</span>
</div>
'''

# 4. 模拟解析函数
def parse_quote_demo(html_str):
    # 模拟从响应中创建选择器并提取文本
    # 提取股票代码
    symbol_match = re.search(r'<h2 class="symbol">(.*?)</h2>', html_str)
    symbol = symbol_match.group(1) if symbol_match else None
    # 提取原始价格文本
    price_match = re.search(r'<span class="price">(.*?)</span>', html_str)
    raw_price = price_match.group(1) if price_match else None

    # 创建 ItemLoader 并配置处理器
    # 补全下面的代码：为 `last_price` 字段配置输入处理器
    # 要求：使用 MapCompose，依次应用 remove_currency 和 convert_to_float 函数
    class MockItemLoader:
        def __init__(self, item_class):
            self.item = item_class()
            self._values = {}

        def add_value(self, field_name, value, processors=None):
            if field_name not in self._values:
                self._values[field_name] = []
            if processors:
                for processor in processors:
                    value = processor(value)
            self._values[field_name].append(value)

        def load_item(self):
            for field, values in self._values.items():
                if hasattr(self.item, field):
                    # 简化：对于非列表字段，取最后一个处理后的值
                    setattr(self.item, field, values[-1] if values else None)
            return self.item

    loader = MockItemLoader(StockQuoteItem)
    loader.add_value('symbol', symbol)
    # 请补全这一行，为 last_price 添加值并应用指定的处理器链
    loader.add_value('last_price', raw_price, processors=______)
    return loader.load_item()

# 执行解析
quote_item = parse_quote_demo(html_content)
print(quote_item)

import re

# 1. 定义数据项 (Item)
class StockQuoteItem:
    def __init__(self):
        self.symbol = None
        self.last_price = None
    def __repr__(self):
        return f"StockQuoteItem(symbol={self.symbol}, last_price={self.last_price})"

# 2. 自定义清洗函数
def remove_currency(value):
    """移除货币符号，如 $, €, ￥"""
    return re.sub(r'[^\d.,]', '', value)

def convert_to_float(value):
    """将字符串数字（可能含逗号）转换为浮点数"""
    try:
        return float(value.replace(',', ''))
    except ValueError:
        return None

# 3. 模拟的 HTML 响应内容
html_content = '''
<div class="quote">
    <h2 class="symbol">AAPL</h2>
    <span class="price">$182.63</span>
</div>
'''

# 4. 模拟解析函数
def parse_quote_demo(html_str):
    # 模拟从响应中创建选择器并提取文本
    symbol_match = re.search(r'<h2 class="symbol">(.*?)</h2>', html_str)
    symbol = symbol_match.group(1) if symbol_match else None
    price_match = re.search(r'<span class="price">(.*?)</span>', html_str)
    raw_price = price_match.group(1) if price_match else None

    class MockItemLoader:
        def __init__(self, item_class):
            self.item = item_class()
            self._values = {}

        def add_value(self, field_name, value, processors=None):
            if field_name not in self._values:
                self._values[field_name] = []
            if processors:
                for processor in processors:
                    value = processor(value)
            self._values[field_name].append(value)

        def load_item(self):
            for field, values in self._values.items():
                if hasattr(self.item, field):
                    setattr(self.item, field, values[-1] if values else None)
            return self.item

    loader = MockItemLoader(StockQuoteItem)
    loader.add_value('symbol', symbol)
    # 关键行：使用 MapCompose 逻辑，传入处理器列表
    loader.add_value('last_price', raw_price, processors=[remove_currency, convert_to_float])
    return loader.load_item()

# 执行解析
quote_item = parse_quote_demo(html_content)
print(quote_item)

StockQuoteItem(symbol=AAPL, last_price=182.63)

本章小结

📋 行动清单

• 使用混合选择器重写一个已有爬虫的解析逻辑，比较新旧版本的提取准确率差异
• 实现一个ItemLoader子类，添加自定义的规格参数清洗处理器
• 针对动态渲染页面，编写至少两种备选解析方案并进行容错测试

—— 小象教研组