2.6.4 Spider Context 上下文

Spider Context 是 Bricks 爬虫框架中的核心概念，它贯穿整个爬虫生命周期，承载着爬虫执行过程中的所有状态和数据。

核心概念

1. Context 的作用

Context 对象在爬虫执行过程中扮演着数据载体的角色：

状态管理：保存爬虫当前的执行状态
数据传递：在不同阶段间传递数据
资源访问：提供对爬虫实例、请求、响应等资源的访问
流程控制：控制爬虫的执行流程

2. Context 的生命周期

创建 Context → 设置种子 → 构造请求 → 发送请求 → 接收响应 → 解析数据 → 处理结果 → 完成/失败

Context 属性详解

1. 基础属性

class Context:
    # 爬虫实例
    target: Spider
    
    # 种子数据
    seeds: dict
    
    # 请求对象
    request: Request
    
    # 响应对象
    response: Response
    
    # 解析结果
    items: list
    
    # 异常信息
    exception: Exception
    
    # 自定义数据
    custom_data: dict

2. 状态属性

# 执行状态
context.is_success  # 是否成功
context.is_failure  # 是否失败
context.is_retry    # 是否重试
 
# 重试信息
context.retry_times  # 重试次数
context.max_retry   # 最大重试次数

Context 使用示例

1. 在 Air Spider 中使用

from bricks.spider import air
from bricks import Request
 
class MySpider(air.Spider):
    def make_seeds(self, context, **kwargs):
        """初始化阶段的 context"""
        print(f"Target: {context.target}")  # 爬虫实例
        print(f"Seeds: {context.seeds}")    # 空字典
        
        return [{"page": i} for i in range(1, 4)]
    
    def make_request(self, context) -> Request:
        """请求构造阶段的 context"""
        print(f"Seeds: {context.seeds}")    # 当前种子数据
        print(f"Target: {context.target}")  # 爬虫实例
        
        page = context.seeds["page"]
        return Request(url=f"https://httpbin.org/json?page={page}")
    
    def parse(self, context):
        """解析阶段的 context"""
        print(f"Request: {context.request.url}")      # 请求对象
        print(f"Response: {context.response.status_code}")  # 响应对象
        print(f"Seeds: {context.seeds}")              # 种子数据
        
        return context.response.json()
    
    def item_pipeline(self, context):
        """处理阶段的 context"""
        print(f"Items: {context.items}")     # 解析结果
        print(f"Seeds: {context.seeds}")     # 种子数据
        
        # 标记成功
        context.success()

2. 在 Form/Template Spider 中使用

from bricks.spider.form import Spider, Config, Download, Parse, Pipeline
 
def custom_parser(context):
    """自定义解析函数"""
    response = context.response
    seeds = context.seeds
    
    print(f"解析 URL: {response.url}")
    print(f"种子数据: {seeds}")
    
    return response.json()
 
def custom_pipeline(context):
    """自定义处理函数"""
    items = context.items
    seeds = context.seeds
    
    print(f"处理 {len(items)} 条数据")
    print(f"来源页面: {seeds.get('page', 'unknown')}")
    
    # 保存自定义数据
    context.custom_data = {"processed_count": len(items)}
    
    return items
 
@dataclass
class MyConfig(Config):
    spider = [
        Download(url="https://httpbin.org/json?page={page}"),
        Parse(func=custom_parser),
        Pipeline(func=custom_pipeline, success=True)
    ]

Context 方法详解

1. 状态控制方法

# 标记成功
context.success()
 
# 标记失败
context.failure(exception=None)
 
# 标记重试
context.retry(exception=None)
 
# 检查状态
if context.is_success:
    print("处理成功")
elif context.is_failure:
    print("处理失败")
elif context.is_retry:
    print("需要重试")

2. 数据提交方法

# 提交新种子（仅在 Form Spider 中可用）
context.submit(
    {"url": "https://example.com/page1"},
    {"url": "https://example.com/page2"}
)
 
# 批量提交
new_seeds = [{"url": f"https://example.com/page{i}"} for i in range(1, 6)]
context.submit(*new_seeds)

3. 自定义数据操作

# 设置自定义数据
context.custom_data = {"start_time": time.time()}
 
# 更新自定义数据
context.custom_data.update({"end_time": time.time()})
 
# 获取自定义数据
start_time = context.custom_data.get("start_time")

高级用法

1. Context 传递数据

class DataPassingSpider(air.Spider):
    def make_request(self, context) -> Request:
        # 在种子中添加额外信息
        context.seeds["request_time"] = time.time()
        
        return Request(url=context.seeds["url"])
    
    def parse(self, context):
        # 获取请求时间
        request_time = context.seeds.get("request_time")
        response_time = time.time()
        
        # 计算响应时间
        duration = response_time - request_time
        
        return {
            "url": context.response.url,
            "duration": duration,
            "data": context.response.json()
        }

2. Context 错误处理

class ErrorHandlingSpider(air.Spider):
    def parse(self, context):
        try:
            return context.response.json()
        except Exception as e:
            # 记录错误信息
            context.custom_data["parse_error"] = str(e)
            # 标记失败
            context.failure(e)
            return []
    
    def item_pipeline(self, context):
        # 检查是否有解析错误
        if "parse_error" in context.custom_data:
            print(f"解析错误: {context.custom_data['parse_error']}")
            return
        
        # 正常处理
        print(f"处理数据: {context.items}")
        context.success()

3. Context 条件控制

def conditional_download(context):
    """条件下载函数"""
    seeds = context.seeds
    
    # 根据种子数据决定是否下载
    if seeds.get("skip_download"):
        return None
    
    # 根据时间决定是否下载
    if time.time() - seeds.get("last_update", 0) < 3600:
        return None
    
    return Request(url=seeds["url"])
 
def conditional_parse(context):
    """条件解析函数"""
    response = context.response
    
    # 检查响应状态
    if response.status_code != 200:
        context.failure(Exception(f"HTTP {response.status_code}"))
        return []
    
    # 检查内容类型
    content_type = response.headers.get("content-type", "")
    if "application/json" not in content_type:
        context.failure(Exception("不是JSON响应"))
        return []
    
    return response.json()

4. Context 统计信息

class StatisticsSpider(air.Spider):
    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.stats = {
            "total_requests": 0,
            "successful_requests": 0,
            "failed_requests": 0,
            "total_items": 0
        }
    
    def make_request(self, context) -> Request:
        self.stats["total_requests"] += 1
        return Request(url=context.seeds["url"])
    
    def parse(self, context):
        if context.response.status_code == 200:
            self.stats["successful_requests"] += 1
            data = context.response.json()
            return data
        else:
            self.stats["failed_requests"] += 1
            context.failure(Exception(f"HTTP {context.response.status_code}"))
            return []
    
    def item_pipeline(self, context):
        items = context.items
        self.stats["total_items"] += len(items)
        
        # 将统计信息保存到 context
        context.custom_data["spider_stats"] = self.stats.copy()
        
        print(f"统计信息: {self.stats}")
        context.success()

Context 最佳实践

1. 数据验证

def validate_context(context):
    """验证 context 数据"""
    # 验证种子数据
    if not context.seeds.get("url"):
        raise ValueError("种子数据缺少 URL")
    
    # 验证响应数据
    if hasattr(context, "response") and context.response.status_code >= 400:
        raise ValueError(f"HTTP 错误: {context.response.status_code}")
    
    # 验证解析结果
    if hasattr(context, "items") and not isinstance(context.items, list):
        raise ValueError("解析结果必须是列表")

2. 资源清理

def cleanup_context(context):
    """清理 context 资源"""
    # 清理大型数据
    if hasattr(context, "response") and len(context.response.content) > 1024*1024:
        context.response._content = b""  # 清理响应内容
    
    # 清理临时文件
    temp_files = context.custom_data.get("temp_files", [])
    for file_path in temp_files:
        if os.path.exists(file_path):
            os.remove(file_path)

3. 调试支持

def debug_context(context):
    """调试 context 信息"""
    print(f"=== Context Debug Info ===")
    print(f"Seeds: {context.seeds}")
    
    if hasattr(context, "request"):
        print(f"Request URL: {context.request.url}")
    
    if hasattr(context, "response"):
        print(f"Response Status: {context.response.status_code}")
        print(f"Response Size: {len(context.response.content)} bytes")
    
    if hasattr(context, "items"):
        print(f"Items Count: {len(context.items)}")
    
    if context.custom_data:
        print(f"Custom Data: {context.custom_data}")
    
    print(f"========================")

Context 是 Bricks 爬虫框架的核心，理解和正确使用 Context 对于开发高效、稳定的爬虫至关重要。

2.6.3 Template Spider 模板爬虫 2.6.5 Spider 生命周期