#Scrapy五大核心组件详解 - Engine、Scheduler、Downloader、Spiders、Pipeline深度解析

"""
Engine详细工作流程：

1. 初始化爬虫实例
2. 从Spider获取初始请求
3. 将请求发送至Scheduler排队
4. 从Scheduler获取下一个请求
5. 将请求发送给Downloader处理
6. 接收Downloader的响应
7. 将响应发送给对应的Spider解析
8. 处理Spider返回的Items和新Requests
9. 将Items发送给Pipeline处理
10. 将新Requests发送回Scheduler排队
11. 重复步骤4-10直至无更多请求
12. 执行清理和关闭操作
"""

# settings.py 中的Engine相关配置
"""
# 引擎相关设置
ENGINE_STOP_ON_IDLE = True          # 空闲时停止引擎
CLOSESPIDER_TIMEOUT = 3600          # 超时关闭爬虫
CLOSESPIDER_PAGECOUNT = 1000        # 页面数量限制
CLOSESPIDER_ITEMCOUNT = 10000       # Item数量限制

# 信号处理
SPIDER_MIDDLEWARES = {
    'myproject.middlewares.CustomSpiderMiddleware': 543,
}

DOWNLOADER_MIDDLEWARES = {
    'myproject.middlewares.CustomDownloaderMiddleware': 543,
}
"""

# Engine信号示例
from scrapy import signals
from pydispatch import dispatcher

class EngineSignalsExample:
    """Engine信号处理示例"""
    
    def __init__(self):
        # 注册信号处理器
        dispatcher.connect(self.spider_opened, signal=signals.spider_opened)
        dispatcher.connect(self.spider_closed, signal=signals.spider_closed)
        dispatcher.connect(self.request_scheduled, signal=signals.request_scheduled)
        dispatcher.connect(self.response_received, signal=signals.response_received)
    
    def spider_opened(self, spider):
        """爬虫开启信号"""
        print(f"Spider {spider.name} opened")
    
    def spider_closed(self, spider, reason):
        """爬虫关闭信号"""
        print(f"Spider {spider.name} closed, reason: {reason}")
    
    def request_scheduled(self, request, spider):
        """请求调度信号"""
        print(f"Request scheduled: {request.url}")
    
    def response_received(self, response, request, spider):
        """响应接收信号"""
        print(f"Response received: {response.status} for {request.url}")

"""
默认Scheduler使用内存队列：
- 队列类型：FIFO/优先级队列
- 去重机制：基于Request指纹
- 内存管理：定期清理过期请求

分布式Scheduler（Scrapy-Redis）：
- 队列类型：Redis队列
- 去重机制：Redis集合
- 持久化：Redis持久化机制
"""

# settings.py 中的Scheduler配置
"""
# 调度器配置
SCHEDULER = 'scrapy.core.scheduler.Scheduler'  # 默认调度器
SCHEDULER_DISK_QUEUE = 'scrapy.squeues.PickleFifoDiskQueue'  # 磁盘队列
SCHEDULER_MEMORY_QUEUE = 'scrapy.squeues.JsonMemoryQueue'    # 内存队列

# 去重配置
DUPEFILTER_CLASS = 'scrapy.dupefilters.RFPDupeFilter'  # 默认去重过滤器
DUPEFILTER_DEBUG = False  # 去重调试模式

# 队列配置
SCHEDULER_PRIORITY_QUEUE = 'scrapy.pqueues.ScrapyPriorityQueue'  # 优先级队列
JOBDIR = 'crawls/spider_name/'  # 持久化存储目录
"""

# 自定义调度器示例
from scrapy.core.scheduler import Scheduler
from scrapy.http import Request
import redis

class CustomScheduler(Scheduler):
    """自定义调度器"""
    
    def __init__(self, dupefilter, jobdir=None, dqclass=None, mqclass=None, logunser=False, stats=None, pqclass=None):
        super().__init__(dupefilter, jobdir, dqclass, mqclass, logunser, stats, pqclass)
        # 初始化自定义逻辑
        self.redis_client = redis.Redis(host='localhost', port=6379, db=0)
    
    def enqueue_request(self, request):
        """自定义请求入队逻辑"""
        if not request.dont_filter and self.df.request_seen(request):
            # 去重逻辑
            self.df.log(request, self.stats)
            return False
        
        # 自定义入队逻辑
        self.queue.push(request)
        self.stats.inc_value('scheduler/enqueued')
        return True
    
    def next_request(self):
        """自定义下一个请求获取逻辑"""
        request = self.queue.pop()
        if request:
            self.stats.inc_value('scheduler/dequeued')
        return request

"""
Downloader包含以下子组件：

1. Download Handlers（下载处理器）
   - HTTP/HTTPS处理器
   - FTP处理器
   - S3处理器等

2. Download Middleware（下载中间件）
   - 请求/响应处理
   - 反爬策略实施
   - 用户代理管理

3. Connection Pool（连接池）
   - TCP连接复用
   - 并发控制
   - 超时管理
"""

# settings.py 中的Downloader配置
"""
# 下载器配置
CONCURRENT_REQUESTS = 16  # 并发请求数
CONCURRENT_REQUESTS_PER_DOMAIN = 8  # 单域名并发数
CONCURRENT_REQUESTS_PER_IP = 0  # 单IP并发数（0表示不限制）

# 下载延迟
DOWNLOAD_DELAY = 1  # 下载延迟（秒）
RANDOMIZE_DOWNLOAD_DELAY = 0.5  # 随机延迟比例（0.5表示±50%）

# 超时设置
DOWNLOAD_TIMEOUT = 180  # 下载超时（秒）
DOWNLOAD_WARNSIZE = 33554432  # 警告大小（32MB）
DOWNLOAD_MAXSIZE = 1048576000  # 最大大小（1GB）

# 重试设置
RETRY_TIMES = 3  # 重试次数
RETRY_HTTP_CODES = [500, 502, 503, 504, 408, 429]  # 重试HTTP状态码

# 自动限速
AUTOTHROTTLE_ENABLED = True  # 启用自动限速
AUTOTHROTTLE_START_DELAY = 1  # 起始延迟
AUTOTHROTTLE_MAX_DELAY = 10  # 最大延迟
AUTOTHROTTLE_TARGET_CONCURRENCY = 1.0  # 目标并发数
AUTOTHROTTLE_DEBUG = False  # 调试模式
"""

# 自定义下载中间件
import random
from scrapy.downloadermiddlewares.retry import RetryMiddleware
from scrapy.utils.response import response_status_message

class CustomDownloaderMiddleware:
    """自定义下载中间件示例"""
    
    def __init__(self):
        # 用户代理池
        self.user_agents = [
            'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
            'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36',
            'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36'
        ]
    
    def process_request(self, request, spider):
        """处理请求"""
        # 随机更换User-Agent
        request.headers['User-Agent'] = random.choice(self.user_agents)
        
        # 添加其他请求头
        request.headers.setdefault('Accept', 'text/html,application/xhtml+xml,*/*')
        request.headers.setdefault('Accept-Language', 'en-US,en;q=0.9')
        
        return None  # 返回None表示继续处理
    
    def process_response(self, request, response, spider):
        """处理响应"""
        # 检查响应状态
        if response.status in [403, 429]:
            # 记录被封IP信息
            spider.logger.warning(f"可能被封IP: {request.url}, Status: {response.status}")
        
        return response
    
    def process_exception(self, request, exception, spider):
        """处理异常"""
        spider.logger.error(f"请求异常: {request.url}, Error: {str(exception)}")
        return None

import scrapy

class BasicSpider(scrapy.Spider):
    """基础爬虫示例"""
    name = 'basic_spider'
    allowed_domains = ['example.com']
    start_urls = ['http://example.com']
    
    custom_settings = {
        'DOWNLOAD_DELAY': 1,
        'RANDOMIZE_DOWNLOAD_DELAY': 0.5
    }
    
    def start_requests(self):
        """自定义起始请求"""
        for url in self.start_urls:
            yield scrapy.Request(
                url=url,
                callback=self.parse,
                headers={'User-Agent': 'Custom Bot 1.0'},
                meta={'page': 1}
            )
    
    def parse(self, response):
        """解析响应的主要方法"""
        # 提取数据
        for item in response.css('div.item'):
            yield {
                'title': item.css('h2::text').get(),
                'price': item.css('span.price::text').get(),
                'url': response.url,
                'page': response.meta.get('page')
            }
        
        # 提取下一页链接
        next_page = response.css('a.next-page::attr(href)').get()
        if next_page:
            yield response.follow(
                next_page,
                callback=self.parse,
                meta={'page': response.meta.get('page', 1) + 1}
            )

import scrapy
from scrapy.spiders import CrawlSpider, Rule
from scrapy.linkextractors import LinkExtractor

class CrawlSpiderExample(CrawlSpider):
    """爬行爬虫示例"""
    name = 'crawl_spider'
    allowed_domains = ['example.com']
    start_urls = ['http://example.com']
    
    # 定义爬取规则
    rules = (
        Rule(
            LinkExtractor(
                allow=r'/category/\w+',  # 允许的URL模式
                restrict_css='.category-links'  # 限制在特定CSS选择器内
            ),
            callback='parse_category',
            follow=True  # 是否跟进链接
        ),
        Rule(
            LinkExtractor(
                allow=r'/product/\d+',  # 产品页面
                deny=r'/admin/'  # 排除管理页面
            ),
            callback='parse_product',
            follow=False  # 不跟进此规则的链接
        ),
    )
    
    def parse_category(self, response):
        """解析分类页面"""
        for product_link in response.css('.product-link::attr(href)'):
            yield response.follow(product_link, callback=self.parse_product)
    
    def parse_product(self, response):
        """解析产品页面"""
        yield {
            'name': response.css('h1.product-title::text').get(),
            'price': response.css('.price::text').get(),
            'description': response.css('.description::text').get(),
            'url': response.url
        }

# 自定义Spider中间件
class CustomSpiderMiddleware:
    """自定义Spider中间件"""
    
    def process_spider_input(self, response, spider):
        """处理Spider输入（响应）"""
        # 可以在这里修改响应对象
        response.meta['processed_by_spider_middleware'] = True
        return None
    
    def process_spider_output(self, response, result, spider):
        """处理Spider输出（Items和Requests）"""
        for item_or_request in result:
            if hasattr(item_or_request, 'get'):  # Item
                item_or_request['processed_time'] = time.time()
            yield item_or_request
    
    def process_spider_exception(self, response, exception, spider):
        """处理Spider异常"""
        spider.logger.error(f"Spider exception: {exception}")
        return []
    
    def process_start_requests(self, start_requests, spider):
        """处理起始请求"""
        for req in start_requests:
            req.meta['start_request'] = True
            yield req

import json
import pymongo
from itemadapter import ItemAdapter
from scrapy.exceptions import DropItem

class ValidationPipeline:
    """数据验证管道"""
    
    def process_item(self, item, spider):
        adapter = ItemAdapter(item)
        
        # 验证必填字段
        if not adapter.get('title'):
            raise DropItem("缺少标题字段")
        
        if not adapter.get('price'):
            raise DropItem("缺少价格字段")
        
        # 验证数据类型
        try:
            price = float(adapter['price'])
            if price <= 0:
                raise DropItem("价格必须大于0")
        except (ValueError, TypeError):
            raise DropItem("价格格式不正确")
        
        return item

class CleaningPipeline:
    """数据清洗管道"""
    
    def process_item(self, item, spider):
        adapter = ItemAdapter(item)
        
        # 清理字符串字段
        for field_name, value in adapter.items():
            if isinstance(value, str):
                adapter[field_name] = value.strip()
        
        # 格式化价格
        if adapter.get('price'):
            # 移除货币符号并转换为浮点数
            price_str = str(adapter['price'])
            import re
            numbers = re.findall(r'\d+\.?\d*', price_str)
            if numbers:
                adapter['price'] = float(numbers[0])
        
        return item

class JsonWriterPipeline:
    """JSON写入管道"""
    
    def open_spider(self, spider):
        self.file = open('items.json', 'w', encoding='utf-8')
    
    def close_spider(self, spider):
        self.file.close()
    
    def process_item(self, item, spider):
        line = json.dumps(ItemAdapter(item).asdict(), ensure_ascii=False) + "\n"
        self.file.write(line)
        return item

class MongoPipeline:
    """MongoDB存储管道"""
    
    collection_name = 'scrapy_items'
    
    def __init__(self, mongo_uri, mongo_db):
        self.mongo_uri = mongo_uri
        self.mongo_db = mongo_db
    
    @classmethod
    def from_crawler(cls, crawler):
        return cls(
            mongo_uri=crawler.settings.get('MONGO_URI'),
            mongo_db=crawler.settings.get('MONGO_DATABASE', 'items')
        )
    
    def open_spider(self, spider):
        self.client = pymongo.MongoClient(self.mongo_uri)
        self.db = self.client[self.mongo_db]
    
    def close_spider(self, spider):
        self.client.close()
    
    def process_item(self, item, spider):
        self.db[self.collection_name].insert_one(ItemAdapter(item).asdict())
        return item

# settings.py 中的Pipeline配置
"""
ITEM_PIPELINES = {
    'myproject.pipelines.ValidationPipeline': 300,
    'myproject.pipelines.CleaningPipeline': 400,
    'myproject.pipelines.MongoPipeline': 500,
}

# 管道执行顺序：数字越小优先级越高
# 300 -> 400 -> 500 依次执行
"""

# 监控组件性能的指标
"""
Engine指标：
- engine/started_crawls: 爬虫启动次数
- engine/stopped_crawls: 爬虫停止次数
- engine/max_active_crawls: 最大活跃爬虫数

Scheduler指标：
- scheduler/enqueued: 入队请求数
- scheduler/dequeued: 出队请求数
- scheduler/dropped: 丢弃请求数
- scheduler/enqueued/memory: 内存队列请求数
- scheduler/enqueued/disk: 磁盘队列请求数

Downloader指标：
- downloader/request_count: 请求总数
- downloader/response_count: 响应总数
- downloader/response_status_count/200: 各状态码响应数
- downloader/mean_response_size: 平均响应大小
- downloader/exception_count: 异常数

Spider指标：
- spider/output: Spider输出项数
- spider/exceptions: Spider异常数

Item Pipeline指标：
- item_scraped_count: 保存的Item数
- dropped_item_count: 丢弃的Item数
"""

# 内存优化配置
"""
# 限制Item并发处理数
CONCURRENT_ITEMS = 100

# 启用垃圾回收
FEED_EXPORTERS = {
    'json': 'scrapy.exporters.JsonItemExporter',
}

# 限制爬虫运行时间
CLOSESPIDER_TIMEOUT = 3600

# 限制处理的页面数
CLOSESPIDER_PAGECOUNT = 10000
"""

# 性能测试工具
import time
import threading
from scrapy.crawler import CrawlerRunner
from scrapy.utils.project import get_project_settings

def performance_test(spider_class, test_duration=60):
    """性能测试函数"""
    settings = get_project_settings()
    runner = CrawlerRunner(settings)
    
    start_time = time.time()
    d = runner.crawl(spider_class)
    
    def check_duration():
        if time.time() - start_time > test_duration:
            runner.crawling_engine.stop()
        else:
            threading.Timer(1.0, check_duration).start()
    
    check_duration()
    d.addBoth(lambda _: time.time() - start_time)
    return d.result