wxpath - 声明式网络图遍历与 XPath

新功能：TUI - 交互式终端界面（由 Textual 驱动），用于测试 wxpath 表达式和导出数据。

Wxpath TUI Demo screenshot

安装

需要 Python 3.10+。

pip install wxpath
# 如需 TUI 支持：
pip install "wxpath[tui]"
# 立即通过 uv 启动 TUI：
uvx --from "wxpath[tui]" wxpath-tui

什么是 wxpath？

wxpath 是一个声明式网络爬虫，其遍历逻辑直接通过 XPath 表达。无需编写命令式的爬取循环，wxpath 允许你在单个表达式中描述要跟随的链接以及要提取的内容。wxpath 并发地执行该表达式，采用广度优先（ish）的方式，并在发现结果时即时流式输出。

以下表达式获取一个页面，提取链接，并并发地流式输出——无需爬取循环：

import wxpath

expr = "url('https://quotes.toscrape.com')//a/@href"

for link in wxpath.wxpath_async_blocking_iter(expr):
    print(link)

通过引入 url(...) 运算符和 /// 语法，wxpath 引擎能够执行递归（或分页）的网络爬取与提取：

import wxpath

path_expr = """
url('https://quotes.toscrape.com')
  ///url(//a/@href)
    //a/@href
"""

for item in wxpath.wxpath_async_blocking_iter(path_expr, max_depth=1):
    print(item)

为什么使用 wxpath？

大多数网页抓取工具强迫你先编写爬取控制流程，再处理数据提取。

wxpath 将这两个步骤合二为一：

你以声明式方式描述遍历
提取逻辑内联书写
引擎负责调度、并发和去重

适合 RAG 的输出

直接从图中提取干净、结构化的 JSON 层级——为你的 LLM 提供信号而非噪音。详情请参考 LangChain 集成。

确定性

wxpath 是确定性的（注意：并非由 LLM 驱动）。虽然我们无法保证网络的稳定性，但能保证遍历的确定性。

文档（编写中）

文档现已可用此处。

示例

import wxpath
from wxpath.settings import CRAWLER_SETTINGS

# 自定义请求头体现礼貌；某些站点（如维基百科）需要此设置
CRAWLER_SETTINGS.headers = {'User-Agent': 'my-app/0.4.0 (contact: [email protected])'}

# 爬取、提取字段、构建知识图谱
path_expr = """
url('https://en.wikipedia.org/wiki/Expression_language')
  ///url(
        //main//a/@href[
            starts-with(., '/wiki/') and not(contains(., ':'))
        ]
    )
    /map{
        'title': (//span[contains(@class, "mw-page-title-main")]/text())[1] ! string(.),
        'url': string(base-uri(.)),
        'short_description': //div[contains(@class, 'shortdescription')]/text() ! string(.),
        'forward_links': //div[@id="mw-content-text"]//a/@href ! string(.)
    }
"""

for item in wxpath.wxpath_async_blocking_iter(path_expr, max_depth=1):
    print(item)

注意： 某些站点（包括维基百科）若无合适的请求头可能会阻止请求。
参见进阶：引擎与爬虫配置以设置自定义 User-Agent。

上述表达式的功能如下：

从指定的 URL https://en.wikipedia.org/wiki/Expression_language 开始。
筛选 <main> 区域中开头为 /wiki/ 且不包含冒号（:）的链接。
对于每个找到的链接，
- 跟随链接并提取该页面的标题、URL 和简短描述。
- 重复步骤 2，直到达到最大深度。
在发现数据时即时流式输出。

`url(...)` 和 `///url(...)` 详解

url(...) 是一个自定义运算符，用于获取用户指定或内部生成的 URL 的内容，并将其作为 lxml.html.HtmlElement 返回，以便进行后续的 XPath 处理。
///url(...) 表示深度爬取。它指示运行时引擎继续跟随链接，直到达到指定的 max_depth。与重复的 url() 跳转不同，它允许单个表达式描述更深入的图探索。警告：请谨慎使用，并通过 max_depth 或 XPath 谓词加以约束，以避免遍历爆炸。

语言设计

详见 DESIGN.md 了解语言设计的详细信息。你将看到核心概念并从零开始设计该语言。

通用流程

wxpath 将表达式评估为一系列遍历和提取步骤（内部称为“Segment”）。

url(...) 创建爬取任务，可以是静态的（通过固定 URL）或动态的（通过 XPath 表达式派生出的 URL）。URL 在全局范围内基于尽力而为的原则去重，而不是按深度去重。

XPath 片段作用于获取到的文档（通过紧邻的前一个 url(...) 操作获取）。

///url(...) 表示深度爬取——它以广度优先（ish）的方式执行，直到 max_depth。

结果一旦就绪即被产出。

异步爬取

wxpath 优先使用 asyncio/aiohttp，提供用于爬取和提取数据的异步 API。

import asyncio
from wxpath import wxpath_async

items = []

async def main():
    path_expr = "url('https://en.wikipedia.org/wiki/Expression_language')///url(//@href[starts-with(., '/wiki/')])//a/@href"
    async for item in wxpath_async(path_expr, max_depth=1):
        items.append(item)

asyncio.run(main())

阻塞式并发请求

wxpath 也提供“异步中的同步”API，允许你并发地爬取多个页面，同时保持同步代码的简洁性。这对于在严格同步的执行环境（即不在 asyncio 事件循环内）中进行性能敏感的爬取特别有用。

from wxpath import wxpath_async_blocking_iter

path_expr = "url('https://en.wikipedia.org/wiki/Expression_language')///url(//@href[starts-with(., '/wiki/')])//a/@href"
items = list(wxpath_async_blocking_iter(path_expr, max_depth=1))

礼貌爬取

wxpath 默认通过 WXPathEngine(..., robotstxt=True) 构造函数尊重 robots.txt。

输出类型

wxpath 的 Python API 输出结构化的对象。

根据表达式的不同，结果可能包括：

lxml.* 和 lxml.html.* 对象
elementpath.datatypes.* 对象（用于 XPath 3.1 特性）
WxStr（带有来源信息的字符串值）
字典/映射（map）
列表或其他 XPath 原生值

CLI 将这些对象展平为纯 JSON 以便显示。 Python API 默认保留结构。

默认使用 XPath 3.1

wxpath 使用 elementpath 库提供 XPath 3.1 支持，使你能使用高级 XPath 特性，如映射、数组等。这让你可以编写功能更强大的 XPath 查询。

path_expr = """
    url('https://en.wikipedia.org/wiki/Expression_language')
    ///url(//div[@id='mw-content-text']//a/@href)
    /map{ 
        'title':(//span[contains(@class, "mw-page-title-main")]/text())[1], 
        'short_description':(//div[contains(@class, "shortdescription")]/text())[1],
        'url'://link[@rel='canonical']/@href[1]
    }
"""
# [...
# {'title': 'Computer language',
# 'short_description': 'Formal language for communicating with a computer',
# 'url': 'https://en.wikipedia.org/wiki/Computer_language'},
# {'title': 'Machine-readable medium and data',
# 'short_description': 'Medium capable of storing data in a format readable by a machine',
# 'url': 'https://en.wikipedia.org/wiki/Machine-readable_medium_and_data'},
# {'title': 'Domain knowledge',
# 'short_description': 'Specialist knowledge within a specific field',
# 'url': 'https://en.wikipedia.org/wiki/Domain_knowledge'},
# ...]

进度条

wxpath 提供进度条（通过 tqdm）来追踪爬取进度。这对于长时间运行的爬取特别有用。

通过设置 engine.run(..., progress=True) 启用，或在任何 wxpath_async*(...) 函数中传入 progress=True。

items = wxpath.wxpath_async_blocking("...", progress=True)
> 100%|██████████████████████████████████████████████████████████▎| 469/471 [00:05<00:00, 72.00it/s, depth=2, yielded=457]

CLI

wxpath 提供命令行界面（CLI），方便你直接从终端快速试验和执行 wxpath 表达式。

以下示例演示如何从“Expression language”页面开始爬取维基百科，提取指向其他维基页面的链接，并从每个链接页面获取特定字段。

注意：由于网页内容的不断变化，输出可能随时间而异。

> wxpath --depth 1 \
    --header "User-Agent: my-app/0.1 (contact: [email protected])" \
    "url('https://en.wikipedia.org/wiki/Expression_language') \
    ///url(//div[@id='mw-content-text']//a/@href[starts-with(., '/wiki/') \
        and not(matches(@href, '^(?:/wiki/)?(?:Wikipedia|File|Template|Special|Template_talk|Help):'))]) \
    /map{ \
        'title':(//span[contains(@class, 'mw-page-title-main')]/text())[1], \
        'short_description':(//div[contains(@class, 'shortdescription')]/text())[1], \
        'url':string(base-uri(.)), \
        'backlink':wx:backlink(.), \
        'depth':wx:depth(.) \
        }"

{"title": "Computer language", "short_description": "Formal language for communicating with a computer", "url": "https://en.wikipedia.org/wiki/Computer_language", "backlink": "https://en.wikipedia.org/wiki/Expression_language", "depth": 1.0}
{"title": "Machine-readable medium and data", "short_description": "Medium capable of storing data in a format readable by a machine", "url": "https://en.wikipedia.org/wiki/Machine_readable", "backlink": "https://en.wikipedia.org/wiki/Expression_language", "depth": 1.0}
{"title": "Domain knowledge", "short_description": "Specialist knowledge within a specific field", "url": "https://en.wikipedia.org/wiki/Domain_knowledge", "backlink": "https://en.wikipedia.org/wiki/Expression_language", "depth": 1.0}
{"title": "Advanced Boolean Expression Language", "short_description": "Hardware description language and software", "url": "https://en.wikipedia.org/wiki/Advanced_Boolean_Expression_Language", "backlink": "https://en.wikipedia.org/wiki/Expression_language", "depth": 1.0}
{"title": "Data Analysis Expressions", "short_description": "Formula and data query language", "url": "https://en.wikipedia.org/wiki/Data_Analysis_Expressions", "backlink": "https://en.wikipedia.org/wiki/Expression_language", "depth": 1.0}
{"title": "Jakarta Expression Language", "short_description": "Computer programming language", "url": "https://en.wikipedia.org/wiki/Jakarta_Expression_Language", "backlink": "https://en.wikipedia.org/wiki/Expression_language", "depth": 1.0}
{"title": "Rights Expression Language", "short_description": [], "url": "https://en.wikipedia.org/wiki/Rights_Expression_Language", "backlink": "https://en.wikipedia.org/wiki/Expression_language", "depth": 1.0}
{"title": "Computer science", "short_description": "Study of computation", "url": "https://en.wikipedia.org/wiki/Computer_science", "backlink": "https://en.wikipedia.org/wiki/Expression_language", "depth": 1.0}

命令行选项：

--depth                <depth>       最大爬取深度
--verbose              [true|false]  提供浅层 CLI 信息
--debug                [true|false]  提供详细的运行时输出和信息
--concurrency          <concurrency> 并发请求数
--concurrency-per-host <concurrency> 每主机的并发请求数
--header               "Key:Value"   添加自定义请求头（例如 'Key:Value'）。可多次使用。
--respect-robots       [true|false] （默认：True）尊重 robots.txt
--cache                [true|false] （默认：False）将爬取结果持久化到本地数据库

TUI

wxpath 提供终端界面（TUI），用于交互式表达式测试和数据提取。

详见 TUI 快速入门。

持久化和缓存

wxpath 可选择将爬取结果持久化到本地数据库。当你爬取大量 URL、需要暂停爬取、更改提取表达式或重新启动爬取时，这尤其有用。

wxpath 支持两种后端：sqlite 和 redis。SQLite 适合小规模爬取，且使用单个 worker（即 engine.crawler.concurrency == 1）。Redis 适合大规模爬取，且使用多个 worker。如果使用 sqlite 后端且 min(engine.crawler.concurrency, engine.crawler.per_host) > 1，你将看到一条警告。

要使用该功能，必须安装相应的可选依赖：

pip install wxpath[cache-sqlite]
pip install wxpath[cache-redis]

安装依赖后，必须启用缓存：

from wxpath.settings import SETTINGS

# 启用缓存；默认使用 sqlite
SETTINGS.http.client.cache.enabled = True

# 使用 redis 后端
SETTINGS.http.client.cache.enabled = True
SETTINGS.http.client.cache.backend = "redis"
SETTINGS.http.client.cache.redis.address = "redis://localhost:6379/0"

# 照常运行 wxpath
items = list(wxpath_async_blocking_iter('...', max_depth=1, engine=engine))

设置

详见 settings.py 了解设置的详细信息。

钩子（实验性）

wxpath 支持可插拔的钩子系统，允许你修改爬取和提取行为。你可以注册钩子来预处理 URL、后处理 HTML、过滤提取的值等。钩子将按注册顺序执行。钩子可能会影响性能。


from wxpath import hooks

@hooks.register
class OnlyEnglish:
    def post_parse(self, ctx, elem):
        lang = elem.xpath('string(/html/@lang)').lower()[:2]
        return elem if lang in ("en", "") else None

异步使用

注意：钩子可以是同步或异步的，但项目中所有钩子应保持同一风格。不支持混合同步和异步的钩子，可能导致意外行为。


from wxpath import hooks

@hooks.register
class OnlyEnglish:
    async def post_parse(self, ctx, elem):
        lang = elem.xpath('string(/html/@lang)').lower()[:2]
        return elem if lang in ("en", "") else None

预定义钩子

JSONLWriter（别名 NDJSONWriter）是一个内置钩子，将提取的数据写入以换行符分隔的 JSON 文件。这对于以结构化格式存储结果、便于后续处理非常有用。

from wxpath import hooks
hooks.register(hooks.JSONLWriter)

安装

需要 Python 3.10+。

pip install wxpath

如需持久化/缓存功能，wxpath 支持以下后端：

pip install wxpath[cache-sqlite]
pip install wxpath[cache-redis]

对比

详见 COMPARISONS.md 了解与其他网页抓取工具的对比。

进阶：引擎与爬虫配置

你可以像这样修改引擎和爬虫的行为：

from wxpath import wxpath_async_blocking_iter
from wxpath.core.runtime import WXPathEngine
from wxpath.http.client.crawler import Crawler

crawler = Crawler(
    concurrency=8,
    per_host=2,
    timeout=10,
    respect_robots=False,
    headers={
        "User-Agent": "my-app/0.1.0 (contact: [email protected])", # 维基百科等站点会感激这个设置
    },
)

# 如果未指定 `crawler`，则会使用提供的 concurrency、per_host 和 respect_robots 值创建默认 Crawler，或使用默认值。
engine = WXPathEngine(
    # concurrency: int = 16, 
    # per_host: int = 8,
    # respect_robots: bool = True,
    # allowed_response_codes: set[int] = {200},
    # allow_redirects: bool = True,
    crawler=crawler,
)

path_expr = "url('https://en.wikipedia.org/wiki/Expression_language')//url(//main//a/@href)"

items = list(wxpath_async_blocking_iter(path_expr, max_depth=1, engine=engine))

运行时 API (`wxpath_async*`) 选项

max_depth: int = 1
progress: bool = False
engine: WXPathEngine | None = None
yield_errors: bool = False

设置

你也可以使用 settings.py 启用缓存、限速、并发等。

项目理念

原则

实现声明式的爬取与抓取，无需模板代码
保持轻量、可组合
异步支持以实现高性能爬取

目标

在每次爬取中基于尽力而为的原则去重 URL。
爬取在边界耗尽或达到 max_depth 时终止。
并发执行请求。
结果一旦可用即流式输出。

当前限制

以下功能尚不支持：

自动代理轮换
基于浏览器的渲染（JavaScript 执行）
严格的结果排序

警告！！！

本项目处于早期开发阶段。核心概念已经稳定，但 API 和功能可能发生变化。请报告问题——特别是死锁爬取或意外行为——以及你希望看到的功能（无法保证一定会实现）。

爬取网站时请保持礼貌。默认启用了一个受 scrapy 启发的限速器。
深度爬取（///）需要用户自律，以避免无限制扩展（遍历爆炸）。
在某些情况下（例如，所有任务都在等待被阻塞的请求）可能出现死锁或挂起。如遇到此类行为，请报告问题。
考虑使用超时、max_depth 和 XPath 谓词及过滤器来限制爬取范围。

商业支持/咨询

如果你希望使用 wxpath 构建或运营爬虫/数据管道（提取、调度、监控、故障修复），或有其他网页抓取需求，请联系：[email protected]。

捐赠

如果你喜欢 wxpath 并希望支持其开发，请考虑捐赠。

版本控制

wxpath 遵循 semver：<MAJOR>.<MINOR>.<PATCH>。

但在 1.0.0 之前，遵循 0.<MAJOR>.<MINOR|PATCH>。

许可证

AGPL-3.0

wxpath

最受欢迎