Scrapy爬虫模块 - Python核心技能与应用

爬虫模块介绍

什么是Scrapy爬虫？

Scrapy爬虫是框架的核心组件，负责定义爬取逻辑、解析网页内容、提取数据以及生成新的请求。每个爬虫都是一个独立的Python类，继承自scrapy.Spider基类。

# 爬虫模块的主要职责
- 定义起始URL和爬取范围
- 解析网页内容，提取结构化数据
- 生成新的请求，实现深度爬取
- 处理页面间的链接关系
- 实现数据清洗和验证逻辑
- 与管道模块协同处理数据

故事化案例：侦探调查案件

想象一下，你是一名经验丰富的侦探，正在调查一个复杂的案件：

确定调查的起始地点和线索（定义起始URL）
收集现场的证据和相关信息（解析网页内容）
分析证据之间的联系，发现新的线索（提取数据）
根据线索扩大调查范围（生成新的请求）
整理证据，形成调查报告（数据清洗和存储）
与其他部门协作完成调查（与管道模块协同）

在这个类比中，Scrapy爬虫就是侦探，它需要：

# 爬虫与侦探调查的类比
爬虫定义起始URL → 侦探确定调查起点
爬虫解析网页 → 侦探收集现场证据
爬虫提取数据 → 侦探分析证据信息
爬虫生成新请求 → 侦探扩大调查范围
爬虫数据清洗 → 侦探整理调查报告
爬虫与管道协同 → 侦探与部门协作

爬虫的生命周期

一个Scrapy爬虫从启动到结束会经历完整的生命周期：

爬虫生命周期阶段

初始化阶段：爬虫类被实例化，设置初始参数
启动阶段：调用start_requests()方法生成初始请求
爬取阶段：处理响应，解析数据，生成新请求
数据处理阶段：提取的数据通过管道进行处理
结束阶段：爬虫完成所有任务，调用closed()方法

爬虫类型与模式

常见爬虫类型

Scrapy支持多种爬虫类型，可以根据不同的爬取需求选择合适的模式：

通用爬虫 (Spider)

• 最基本的爬虫类型
• 适合简单的页面爬取
• 需要手动处理分页和链接
• 灵活性最高，控制力强

爬取规则爬虫 (CrawlSpider)

• 基于规则的自动爬取
• 使用LinkExtractor提取链接
• 适合结构化网站爬取
• 减少重复代码编写

XML内容爬虫 (XMLFeedSpider)

• 专门处理XML格式数据
• 支持RSS、Atom等订阅源
• 自动解析XML节点
• 适合新闻、博客等站点

站点地图爬虫 (SitemapSpider)

• 基于网站地图爬取
• 自动解析sitemap.xml
• 适合搜索引擎优化站点
• 爬取效率高，覆盖全面

CSV数据爬虫 (CSVFeedSpider)

• 处理CSV格式数据
• 自动解析表格数据
• 适合数据导出站点
• 支持自定义分隔符

自定义爬虫

• 继承基类自定义功能
• 支持复杂业务逻辑
• 适合特殊需求场景
• 灵活性最高

爬虫模式选择指南

根据不同的爬取场景，选择合适的爬虫类型：

# 爬虫类型选择指南

# 场景1：简单页面爬取
# 适用：通用爬虫 (Spider)
# 特点：手动控制，灵活性高

# 场景2：结构化网站爬取
# 适用：爬取规则爬虫 (CrawlSpider)
# 特点：自动链接提取，减少重复代码

# 场景3：新闻博客站点
# 适用：XML内容爬虫 (XMLFeedSpider)
# 特点：自动解析订阅源，适合动态内容

# 场景4：SEO优化站点
# 适用：站点地图爬虫 (SitemapSpider)
# 特点：基于sitemap，爬取效率高

# 场景5：数据导出站点
# 适用：CSV数据爬虫 (CSVFeedSpider)
# 特点：处理表格数据，适合批量处理

# 场景6：特殊需求场景
# 适用：自定义爬虫
# 特点：完全自定义，适合复杂业务

代码示例

基础爬虫示例

以下是一个简单的Scrapy爬虫示例，演示基本的爬取逻辑：

# spiders/example_spider.py
import scrapy
from scrapy.linkextractors import LinkExtractor
from scrapy.spiders import CrawlSpider, Rule

class ExampleSpider(scrapy.Spider):
    name = 'example'
    allowed_domains = ['example.com']
    start_urls = ['http://www.example.com/']
    
    def parse(self, response):
        """解析起始页面的回调函数"""
        # 提取页面标题
        title = response.css('title::text').get()
        
        # 提取所有链接
        links = response.css('a::attr(href)').getall()
        
        # 生成数据项
        yield {
            'url': response.url,
            'title': title,
            'links_count': len(links),
            'links': links[:10]  # 只保留前10个链接
        }
        
        # 生成新的请求（深度爬取）
        for link in links[:5]:  # 只爬取前5个链接
            if link.startswith('http'):
                yield scrapy.Request(
                    url=link,
                    callback=self.parse_detail
                )
    
    def parse_detail(self, response):
        """解析详情页面的回调函数"""
        # 提取详细信息
        yield {
            'url': response.url,
            'title': response.css('title::text').get(),
            'content_length': len(response.text),
            'status_code': response.status
        }

# 使用CrawlSpider的示例
class ExampleCrawlSpider(CrawlSpider):
    name = 'example_crawl'
    allowed_domains = ['example.com']
    start_urls = ['http://www.example.com/']
    
    # 定义爬取规则
    rules = (
        # 提取所有链接，并调用parse_items方法
        Rule(LinkExtractor(), callback='parse_items', follow=True),
    )
    
    def parse_items(self, response):
        """解析每个页面的回调函数"""
        item = {}
        item['url'] = response.url
        item['title'] = response.css('title::text').get()
        
        # 提取正文内容
        item['content'] = response.css('p::text').getall()[:3]  # 前3段
        
        # 提取图片链接
        item['images'] = response.css('img::attr(src)').getall()
        
        yield item

高级爬虫示例

以下是一个更复杂的爬虫示例，包含错误处理、数据验证等功能：

# spiders/advanced_spider.py
import scrapy
import json
from urllib.parse import urljoin
from scrapy.exceptions import CloseSpider
from items import NewsItem

class AdvancedSpider(scrapy.Spider):
    name = 'advanced_news'
    
    # 配置参数
    custom_settings = {
        'CONCURRENT_REQUESTS': 8,
        'DOWNLOAD_DELAY': 1,
        'AUTOTHROTTLE_ENABLED': True,
        'ITEM_PIPELINES': {
            'pipelines.NewsPipeline': 300,
        }
    }
    
    def __init__(self, category=None, *args, **kwargs):
        super(AdvancedSpider, self).__init__(*args, **kwargs)
        self.category = category or 'technology'
        self.page_count = 0
        self.max_pages = 10  # 最大爬取页数
    
    def start_requests(self):
        """生成起始请求"""
        base_url = f'https://news.example.com/{self.category}'
        yield scrapy.Request(
            url=base_url,
            callback=self.parse_list,
            meta={'page': 1}
        )
    
    def parse_list(self, response):
        """解析新闻列表页"""
        # 检查页面是否有效
        if response.status != 200:
            self.logger.warning(f'页面访问失败: {response.url}')
            return
        
        # 提取新闻链接
        news_links = response.css('.news-item a::attr(href)').getall()
        
        if not news_links:
            self.logger.info('没有找到新闻链接，可能已到达最后一页')
            return
        
        # 生成详情页请求
        for link in news_links:
            absolute_url = urljoin(response.url, link)
            yield scrapy.Request(
                url=absolute_url,
                callback=self.parse_detail,
                errback=self.handle_error
            )
        
        # 生成下一页请求
        self.page_count += 1
        if self.page_count < self.max_pages:
            next_page = response.css('.next-page::attr(href)').get()
            if next_page:
                next_url = urljoin(response.url, next_page)
                yield scrapy.Request(
                    url=next_url,
                    callback=self.parse_list,
                    meta={'page': self.page_count + 1}
                )
    
    def parse_detail(self, response):
        """解析新闻详情页"""
        # 数据验证
        if not self._validate_response(response):
            return
        
        # 创建数据项
        item = NewsItem()
        
        # 提取字段
        item['title'] = response.css('h1.news-title::text').get()
        item['content'] = ' '.join(response.css('.news-content p::text').getall())
        item['author'] = response.css('.author::text').get()
        item['publish_time'] = response.css('.publish-time::text').get()
        item['source_url'] = response.url
        
        # 数据清洗
        item = self._clean_item(item)
        
        # 数据验证
        if self._validate_item(item):
            yield item
        else:
            self.logger.warning(f'数据验证失败: {response.url}')
    
    def handle_error(self, failure):
        """错误处理"""
        self.logger.error(f'请求失败: {failure.value}')
        
        # 可以根据错误类型进行不同处理
        if failure.check(scrapy.exceptions.IgnoreRequest):
            self.logger.warning('请求被忽略')
        elif failure.check(scrapy.exceptions.DropItem):
            self.logger.warning('数据项被丢弃')
    
    def _validate_response(self, response):
        """验证响应有效性"""
        if response.status != 200:
            return False
        
        # 检查页面内容是否有效
        title = response.css('title::text').get()
        if not title or len(title.strip()) < 5:
            return False
        
        return True
    
    def _clean_item(self, item):
        """数据清洗"""
        # 清理标题
        if item['title']:
            item['title'] = item['title'].strip()
        
        # 清理内容
        if item['content']:
            item['content'] = ' '.join(item['content'].split())
        
        return item
    
    def _validate_item(self, item):
        """数据验证"""
        # 检查必填字段
        required_fields = ['title', 'content']
        for field in required_fields:
            if not item.get(field):
                return False
        
        # 检查字段长度
        if len(item['title']) < 5 or len(item['content']) < 50:
            return False
        
        return True
    
    def closed(self, reason):
        """爬虫结束时的处理"""
        self.logger.info(f'爬虫结束，原因: {reason}')
        self.logger.info(f'总共爬取了 {self.page_count} 页')

数据项定义示例

定义爬虫使用的数据项模型：

# items.py - 数据项定义
import scrapy
from scrapy.loader import ItemLoader
from scrapy.loader.processors import TakeFirst, MapCompose, Join

class NewsItem(scrapy.Item):
    # 定义字段
    title = scrapy.Field()
    content = scrapy.Field()
    author = scrapy.Field()
    publish_time = scrapy.Field()
    source_url = scrapy.Field()
    category = scrapy.Field()
    tags = scrapy.Field()
    
    # 字段处理处理器
    title_out = TakeFirst()
    content_out = Join()
    author_out = TakeFirst()

class NewsItemLoader(ItemLoader):
    """自定义ItemLoader"""
    default_item_class = NewsItem
    
    # 字段处理
    title_in = MapCompose(str.strip)
    content_in = MapCompose(str.strip)
    author_in = MapCompose(str.strip)
    
    # 输出处理
    title_out = TakeFirst()
    content_out = Join(' ')
    author_out = TakeFirst()

# 使用ItemLoader的示例
def parse_with_loader(self, response):
    loader = NewsItemLoader(item=NewsItem(), response=response)
    
    # 使用CSS选择器提取
    loader.add_css('title', 'h1.news-title::text')
    loader.add_css('content', '.news-content p::text')
    loader.add_css('author', '.author::text')
    loader.add_value('source_url', response.url)
    
    # 返回处理后的数据项
    yield loader.load_item()

练习题

基础练习题

描述Scrapy爬虫的主要功能和作用。
解释爬虫的生命周期各个阶段。
如何定义起始URL和爬取范围？
爬虫如何与下载器、管道模块协同工作？
什么是回调函数？在爬虫中如何使用？

进阶练习题

设计一个支持深度优先和广度优先爬取的爬虫。
如何实现爬虫的错误处理和重试机制？
解释ItemLoader的作用和使用方法。
设计一个支持动态参数配置的爬虫。
如何优化爬虫的性能和内存使用？

实践练习题

创建一个新闻网站爬虫，提取标题、内容和发布时间。
编写一个电商网站爬虫，支持分页和商品详情爬取。
实现一个图片爬虫，自动下载网页中的图片。
设计一个API数据爬虫，处理JSON格式的响应。
创建一个多语言网站爬虫，支持内容翻译。

思考题

在大规模爬虫项目中，如何管理多个爬虫？
爬虫如何应对网站的反爬虫机制？
设计一个支持增量爬取的爬虫架构。
在分布式环境中，爬虫应该如何设计？
未来爬虫技术的发展趋势是什么？