《Python实战进阶》No20: 网络爬虫开发：Scrapy框架详解

news2025/3/11 15:23:17

No20: 网络爬虫开发：Scrapy框架详解

摘要

本文深入解析Scrapy核心架构，通过中间件链式处理、布隆过滤器增量爬取、Splash动态渲染、分布式指纹策略四大核心技术，结合政府数据爬取与动态API逆向工程实战案例，构建企业级爬虫系统。提供完整代码与运行结果，包含法律合规设计与反爬对抗方案。

Scrapy是适用于Python的一个快速、高层次的屏幕抓取和web抓取框架，用于抓取web站点并从页面中提取结构化的数据。Scrapy用途广泛，可以用于数据挖掘、监测和自动化测试。
Scrapy吸引人的地方在于它是一个框架，任何人都可以根据需求方便的修改。它也提供了多种类型爬虫的基类，如BaseSpider、sitemap爬虫等，最新版本又提供了web2.0爬虫的支持。

在这里插入图片描述

运行环境与依赖说明

# 环境要求
Python 3.8+
Redis 6.0+ (分布式场景)

# 依赖安装
pip install scrapy==2.8.0 scrapy-splash==0.9.0 scrapy-redis==0.7.2
pip install redis==4.5.5 requests==2.31.0 beautifulsoup4==4.12.2

核心概念与实战代码

1. 中间件链式处理机制

核心逻辑：通过process_request和process_response实现请求/响应预处理

# 中间件实现示例：随机User-Agent
class RandomUserAgentMiddleware:
    def __init__(self):
        self.user_agents = [
            "Mozilla/5.0 (Windows NT 10.0...)",
            "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15...)"
        ]
    
    def process_request(self, request, spider):
        request.headers['User-Agent'] = random.choice(self.user_agents)
        spider.logger.info(f"Using User-Agent: {request.headers['User-Agent']}")

# settings.py配置
DOWNLOADER_MIDDLEWARES = {
    'myproject.middlewares.RandomUserAgentMiddleware': 543,
}

输出日志：

[scrapy.core.engine] INFO: Using User-Agent: Mozilla/5.0 (Windows NT 10.0...

2. 增量爬取与布隆过滤器

实现方案：基于Scrapy-Redis的BloomFilter去重

# 布隆过滤器配置（settings.py）
SCHEDULER = "scrapy_redis.scheduler.Scheduler"
DUPEFILTER_CLASS = "scrapy_redis.dupefilter.BloomFilter"
REDIS_URL = 'redis://localhost:6379'

# 启动爬虫（保留历史指纹）
scrapy crawl gov_spider -s SCHEDULER_PERSIST=True

运行效果：

[scrapy_redis.scheduler] INFO: Resuming crawl (5678 requests scheduled)

3. Splash渲染与自动代理池

动态页面处理：集成Splash渲染JavaScript

# 使用SplashRequest爬取动态页面
import scrapy
from scrapy_splash import SplashRequest

class DynamicSpider(scrapy.Spider):
    name = 'dynamic_spider'
    
    def start_requests(self):
        yield SplashRequest(
            url="https://example.com/ajax-page",
            callback=self.parse,
            args={'wait': 2}
        )

    def parse(self, response):
        yield {
            'content': response.xpath('//div[@class="dynamic-content"]/text()').get()
        }

输出结果：

{'content': '这是动态加载的内容'}

4. 分布式爬虫指纹策略

分布式架构：通过Redis共享指纹和队列

# 分布式配置（settings.py）
SCHEDULER = "scrapy_redis.scheduler.Scheduler"
SCHEDULER_QUEUE_CLASS = 'scrapy_redis.queue.FifoQueue'
REDIS_HOST = '192.168.1.100'

运行命令：

# 在多台机器上启动
scrapy runspider myspider.py -a slave_id=1
scrapy runspider myspider.py -a slave_id=2

在这里插入图片描述

实战案例

案例1：政府公开数据结构化爬取

目标：爬取国家统计局季度GDP数据

# items.py定义结构
class GovDataItem(scrapy.Item):
    quarter = scrapy.Field()
    gdp = scrapy.Field()
    growth_rate = scrapy.Field()

# spiders/gov_spider.py
class GovSpider(scrapy.Spider):
    name = 'gov_spider'
    start_urls = ['http://www.stats.gov.cn/tjsj/zxfb/']

    def parse(self, response):
        for row in response.css('table.data-table tr'):
            yield GovDataItem({
                'quarter': row.xpath('td[1]/text()').get(),
                'gdp': row.xpath('td[2]/text()').get(),
                'growth_rate': row.xpath('td[3]/text()').get()
            })

输出示例：

[
  {"quarter": "2023-Q1", "gdp": "284997", "growth_rate": "4.5%"},
  {"quarter": "2023-Q2", "gdp": "308038", "growth_rate": "6.3%"}
]

案例2：动态API逆向工程

目标：破解某电商商品列表加密参数

# 逆向分析加密参数
import requests
from bs4 import BeautifulSoup

def get_token():
    response = requests.get('https://api.example.com/init')
    soup = BeautifulSoup(response.text, 'html.parser')
    return soup.find('script')['data-token']

# 在Scrapy中使用
class ApiSpider(scrapy.Spider):
    def parse(self, response):
        token = get_token()
        yield scrapy.FormRequest(
            url="https://api.example.com/data",
            formdata={'token': token, 'page': '1'},
            callback=self.parse_data
        )

响应示例：

{
  "data": [
    {"id": 1001, "name": "智能手机", "price": 2999},
    {"id": 1002, "name": "笔记本电脑", "price": 8999}
  ]
}

扩展思考

1. 法律合规设计

# robots.txt遵守中间件
class RobotsTxtMiddleware:
    def process_request(self, request, spider):
        if not spider.allowed_domains:
            return
        # 校验robots协议
        rp = RobotFileParser()
        rp.set_url(f"http://{spider.allowed_domains[0]}/robots.txt")
        rp.read()
        if not rp.can_fetch("*", request.url):
            spider.logger.warning(f"Blocked by robots.txt: {request.url}")
            return scrapy.Request(url=request.url, dont_filter=True, callback=lambda _: None)

2. 反爬虫对抗测试框架

测试方案：模拟IP封禁、验证码场景

# 使用Selenium测试反爬
from selenium import webdriver
from selenium.webdriver.common.by import By

driver = webdriver.Chrome()
driver.get("https://example.com/protected-page")

# 检测是否出现验证码
if "验证码" in driver.page_source:
    print("触发验证码防护")
    # 调用第三方验证码识别API
    captcha = solve_captcha(driver.find_element(By.ID, "captcha-img"))
    driver.find_element(By.ID, "captcha-input").send_keys(captcha)