Scrapy 框架的核心价值在于四大组件（Spider/Item/Pipeline/Downloader）的协同工作，优秀的案例必然是组件职责清晰、配置优化合理、可复用性强的。

案例总览（梯度进阶，覆盖核心组件）

案例序号	案例名称	核心目标	覆盖核心组件	难度	适用场景
1	博客园文章静态爬取（双格式存储）	掌握组件基础协同、数据结构化与持久化	Spider/Item/Pipeline/Downloader（基础配置）	入门	静态网页、小批量数据爬取
2	豆瓣电影Top250 分页爬取	掌握Spider分页跟进、数据去重	Spider（核心进阶）/Item/Pipeline/Downloader	进阶	静态分页网页、批量数据爬取
3	知乎专栏文章 MySQL 存储	掌握Pipeline数据库持久化、配置解耦	Pipeline（核心进阶）/其余组件全覆盖	进阶	数据长期存储、批量结构化数据
4	京东商品列表 动态网页爬取	掌握Downloader拓展（处理AJAX渲染）	Downloader（进阶拓展）/其余组件全覆盖	高阶	动态渲染网页、反爬较弱场景

案例1：博客园文章静态爬取（双格式存储）

案例场景

爬取博客园首页推荐文章的标题、作者、发布时间、文章链接，将数据结构化后，同时保存为JSON文件和Excel文件，核心是验证四大组件的基础协同工作流程。

核心准备

1. 创建Scrapy项目：scrapy startproject cnblog_static_spider
2. 安装依赖（Excel存储需额外安装）：pip install pandas openpyxl

步骤1：Item 组件（数据结构化，规范字段与清洗）

编辑项目中的items.py，定义字段并添加基础数据清洗逻辑，避免脏数据传递到Pipeline。

import scrapy
from scrapy.loader.processors import MapCompose, TakeFirst

# 自定义清洗函数（对应Item字段的前置处理）
def clean_string(value):
    """去除字符串前后空格、换行符、制表符"""
    if isinstance(value, str):
        return value.strip().replace("\n", "").replace("\t", "")
    return value

def clean_url(value):
    """补全相对链接为绝对链接（防止博客园内部相对路径）"""
    if value and not value.startswith("http"):
        return f"https://www.cnblogs.com{value}"
    return value or "未知链接"

# 定义文章Item（严格对应爬取字段，职责：结构化+前置清洗）
class CnblogStaticArticleItem(scrapy.Item):
    title = scrapy.Field(
        input_processor=MapCompose(clean_string),  # 调用自定义清洗函数
        output_processor=TakeFirst()  # 提取单个结果，避免返回列表
    )
    author = scrapy.Field(
        input_processor=MapCompose(clean_string),
        output_processor=TakeFirst()
    )
    publish_time = scrapy.Field(
        input_processor=MapCompose(clean_string),
        output_processor=TakeFirst(),
        default="未知时间"  # 可选字段设置默认值，避免KeyError
    )
    link = scrapy.Field(
        input_processor=MapCompose(clean_url),
        output_processor=TakeFirst()
    )

步骤2：Spider 组件（爬取与解析，核心业务逻辑）

在spiders/目录下创建cnblog_article_spider.py，定义爬取规则，解析响应并生成Item对象，职责是发起请求、提取数据、传递Item。

import scrapy
from cnblog_static_spider.items import CnblogStaticArticleItem

class CnblogArticleSpider(scrapy.Spider):
    # 爬虫唯一标识（运行时需指定）
    name = "cnblog_static_article"
    # 允许爬取的域名（限制Downloader的爬取范围，避免资源浪费）
    allowed_domains = ["cnblogs.com"]
    # 初始请求URL（传递给Downloader下载）
    start_urls = ["https://www.cnblogs.com/"]

    # 核心解析方法：处理Downloader返回的Response，提取数据
    def parse(self, response):
        # 1. 用XPath定位文章列表（Scrapy内置XPath/CSS解析，高效稳定）
        article_list = response.xpath('//div[@class="post-item"]')

        # 2. 遍历解析单篇文章数据
        for article in article_list:
            # 初始化Item对象（承接结构化数据）
            item = CnblogStaticArticleItem()

            # 3. 提取字段数据（赋值给Item，自动触发清洗逻辑）
            item["title"] = article.xpath('.//a[@class="post-item-title"]/text()').extract()
            item["author"] = article.xpath('.//a[@class="post-item-author"]/text()').extract()
            item["publish_time"] = article.xpath('.//span[@class="post-date"]/text()').extract()
            item["link"] = article.xpath('.//a[@class="post-item-title"]/@href').extract()

            # 4. 传递Item给Pipeline（通过yield返回给引擎，再转发到Pipeline）
            yield item

步骤3：Downloader 组件（配置优化，提升下载稳定性）

无需编写代码，仅在settings.py中配置Downloader的核心参数，职责是高效、安全地下载网页响应。

# 1. 配置下载延迟（避免高频请求被反爬，单位：秒）
DOWNLOAD_DELAY = 2

# 2. 配置并发请求数（全局/单域名，平衡速度与反爬）
CONCURRENT_REQUESTS = 16  # 全局最大并发
CONCURRENT_REQUESTS_PER_DOMAIN = 8  # 单域名最大并发

# 3. 配置请求头（模拟浏览器，避免被识别为爬虫）
USER_AGENT = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"

# 4. 配置超时时间（请求超时后放弃，避免阻塞）
DOWNLOAD_TIMEOUT = 30

# 5. 禁用Cookie（静态网页无需保持会话，减少反爬风险）
COOKIES_ENABLED = False

步骤4：Pipeline 组件（双格式持久化，解耦数据处理）

编辑pipelines.py，实现两个Pipeline（JSON存储+Excel存储），按优先级执行，职责是数据后处理、去重、持久化。

import json
import pandas as pd
from openpyxl import load_workbook

# Pipeline 1：JSON文件存储（优先级300，先执行）
class CnblogJsonPipeline:
    """将Item数据保存到cnblog_articles.json（追加模式，避免覆盖）"""
    def open_spider(self, spider):
        """爬虫启动时执行：打开JSON文件，初始化列表"""
        self.file = open("cnblog_articles.json", "a", encoding="utf-8")
        self.item_list = []

    def process_item(self, item, spider):
        """核心处理方法：接收Item，添加到列表"""
        self.item_list.append(dict(item))
        return item  # 返回Item，传递给下一个Pipeline

    def close_spider(self, spider):
        """爬虫关闭时执行：将列表数据写入JSON文件"""
        # 格式化写入，保证可读性
        json.dump(self.item_list, self.file, ensure_ascii=False, indent=4)
        self.file.close()
        spider.logger.info(f"JSON数据已保存：cnblog_articles.json，共{len(self.item_list)}条")

# Pipeline 2：Excel文件存储（优先级400，后执行）
class CnblogExcelPipeline:
    """将Item数据保存到cnblog_articles.xlsx（追加模式，支持重复爬取）"""
    def __init__(self):
        self.excel_path = "cnblog_articles.xlsx"
        self.item_list = []

    def process_item(self, item, spider):
        self.item_list.append(dict(item))
        return item

    def close_spider(self, spider):
        """爬虫关闭时执行：将数据写入Excel，支持追加"""
        df = pd.DataFrame(self.item_list)

        try:
            # 若文件已存在，追加写入（不覆盖原有数据）
            book = load_workbook(self.excel_path)
            with pd.ExcelWriter(self.excel_path, engine="openpyxl", mode="a", if_sheet_exists="overlay") as writer:
                writer.book = book
                startrow = writer.sheets["Sheet1"].max_row  # 从下一行开始写入
                df.to_excel(writer, sheet_name="Sheet1", startrow=startrow, index=False, header=False)
        except FileNotFoundError:
            # 若文件不存在，创建新文件并写入
            df.to_excel(self.excel_path, sheet_name="Sheet1", index=False)

        spider.logger.info(f"Excel数据已保存：{self.excel_path}，共{len(self.item_list)}条")

步骤5：启用Pipeline，运行爬虫

1. 在settings.py中启用Pipeline，配置优先级（数字越小，优先级越高）：

ITEM_PIPELINES = {
    "cnblog_static_spider.pipelines.CnblogJsonPipeline": 300,
    "cnblog_static_spider.pipelines.CnblogExcelPipeline": 400,
}

2. 终端运行爬虫（进入项目根目录）：

scrapy crawl cnblog_static_article

案例验证与核心亮点

1. 验证结果：项目根目录生成cnblog_articles.json和cnblog_articles.xlsx，数据无空值、格式统一；
2. 核心亮点：四大组件职责清晰，Item实现前置清洗，Pipeline实现解耦存储，Downloader配置兼顾速度与反爬，符合Scrapy架构设计思想。

案例2：豆瓣电影Top250 分页爬取（带数据去重）

案例场景

爬取豆瓣电影Top250的电影名称、评分、简介、详情链接，自动跟进分页（共10页），通过Pipeline去重（避免重复爬取同一部电影），最终保存为Excel文件，核心是掌握Spider的分页跟进能力和Pipeline的去重逻辑。

核心难点与解决思路

• 分页跟进：豆瓣Top250分页URL规律为https://movie.douban.com/top250?start=0&filter=（start从0开始，每次+25），通过循环生成分页URL，或解析下一页链接跟进；
• 数据去重：以电影名称+评分为唯一标识，通过Pipeline中的集合存储已爬取数据，实现去重。

关键组件实现（仅展示核心差异部分，其余组件类似案例1）

1. Spider 组件（分页跟进，核心进阶）

在spiders/目录下创建douban_top250_spider.py：

import scrapy
from cnblog_static_spider.items import DoubanTop250Item  # 需提前在items.py定义对应字段

class DoubanTop250Spider(scrapy.Spider):
    name = "douban_top250"
    allowed_domains = ["movie.douban.com"]
    # 初始URL（第一页）
    start_urls = ["https://movie.douban.com/top250"]

    def parse(self, response):
        # 1. 解析当前页电影数据
        movie_list = response.xpath('//div[@class="item"]')
        for movie in movie_list:
            item = DoubanTop250Item()
            item["movie_name"] = movie.xpath('.//span[@class="title"][1]/text()').extract()
            item["score"] = movie.xpath('.//span[@class="rating_num"]/text()').extract()
            item["intro"] = movie.xpath('.//span[@class="inq"]/text()').extract()
            item["detail_link"] = movie.xpath('.//div[@class="hd"]/a/@href').extract()

            yield item

        # 2. 分页跟进（解析下一页链接，递归爬取）
        next_page = response.xpath('//span[@class="next"]/a/@href').extract_first()
        if next_page:
            # 补全绝对URL，发起新请求（传递给Downloader下载）
            next_page_url = f"https://movie.douban.com/top250{next_page}"
            yield scrapy.Request(
                url=next_page_url,
                callback=self.parse  # 回调自身，继续解析下一页
            )

2. Pipeline 组件（数据去重，核心进阶）

在pipelines.py中添加去重Pipeline，优先级高于存储Pipeline：

class DoubanDuplicatePipeline:
    """电影数据去重Pipeline：基于电影名称+评分去重"""
    def __init__(self):
        # 初始化集合，存储已爬取的（名称+评分）唯一标识
        self.movie_identifiers = set()

    def process_item(self, item, spider):
        # 构建唯一标识（避免单一字段重复）
        movie_name = item.get("movie_name", "")
        movie_score = item.get("score", "")
        identifier = f"{movie_name}_{movie_score}"

        if identifier in self.movie_identifiers:
            # 重复数据，丢弃并记录日志
            raise scrapy.exceptions.DropItem(f"重复电影数据：{movie_name}（评分：{movie_score}）")
        else:
            # 非重复数据，添加到集合并传递给下一个Pipeline
            self.movie_identifiers.add(identifier)
            return item

3. 运行与验证

1. 启用Pipeline：在settings.py中配置去重Pipeline优先级最高；
2. 运行爬虫：scrapy crawl douban_top250；
3. 验证结果：Excel文件中无重复电影，共250条数据，分页跟进正常。

案例3：知乎专栏文章 MySQL 存储（进阶持久化）

案例场景

爬取知乎指定专栏的文章标题、作者、发布时间、阅读量、文章链接，将结构化数据存入MySQL数据库，核心是掌握Pipeline的数据库持久化逻辑，实现数据的长期存储和后续分析。

核心准备

1. 安装MySQL驱动：pip install pymysql
2. MySQL中创建数据库和表：

-- 创建数据库
CREATE DATABASE IF NOT EXISTS zhihu_spider DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;

-- 切换数据库
USE zhihu_spider;

-- 创建文章表
CREATE TABLE IF NOT EXISTS zhihu_column_article (
    id INT AUTO_INCREMENT PRIMARY KEY COMMENT '自增ID',
    title VARCHAR(255) NOT NULL COMMENT '文章标题',
    author VARCHAR(100) NOT NULL COMMENT '文章作者',
    publish_time VARCHAR(50) COMMENT '发布时间',
    read_count INT DEFAULT 0 COMMENT '阅读量',
    link VARCHAR(255) UNIQUE COMMENT '文章详情链接（唯一约束）',
    create_time DATETIME DEFAULT CURRENT_TIMESTAMP COMMENT '入库时间'
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT '知乎专栏文章表';

关键组件实现（Pipeline 数据库持久化）

在pipelines.py中实现MySQL存储Pipeline，核心是数据库连接池管理、数据插入、异常处理：

import pymysql
from twisted.enterprise import adbapi  # Scrapy内置异步数据库连接库

class ZhihuMysqlPipeline:
    """知乎专栏文章MySQL存储Pipeline（异步插入，提升效率）"""
    def __init__(self, db_pool):
        self.db_pool = db_pool

    @classmethod
    def from_settings(cls, settings):
        """从settings.py中读取数据库配置，初始化连接池"""
        db_params = {
            "host": settings.get("MYSQL_HOST"),
            "port": settings.get("MYSQL_PORT"),
            "user": settings.get("MYSQL_USER"),
            "password": settings.get("MYSQL_PASSWORD"),
            "db": settings.get("MYSQL_DB"),
            "charset": "utf8mb4",
            "cursorclass": pymysql.cursors.DictCursor  # 游标类型
        }
        # 创建异步数据库连接池
        db_pool = adbapi.ConnectionPool("pymysql", **db_params)
        return cls(db_pool)

    def process_item(self, item, spider):
        """异步插入数据（避免阻塞爬虫进程）"""
        query = self.db_pool.runInteraction(self.insert_item, item)
        # 处理插入异常
        query.addErrback(self.handle_error, item, spider)
        return item

    def insert_item(self, cursor, item):
        """执行数据插入SQL（避免SQL注入，使用参数化查询）"""
        insert_sql = """
            INSERT INTO zhihu_column_article (title, author, publish_time, read_count, link)
            VALUES (%s, %s, %s, %s, %s)
            ON DUPLICATE KEY UPDATE  # 唯一约束冲突时，更新阅读量（避免重复插入）
                read_count = VALUES(read_count),
                publish_time = VALUES(publish_time)
        """
        # 提取Item数据，转换格式
        data = (
            item.get("title", ""),
            item.get("author", ""),
            item.get("publish_time", ""),
            int(item.get("read_count", 0)),
            item.get("link", "")
        )
        # 执行SQL
        cursor.execute(insert_sql, data)

    def handle_error(self, failure, item, spider):
        """处理数据库插入异常，记录日志"""
        spider.logger.error(f"数据库插入失败：{failure}，数据：{item}")

配置与运行

1. 在settings.py中添加MySQL配置（解耦，方便修改）：

# MySQL 数据库配置
MYSQL_HOST = "127.0.0.1"
MYSQL_PORT = 3306
MYSQL_USER = "root"  # 你的MySQL用户名
MYSQL_PASSWORD = "123456"  # 你的MySQL密码
MYSQL_DB = "zhihu_spider"

2. 启用MySQL Pipeline，运行爬虫；
3. 验证结果：MySQL中zhihu_column_article表已存入数据，无重复，字段格式统一。

案例4：京东商品列表 动态网页爬取（Downloader 拓展）

案例场景

爬取京东某商品分类列表的商品名称、价格、销量、商品链接，由于京东商品列表是AJAX动态渲染（静态HTML中无完整数据），需通过Scrapy-Splash拓展Downloader，实现动态网页的下载和解析，核心是掌握Downloader的进阶拓展能力。

核心准备

1. 安装Scrapy-Splash：pip install scrapy-splash
2. 启动Splash服务（需安装Docker，拉取镜像并运行）：

# 拉取Splash镜像
docker pull scrapinghub/splash

# 运行Splash服务（端口8050）
docker run -p 8050:8050 scrapinghub/splash

关键配置与实现（Downloader 拓展）

1. 在settings.py中配置Splash中间件（拓展Downloader，处理动态渲染）：

# 启用Splash下载中间件
DOWNLOADER_MIDDLEWARES = {
    "scrapy_splash.SplashCookiesMiddleware": 723,
    "scrapy_splash.SplashMiddleware": 725,
    "scrapy.downloadermiddlewares.httpcompression.HttpCompressionMiddleware": 810,
}

# Splash服务地址
SPLASH_URL = "http://127.0.0.1:8050"

# 配置Splash去重过滤器
DUPEFILTER_CLASS = "scrapy_splash.SplashAwareDupeFilter"

# 配置缓存策略
HTTPCACHE_STORAGE = "scrapy_splash.SplashAwareFSCacheStorage"

2. Spider 组件（发起Splash请求，解析动态数据）：

import scrapy
from scrapy_splash import SplashRequest
from cnblog_static_spider.items import JdProductItem

class JdProductSpider(scrapy.Spider):
    name = "jd_product"
    allowed_domains = ["jd.com"]

    # 定义Splash Lua脚本（实现动态页面渲染，等待数据加载）
    splash_lua_script = """
        function main(splash, args)
            splash:go(args.url)  # 访问目标URL
            splash:wait(3)  # 等待3秒，让页面动态渲染完成
            splash:scroll_position({y=1000})  # 滚动页面，加载更多商品
            splash:wait(2)  # 再等待2秒，确保数据加载完整
            return splash:html()  # 返回渲染后的完整HTML
        end
    """

    def start_requests(self):
        """发起Splash请求（替代普通scrapy.Request，实现动态渲染）"""
        start_url = "https://list.jd.com/list.html?cat=652,12345,12346"  # 京东商品分类URL
        yield SplashRequest(
            url=start_url,
            callback=self.parse,
            endpoint="execute",  # 执行Lua脚本
            args={
                "lua_source": self.splash_lua_script,
                "url": start_url
            }
        )

    def parse(self, response):
        """解析Splash返回的动态渲染HTML（与静态解析逻辑一致）"""
        product_list = response.xpath('//div[@class="gl-item"]')
        for product in product_list:
            item = JdProductItem()
            item["product_name"] = product.xpath('.//div[@class="p-name"]/a/em/text()').extract()
            item["price"] = product.xpath('.//div[@class="p-price"]/strong/i/text()').extract()
            item["sales"] = product.xpath('.//div[@class="p-commit"]/strong/a/text()').extract()
            item["link"] = product.xpath('.//div[@class="p-name"]/a/@href').extract()

            yield item

运行与验证

1. 确保Splash服务正常运行（http://127.0.0.1:8050可访问）；
2. 运行爬虫：scrapy crawl jd_product；
3. 验证结果：成功提取商品数据，无空值，说明Downloader拓展成功，动态页面渲染完成。