Scrapy实战(2)：编写财经爬虫逻辑

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

本节我们将深入 Scrapy 的核心工作流程，掌握 Spider 编写、选择器提取、Item Pipeline 清洗存储与分页跟进，最终完成一个完整的爬虫项目。

💡 核心导读

本节将深入Scrapy框架的核心工作流程，重点掌握Spider开发、数据提取与管道处理三大核心环节。通过实战演练，你将学会：1) 使用XPath和CSS选择器精确提取网页数据 2) 配置Item Pipeline实现数据清洗与存储 3) 处理分页与动态内容抓取。最终完成一个完整的新闻网站爬虫项目。

Spider开发进阶

创建结构化爬虫

import sqlite3

class NewsItem(dict):
    pass

class CleanPipeline:
    def process_item(self, item, spider):
        if 'title' in item:
            item['title'] = item['title'].replace('\xa0', ' ').strip()
        print(f"Cleaned item title: {item.get('title')}")
        return item

class SQLitePipeline:
    def open_spider(self, spider):
        self.conn = sqlite3.connect(':memory:')
        self.cursor = self.conn.cursor()
        self.cursor.execute('''
            CREATE TABLE IF NOT EXISTS articles
            (title TEXT, publish_date TEXT, content TEXT)
        ''')
        self.conn.commit()
        print("SQLite database and table created.")
    
    def process_item(self, item, spider):
        self.cursor.execute('''
            INSERT INTO articles (title, publish_date, content)
            VALUES (?, ?, ?)
        ''', (item.get('title'), item.get('publish_date'), item.get('content')))
        self.conn.commit()
        print(f"Item inserted into SQLite: {item.get('title')}")
        self.cursor.execute('SELECT COUNT(*) FROM articles')
        count = self.cursor.fetchone()[0]
        print(f"Total articles in database: {count}")
        return item
    
    def close_spider(self, spider):
        self.conn.close()
        print("SQLite connection closed.")

class MockSpider:
    name = 'mock_spider'

spider = MockSpider()
clean_pipe = CleanPipeline()
sqlite_pipe = SQLitePipeline()

test_item = NewsItem()
test_item['title'] = '  测试标题\xa0有空格  '
test_item['publish_date'] = '2023-10-27'
test_item['content'] = '测试内容'

sqlite_pipe.open_spider(spider)
cleaned_item = clean_pipe.process_item(test_item, spider)
sqlite_pipe.process_item(cleaned_item, spider)
sqlite_pipe.close_spider(spider)

选择器对比表

SQLite database and table created.
Cleaned item title: 测试标题 有空格
Item inserted into SQLite: 测试标题 有空格
Total articles in database: 1
SQLite connection closed.

数据处理管道

多级管道配置

class DynamicRenderMiddleware:
    def process_request(self, request, spider):
        if request.meta.get('dynamic', False):
            print(f"Dynamic rendering simulated for: {request.url}")
            simulated_html = """
            <html>
                <body>
                    <h1>动态加载的内容</h1>
                    <div class="stock-price">价格: 150.25</div>
                </body>
            </html>
            """
            class HtmlResponse:
                def __init__(self, url, body, encoding):
                    self.url = url
                    self.body = body
                    self.encoding = encoding
                def css(self, selector):
                    class SelectorResult:
                        def get(self):
                            if selector == 'div.stock-price::text':
                                return '价格: 150.25'
                            return None
                    return SelectorResult()
            return HtmlResponse(url=request.url, body=simulated_html.encode('utf-8'), encoding='utf-8')
        return None

middleware = DynamicRenderMiddleware()

class Request:
    def __init__(self, url, meta=None):
        self.url = url
        self.meta = meta or {}

dynamic_request = Request('https://example.com/dynamic', meta={'dynamic': True})

response = middleware.process_request(dynamic_request, None)
if response:
    print(f"Response received. Extracted price: {response.css('div.stock-price::text').get()}")
else:
    print("No dynamic rendering applied.")

数据流示意图

Dynamic rendering simulated for: https://example.com/dynamic
Response received. Extracted price: 价格: 150.25

动态内容处理

中间件配置示例

class FinancialSpider:
    name = "finance_news"
    start_urls = ["https://example-news.com/stock", "https://example-news.com/bond"]
    
    def parse(self, response):
        print(f"Parsing financial news from: {response.url}")
        mock_tickers = ['AAPL', 'GOOGL', 'MSFT']
        for ticker in mock_tickers:
            item = {
                'ticker': ticker,
                'price': 100.0 + len(ticker) * 10,
                'news_headline': f'News about {ticker}'
            }
            print(f"Extracted financial data: {item}")
            yield item

spider = FinancialSpider()
print(f"Spider name: {spider.name}")
print(f"Start URLs: {spider.start_urls}")

class HtmlResponse:
    def __init__(self, url, body):
        self.url = url
        self.body = body

mock_response = HtmlResponse(url='https://example-news.com/stock', body=b'<html></html>')
results = list(spider.parse(mock_response))
print(f"Generated {len(results)} items from parse method.")

Spider name: finance_news
Start URLs: ['https://example-news.com/stock', 'https://example-news.com/bond']
Parsing financial news from: https://example-news.com/stock
Extracted financial data: {'ticker': 'AAPL', 'price': 140.0, 'news_headline': 'News about AAPL'}
Extracted financial data: {'ticker': 'GOOGL', 'price': 150.0, 'news_headline': 'News about GOOGL'}
Extracted financial data: {'ticker': 'MSFT', 'price': 140.0, 'news_headline': 'News about MSFT'}
Generated 3 items from parse method.

📝 动手练一练

第 1 题 · 概念应用

小象正在编写一个爬虫，用于采集多家财经新闻网站的文章，以构建一个量化交易事件数据库。他设计了一个名为 FinanceNewsSpider 的爬虫类，并希望遵循 Scrapy 的最佳实践来构建数据流。以下是他对数据流设计的描述，请判断哪一项是正确的？

• A. 在 parse 方法中，使用 for 循环提取出当前页面的所有新闻条目，将它们打包成一个 items 列表，最后通过 return items 一次性返回——这样引擎只会处理列表中的第一个条目，其余条目不会进入 Item Pipeline。
• B. 在 parse 方法中，每提取并封装好一条新闻数据（一个 Item 对象），就立即调用一个自定义的 save_to_database(item) 方法将其存入数据库，以确保数据不丢失。
• C. 在 parse 方法中，使用 yield 关键字，将提取到的每个新闻 Item 对象逐个生成并交给 Scrapy 引擎，引擎会将其自动送入配置好的 Item Pipeline 进行处理。
• D. 为了提升效率，应该在 parse 方法内部直接打印提取到的数据，Scrapy 的日志系统会自动捕获这些打印信息并将其存储到指定文件中。

第 2 题 · 代码填空

在下面的模拟场景中，我们实现了一个极简的 Item Pipeline 组件，用于清洗和验证爬取到的“股票研究报告”数据。该管道需要完成两项任务：1) 确保报告标题不为空且去除首尾空格；2) 将字符串格式的“目标价”转换为浮点数。请补全管道类 ResearchReportPipeline 中处理单个数据项的核心方法名。

import json

class MockItem:
    """模拟Scrapy中的Item对象"""
    def __init__(self, data):
        self._data = data
    def get(self, key, default=None):
        return self._data.get(key, default)
    def __setitem__(self, key, value):
        self._data[key] = value
    def __getitem__(self, key):
        return self._data[key]

class DropItem(Exception):
    """模拟 scrapy.exceptions.DropItem：管道主动丢弃不合格的数据项"""

class ResearchReportPipeline:
    """模拟的Item Pipeline，用于清洗和验证数据"""

    def ______(self, item, spider):
        """
        处理单个item的核心方法。
        item: 包含提取数据的字典或Item对象。
        spider: 产生该item的爬虫实例。
        返回值: 处理后的item；不合格的item应raise DropItem丢弃。
        """
        # 1. 清洗标题：先去首尾空格，再校验非空（None 或纯空白都要丢弃）
        title = item.get('title')
        if not title or not title.strip():
            raise DropItem(f"标题缺失或为空白，股票代码：{item.get('ticker')}")
        item['title'] = title.strip()

        # 2. 转换目标价
        target_price_str = item.get('target_price')
        if target_price_str:
            try:
                # 移除可能存在的货币符号和逗号，如 "$1,250.50"
                cleaned = target_price_str.replace('$', '').replace(',', '')
                item['target_price'] = float(cleaned)
            except ValueError:
                print(f"无法将目标价 '{target_price_str}' 转换为浮点数")
                item['target_price'] = None
        return item

# 模拟使用管道
pipeline = ResearchReportPipeline()
mock_spider = type('MockSpider', (), {'name': 'test_spider'})()

# 创建一个测试数据项
test_item = MockItem({
    'ticker': 'AAPL',
    'title': '  苹果公司季度业绩超预期，维持买入评级  ',
    'target_price': '$195.50',
    'analyst': '摩根士丹利'
})

print("处理前:", test_item._data)
processed_item = pipeline.process_item(test_item, mock_spider)
print("处理后:", processed_item._data)

# 缺标题的不合格项应被丢弃
bad_item = MockItem({'ticker': 'TSLA', 'title': '   ', 'target_price': '$300'})
try:
    pipeline.process_item(bad_item, mock_spider)
except DropItem as e:
    print(f"已丢弃: {e}")

import json

class MockItem:
    """模拟Scrapy中的Item对象"""
    def __init__(self, data):
        self._data = data
    def get(self, key, default=None):
        return self._data.get(key, default)
    def __setitem__(self, key, value):
        self._data[key] = value
    def __getitem__(self, key):
        return self._data[key]

class DropItem(Exception):
    """模拟 scrapy.exceptions.DropItem：管道主动丢弃不合格的数据项"""

class ResearchReportPipeline:
    """模拟的Item Pipeline，用于清洗和验证数据"""

    def process_item(self, item, spider):
        """
        处理单个item的核心方法。
        item: 包含提取数据的字典或Item对象。
        spider: 产生该item的爬虫实例。
        返回值: 处理后的item；不合格的item应raise DropItem丢弃。
        """
        # 1. 清洗标题：先去首尾空格，再校验非空（None 或纯空白都要丢弃）
        title = item.get('title')
        if not title or not title.strip():
            raise DropItem(f"标题缺失或为空白，股票代码：{item.get('ticker')}")
        item['title'] = title.strip()

        # 2. 转换目标价
        target_price_str = item.get('target_price')
        if target_price_str:
            try:
                # 移除可能存在的货币符号和逗号，如 "$1,250.50"
                cleaned = target_price_str.replace('$', '').replace(',', '')
                item['target_price'] = float(cleaned)
            except ValueError:
                print(f"无法将目标价 '{target_price_str}' 转换为浮点数")
                item['target_price'] = None
        return item

# 模拟使用管道
pipeline = ResearchReportPipeline()
mock_spider = type('MockSpider', (), {'name': 'test_spider'})()

# 创建一个测试数据项
test_item = MockItem({
    'ticker': 'AAPL',
    'title': '  苹果公司季度业绩超预期，维持买入评级  ',
    'target_price': '$195.50',
    'analyst': '摩根士丹利'
})

print("处理前:", test_item._data)
processed_item = pipeline.process_item(test_item, mock_spider)
print("处理后:", processed_item._data)

# 缺标题的不合格项应被丢弃
bad_item = MockItem({'ticker': 'TSLA', 'title': '   ', 'target_price': '$300'})
try:
    pipeline.process_item(bad_item, mock_spider)
except DropItem as e:
    print(f"已丢弃: {e}")

处理前: {'ticker': 'AAPL', 'title': '  苹果公司季度业绩超预期，维持买入评级  ', 'target_price': '$195.50', 'analyst': '摩根士丹利'}
处理后: {'ticker': 'AAPL', 'title': '苹果公司季度业绩超预期，维持买入评级', 'target_price': 195.5, 'analyst': '摩根士丹利'}
已丢弃: 标题缺失或为空白，股票代码：TSLA

本章小结

📋 行动清单

• 创建一个爬取豆瓣电影Top250的Spider，要求提取电影名称、评分和短评数量
• 实现管道系统将抓取数据同时存储到CSV文件和MySQL数据库
• 使用Selenium中间件抓取需要登录才能访问的电商网站商品列表

「小象量化教研组」提示：运行爬虫时请遵守robots.txt规则，设置合理的DOWNLOAD_DELAY参数避免对目标网站造成压力。

—— 小象教研组