bricks
开发指南
2.4 解析篇
2.4.5 配置式爬虫中的解析

2.4.5 配置式爬虫中的解析

在 Bricks 框架的配置式爬虫中，解析器的使用更加简洁和声明式。通过配置 Parse 节点，可以轻松实现各种数据解析需求，无需编写复杂的解析代码。

Parse 节点概述

Parse 节点是配置式爬虫中专门用于数据解析的配置节点，它封装了所有内置解析器的功能，并提供了统一的配置接口。

Parse 节点参数

参数名参数类型参数描述默认值
funcUnion[str, Callable]解析引擎或自定义函数必传
argsOptional[list]传递给解析引擎的位置参数None
kwargsOptional[dict]传递给解析引擎的关键字参数None
layoutOptional[Layout]数据布局配置None
archivebool是否在此节点存档False

内置解析引擎

配置式爬虫支持以下内置解析引擎，只需在 func 参数中指定引擎名称即可：

支持的引擎

引擎名称描述对应解析器
"json"JSON/JMESPath 解析JsonExtractor
"xpath"XPath 解析XpathExtractor
"jsonpath"JSONPath 解析JsonpathExtractor
"regex"正则表达式解析RegexExtractor

基本使用示例

1. JSON 数据解析

from dataclasses import dataclass
from bricks.spider.template import Spider, Config, Download, Parse
 
@dataclass
class MyConfig(Config):
    download = [
        Download(url="https://api.example.com/users")
    ]
    parse = [
        Parse(
            func="json",
            kwargs={
                "rules": {
                    "users[*]": {
                        "user_id": "id",
                        "username": "name",
                        "email": "email",
                        "age": "age"
                    }
                }
            }
        )
    ]
 
class MySpider(Spider):
    @property
    def config(self):
        return MyConfig()

2. HTML 页面解析

@dataclass
class NewsConfig(Config):
    download = [
        Download(url="https://news.example.com/articles")
    ]
    parse = [
        Parse(
            func="xpath",
            kwargs={
                "rules": {
                    "//article": {
                        "title": ".//h2/text()",
                        "content": ".//p/text()",
                        "author": ".//span[@class='author']/text()",
                        "publish_time": ".//time/@datetime",
                        "url": ".//a/@href"
                    }
                }
            }
        )
    ]
 
class NewsSpider(Spider):
    @property
    def config(self):
        return NewsConfig()

3. 正则表达式解析

@dataclass
class TextConfig(Config):
    download = [
        Download(url="https://example.com/contact")
    ]
    parse = [
        Parse(
            func="regex",
            kwargs={
                "rules": {
                    "phones": r"1[3-9]\d{9}",
                    "emails": r"\w+@\w+\.\w+",
                    "urls": r"https?://[^\s]+"
                }
            }
        )
    ]
 
class TextSpider(Spider):
    @property
    def config(self):
        return TextConfig()

高级配置

使用 Layout 进行数据处理

Layout 对象提供了强大的数据处理功能：

from bricks.spider.form import Layout
 
@dataclass
class AdvancedConfig(Config):
    download = [
        Download(url="https://api.example.com/products")
    ]
    parse = [
        Parse(
            func="json",
            kwargs={
                "rules": {
                    "products[*]": {
                        "id": "product_id",
                        "name": "title",
                        "price": "price.current",
                        "original_price": "price.original",
                        "category": "category.name",
                        "in_stock": "availability"
                    }
                }
            },
            layout=Layout(
                # 字段重命名
                rename={
                    "id": "product_id",
                    "name": "product_name"
                },
                # 设置默认值
                default={
                    "in_stock": True,
                    "category": "未分类"
                },
                # 数据类型转换
                factory={
                    "price": float,
                    "original_price": float,
                    "product_id": int
                },
                # 字段过滤
                show=["product_id", "product_name", "price", "category"]
            )
        )
    ]

多步骤解析

配置式爬虫支持多个解析步骤的组合：

@dataclass
class MultiStepConfig(Config):
    download = [
        Download(url="https://example.com/page1")
    ]
    parse = [
        # 第一步：提取基本信息
        Parse(
            func="xpath",
            kwargs={
                "rules": {
                    "basic_info": "//div[@class='info']/text()",
                    "detail_urls": "//a[@class='detail']/@href"
                }
            }
        ),
        # 第二步：处理详情链接
        Parse(
            func="json",
            kwargs={
                "rules": {
                    "detail_urls[*]": {
                        "url": "@",
                        "full_url": "f'https://example.com{@}'"
                    }
                }
            }
        )
    ]

自定义解析函数

除了内置解析引擎，还可以使用自定义解析函数：

1. 简单自定义函数

def custom_parser(context):
    """自定义解析函数"""
    response = context.response
    text = response.text
 
    # 自定义解析逻辑
    lines = text.split('\n')
    data = []
    for line in lines:
        if line.strip():
            parts = line.split(',')
            if len(parts) >= 3:
                data.append({
                    "name": parts[0].strip(),
                    "value": parts[1].strip(),
                    "type": parts[2].strip()
                })
 
    return data
 
@dataclass
class CustomConfig(Config):
    download = [
        Download(url="https://example.com/data.csv")
    ]
    parse = [
        Parse(func=custom_parser)
    ]

2. 带参数的自定义函数

def advanced_parser(context, min_price=0, max_price=float('inf')):
    """带参数的自定义解析函数"""
    response = context.response
    data = response.json()
 
    filtered_products = []
    for product in data.get('products', []):
        price = product.get('price', 0)
        if min_price <= price <= max_price:
            filtered_products.append({
                "id": product.get('id'),
                "name": product.get('name'),
                "price": price,
                "discounted": price < product.get('original_price', price)
            })
 
    return filtered_products
 
@dataclass
class ParameterConfig(Config):
    download = [
        Download(url="https://api.example.com/products")
    ]
    parse = [
        Parse(
            func=advanced_parser,
            kwargs={
                "min_price": 100,
                "max_price": 1000
            }
        )
    ]

解析结果处理

使用 Pipeline 节点处理解析结果

解析完成后，可以使用 Pipeline 节点对结果进行进一步处理：

from bricks.spider.template import Pipeline
 
def process_items(context):
    """处理解析结果"""
    items = context.items
 
    for item in items:
        # 数据清洗
        if 'price' in item:
            item['price'] = round(float(item['price']), 2)
 
        # 添加时间戳
        item['crawl_time'] = datetime.now().isoformat()
 
        # 数据验证
        if not item.get('name'):
            continue
 
        # 保存到数据库或文件
        print(f"处理商品: {item['name']}, 价格: {item['price']}")
 
@dataclass
class ProcessConfig(Config):
    download = [
        Download(url="https://api.example.com/products")
    ]
    parse = [
        Parse(
            func="json",
            kwargs={
                "rules": {
                    "products[*]": {
                        "name": "title",
                        "price": "price",
                        "description": "desc"
                    }
                }
            }
        )
    ]
    pipeline = [
        Pipeline(
            func=process_items,
            success=True  # 处理成功后标记任务完成
        )
    ]

错误处理和调试

1. 解析失败处理

def safe_parser(context):
    """安全的解析函数，包含错误处理"""
    try:
        response = context.response
        data = response.json()
 
        if not data:
            return []
 
        return data.get('items', [])
 
    except Exception as e:
        print(f"解析失败: {e}")
        return []
 
@dataclass
class SafeConfig(Config):
    parse = [
        Parse(func=safe_parser)
    ]

2. 调试解析过程

def debug_parser(context):
    """调试解析过程"""
    response = context.response
 
    print(f"响应状态码: {response.status_code}")
    print(f"响应URL: {response.url}")
    print(f"响应长度: {len(response.text)}")
 
    # 尝试不同的解析方法
    try:
        # 尝试JSON解析
        json_data = response.json()
        print(f"JSON解析成功，数据类型: {type(json_data)}")
        return json_data
    except:
        print("JSON解析失败，尝试XPath解析")
 
    try:
        # 尝试XPath解析
        titles = response.xpath("//title/text()")
        print(f"XPath解析结果: {titles}")
        return {"titles": titles}
    except:
        print("XPath解析失败")
 
    return []

最佳实践

  1. 选择合适的解析引擎：

    • API 数据使用 "json"
    • HTML 页面使用 "xpath"
    • 复杂 JSON 查询使用 "jsonpath"
    • 文本模式匹配使用 "regex"

  2. 合理使用 Layout：

    • 使用 rename 统一字段命名
    • 使用 default 设置合理默认值
    • 使用 factory 进行数据类型转换
    • 使用 show 过滤不需要的字段

  3. 错误处理：

    • 在自定义解析函数中添加异常处理
    • 设置合理的默认值
    • 记录解析失败的情况

  4. 性能优化：

    • 使用精确的选择器表达式
    • 避免过度复杂的解析逻辑
    • 合理使用 archive 参数进行断点续传

模板爬虫 vs 表单爬虫

Bricks 框架提供了两种配置式爬虫：模板爬虫（Template Spider）和表单爬虫（Form Spider），它们在解析器使用上有所不同。

模板爬虫（Template Spider）

模板爬虫使用 bricks.spider.template.Spider，适用于简单的线性爬取流程：

from bricks.spider.template import Spider, Config, Download, Parse, Pipeline
 
@dataclass
class TemplateConfig(Config):
    # 初始化种子
    init = [
        Init(
            func=lambda: [{"page": i} for i in range(1, 6)],
        )
    ]
 
    # 下载配置
    download = [
        Download(url="https://api.example.com/data?page={page}")
    ]
 
    # 解析配置
    parse = [
        Parse(
            func="json",
            kwargs={
                "rules": {
                    "items[*]": {
                        "id": "id",
                        "title": "title",
                        "content": "content"
                    }
                }
            }
        )
    ]
 
    # 管道配置
    pipeline = [
        Pipeline(
            func=lambda context: print(f"处理了 {len(context.items)} 条数据"),
            success=True
        )
    ]
 
class TemplateSpider(Spider):
    @property
    def config(self):
        return TemplateConfig()

表单爬虫（Form Spider）

表单爬虫使用 bricks.spider.form.Spider，支持更复杂的流程控制：

from bricks.spider.form import Spider, Config, Download, Parse, Pipeline, Task
 
@dataclass
class FormConfig(Config):
    # 流程节点配置（按顺序执行）
    spider = [
        # 第一步：下载列表页
        Download(url="https://example.com/list?page={page}"),
 
        # 第二步：解析列表页
        Parse(
            func="xpath",
            kwargs={
                "rules": {
                    "//a[@class='item-link']": {
                        "detail_url": "@href",
                        "title": "text()"
                    }
                }
            }
        ),
 
        # 第三步：自定义任务处理
        Task(
            func=lambda context: context.submit(
                *[{"url": item["detail_url"]} for item in context.items]
            )
        ),
 
        # 第四步：下载详情页
        Download(url="{url}"),
 
        # 第五步：解析详情页
        Parse(
            func="xpath",
            kwargs={
                "rules": {
                    "//div[@class='content']": {
                        "content": ".//text()",
                        "images": ".//img/@src"
                    }
                }
            }
        ),
 
        # 第六步：数据处理
        Pipeline(
            func=lambda context: save_to_database(context.items),
            success=True
        )
    ]
 
class FormSpider(Spider):
    @property
    def config(self):
        return FormConfig()

动态解析规则

配置式爬虫支持动态解析规则，可以根据种子数据动态调整解析逻辑：

1. 使用模板变量

@dataclass
class DynamicConfig(Config):
    download = [
        Download(url="https://api.example.com/{category}/items")
    ]
    parse = [
        Parse(
            func="json",
            kwargs={
                "rules": {
                    "data[*]": {
                        "id": "id",
                        "name": "name",
                        "category": "'{category}'",  # 使用种子中的category
                        "price": "price",
                        "url": "f'https://example.com/{category}/item/{id}'"
                    }
                }
            }
        )
    ]

2. 条件解析

def conditional_parser(context):
    """根据响应内容选择不同的解析策略"""
    response = context.response
    seeds = context.seeds
 
    # 根据种子中的类型选择解析方式
    data_type = seeds.get("type", "json")
 
    if data_type == "json":
        return response.extract(
            engine="json",
            rules={
                "items[*]": {
                    "id": "id",
                    "title": "title"
                }
            }
        )
    elif data_type == "html":
        return response.extract(
            engine="xpath",
            rules={
                "//div[@class='item']": {
                    "id": "@data-id",
                    "title": ".//h3/text()"
                }
            }
        )
    else:
        return []
 
@dataclass
class ConditionalConfig(Config):
    parse = [
        Parse(func=conditional_parser)
    ]

解析器链式调用

可以在一个解析节点中组合使用多个解析器：

def chain_parser(context):
    """链式解析器调用"""
    response = context.response
 
    # 第一步：使用正则提取JSON数据
    json_text = response.re_first(r'var data = ({.*?});')
    if not json_text:
        return []
 
    # 第二步：解析JSON数据
    from bricks.lib.response import Response
    json_response = Response(content=json_text)
 
    # 第三步：使用JMESPath提取具体字段
    items = json_response.extract(
        engine="json",
        rules={
            "products[*]": {
                "id": "productId",
                "name": "productName",
                "price": "price.current",
                "images": "images[*].url"
            }
        }
    )
 
    return items
 
@dataclass
class ChainConfig(Config):
    parse = [
        Parse(func=chain_parser)
    ]

解析结果的后处理

使用 Layout 进行高级数据处理

from bricks.spider.form import Layout
 
# 自定义数据处理函数
def price_converter(price_str):
    """价格转换函数"""
    if isinstance(price_str, str):
        # 移除货币符号和逗号
        clean_price = price_str.replace('$', '').replace(',', '')
        try:
            return float(clean_price)
        except ValueError:
            return 0.0
    return price_str
 
def category_mapper(category_id):
    """分类映射函数"""
    category_map = {
        1: "电子产品",
        2: "服装",
        3: "图书",
        4: "家居"
    }
    return category_map.get(category_id, "其他")
 
@dataclass
class AdvancedLayoutConfig(Config):
    parse = [
        Parse(
            func="json",
            kwargs={
                "rules": {
                    "products[*]": {
                        "id": "id",
                        "name": "title",
                        "price_raw": "price",
                        "category_id": "categoryId",
                        "description": "desc",
                        "stock": "inventory.quantity"
                    }
                }
            },
            layout=Layout(
                # 字段重命名
                rename={
                    "price_raw": "price",
                    "category_id": "category"
                },
                # 默认值设置
                default={
                    "stock": 0,
                    "description": "暂无描述"
                },
                # 数据类型转换和处理
                factory={
                    "price": price_converter,
                    "category": category_mapper,
                    "stock": int,
                    "id": str
                },
                # 只显示需要的字段
                show=["id", "name", "price", "category", "stock"]
            )
        )
    ]

调试和监控

1. 解析过程监控

def monitoring_parser(context):
    """带监控的解析函数"""
    import time
    start_time = time.time()
 
    response = context.response
    seeds = context.seeds
 
    print(f"开始解析 URL: {response.url}")
    print(f"种子数据: {seeds}")
    print(f"响应大小: {len(response.content)} bytes")
 
    try:
        # 执行解析
        items = response.extract(
            engine="json",
            rules={
                "data[*]": {
                    "id": "id",
                    "title": "title"
                }
            }
        )
 
        end_time = time.time()
        print(f"解析完成，耗时: {end_time - start_time:.2f}s")
        print(f"提取到 {len(items)} 条数据")
 
        return items
 
    except Exception as e:
        print(f"解析失败: {e}")
        return []
 
@dataclass
class MonitoringConfig(Config):
    parse = [
        Parse(func=monitoring_parser)
    ]

2. 解析结果验证

def validating_parser(context):
    """带验证的解析函数"""
    response = context.response
 
    # 执行解析
    items = response.extract(
        engine="json",
        rules={
            "products[*]": {
                "id": "id",
                "name": "name",
                "price": "price"
            }
        }
    )
 
    # 验证解析结果
    valid_items = []
    for item in items:
        # 检查必需字段
        if not item.get("id") or not item.get("name"):
            print(f"跳过无效数据: {item}")
            continue
 
        # 检查数据类型
        try:
            item["price"] = float(item.get("price", 0))
        except (ValueError, TypeError):
            print(f"价格数据无效: {item}")
            continue
 
        valid_items.append(item)
 
    print(f"原始数据: {len(items)} 条，有效数据: {len(valid_items)} 条")
    return valid_items
 
@dataclass
class ValidatingConfig(Config):
    parse = [
        Parse(func=validating_parser)
    ]

通过配置式爬虫的解析功能，可以大大简化数据提取的复杂度，提高开发效率和代码可维护性。无论是简单的数据提取还是复杂的多步骤处理，都能通过声明式的配置来实现。

2.4.4 解析器使用指南2.5.1 RPC 概述