wxpath – XPathで宣言的に Web クロール

wxpath は、クロール経路を直接 XPath で記述できる宣言型ウェブクローラです。
命令的なループを書かずに「何をたどり」「何を抽出する」ことを一つの式で表現します。
エンジンはその式を並行して実行（幅優先っぽい）し、発見次第結果をストリームとして返します。

NOTE – プロジェクトはまだ開発初期です。コア概念は安定していますが、API と機能は変更される可能性があります。
デッドロックしたクロールや予期せぬ挙動の報告・新機能提案をお願いします。実装保証はありません。

目次

• 例
• 言語設計
• url(...) と ///url(...) の解説
• 一般的な流れ
• 非同期クロール
• 礼儀正しいクロール
• 出力タイプ
• XPath 3.1 対応
• プログレスバー
• CLI
• 永続化とキャッシュ
• 設定
• フック（実験的）
• インストール
• その他の例
• 比較
• 高度な: エンジン & クローラ設定
• プロジェクト哲学
• 警告
• 商用サポート / コンサルティング
• バージョニング
• ライセンス

例

import wxpath
from wxpath.settings import CRAWLER_SETTINGS

# 礼儀正しさのためのカスタムヘッダー（例：Wikipedia が要求）
CRAWLER_SETTINGS.headers = {
    'User-Agent': 'my-app/0.4.0 (contact: you@example.com)'
}

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)

出力

{
  "title": "Computer language",
  "url": "https://en.wikipedia.org/wiki/Computer_language",
  "short_description": "Formal language for communicating with a computer",
  "forward_links": [
    "/wiki/Formal_language",
    "/wiki/Communication",
    ...
  ]
}

式は以下を行います：

1. https://en.wikipedia.org/wiki/Expression_language

   を起点にする。

2. <main>

   内の

   /wiki/

   で始まり、コロン（

   :）

   を含まないリンクをフィルタ。
3. 各リンクへ遷移し、以下を抽出：
   • title
   • URL
   • short description

4. max_depth

   に到達したら停止。

データは取得次第すぐにストリームで返されます。

言語設計

詳細は

DESIGN.md

url(...) と ///url(...) の解説

url(...)

lxml.html.HtmlElement

///url(...)

max_depth

url()

max_depth

一般的な流れ

1. 評価 – wxpath は式を セグメント（遷移／抽出ステップ）のリストへパース。
2. クロールタスク –

   url(...)

   が静的/動的 URL を生成し、全体で重複除外を試みる。
3. XPath セグメント – 前の

   url(...)

   から取得したドキュメントに対して実行。
4. 深いクロール –

   ///url(...)

   が幅優先で

   max_depth

   まで展開。
5. Yield – 準備が整い次第結果を返す。

非同期クロール

wxpath は

asyncio

aiohttp

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())

ブロッキング、同時実行リクエスト

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))

礼儀正しいクロール

デフォルトで

robots.txt

engine = WXPathEngine(..., robotstxt=True)

CLI フラグで有効／無効を切り替え可能です。

出力タイプ

Python API は以下のような構造化オブジェクトを返します：

• lxml.*

  ,

  lxml.html.*

• elementpath.datatypes.*

  （XPath 3.1 機能用）

• WxStr

  – 由来情報付き文字列
• 辞書 / マップ
• リストやその他 XPath ネイティブ値

CLI はこれらを平坦化し、JSON として表示します。

XPath 3.1 対応

wxpath は elementpath ライブラリを利用し、マップ・配列などの高度な機能をサポートします。

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"
}

プログレスバー

tqdm

items = wxpath.wxpath_async_blocking("…", progress=True)

出力例

> 100%|██████████████████████████████████████████████████████▎|
469/471 [00:05<00:00, 72.00it/s, depth=2, yielded=457]

CLI

ターミナルから直接実験できるコマンドラインインターフェースです。

wxpath --depth 1 \
    --header "User-Agent: my-app/0.1 (contact: you@example.com)" \
    "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(.) \
     }"

JSONL 出力

{"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}
...

よく使う CLI オプション

--depth <depth>

--concurrency <n>

--concurrency-per-host <n>

--header "Key:Value"

永続化とキャッシュ

大規模クロールや途中で中断・再開したい場合に有効です。

バックエンド

engine.crawler.concurrency == 1

SQLite を使用しながら

min(engine.crawler.concurrency, engine.crawler.per_host) > 1

の場合は警告が表示されます。

インストール

pip install wxpath[cache-sqlite]   # または cache-redis

キャッシュを有効化

from wxpath.settings import SETTINGS

# SQLite がデフォルト
SETTINGS.http.client.cache.enabled = True

# Redis バックエンドの場合
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 はクロール／抽出挙動を変更できるプラグイン可能なフックシステムを備えています。

フックの登録

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

フックは同期/非同期のいずれかで実装し、すべて同じスタイルである必要があります。混在は未サポートです。

既定フック

JSONLWriter

NDJSONWriter

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

インストール

Python 3.10+ が必要です。

pip install wxpath[cache-sqlite]   # もしくは cache-redis

その他の例

EXAMPLES.md

比較

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: you@example.com)",
    },
)

engine = WXPathEngine(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))

プロジェクト哲学

原則

• ボイラープレートなしで宣言的クロール／スクレイピングを実現
• 軽量かつ組み合わせやすい設計
• 高速クローリングのため非同期化

目標

• URL のグローバル重複除外（最善策）
• フロンティアが枯渇または

  max_depth

  に到達したら終了
• 同時リクエストを実装
• 発見次第結果をストリーム化

警告

• ウェブサイトのポリシーに従い、デフォルトでスクレイピング制限（スロットリング）を有効にしています。

• ///

  で深くクロールすると探索爆発が起こる可能性があります。predicate と

  max_depth

  を活用してください。
• エッジケースでデッドロック／ハングが発生する場合があります。報告してください。
• タイムアウト、

  max_depth

  、XPath フィルタを使ってクロール範囲を制限しましょう。

商用サポート / コンサルティング

wxpath でクローラやデータフィードの構築・運用支援が必要な場合は
rodrigopala91@gmail.com までご連絡ください。

寄付

wxpath が役立ったと感じたら、開発への寄付を検討してください。

バージョニング

wxpath はセマンティックバージョニングに従います：

<MAJOR>.<MINOR>.<PATCH>

0.<MAJOR>.<MINOR|PATCH>

ライセンス

MIT © 2026