#Scrapy ImagesPipeline与FilesPipeline完全指南 - 多媒体资源下载与处理技术详解

"""
ImagesPipeline的工作流程：
1. 从Item中提取图片URL
2. 发起下载请求
3. 验证下载的图片
4. 处理图片（调整大小、格式转换等）
5. 保存到指定目录
6. 更新Item中的图片路径信息
"""

"""
ImagesPipeline主要功能：
1. 自动下载图片
2. 图片格式验证
3. 图片尺寸调整
4. 图片去重处理
5. 文件命名管理
6. 下载状态跟踪
"""

"""
FilesPipeline的工作流程：
1. 从Item中提取文件URL
2. 发起下载请求
3. 验证下载的文件
4. 保存到指定目录
5. 更新Item中的文件路径信息
"""

"""
FilesPipeline vs ImagesPipeline：

FilesPipeline：
- 通用文件下载
- 不进行图片特定处理
- 支持所有文件类型
- 处理速度较快

ImagesPipeline：
- 专门处理图片
- 提供图片处理功能
- 只支持图片格式
- 包含图片验证
"""

# settings.py - ImagesPipeline基础配置
IMAGES_URLS_FIELD = 'image_urls'      # 定义Item中存放图片URL的字段名
IMAGES_RESULT_FIELD = 'images'        # 定义Item中存放下载结果的字段名
IMAGES_STORE = 'images'               # 图片存储根目录
IMAGES_EXPIRES = 90                   # 图片缓存过期天数
IMAGES_MIN_WIDTH = 0                  # 图片最小宽度（小于该值会被过滤）
IMAGES_MIN_HEIGHT = 0                 # 图片最小高度（小于该值会被过滤）

# settings.py - ImagesPipeline高级配置
IMAGES_THUMBS = {                     # 生成缩略图
    'small': (50, 50),
    'medium': (100, 100),
    'large': (200, 200),
}

IMAGES_URLS_FIELD = 'image_urls'
IMAGES_RESULT_FIELD = 'images'
IMAGES_STORE = 'images'

# 下载设置
MEDIA_ALLOW_REDIRECTS = True          # 允许重定向
IMAGES_IGNORE_EXTENSIONS = [          # 忽略的文件扩展名
    'exe', 'zip', 'rar', 'pdf', 'doc'
]

import scrapy
from scrapy.pipelines.images import ImagesPipeline
from scrapy.exceptions import DropItem
from PIL import Image
import os

class CustomImagesPipeline(ImagesPipeline):
    """
    自定义ImagesPipeline实现
    """
    
    def get_media_requests(self, item, info):
        """
        为每个图片URL生成下载请求
        """
        for image_url in item.get('image_urls', []):
            # 添加自定义头部信息
            headers = {
                'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
                'Referer': item.get('url', '')  # 添加来源页面作为Referer
            }
            
            # 可以为不同的图片URL设置不同的meta信息
            request = scrapy.Request(
                url=image_url,
                headers=headers,
                meta={
                    'item_id': item.get('id'),
                    'image_name': self.generate_filename(image_url, item)
                }
            )
            
            yield request
    
    def file_path(self, request, response=None, info=None):
        """
        自定义文件保存路径
        """
        # 从meta中获取自定义文件名
        image_name = request.meta.get('image_name')
        if image_name:
            return image_name
        
        # 默认文件路径生成
        image_guid = request.url.split('/')[-1]
        if not image_guid.endswith(('.jpg', '.jpeg', '.png', '.gif', '.bmp')):
            image_guid += '.jpg'  # 默认添加jpg扩展名
        
        return f'full/{image_guid}'
    
    def thumb_path(self, request, thumb_id, response=None, info=None):
        """
        自定义缩略图保存路径
        """
        image_guid = request.url.split('/')[-1]
        return f'thumbs/{thumb_id}/{image_guid}'
    
    def file_path(self, request, response=None, info=None, *, item=None):
        """
        生成文件路径
        """
        # 从URL提取文件名
        image_url = request.url
        filename = image_url.split('/')[-1]
        
        # 如果URL没有文件扩展名，尝试从Content-Type推断
        if '.' not in filename:
            content_type = response.headers.get('Content-Type', b'').decode('utf-8').lower()
            if 'jpeg' in content_type or 'jpg' in content_type:
                filename += '.jpg'
            elif 'png' in content_type:
                filename += '.png'
            elif 'gif' in content_type:
                filename += '.gif'
            else:
                filename += '.jpg'  # 默认扩展名
        
        # 创建基于日期的目录结构
        import time
        date_dir = time.strftime('%Y/%m/%d')
        return f'{date_dir}/{filename}'
    
    def item_completed(self, results, item, info):
        """
        处理下载完成后的结果
        """
        image_paths = []
        image_infos = []
        
        for ok, result in results:
            if ok:
                path = result['path']
                image_paths.append(path)
                
                # 保存额外的图片信息
                image_info = {
                    'path': path,
                    'url': result['url'],
                    'checksum': result['checksum'],
                    'status': result.get('status', 'downloaded')
                }
                
                # 如果需要，可以检查图片尺寸
                full_path = os.path.join(self.store, path)
                if os.path.exists(full_path):
                    try:
                        with Image.open(full_path) as img:
                            image_info['width'] = img.width
                            image_info['height'] = img.height
                            image_info['format'] = img.format
                    except Exception as e:
                        info.spider.logger.error(f"Error getting image info: {e}")
                
                image_infos.append(image_info)
            else:
                info.spider.logger.warning(f"Failed to download image: {result}")
        
        if not image_paths:
            raise DropItem("No images downloaded for item")
        
        # 将图片路径和信息添加到Item中
        item['images'] = image_infos
        item['image_paths'] = image_paths
        
        return item

# settings.py - FilesPipeline基础配置
FILES_URLS_FIELD = 'file_urls'        # 定义Item中存放文件URL的字段名
FILES_RESULT_FIELD = 'files'          # 定义Item中存放下载结果的字段名
FILES_STORE = 'files'                 # 文件存储根目录
FILES_EXPIRES = 90                    # 文件缓存过期天数
FILES_ALLOW_REDIRECTS = True          # 允许重定向

# settings.py - FilesPipeline高级配置
FILES_URLS_FIELD = 'file_urls'
FILES_RESULT_FIELD = 'files'
FILES_STORE = 'downloads/files'

# 文件过滤设置
FILES_ALLOW_REDIRECTS = True

# 自定义文件过滤规则
FILES_IGNORE_EXTENSIONS = [
    'tmp', 'bak', 'swp', 'part',      # 临时文件
    'exe', 'bat', 'sh',               # 可执行文件
    'zip', 'rar', '7z',               # 压缩文件（如果不想下载）
]

# 自定义文件大小限制（在Pipeline中实现）
MAX_FILE_SIZE = 50 * 1024 * 1024      # 50MB

import scrapy
from scrapy.pipelines.files import FilesPipeline
from scrapy.exceptions import DropItem
import os
from urllib.parse import urlparse

class CustomFilesPipeline(FilesPipeline):
    """
    自定义FilesPipeline实现
    """
    
    def get_media_requests(self, item, info):
        """
        为每个文件URL生成下载请求
        """
        for file_url in item.get('file_urls', []):
            # 解析URL获取文件信息
            parsed_url = urlparse(file_url)
            filename = os.path.basename(parsed_url.path)
            
            headers = {
                'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
                'Referer': item.get('url', ''),
            }
            
            request = scrapy.Request(
                url=file_url,
                headers=headers,
                meta={
                    'item_id': item.get('id'),
                    'filename': filename,
                    'original_url': file_url
                }
            )
            
            yield request
    
    def file_path(self, request, response=None, info=None, *, item=None):
        """
        自定义文件保存路径
        """
        # 从meta中获取信息
        filename = request.meta.get('filename', 'unnamed_file')
        item_id = request.meta.get('item_id', 'unknown')
        
        # 如果文件名没有扩展名，尝试从URL或Content-Type推断
        if '.' not in filename:
            original_url = request.meta.get('original_url', '')
            content_type = response.headers.get('Content-Type', b'').decode('utf-8').lower()
            
            extension = self._infer_extension(original_url, content_type)
            if extension:
                filename += extension
        
        # 创建基于item_id的目录结构
        return f'{item_id}/{filename}'
    
    def _infer_extension(self, url, content_type):
        """
        从URL或Content-Type推断文件扩展名
        """
        # 从URL推断
        if url:
            parsed = urlparse(url)
            path = parsed.path
            if '.' in path:
                ext = os.path.splitext(path)[1]
                if ext.lower() in ['.pdf', '.doc', '.docx', '.xls', '.xlsx', '.ppt', '.pptx', '.txt', '.zip']:
                    return ext
        
        # 从Content-Type推断
        if content_type:
            if 'pdf' in content_type:
                return '.pdf'
            elif 'word' in content_type or 'document' in content_type:
                return '.doc'
            elif 'excel' in content_type or 'spreadsheet' in content_type:
                return '.xls'
            elif 'powerpoint' in content_type or 'presentation' in content_type:
                return '.ppt'
            elif 'plain' in content_type or 'text' in content_type:
                return '.txt'
            elif 'zip' in content_type:
                return '.zip'
            elif 'jpeg' in content_type or 'jpg' in content_type:
                return '.jpg'
            elif 'png' in content_type:
                return '.png'
            elif 'gif' in content_type:
                return '.gif'
        
        return ''
    
    def file_path(self, request, response=None, info=None, *, item=None):
        """
        生成文件路径
        """
        filename = request.meta.get('filename', 'unnamed_file')
        item_id = request.meta.get('item_id', 'unknown')
        
        # 确保文件名安全
        filename = self._sanitize_filename(filename)
        
        # 检查文件大小限制（如果在headers中有Content-Length）
        content_length = response.headers.get('Content-Length')
        if content_length:
            size = int(content_length)
            max_size = 50 * 1024 * 1024  # 50MB
            if size > max_size:
                info.spider.logger.warning(f"File too large: {size} bytes, skipping {request.url}")
                return None  # 返回None表示跳过此文件
        
        return f'{item_id}/{filename}'
    
    def _sanitize_filename(self, filename):
        """
        清理文件名，移除不安全字符
        """
        # 移除不安全字符
        unsafe_chars = '<>:"/\\|?*'
        for char in unsafe_chars:
            filename = filename.replace(char, '_')
        
        # 限制文件名长度
        if len(filename) > 200:
            name, ext = os.path.splitext(filename)
            filename = name[:190] + ext
        
        return filename
    
    def item_completed(self, results, item, info):
        """
        处理下载完成后的结果
        """
        file_paths = []
        file_infos = []
        
        for ok, result in results:
            if ok:
                path = result['path']
                if path is not None:  # 检查是否被跳过
                    file_info = {
                        'path': path,
                        'url': result['url'],
                        'checksum': result['checksum'],
                        'status': result.get('status', 'downloaded')
                    }
                    
                    # 获取文件大小
                    full_path = os.path.join(self.store, path)
                    if os.path.exists(full_path):
                        file_info['size'] = os.path.getsize(full_path)
                    
                    file_infos.append(file_info)
                    file_paths.append(path)
            else:
                info.spider.logger.warning(f"Failed to download file: {result}")
        
        # 将文件路径和信息添加到Item中
        item['files'] = file_infos
        item['file_paths'] = file_paths
        
        return item

from PIL import Image
import io

class ImageResizePipeline(ImagesPipeline):
    """
    图片尺寸调整Pipeline
    """
    
    def get_media_requests(self, item, info):
        for image_url in item.get('image_urls', []):
            # 在meta中传递尺寸要求
            yield scrapy.Request(
                image_url,
                meta={
                    'target_sizes': [(800, 600), (1024, 768)]  # 目标尺寸列表
                }
            )
    
    def convert_image(self, image, size=None):
        """
        转换图片格式和尺寸
        """
        # 转换为RGB模式（处理RGBA、P等模式）
        if image.mode in ('RGBA', 'LA', 'P'):
            # 检查是否有透明度通道
            if image.mode == 'P':
                image = image.convert('RGBA')
            
            # 创建白色背景
            background = Image.new('RGB', image.size, (255, 255, 255))
            
            # 如果原图有透明度，将其粘贴到白色背景上
            if image.mode in ('RGBA', 'LA'):
                background.paste(image, mask=image.split()[-1])  # 使用alpha通道作为掩码
            else:
                background.paste(image)
            
            image = background
        
        # 调整尺寸
        if size:
            image = image.resize(size, Image.Resampling.LANCZOS)
        
        return image
    
    def thumb_path(self, request, thumb_id, response=None, info=None):
        """
        生成缩略图路径
        """
        image_guid = request.url.split('/')[-1]
        return f'thumbnails/{thumb_id}/{image_guid}'

class ImageFormatConverterPipeline(ImagesPipeline):
    """
    图片格式转换Pipeline
    """
    
    def file_path(self, request, response=None, info=None, *, item=None):
        """
        确保统一的图片格式
        """
        image_guid = request.url.split('/')[-1]
        name, ext = os.path.splitext(image_guid)
        
        # 统一转换为webp格式以节省空间
        return f'converted/{name}.webp'
    
    def image_path(self, url):
        """
        生成图片路径
        """
        return self.file_path(scrapy.Request(url))
    
    def convert_image(self, image, size=None):
        """
        转换图片格式
        """
        # 转换为RGB模式
        if image.mode in ('RGBA', 'LA', 'P'):
            if image.mode == 'P':
                image = image.convert('RGBA')
            
            background = Image.new('RGB', image.size, (255, 255, 255))
            if image.mode in ('RGBA', 'LA'):
                background.paste(image, mask=image.split()[-1])
            else:
                background.paste(image)
            image = background
        
        # 调整尺寸
        if size:
            image = image.resize(size, Image.Resampling.LANCZOS)
        
        return image
    
    def get_images(self, response, request, info):
        """
        获取并处理图片
        """
        for key, image, buf in super().get_images(response, request, info):
            # 可以在这里添加额外的图片处理逻辑
            yield key, image, buf

class ImageOptimizationPipeline(ImagesPipeline):
    """
    图片质量优化Pipeline
    """
    
    def file_path(self, request, response=None, info=None):
        """
        生成优化后的文件路径
        """
        image_guid = request.url.split('/')[-1]
        return f'optimized/{image_guid}'
    
    def convert_image(self, image, size=None):
        """
        优化图片质量
        """
        # 转换模式
        if image.mode in ('RGBA', 'LA', 'P'):
            if image.mode == 'P':
                image = image.convert('RGBA')
            
            background = Image.new('RGB', image.size, (255, 255, 255))
            if image.mode in ('RGBA', 'LA'):
                background.paste(image, mask=image.split()[-1])
            else:
                background.paste(image)
            image = background
        
        # 调整尺寸
        if size:
            image = image.resize(size, Image.Resampling.LANCZOS)
        
        return image
    
    def write_images(self, image_paths, info):
        """
        写入优化后的图片
        """
        for path, image, buf in image_paths:
            # 保存为优化格式
            if path.endswith('.jpg') or path.endswith('.jpeg'):
                # JPEG优化
                image.save(buf, 'JPEG', quality=85, optimize=True)
            elif path.endswith('.png'):
                # PNG优化
                image.save(buf, 'PNG', optimize=True)
            else:
                # 默认优化
                image.save(buf, 'JPEG', quality=85, optimize=True)
            
            yield path, image, buf

import magic  # 需要安装 python-magic

class FileTypeValidationPipeline(FilesPipeline):
    """
    文件类型验证Pipeline
    """
    
    def __init__(self, store_uri, download_func=None, settings=None):
        super().__init__(store_uri, download_func, settings)
        self.allowed_types = {
            'image/jpeg',
            'image/png', 
            'image/gif',
            'application/pdf',
            'text/plain',
            'application/msword',
            'application/vnd.openxmlformats-officedocument.wordprocessingml.document'
        }
    
    def file_path(self, request, response=None, info=None, *, item=None):
        """
        验证文件类型后再生成路径
        """
        # 在这里可以进行文件类型验证
        return super().file_path(request, response, info, item=item)
    
    def validate_file(self, response, request):
        """
        验证下载的文件
        """
        if response.status != 200:
            return {'url': request.url, 'status': response.status}
        
        # 检查Content-Type
        content_type = response.headers.get('Content-Type', b'').decode('utf-8').lower()
        
        # 使用python-magic进行真正的文件类型检测
        try:
            file_type = magic.from_buffer(response.body, mime=True)
            if file_type not in self.allowed_types:
                return {'url': request.url, 'status': 'invalid_type', 'file_type': file_type}
        except Exception as e:
            # 如果magic不可用，回退到Content-Type检查
            pass
        
        # 检查文件大小
        max_size = 50 * 1024 * 1024  # 50MB
        if len(response.body) > max_size:
            return {'url': request.url, 'status': 'too_large'}
        
        return {'url': request.url, 'status': 'ok'}

import hashlib
import zipfile
import tarfile

class FileContentProcessorPipeline(FilesPipeline):
    """
    文件内容处理Pipeline
    """
    
    def item_completed(self, results, item, info):
        """
        处理下载完成的文件内容
        """
        processed_results = []
        
        for ok, result in results:
            if ok:
                # 计算文件哈希值用于去重
                full_path = os.path.join(self.store, result['path'])
                if os.path.exists(full_path):
                    file_hash = self.calculate_file_hash(full_path)
                    result['hash'] = file_hash
                    
                    # 如果是压缩文件，可以进行额外处理
                    if self.is_archive_file(full_path):
                        result['is_archive'] = True
                        result['extract_path'] = self.extract_archive(full_path, result['path'])
                
                processed_results.append((ok, result))
            else:
                processed_results.append((ok, result))
        
        return super().item_completed(processed_results, item, info)
    
    def calculate_file_hash(self, file_path):
        """
        计算文件哈希值
        """
        hash_sha256 = hashlib.sha256()
        with open(file_path, "rb") as f:
            for chunk in iter(lambda: f.read(4096), b""):
                hash_sha256.update(chunk)
        return hash_sha256.hexdigest()
    
    def is_archive_file(self, file_path):
        """
        检查是否为压缩文件
        """
        archive_extensions = ['.zip', '.tar', '.gz', '.rar', '.7z']
        _, ext = os.path.splitext(file_path.lower())
        return ext in archive_extensions
    
    def extract_archive(self, archive_path, base_path):
        """
        解压压缩文件
        """
        extract_dir = base_path.rsplit('/', 1)[0] + '/extracted/' + base_path.split('/')[-1].rsplit('.', 1)[0]
        extract_full_path = os.path.join(self.store, extract_dir)
        
        os.makedirs(extract_full_path, exist_ok=True)
        
        try:
            if archive_path.endswith('.zip'):
                with zipfile.ZipFile(archive_path, 'r') as zip_ref:
                    zip_ref.extractall(extract_full_path)
            elif archive_path.endswith(('.tar', '.tar.gz', '.tgz')):
                with tarfile.open(archive_path, 'r') as tar_ref:
                    tar_ref.extractall(extract_full_path)
            # 可以添加更多格式支持
            
            return extract_dir
        except Exception as e:
            # 解压失败，记录错误但不影响整体流程
            return None

from scrapy.utils.misc import arg_to_iter
from itemadapter import ItemAdapter
import asyncio
from concurrent.futures import ThreadPoolExecutor

class ParallelDownloadPipeline(FilesPipeline):
    """
    并行下载优化Pipeline
    """
    
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.executor = ThreadPoolExecutor(max_workers=4)
    
    def open_spider(self, spider):
        """
        爬虫开启时的初始化
        """
        spider.logger.info("Parallel Download Pipeline started")
    
    def close_spider(self, spider):
        """
        爬虫关闭时的清理
        """
        self.executor.shutdown(wait=True)
        spider.logger.info("Parallel Download Pipeline closed")
    
    def get_media_requests(self, item, info):
        """
        生成媒体请求，支持批量处理
        """
        urls = ItemAdapter(item).get(self.files_urls_field, [])
        for url in arg_to_iter(urls):
            yield scrapy.Request(url)
    
    def item_completed(self, results, item, info):
        """
        完成项目处理
        """
        adapter = ItemAdapter(item)
        
        # 收集成功的下载结果
        files = []
        for ok, result in results:
            if ok:
                files.append(result)
            else:
                info.spider.logger.warning(f"Failed to download: {result}")
        
        if files:
            adapter[self.files_result_field] = files
        
        return item

import time
from scrapy.utils.misc import arg_to_iter

class IntelligentRetryPipeline(FilesPipeline):
    """
    智能重试机制Pipeline
    """
    
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.retry_counts = {}  # 记录重试次数
        self.retry_delays = {}  # 记录重试延迟
        self.max_retries = 3
        self.base_delay = 1  # 基础延迟1秒
    
    def get_media_requests(self, item, info):
        """
        生成媒体请求，包含重试逻辑
        """
        urls = ItemAdapter(item).get(self.files_urls_field, [])
        
        for url in arg_to_iter(urls):
            # 获取当前URL的重试次数
            retry_count = self.retry_counts.get(url, 0)
            
            headers = {
                'User-Agent': self._get_random_user_agent(),
                'Referer': item.get('url', ''),
            }
            
            # 如果是重试请求，添加延迟
            if retry_count > 0:
                time.sleep(self.base_delay * (2 ** retry_count))  # 指数退避
            
            request = scrapy.Request(
                url=url,
                headers=headers,
                meta={
                    'retry_count': retry_count,
                    'original_url': url
                }
            )
            
            yield request
    
    def media_failed(self, failure, request, info):
        """
        处理下载失败
        """
        original_url = request.meta.get('original_url', request.url)
        retry_count = request.meta.get('retry_count', 0)
        
        if retry_count < self.max_retries:
            # 增加重试次数
            self.retry_counts[original_url] = retry_count + 1
            
            info.spider.logger.warning(
                f"Download failed for {original_url}, retry {retry_count + 1}/{self.max_retries}"
            )
            
            # 重新调度请求
            return self.get_media_requests(
                {'file_urls': [original_url]}, info
            )
        else:
            # 达到最大重试次数，放弃
            info.spider.logger.error(f"Failed to download {original_url} after {self.max_retries} attempts")
            self.retry_counts.pop(original_url, None)  # 清除重试记录
            
            # 返回失败结果
            return {'url': original_url, 'status': 'failed_after_retries'}
    
    def _get_random_user_agent(self):
        """
        获取随机User-Agent
        """
        user_agents = [
            'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36',
            'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36',
            'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36'
        ]
        import random
        return random.choice(user_agents)

import boto3
from botocore.exceptions import ClientError

class DistributedStoragePipeline(FilesPipeline):
    """
    分布式存储Pipeline（以S3为例）
    """
    
    def __init__(self, store_uri, download_func=None, settings=None):
        super().__init__(store_uri, download_func, settings)
        
        # S3配置
        self.use_s3 = settings.getbool('FILES_STORE_S3', False)
        if self.use_s3:
            self.s3_client = boto3.client(
                's3',
                aws_access_key_id=settings.get('AWS_ACCESS_KEY_ID'),
                aws_secret_access_key=settings.get('AWS_SECRET_ACCESS_KEY'),
                region_name=settings.get('AWS_REGION_NAME')
            )
            self.s3_bucket = settings.get('S3_BUCKET_NAME')
    
    def file_path(self, request, response=None, info=None, *, item=None):
        """
        生成文件路径
        """
        if self.use_s3:
            # S3使用不同的路径结构
            return f"scrapy-files/{int(time.time())}/{request.url.split('/')[-1]}"
        else:
            return super().file_path(request, response, info, item=item)
    
    def file_path(self, request, response=None, info=None, *, item=None):
        """
        生成文件路径
        """
        image_guid = request.url.split('/')[-1]
        return f'full/{image_guid}'
    
    def file_path(self, request, response=None, info=None, *, item=None):
        """
        生成文件路径（最终实现）
        """
        image_guid = request.url.split('/')[-1]
        return f'full/{image_guid}'
    
    def persist_file(self, path, buf, info, *, item=None):
        """
        持久化文件，支持本地和S3存储
        """
        if self.use_s3:
            try:
                self.s3_client.put_object(
                    Bucket=self.s3_bucket,
                    Key=path,
                    Body=buf.getvalue(),
                    ContentType='application/octet-stream'
                )
                info.spider.logger.info(f"File uploaded to S3: {path}")
            except ClientError as e:
                info.spider.logger.error(f"S3 upload failed: {e}")
                # 可以选择降级到本地存储
                super().persist_file(path, buf, info, item=item)
        else:
            super().persist_file(path, buf, info, item=item)

import gc
import weakref

class MemoryOptimizedPipeline(FilesPipeline):
    """
    内存优化Pipeline
    """
    
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.download_cache = {}  # 下载缓存
        self.max_cache_size = 100  # 最大缓存大小
        self.processed_count = 0
    
    def get_media_requests(self, item, info):
        """
        检查缓存后再生成请求
        """
        urls = ItemAdapter(item).get(self.files_urls_field, [])
        
        for url in arg_to_iter(urls):
            # 检查是否已在缓存中
            if url in self.download_cache:
                info.spider.logger.debug(f"Cache hit for {url}")
                continue
            
            yield scrapy.Request(url, meta={'item': item})
    
    def media_to_download(self, request, info, *, item=None):
        """
        优化下载过程
        """
        # 检查是否需要跳过（如已在缓存中）
        url = request.url
        if url in self.download_cache:
            return self.download_cache[url]
        
        return super().media_to_download(request, info, item=item)
    
    def item_completed(self, results, item, info):
        """
        完成项目处理，清理内存
        """
        self.processed_count += 1
        
        # 定期清理缓存和执行垃圾回收
        if self.processed_count % 50 == 0:
            # 限制缓存大小
            if len(self.download_cache) > self.max_cache_size:
                # 移除最早的缓存项
                oldest_key = next(iter(self.download_cache))
                del self.download_cache[oldest_key]
            
            # 执行垃圾回收
            collected = gc.collect()
            info.spider.logger.debug(f"Garbage collected: {collected} objects")
        
        return super().item_completed(results, item, info)

import asyncio
import threading
from concurrent.futures import ThreadPoolExecutor, as_completed

class ConcurrentControlledPipeline(FilesPipeline):
    """
    并发控制Pipeline
    """
    
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.max_concurrent_downloads = 5
        self.current_downloads = threading.Semaphore(self.max_concurrent_downloads)
        self.executor = ThreadPoolExecutor(max_workers=self.max_concurrent_downloads)
    
    def get_media_requests(self, item, info):
        """
        控制并发的媒体请求
        """
        urls = ItemAdapter(item).get(self.files_urls_field, [])
        
        for url in arg_to_iter(urls):
            # 使用信号量控制并发
            def make_request_with_semaphore():
                with self.current_downloads:
                    return scrapy.Request(url)
            
            yield make_request_with_semaphore()
    
    def media_to_download(self, request, info, *, item=None):
        """
        控制下载并发
        """
        # 可以在这里添加额外的并发控制逻辑
        return super().media_to_download(request, info, item=item)

class RobustImagesPipeline(ImagesPipeline):
    def get_media_requests(self, item, info):
        for image_url in item.get('image_urls', []):
            # 添加重试机制和错误处理
            request = scrapy.Request(
                image_url,
                headers={
                    'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
                    'Referer': item.get('source_url', ''),
                },
                meta={
                    'max_retry_times': 3,
                    'dont_redirect': True,
                    'download_timeout': 30,
                }
            )
            yield request

import hashlib
import time

class UniqueFilenamePipeline(FilesPipeline):
    def file_path(self, request, response=None, info=None, *, item=None):
        # 使用URL的哈希值确保唯一性
        url_hash = hashlib.md5(request.url.encode()).hexdigest()[:8]
        original_filename = request.url.split('/')[-1]
        name, ext = os.path.splitext(original_filename)
        
        if not ext:
            ext = '.dat'  # 默认扩展名
        
        timestamp = str(int(time.time()))
        return f'{timestamp}_{url_hash}_{name}{ext}'

import shutil

class SpaceAwarePipeline(FilesPipeline):
    def open_spider(self, spider):
        # 检查可用磁盘空间
        total, used, free = shutil.disk_usage(self.store)
        if free < 1024 * 1024 * 100:  # 少于100MB
            spider.logger.warning("Low disk space warning!")
    
    def item_completed(self, results, item, info):
        # 检查存储使用情况
        total, used, free = shutil.disk_usage(self.store)
        usage_percent = (used / total) * 100
        
        if usage_percent > 80:  # 使用率超过80%
            info.spider.logger.warning(f"Storage usage is high: {usage_percent:.1f}%")
        
        return super().item_completed(results, item, info)