CamoufoxをCapSolverと統合する方法

# Pythonパッケージをインストール
pip install -U camoufox[geoip]

# Camoufoxブラウザをダウンロード
camoufox fetch

from camoufox.sync_api import Camoufox

with Camoufox(humanize=True) as browser:
    page = browser.new_page()
    page.goto("https://example.com")

pip install camoufox[geoip] httpx

import asyncio
import httpx
from camoufox.async_api import AsyncCamoufox

CAPSOLVER_API_KEY = "YOUR_API_KEY"
CAPSOLVER_API = "https://api.capsolver.com"


async def create_task(task_payload: dict) -> str:
    """CAPTCHA解決タスクを作成し、タスクIDを返します。"""
    async with httpx.AsyncClient() as client:
        response = await client.post(
            f"{CAPSOLVER_API}/createTask",
            json={
                "clientKey": CAPSOLVER_API_KEY,
                "task": task_payload
            }
        )
        result = response.json()
        if result.get("errorId") != 0:
            raise Exception(f"CapSolverエラー: {result.get('errorDescription')}")
        return result["taskId"]


async def get_task_result(task_id: str, max_attempts: int = 120) -> dict:
    """タスク結果をポーリングして解決またはタイムアウトまで待機します。"""
    async with httpx.AsyncClient() as client:
        for _ in range(max_attempts):
            response = await client.post(
                f"{CAPSOLVER_API}/getTaskResult",
                json={
                    "clientKey": CAPSOLVER_API_KEY,
                    "taskId": task_id
                }
            )
            result = response.json()

            if result.get("status") == "ready":
                return result["solution"]
            elif result.get("status") == "failed":
                raise Exception(f"タスク失敗: {result.get('errorDescription')}")

            await asyncio.sleep(1)

    raise TimeoutError("CAPTCHA解決がタイムアウトしました")


async def solve_captcha(task_payload: dict) -> dict:
    """CAPTCHA解決ワークフローを完了します。"""
    task_id = await create_task(task_payload)
    return await get_task_result(task_id)

from camoufox.sync_api import Camoufox

with Camoufox(
    addons=["/path/to/capsolver-extension"],
    headless=False  # 拡張機能はヘッドレスモードでは動作しません
) as browser:
    page = browser.new_page()
    # 拡張機能が自動的にCAPTCHAを検出および解決します

import asyncio
from camoufox.async_api import AsyncCamoufox

CAPSOLVER_API_KEY = "YOUR_API_KEY"
CAPSOLVER_API = "https://api.capsolver.com"


async def solve_turnstile(site_key: str, page_url: str) -> str:
    """Cloudflare Turnstileを解決し、トークンを返します。"""
    import httpx

    async with httpx.AsyncClient() as client:
        # タスクを作成
        response = await client.post(
            f"{CAPSOLVER_API}/createTask",
            json={
                "clientKey": CAPSOLVER_API_KEY,
                "task": {
                    "type": "AntiTurnstileTaskProxyLess",
                    "websiteURL": page_url,
                    "websiteKey": site_key,
                }
            }
        )
        task_id = response.json()["taskId"]

        # 結果をポーリング
        while True:
            result = await client.post(
                f"{CAPSOLVER_API}/getTaskResult",
                json={
                    "clientKey": CAPSOLVER_API_KEY,
                    "taskId": task_id
                }
            )
            data = result.json()

            if data.get("status") == "ready":
                return data["solution"]["token"]

            await asyncio.sleep(1)


async def main():
    target_url = "https://example.com/protected-page"
    turnstile_site_key = "0x4XXXXXXXXXXXXXXXXX"  # ページソースから確認してください

    async with AsyncCamoufox(
        humanize=True,
        headless=False,
        os="windows"
    ) as browser:
        page = await browser.new_page()
        await page.goto(target_url)

        # Turnstileが読み込まれるのを待機
        await page.wait_for_selector('input[name="cf-turnstile-response"]', timeout=10000)

        # CAPTCHAを解決
        token = await solve_turnstile(turnstile_site_key, target_url)
        print(f"Turnstileトークンを取得: {token[:50]}...")

        # トークンを注入
        await page.evaluate(f'''
            document.querySelector('input[name="cf-turnstile-response"]').value = "{token}";

            // もし存在すれば、非表示のコールバックを設定
            const callback = document.querySelector('[data-callback]');
            if (callback) {{
                const callbackName = callback.getAttribute('data-callback');
                if (window[callbackName]) {{
                    window[callbackName]('{token}');
                }}
            }}
        ''')

        # フォームを送信
        await page.click('button[type="submit"]')
        await page.wait_for_load_state("networkidle")

        print("Turnstileを成功裏に回避しました！")


if __name__ == "__main__":
    asyncio.run(main())

import asyncio
from camoufox.async_api import AsyncCamoufox

CAPSOLVER_API_KEY = "YOUR_API_KEY"
CAPSOLVER_API = "https://api.capsolver.com"


async def solve_recaptcha_v2(site_key: str, page_url: str) -> str:
    """reCAPTCHA v2を解決し、トークンを返します。"""
    import httpx

    async with httpx.AsyncClient() as client:
        # タスクを作成
        response = await client.post(
            f"{CAPSOLVER_API}/createTask",
            json={
                "clientKey": CAPSOLVER_API_KEY,
                "task": {
                    "type": "ReCaptchaV2TaskProxyLess",
                    "websiteURL": page_url,
                    "websiteKey": site_key,
                }
            }
        )
        result = response.json()

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

        task_id = result["taskId"]

        # 結果をポーリング
        while True:
            result = await client.post(
                f"{CAPSOLVER_API}/getTaskResult",
                json={
                    "clientKey": CAPSOLVER_API_KEY,
                    "taskId": task_id
                }
            )
            data = result.json()

            if data.get("status") == "ready":
                return data["solution"]["gRecaptchaResponse"]
            elif data.get("status") == "failed":
                raise Exception(f"失敗: {data.get('errorDescription')}")

            await asyncio.sleep(2)


async def main():
    target_url = "https://example.com/login"
    recaptcha_site_key = "6LcXXXXXXXXXXXXXXXXXXXXXXXXX"  # ページソースから確認してください

    async with AsyncCamoufox(
        humanize=True,
        headless=False,
        os=["windows", "macos"]  # ランダムなOSの選択
    ) as browser:
        page = await browser.new_page()
        await page.goto(target_url)

        # 人間のような遅延でフォームフィールドを埋めます
        await page.fill('input[name="username"]', "[email protected]")
        await asyncio.sleep(0.5)  # 人間のような一時停止
        await page.fill('input[name="password"]', "password123")

        # CAPTCHAを解決
        print("reCAPTCHA v2を解決中...")
        token = await solve_recaptcha_v2(recaptcha_site_key, target_url)
        print(f"トークンを取得: {token[:50]}...")

        # reCAPTCHAの応答フィールドにトークンを注入
        await page.evaluate(f'''
            document.getElementById('g-recaptcha-response').innerHTML = '{token}';
            document.getElementById('g-recaptcha-response').style.display = 'block';
        ''')

        # フォームを送信
        await page.click('button[type="submit"]')
        await page.wait_for_load_state("networkidle")

        print("ログインに成功しました！")


if __name__ == "__main__":
    asyncio.run(main())

import asyncio
from camoufox.async_api import AsyncCamoufox

CAPSOLVER_API_KEY = "YOUR_API_KEY"
CAPSOLVER_API = "https://api.capsolver.com"


async def solve_recaptcha_v3(
    site_key: str,
    page_url: str,
    action: str = "verify",
    min_score: float = 0.7
) -> str:
    """指定されたアクションと最小スコアでreCAPTCHA v3を解決します。"""
    import httpx

    async with httpx.AsyncClient() as client:
        response = await client.post(
            f"{CAPSOLVER_API}/createTask",
            json={
                "clientKey": CAPSOLVER_API_KEY,
                "task": {
                    "type": "ReCaptchaV3TaskProxyLess",
                    "websiteURL": page_url,
                    "websiteKey": site_key,
                    "pageAction": action,
                    "minScore": min_score
                }
            }
        )
        result = response.json()

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

        task_id = result["taskId"]

        while True:
            result = await client.post(
                f"{CAPSOLVER_API}/getTaskResult",
                json={
                    "clientKey": CAPSOLVER_API_KEY,
                    "taskId": task_id
                }
            )
            data = result.json()

            if data.get("status") == "ready":
                return data["solution"]["gRecaptchaResponse"]
            elif data.get("status") == "failed":
                raise Exception(f"失敗: {data.get('errorDescription')}")

            await asyncio.sleep(1)


async def main():
    target_url = "https://example.com/search"
    recaptcha_v3_key = "6LcXXXXXXXXXXXXXXXXXXXXXXXXX"

    async with AsyncCamoufox(
        humanize=2.0,  # 最大2秒のヒューマナイズされた動き
        headless=True,  # v3ではヘッドレスで動作可能
        geoip=True,  # プロキシからジオロケーションを自動検出
    ) as browser:
        page = await browser.new_page()
        await page.goto(target_url)

        # "search"アクションでreCAPTCHA v3を解決
        print("reCAPTCHA v3を解決中...")
        token = await solve_recaptcha_v3(
            recaptcha_v3_key,
            target_url,
            action="search",
            min_score=0.9  # 高スコアを要求
        )

        # トークンでコールバックを実行
        await page.evaluate(f'''
            // サイトのコールバック経由でトークンを送信

    print("reCAPTCHA v3を回避しました！")

---

## ベストプラクティス

### 1. プロキシローテーションとGeoIP

CamoufoxのGeoIP機能を使用して、プロキシの場所に自動的にフィンガープリントを一致させます：

```python
async with AsyncCamoufox(
    geoip=True,  # プロキシIPから自動検出
    proxy={
        "server": "http://proxy.example.com:8080",
        "username": "user",
        "password": "pass"
    }
) as browser:
    # フィンガープリントはプロキシの地理的場所に一致する
    pass

from browserforge.fingerprints import Screen

# 一般的な画面サイズに制限
screen = Screen(
    min_width=1280,
    max_width=1920,
    min_height=720,
    max_height=1080
)

async with AsyncCamoufox(
    os="windows",
    screen=screen,
) as browser:
    pass

import random

async def human_delay():
    """人間の行動を模倣するランダムな遅延."""
    await asyncio.sleep(random.uniform(1.0, 3.0))

# アクションの間に使用
await page.click('button')
await human_delay()
await page.fill('input', 'text')

async def solve_with_retry(task_payload: dict, max_retries: int = 3) -> dict:
    """リトライロジックでCAPTCHAを解決."""
    for attempt in range(max_retries):
        try:
            return await solve_captcha(task_payload)
        except TimeoutError:
            if attempt < max_retries - 1:
                print(f"タイムアウト、リトライ中... ({attempt + 1}/{max_retries})")
                await asyncio.sleep(5)
            else:
                raise
        except Exception as e:
            if "balance" in str(e).lower():
                raise  # バランスエラーはリトライしない
            if attempt < max_retries - 1:
                await asyncio.sleep(2)
            else:
                raise