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

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 SplashRequestclass 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 BeautifulSoupdef 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 Bydriver = webdriver.Chrome()
driver.get("https://example.com/protected-page")# 检测是否出现验证码
if "验证码" in driver.page_source:print("触发验证码防护")# 调用第三方验证码识别APIcaptcha = solve_captcha(driver.find_element(By.ID, "captcha-img"))driver.find_element(By.ID, "captcha-input").send_keys(captcha)