CrawlabとCapSolverの統合：自動CAPTCHA解消による分散クローリング

CrawlabとCapSolverの統合: 分散クローリングのための自動CAPTCHA解決

Sora Fujimoto

AI Solutions Architect

09-Jan-2026

How to Solve Captcha with Crawlab Using CapSolver

スケールに応じたウェブクローラーの管理には、現代のアンチボットチャレンジに対応する堅牢なインフラが必要です。Crawlabは強力な分散型ウェブクローラー管理プラットフォームであり、CapSolverはAIを活用したCAPTCHA解決サービスです。これらを組み合わせることで、CAPTCHAチャレンジを自動的に回避する企業向けのクローリングシステムを構築できます。

このガイドでは、CrawlabのスパイダーにCapSolverを統合するための完全で即時使用可能なコード例を提供します。

学ぶ内容

Seleniumを使用したreCAPTCHA v2の解決
Cloudflare Turnstileの解決
Scrapyミドルウェアの統合
Node.js/Puppeteerの統合
スケールに応じたCAPTCHA処理のベストプラクティス

Crawlabとは？

Crawlabは、複数のプログラミング言語でスパイダーを管理するための分散型ウェブクローラー管理プラットフォームです。

主な特徴

言語に依存しない: Python、Node.js、Go、Java、PHPをサポート
フレームワークの柔軟性: Scrapy、Selenium、Puppeteer、Playwrightと動作
分散アーキテクチャ: マスター/ワーカーノードによる水平スケーリング
管理UI: スパイダーの管理とスケジューリング用のWebインターフェース

インストール

bash Copy

# Docker Composeを使用
git clone https://github.com/crawlab-team/crawlab.git
cd crawlab
docker-compose up -d

UIにアクセス: http://localhost:8080 (デフォルト: admin/admin)。

CapSolverとは？

CapSolverは、さまざまなCAPTCHAタイプに対応する高速で信頼性の高い解決サービスを提供するAIを活用したCAPTCHA解決サービスです。

サポートされているCAPTCHAタイプ

reCAPTCHA: v2、v3、Enterprise
Cloudflare: TurnstileとChallenge
AWS WAF: セキュリティ回避
およびその他

APIワークフロー

CAPTCHAパラメータを送信（タイプ、siteKey、URL）
タスクIDを取得
解決結果をポーリング
ページにトークンを挿入

前提条件

Python 3.8+ または Node.js 16+
CapSolver APIキー - ここから登録
Chrome/Chromiumブラウザ

bash Copy

# Python依存関係
pip install selenium requests

Seleniumを使用したreCAPTCHA v2の解決

reCAPTCHA v2を解決する完全なPythonスクリプト:

python Copy

"""
Crawlab + CapSolver: reCAPTCHA v2ソルバー
Seleniumを使用したreCAPTCHA v2チャレンジの解決用完全スクリプト
"""

import os
import time
import json
import requests
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.chrome.options import Options
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC

# 設定
CAPSOLVER_API_KEY = os.getenv('CAPSOLVER_API_KEY', 'YOUR_CAPSOLVER_API_KEY')
CAPSOLVER_API = 'https://api.capsolver.com'


class CapsolverClient:
    """Capsolver APIクライアント（reCAPTCHA v2用）"""

    def __init__(self, api_key: str):
        self.api_key = api_key
        self.session = requests.Session()

    def create_task(self, task: dict) -> str:
        """CAPTCHA解決タスクの作成"""
        payload = {
            "clientKey": self.api_key,
            "task": task
        }
        response = self.session.post(
            f"{CAPSOLVER_API}/createTask",
            json=payload
        )
        result = response.json()

        if result.get('errorId', 0) != 0:
            raise Exception(f"Capsolverエラー: {result.get('errorDescription')}")

        return result['taskId']

    def get_task_result(self, task_id: str, timeout: int = 120) -> dict:
        """タスク結果のポーリング"""
        for _ in range(timeout):
            payload = {
                "clientKey": self.api_key,
                "taskId": task_id
            }
            response = self.session.post(
                f"{CAPSOLVER_API}/getTaskResult",
                json=payload
            )
            result = response.json()

            if result.get('status') == 'ready':
                return result['solution']

            if result.get('status') == 'failed':
                raise Exception("CAPTCHA解決に失敗しました")

            time.sleep(1)

        raise Exception("解決待ちのタイムアウト")

    def solve_recaptcha_v2(self, website_url: str, site_key: str) -> str:
        """reCAPTCHA v2を解決し、トークンを返却"""
        task = {
            "type": "ReCaptchaV2TaskProxyLess",
            "websiteURL": website_url,
            "websiteKey": site_key
        }

        print(f"{website_url}のタスクを作成中...");
        task_id = self.create_task(task)
        print(f"タスク作成: {task_id}")

        print("解決待ち...");
        solution = self.get_task_result(task_id)
        return solution['gRecaptchaResponse']

    def get_balance(self) -> float:
        """アカウント残高の取得"""
        response = self.session.post(
            f"{CAPSOLVER_API}/getBalance",
            json={"clientKey": self.api_key}
        )
        return response.json().get('balance', 0)


class RecaptchaV2Crawler:
    """reCAPTCHA v2対応のSeleniumクローラー"""

    def __init__(self, headless: bool = True):
        self.headless = headless
        self.driver = None
        self.capsolver = CapsolverClient(CAPSOLVER_API_KEY)

    def start(self):
        """ブラウザの初期化"""
        options = Options()
        if self.headless:
            options.add_argument("--headless=new")
        options.add_argument("--no-sandbox")
        options.add_argument("--disable-dev-shm-usage")
        options.add_argument("--window-size=1920,1080")

        self.driver = webdriver.Chrome(options=options)
        print("ブラウザを起動しました")

    def stop(self):
        """ブラウザの終了"""
        if self.driver:
            self.driver.quit()
            print("ブラウザを終了しました")

    def detect_recaptcha(self) -> str:
        """reCAPTCHAの検出とサイトキーの取得"""
        try:
            element = self.driver.find_element(By.CLASS_NAME, "g-recaptcha")
            return element.get_attribute("data-sitekey")
        except:
            return None

    def inject_token(self, token: str):
        """解決したトークンをページに挿入"""
        self.driver.execute_script(f"""
            // g-recaptcha-responseテキストエリアを設定
            var responseField = document.getElementById('g-recaptcha-response');
            if (responseField) {{
                responseField.style.display = 'block';
                responseField.value = '{token}';
            }}

            // すべての隠し応答フィールドを設定
            var textareas = document.querySelectorAll('textarea[name="g-recaptcha-response"]');
            for (var i = 0; i < textareas.length; i++) {{
                textareas[i].value = '{token}';
            }}
        """)
        print("トークンを挿入しました")

    def submit_form(self):
        """フォームの送信"""
        try:
            submit = self.driver.find_element(
                By.CSS_SELECTOR,
                'button[type="submit"], input[type="submit"]'
            )
            submit.click()
            print("フォームを送信しました")
        except Exception as e:
            print(f"フォームの送信に失敗しました: {e}")

    def crawl(self, url: str) -> dict:
        """reCAPTCHA v2処理付きのURLクロール"""
        result = {
            'url': url,
            'success': False,
            'captcha_solved': False
        }

        try:
            print(f"アクセス中: {url}")
            self.driver.get(url)
            time.sleep(2)

            # reCAPTCHAの検出
            site_key = self.detect_recaptcha()

            if site_key:
                print(f"reCAPTCHA v2が検出されました！サイトキー: {site_key}")

                # CAPTCHAの解決
                token = self.capsolver.solve_recaptcha_v2(url, site_key)
                print(f"トークンを取得: {token[:50]}...")

                # トークンの挿入
                self.inject_token(token)
                result['captcha_solved'] = True

                # フォームの送信
                self.submit_form()
                time.sleep(2)

            result['success'] = True
            result['title'] = self.driver.title

        except Exception as e:
            result['error'] = str(e)
            print(f"エラー: {e}")

        return result


def main():
    """メインエントリポイント"""
    # 残高の確認
    client = CapsolverClient(CAPSOLVER_API_KEY)
    print(f"CapSolver残高: ${client.get_balance():.2f}")

    # クローラーの作成
    crawler = RecaptchaV2Crawler(headless=True)

    try:
        crawler.start()

        # ターゲットURLをクロール（ご自身のターゲットに置き換えてください）
        result = crawler.crawl("https://example.com/protected-page")

        print("\n" + "=" * 50)
        print("RESULT:")
        print(json.dumps(result, indent=2))

    finally:
        crawler.stop()


if __name__ == "__main__":
    main()

Cloudflare Turnstileの解決

Cloudflare Turnstileを解決する完全なPythonスクリプト:

python Copy

"""
Crawlab + Capsolver: Cloudflare Turnstile Solver
Turnstileチャレンジの解決用完全スクリプト
"""

import os
import time
import json
import requests
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.chrome.options import Options
from selenium.common.exceptions import NoSuchElementException

# 設定
CAPSOLVER_API_KEY = os.getenv('CAPSOLVER_API_KEY', 'YOUR_CAPSOLVER_API_KEY')
CAPSOLVER_API = 'https://api.capsolver.com'


class TurnstileSolver:
    """Turnstile用のCapsolverクライアント"""

    def __init__(self, api_key: str):
        self.api_key = api_key
        self.session = requests.Session()

    def solve(self, website_url: str, site_key: str) -> str:
        """Turnstile CAPTCHAの解決"""
        print(f"{website_url}のTurnstileを解決中")
        print(f"サイトキー: {site_key}")

        # タスクの作成
        task_data = {
            "clientKey": self.api_key,
            "task": {
                "type": "AntiTurnstileTaskProxyLess",
                "websiteURL": website_url,
                "websiteKey": site_key
            }
        }

        response = self.session.post(f"{CAPSOLVER_API}/createTask", json=task_data)
        result = response.json()

        if result.get('errorId', 0) != 0:
            raise Exception(f"Capsolverエラー: {result.get('errorDescription')}")

        task_id = result['taskId']
        print(f"タスク作成: {task_id}")

        # 結果のポーリング
        for i in range(120):
            result_data = {
                "clientKey": self.api_key,
                "taskId": task_id
            }

            response = self.session.post(f"{CAPSOLVER_API}/getTaskResult", json=result_data)
            result = response.json()

            if result.get('status') == 'ready':
                token = result['solution']['token']
                print("Turnstileが解決されました！")
                return token

            if result.get('status') == 'failed':
                raise Exception("Turnstileの解決に失敗しました")

            time.sleep(1)

        raise Exception("解決待ちのタイムアウト")


class TurnstileCrawler:
    """Turnstile対応のSeleniumクローラー"""

    def __init__(self, headless: bool = True):
        self.headless = headless
        self.driver = None
        self.solver = TurnstileSolver(CAPSOLVER_API_KEY)

    def start(self):
        """ブラウザの初期化"""
        options = Options()
        if self.headless:
            options.add_argument("--headless=new")
        options.add_argument("--no-sandbox")
        options.add_argument("--disable-dev-shm-usage")

        self.driver = webdriver.Chrome(options=options)

    def stop(self):
        """ブラウザの終了"""
        if self.driver:
            self.driver.quit()

    def detect_turnstile(self) -> str:
        """Turnstileの検出とサイトキーの取得"""
        try:
            turnstile = self.driver.find_element(By.CLASS_NAME, "cf-turnstile")
            return turnstile.get_attribute("data-sitekey")
        except NoSuchElementException:
            return None

    def inject_token(self, token: str):
        """Turnstileトークンの挿入"""
        self.driver.execute_script(f"""
            var token = '{token}';

            // cf-turnstile-responseフィールドを検索
            var field = document.querySelector('[name="cf-turnstile-response"]');
            if (field) {{
                field.value = token;
            }}

            // すべてのTurnstile入力フィールドを検索
            var inputs = document.querySelectorAll('input[name*="turnstile"]');
            for (var i = 0; i < inputs.length; i++) {{
                inputs[i].value = token;
            }}
        """)
        print("トークンを挿入しました！")

    def crawl(self, url: str) -> dict:
        """Turnstile処理付きのURLクロール"""
        result = {
            'url': url,
            'success': False,
            'captcha_solved': False,
            'captcha_type': None
        }

        try:
            print(f"アクセス中: {url}")
            self.driver.get(url)
            time.sleep(3)

            # Turnstileの検出
            site_key = self.detect_turnstile()

            if site_key:
                result['captcha_type'] = 'turnstile'
                print(f"Turnstileが検出されました！サイトキー: {site_key}")

                # 解決
                token = self.solver.solve(url, site_key)

                # 挿入
                self.inject_token(token)
                result['captcha_solved'] = True

                time.sleep(2)

            result['success'] = True
            result['title'] = self.driver.title

        except Exception as e:
            print(f"エラー: {e}")
            result['error'] = str(e)

        return result


def main():
    """メインエントリポイント"""
    crawler = TurnstileCrawler(headless=True)

    try:
        crawler.start()

        # ターゲットをクロール（ご自身のターゲットURLに置き換えてください）
        result = crawler.crawl("https://example.com/turnstile-protected")

        print("\n" + "=" * 50)
        print("RESULT:")
        print(json.dumps(result, indent=2))

    finally:
        crawler.stop()


if __name__ == "__main__":
    main()

Scrapy統合

CapSolverミドルウェアを備えた完全なScrapyスパイダー:

python Copy

"""
Crawlab + Capsolver: Scrapyスパイダー
CAPTCHA解決ミドルウェアを備えた完全なScrapyスパイダー
"""

import scrapy
import requests
import time
import os

CAPSOLVER_API_KEY = os.getenv('CAPSOLVER_API_KEY', 'YOUR_CAPSOLVER_API_KEY')
CAPSOLVER_API = 'https://api.capsolver.com'


class CapsolverMiddleware:
    """CAPTCHA解決用のScrapyミドルウェア"""

    def __init__(self):
        self.api_key = CAPSOLVER_API_KEY

    def solve_recaptcha_v2(self, url: str, site_key: str) -> str:
        """reCAPTCHA v2の解決"""

タスクの作成

Copy

    response = requests.post(
        f"{CAPSOLVER_API}/createTask",
        json={
            "clientKey": self.api_key,
            "task": {
                "type": "ReCaptchaV2TaskProxyLess",
                "websiteURL": url,
                "websiteKey": site_key
            }
        }
    )
    task_id = response.json()['taskId']

    # 結果のポーリング
    for _ in range(120):
        result = requests.post(
            f"{CAPSOLVER_API}/getTaskResult",
            json={"clientKey": self.api_key, "taskId": task_id}
        ).json()

        if result.get('status') == 'ready':
            return result['solution']['gRecaptchaResponse']

        time.sleep(1)

    raise Exception("タイムアウト")

class CaptchaSpider(scrapy.Spider):
"""CAPTCHA処理を備えたスパイダー"""

Copy

name = "captcha_spider"
start_urls = ["https://example.com/protected"]

custom_settings = {
    'DOWNLOAD_DELAY': 2,
    'CONCURRENT_REQUESTS': 1,
}

def __init__(self, *args, **kwargs):
    super().__init__(*args, **kwargs)
    self.capsolver = CapsolverMiddleware()

def parse(self, response):
    # reCAPTCHAのチェック
    site_key = response.css('.g-recaptcha::attr(data-sitekey)').get()

    if site_key:
        self.logger.info(f"reCAPTCHAが検出されました: {site_key}")

        # CAPTCHAの解決
        token = self.capsolver.solve_recaptcha_v2(response.url, site_key)

        # トークンを含むフォームの送信
        yield scrapy.FormRequest.from_response(
            response,
            formdata={'g-recaptcha-response': token},
            callback=self.after_captcha
        )
    else:
        yield from self.extract_data(response)

def after_captcha(self, response):
    """CAPTCHA後のページ処理"""
    yield from self.extract_data(response)

def extract_data(self, response):
    """ページからのデータ抽出"""
    yield {
        'title': response.css('title::text').get(),
        'url': response.url,
    }

Scrapy設定 (settings.py)

"""
BOT_NAME = 'captcha_crawler'
SPIDER_MODULES = ['spiders']

Capsolver

CAPSOLVER_API_KEY = 'YOUR_CAPSOLVER_API_KEY'

レートリミット

DOWNLOAD_DELAY = 2
CONCURRENT_REQUESTS = 1
ROBOTSTXT_OBEY = True
"""

Node.js/Puppeteer統合

CAPTCHA解決用の完全なNode.jsスクリプト:

javascript Copy

/**
 * Crawlab + Capsolver: Puppeteer スパイダー
 * CAPTCHA解決用の完全なNode.jsスクリプト
 */

const puppeteer = require('puppeteer');

const CAPSOLVER_API_KEY = process.env.CAPSOLVER_API_KEY || 'YOUR_CAPSOLVER_API_KEY';
const CAPSOLVER_API = 'https://api.capsolver.com';

/**
 * Capsolverクライアント
 */
class Capsolver {
    constructor(apiKey) {
        this.apiKey = apiKey;
    }

    async createTask(task) {
        const response = await fetch(`${CAPSOLVER_API}/createTask`, {
            method: 'POST',
            headers: { 'Content-Type': 'application/json' },
            body: JSON.stringify({
                clientKey: this.apiKey,
                task: task
            })
        });
        const result = await response.json();

        if (result.errorId !== 0) {
            throw new Error(result.errorDescription);
        }

        return result.taskId;
    }

    async getTaskResult(taskId, timeout = 120) {
        for (let i = 0; i < timeout; i++) {
            const response = await fetch(`${CAPSOLVER_API}/getTaskResult`, {
                method: 'POST',
                headers: { 'Content-Type': 'application/json' },
                body: JSON.stringify({
                    clientKey: this.apiKey,
                    taskId: taskId
                })
            });
            const result = await response.json();

            if (result.status === 'ready') {
                return result.solution;
            }

            if (result.status === 'failed') {
                throw new Error('タスク失敗');
            }

            await new Promise(r => setTimeout(r, 1000));
        }

        throw new Error('タイムアウト');
    }

    async solveRecaptchaV2(url, siteKey) {
        const taskId = await this.createTask({
            type: 'ReCaptchaV2TaskProxyLess',
            websiteURL: url,
            websiteKey: siteKey
        });

        const solution = await this.getTaskResult(taskId);
        return solution.gRecaptchaResponse;
    }

    async solveTurnstile(url, siteKey) {
        const taskId = await this.createTask({
            type: 'AntiTurnstileTaskProxyLess',
            websiteURL: url,
            websiteKey: siteKey
        });

        const solution = await this.getTaskResult(taskId);
        return solution.token;
    }
}

/**
 * メインクローリング関数
 */
async function crawlWithCaptcha(url) {
    const capsolver = new Capsolver(CAPSOLVER_API_KEY);

    const browser = await puppeteer.launch({
        headless: true,
        args: ['--no-sandbox', '--disable-setuid-sandbox']
    });

    const page = await browser.newPage();

    try {
        console.log(`クローリング中: ${url}`);
        await page.goto(url, { waitUntil: 'networkidle2' });

        // CAPTCHAタイプの検出
        const captchaInfo = await page.evaluate(() => {
            const recaptcha = document.querySelector('.g-recaptcha');
            if (recaptcha) {
                return {
                    type: 'recaptcha',
                    siteKey: recaptcha.dataset.sitekey
                };
            }

            const turnstile = document.querySelector('.cf-turnstile');
            if (turnstile) {
                return {
                    type: 'turnstile',
                    siteKey: turnstile.dataset.sitekey
                };
            }

            return null;
        });

        if (captchaInfo) {
            console.log(`${captchaInfo.type}が検出されました！`);

            let token;

            if (captchaInfo.type === 'recaptcha') {
                token = await capsolver.solveRecaptchaV2(url, captchaInfo.siteKey);

                // トークンの挿入
                await page.evaluate((t) => {
                    const field = document.getElementById('g-recaptcha-response');
                    if (field) field.value = t;

                    document.querySelectorAll('textarea[name="g-recaptcha-response"]')
                        .forEach(el => el.value = t);
                }, token);

            } else if (captchaInfo.type === 'turnstile') {
                token = await capsolver.solveTurnstile(url, captchaInfo.siteKey);

                // トークンの挿入
                await page.evaluate((t) => {
                    const field = document.querySelector('[name="cf-turnstile-response"]');
                    if (field) field.value = t;
                }, token);
            }

            console.log('CAPTCHAが解決され、挿入されました！');
        }

        // データの抽出
        const data = await page.evaluate(() => ({
            title: document.title,
            url: window.location.href
        }));

        return data;

    } finally {
        await browser.close();
    }
}

// メイン実行
const targetUrl = process.argv[2] || 'https://example.com';

crawlWithCaptcha(targetUrl)
    .then(result => {
        console.log('\n結果:');
        console.log(JSON.stringify(result, null, 2));
    })
    .catch(console.error);

最適な実践方法

1. エラー処理と再試行

python Copy

def solve_with_retry(solver, url, site_key, max_retries=3):
    """再試行ロジックを備えたCAPTCHAの解決"""
    for attempt in range(max_retries):
        try:
            return solver.solve(url, site_key)
        except Exception as e:
            if attempt == max_retries - 1:
                raise
            print(f"試行 {attempt + 1} に失敗: {e}")
            time.sleep(2 ** attempt)  # 指数バックオフ

2. コスト管理

解決前検出: CAPTCHAが存在する場合にのみCapsolverを呼び出す
トークンのキャッシュ: reCAPTCHAトークンは約2分間有効
残高の監視: バッチジョブの実行前に残高を確認する

3. レートリミット

python Copy

# Scrapy設定
DOWNLOAD_DELAY = 3
CONCURRENT_REQUESTS_PER_DOMAIN = 1

4. 環境変数

bash Copy

export CAPSOLVER_API_KEY="your-api-key-here"

問題解決

エラー	原因	解決策
`ERROR_ZERO_BALANCE`	クレジットが不足	Capsolverアカウントにクレジットを追加
`ERROR_CAPTCHA_UNSOLVABLE`	パラメータが不正	サイトキーの抽出を確認
`TimeoutError`	ネットワークの問題	タイムアウトを増加、再試行を追加
`WebDriverException`	ブラウザのクラッシュ	`--no-sandbox`フラグを追加

よくある質問

Q: CAPTCHAトークンの有効期限はどのくらいですか？
A: reCAPTCHAトークン: 約2分。Turnstile: サイトによって異なります。

Q: 平均的な解決時間はどのくらいですか？
A: reCAPTCHA v2: 5-15秒、Turnstile: 1-10秒。

Q: 自前のプロキシを使用できますか？
A: はい、"ProxyLess"の接尾辞のないタスクタイプを使用し、プロキシ設定を提供してください。

結論

CrawlabにCapSolverを統合することで、分散クローリングインフラストラクチャで信頼性の高いCAPTCHA処理が可能になります。上記の完全なスクリプトはCrawlabスパイダーに直接コピーできます。

始める準備はできましたか？ CapSolverに登録して、クローラーを高速化しましょう！

💡 Crawlab統合ユーザー向けの特典:
この統合を記念して、Crawlabユーザー向けに6%のボーナスコードを提供しています — Crawlab。このチュートリアルを通じて登録したCapSolverユーザーは、ダッシュボードで再充電時にコードを入力すると、即座に6%のクレジットが追加されます。

13. ドキュメンテーション

コンプライアンス免責事項：このブログで提供される情報は、情報提供のみを目的としています。CapSolverは、すべての適用される法律および規制の遵守に努めています。CapSolverネットワークの不法、詐欺、または悪用の目的での使用は厳格に禁止され、調査されます。私たちのキャプチャ解決ソリューションは、公共データのクローリング中にキャプチャの問題を解決する際に100%のコンプライアンスを確保しながら、ユーザーエクスペリエンスを向上させます。私たちは、サービスの責任ある使用を奨励します。詳細については、サービス利用規約およびプライバシーポリシーをご覧ください。