Scarnox · Blog

Scrapy のキャプチャミドルウェア: スパイダー内でキャプチャを解決する（2026）

Scrapy はリクエストとコールバックを中心に作られているため、キャプチャはそこにきれいに収まりません。キャプチャを扱う正しい方法は、ダウンローダーミドルウェアを使うことです。すべてのレスポンスを見て、チャレンジに気づき、トークンを取得し、そのトークンを使ってスパイダーにフォームを送信させます。このガイドでは、Scarnox が今日プロキシなしで解決する PopularCaptcha のページ向けに、そのミドルウェアを作ります。

無料トライアルクレジットから始めましょう。クレジットカードは不要です。他のものに向けて使う前に、テストページでコードを試してください。

ミドルウェアの仕組み

ミドルウェアはすべてのレスポンスで動作します。

  1. レスポンスに data-sitekey がなければ、そのレスポンスをそのまま通します。
  2. あれば、ミドルウェアはキャプチャを解決し、トークンを meta に保存した同じリクエストのコピーを返します。
  3. Scrapy はそのリクエストを再度取得します。ミドルウェアは meta にあるトークンを見て、レスポンスを通します。
  4. スパイダーは meta からトークンを読み取り、FormRequest.from_response でフォームを送信します。

解決の呼び出しは deferToThread を通じてスレッドで実行されるため、Scrapy のリアクターをブロックしません。キャプチャが解決されている間も、他のリクエストは進み続けます。

プロジェクトの設定

キーはコードではなく、環境変数に入れてください。

# settings.py
import os

SCARNOX_API_KEY = os.environ["SCARNOX_API_KEY"]
SCARNOX_TIMEOUT = 185

DOWNLOADER_MIDDLEWARES = {
    "myproject.middlewares.PopularCaptchaMiddleware": 543,
}

ROBOTSTXT_OBEY = True
DOWNLOAD_DELAY = 3
CONCURRENT_REQUESTS_PER_DOMAIN = 1

ダウンロードの遅延と、ドメインごとの同時実行数を1にするのは意図的です。サイトへの負荷を低く保ち、遭遇するチャレンジの数を減らします。

ミドルウェア

# myproject/middlewares.py
import requests
from scrapy import Request
from scrapy.http import Response
from twisted.internet.threads import deferToThread

API_URL = "https://api.scarnox.com/api/tasks/create"


class PopularCaptchaMiddleware:
    def __init__(self, api_key: str, timeout: int):
        self.api_key = api_key
        self.timeout = timeout

    @classmethod
    def from_crawler(cls, crawler):
        return cls(
            api_key=crawler.settings.get("SCARNOX_API_KEY"),
            timeout=crawler.settings.getint("SCARNOX_TIMEOUT", 185),
        )

    def _solve(self, sitekey: str, page_url: str) -> str:
        r = requests.post(
            API_URL,
            headers={"Authorization": f"Bearer {self.api_key}"},
            json={
                "captcha_type": "PopularCaptchaTaskProxyless",
                "site_key": sitekey,
                "site_url": page_url,
            },
            timeout=self.timeout,  # ワーカースレッドで実行され、リアクターでは実行されない
        )
        data = r.json()
        if data.get("errorId") != 0:
            raise RuntimeError(f"{data.get('errorCode')}: {data.get('errorDescription')}")
        return data["solution"]["token"]

    def process_response(self, request: Request, response: Response, spider):
        # 解決済み、または自分で組み立てたフォーム送信: そのまま通す。
        if request.meta.get("captcha_token") or request.meta.get("skip_captcha"):
            return response

        sitekey = response.css("[data-sitekey]::attr(data-sitekey)").get()
        if not sitekey:
            return response

        spider.logger.info("PopularCaptcha on %s, solving", response.url)

        def retry_with_token(token: str) -> Request:
            # Request を返すと再スケジュールされます。dont_filter で重複フィルターを回避します。
            return request.replace(
                dont_filter=True,
                meta={**request.meta, "captcha_token": token},
            )

        return deferToThread(self._solve, sitekey, response.url).addCallback(retry_with_token)

process_response から Deferred を返すことは、Scrapy がサポートしています。コールバックがトークンを再試行用のリクエストに変え、Scrapy はそれを他のリクエストと同様にスケジュールします。

スパイダー

# myproject/spiders/search.py
import scrapy
from scrapy.http import FormRequest


class SearchSpider(scrapy.Spider):
    name = "search"
    start_urls = ["https://search.example.com/search?q=web+scraping"]

    def parse(self, response):
        token = response.meta.get("captcha_token")
        if token and response.css("[data-sitekey]"):
            # Scrapy はフォーム自身のフィールドを埋めます。トークンは私たちが追加します。
            yield FormRequest.from_response(
                response,
                formdata={
                    "h-captcha-response": token,
                    "g-recaptcha-response": token,
                },
                callback=self.parse_results,
                meta={"skip_captcha": True},
                dont_filter=True,
            )
            return

        yield from self.parse_results(response)

    def parse_results(self, response):
        for item in response.css("div.result"):
            yield {
                "title": item.css("h3::text").get(),
                "url": item.css("a::attr(href)").get(),
            }

セレクタはプレースホルダーです。スクレイピングするサイトの実際のセレクタに置き換えてください。

skip_captcha フラグは、ミドルウェアがフォームの送信を再び解決しないようにするためのものです。これがないと、送信のたびに解決の費用がかかってしまいます。

実行する

export SCARNOX_API_KEY=scarnox_YOURAPIKEY
scrapy crawl search -O results.json

ログを見てください。PopularCaptcha on ..., solving のような行は、チャレンジが見つかり、トークンが要求されたことを意味します。すべてのページでその行が出るなら、速度を落としてください。頻繁なチャレンジは、サイトがクロールに気づいているサインです。

トークンのタイミング

PopularCaptcha のトークンの有効期限は約2分です。ミドルウェアは、トークンが届くとすぐに再試行のリクエストを要求し、スパイダーは同じコールバックの連鎖内で送信します。連鎖は短く保ってください。トークンからフォーム送信までの間に、遅いダウンロードや長い処理を加えないでください。ウィンドウについての詳細はキャプチャトークンの有効期限を参照してください。

エラー

解決に失敗すると、スレッド内で RuntimeError が発生します。Scrapy はそれを失敗したリクエストとしてログに記録します。失敗した解決には課金されないため、リトライは安価です。上限のあるリトライポリシーには、Scrapy 組み込みの RetryMiddleware を使い、_solve を短いループで包んでください。本番環境で機能するパターンは、キャプチャソルバーのエラー処理のガイドにまとめています。

責任を持ったスクレイピング

法的な背景はウェブスクレイピングは合法か？を参照してください。

料金

Scarnox は成功した解決ごとに課金し、失敗した解決は無料です。料金は現在の料金ページをご覧ください。無料トライアルクレジットから始めましょう。クレジットカードは不要です。

次のステップ