描述 #

CommonCrawl网页内容提取器，支持多种解析策略

核心功能 #

• 多解析器支持：trafilatura/justext/goose3
• 支持本地和远程WARC文件
• 支持二进制数据和base64编码
• 批量处理WARC文件，提取网页正文
• 智能内容提取，过滤广告和导航元素

解析器对比 #

• justext：速度最快，占用资源少，过滤比较严格
• trafilatura：速度适中，内置规则+轻量NLP模型，提取质量高
• goose3：速度较慢，启用多层逻辑提取，主要针对英文新闻

算子参数 #

输入 #

输出 #

提取结果列表，每个元素包含以下字段：

• url: 网页URL
• content: 提取的正文内容
• warc_file: 源WARC文件标识
• extractor: 使用的提取器名称

参数 #

如参数没有默认值，则为必填参数

调用示例 #

下面的代码展示了如何使用 daft 运行算子从CommonCrawl的WARC文件中提取网页正文，支持文件路径、二进制数据和base64编码等多种输入格式。

from __future__ import annotations

import os

import daft
from daft import col
from daft.las.functions.text.commoncrawl_content_extractor import CommonCrawlContentExtractor
from daft.las.functions.udf import las_udf

if __name__ == "__main__":

    if os.getenv("DAFT_RUNNER", "native") == "ray":
        import logging

        import ray

        def configure_logging():
            logging.basicConfig(
                level=logging.INFO,
                format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
                datefmt="%Y-%m-%d %H:%M:%S.%s".format(),
            )
            logging.getLogger("tracing.span").setLevel(logging.WARNING)
            logging.getLogger("daft_io.stats").setLevel(logging.WARNING)
            logging.getLogger("DaftStatisticsManager").setLevel(logging.WARNING)
            logging.getLogger("DaftFlotillaScheduler").setLevel(logging.WARNING)
            logging.getLogger("DaftFlotillaDispatcher").setLevel(logging.WARNING)

        ray.init(dashboard_host="0.0.0.0", runtime_env={"worker_process_setup_hook": configure_logging})
        daft.context.set_runner_ray()
    daft.set_execution_config(actor_udf_ready_timeout=600)
    daft.set_execution_config(min_cpu_per_task=0)

    tos_dir_url = os.getenv("TOS_DIR_URL", "las-cn-beijing-public-online.tos-cn-beijing.volces.com")
    samples = {
        "warc_data": [
            f"https://{tos_dir_url}/public/shared_file_dataset/sample.warc.gz"
        ]
    }

    extractor_type = "trafilatura"
    max_records = 5
    df = daft.from_pydict(samples)
    df = df.with_column(
        "extracted_content",
        las_udf(
            CommonCrawlContentExtractor,
            construct_args={
                "warc_src_type": "warc_url",
                "extractor_type": extractor_type,
                "max_records": max_records,
            },
            num_gpus=0,
            batch_size=1,
            concurrency=1,
        )(col("warc_data")),
    )
    df.show()

    # ╭──────────────────────────────────────────────┬─────────────────────────────────────────────────────────────╮
    # │ warc_files                                   ┆ extracted_content                                           │
    # │ ---                                          ┆ ---                                                         │
    # │ Utf8                                         ┆ List[Struct[url: Utf8, content: Utf8, warc_file: Utf8,      │
    # │                                              ┆ extractor: Utf8]]                                           │
    # ╞══════════════════════════════════════════════╪═════════════════════════════════════════════════════════════╡
    # │ https://las-public-data-qa.tos…              ┆ [{url: http://00852imports.com/detail/5389084.html,         │
    # │                                              ┆ content: 随着互联网的发展，人们对网络速度的要求也越来越高…,         │
    # │                                              ┆ warc_file: sample.warc.gz, extractor: trafilatura},         │
    # │                                              ┆ {url: http://02y3tcpv.gd9.cc/?penglaibexdkcl224396.html,    │
    # │                                              ┆ content: 查看更多相关内容\n\n取消关注在如今的数字时代…,            │
    # │                                              ┆ warc_file: sample.warc.gz, extractor: trafilatura}]         │
    # ╰──────────────────────────────────────────────┴─────────────────────────────────────────────────────────────╯