CAPTCHAを解決する方法: AutoGenエージェント

# CapSolverコアエンジン（必須の依存関係）
pip install git+https://github.com/capsolver-ai/capsolver-core.git

# CapSolverエージェントツール
pip install git+https://github.com/capsolver-ai/capsolver-agent.git

# AutoGenフレームワーク
pip install autogen-agentchat

export CAPSOLVER_API_KEY="your-capsolver-api-key"
export OPENAI_API_KEY="your-openai-api-key"

import asyncio
from autogen import AssistantAgent, UserProxyAgent, config_list_from_json
from capsolver_agent.schema import create_executor

# CapSolverエクスキュータを作成
executor = create_executor(api_key="YOUR_CAPSOLVER_API_KEY")

def solve_captcha(
    captcha_type: str,
    website_url: str,
    website_key: str,
    page_action: str = "",
    min_score: float = 0.7
) -> str:
    """CAPTCHAチャレンジを解決し、検証トークンを返します。
    
    Args:
        captcha_type: CAPTCHAタイプ - 'reCaptchaV2', 'reCaptchaV3', または 'cloudflare'
        website_url: CAPTCHAが表示されているページの完全なURL
        website_key: サイトキー（data-sitekey属性）
        page_action: reCAPTCHA v3用のアクション名（オプション）
        min_score: reCAPTCHA v3の最小スコア（デフォルト0.7）
    
    Returns:
        解決されたCAPTCHAトークンまたはエラーメッセージ
    """
    params = {
        "captcha_type": captcha_type,
        "website_url": website_url,
        "website_key": website_key
    }
    if page_action and captcha_type == "reCaptchaV3":
        params["page_action"] = page_action
        params["min_score"] = min_score
    
    result = asyncio.run(executor.execute("solve_captcha", params))
    
    if result["success"]:
        token = result["solution"]["token"]
        return f"CAPTCHAは正常に解決されました。トークン（最初の80文字）: {token[:80]}..."
    return f"CAPTCHA解決に失敗しました: {result['error']}"

# LLM構成
llm_config = {
    "config_list": [{"model": "gpt-4o", "api_key": "YOUR_OPENAI_KEY"}],
    "temperature": 0
}

# アシスタントエージェント（何をすべきかを決定するAI）
assistant = AssistantAgent(
    name="web_assistant",
    system_message="""ウェブリサーチアシスタントで、CAPTCHA解決機能を持っています。
    タスクでCAPTCHA保護されたウェブサイトにアクセスする必要がある場合、次のパラメータを使用してsolve_captcha関数を呼び出してください:
    - captcha_type: 'reCaptchaV2', 'reCaptchaV3', または 'cloudflare'
    - website_url: ページの完全なURL
    - website_key: ページからのサイトキー
    
    解決後、トークンをユーザーに返してフォーム送信を完了してください。""",
    llm_config=llm_config
)

# ユーザープロキシ（アシスタントの代わりに関数を実行）
user_proxy = UserProxyAgent(
    name="user_proxy",
    human_input_mode="NEVER",
    max_consecutive_auto_reply=5,
    code_execution_config={"work_dir": "workspace", "use_docker": False}
)

# CAPTCHA解決関数を登録
from autogen import register_function

register_function(
    solve_captcha,
    caller=assistant,      # アシスタントがいつ呼び出すかを決定
    executor=user_proxy,   # ユーザープロキシが関数を実行
    name="solve_captcha",
    description="CAPTCHAチャレンジ（reCAPTCHA v2/v3またはCloudflare Turnstile）を解決し、検証トークンを返します"
)

# 会話を開始
user_proxy.initiate_chat(
    assistant,
    message="""reCAPTCHA v2で保護されたページにアクセスする必要があります。
    URL: https://example.com/data-portal
    サイトキー: 6LeIxAcTAAAAAJcZVRqyHh71UMIEGNQ_MXjiZKhI
    
    CAPTCHAを解決し、トークンを私に提供してください。"""
)

from autogen import GroupChat, GroupChatManager

# 特化したエージェント
researcher = AssistantAgent(
    name="researcher",
    system_message="ウェブサイトをリサーチし、データを収集します。検証チャレンジに遭遇した場合はsolve_captchaを使用してください。",
    llm_config=llm_config
)

analyst = AssistantAgent(
    name="analyst",
    system_message="収集されたデータを分析し、洞察を生成します。",
    llm_config=llm_config
)

executor_agent = UserProxyAgent(
    name="executor",
    human_input_mode="NEVER",
    code_execution_config={"work_dir": "workspace", "use_docker": False}
)

# リ searcherにCAPTCHA解決を登録
register_function(
    solve_captcha,
    caller=researcher,
    executor=executor_agent,
    name="solve_captcha",
    description="ウェブサイトのCAPTCHAチャレンジを解決"
)

# グループチャットを作成
group_chat = GroupChat(
    agents=[researcher, analyst, executor_agent],
    messages=[],
    max_round=10
)

manager = GroupChatManager(groupchat=group_chat, llm_config=llm_config)

# グループリサーチタスクを開始
executor_agent.initiate_chat(
    manager,
    message="https://example.com/reportsのデータをリサーチしてください。reCAPTCHA v2（キー: 6Le...）で保護されています。CAPTCHAを解決し、データを収集した後、分析してください。"
)

def solve_captcha_with_retry(
    captcha_type: str,
    website_url: str,
    website_key: str,
    max_retries: int = 3
) -> str:
    """失敗時に自動リトライでCAPTCHAを解決します。"""
    for attempt in range(max_retries):
        params = {
            "captcha_type": captcha_type,
            "website_url": website_url,
            "website_key": website_key
        }
        result = asyncio.run(executor.execute("solve_captcha", params))
        
        if result["success"]:
            return f"解決済み（試行 {attempt+1}）。トークン: {result['solution']['token'][:80]}..."
        
        if attempt < max_retries - 1:
            import time
            time.sleep(3)
    
    return f"{max_retries}回の試行後に失敗しました: {result.get('error', '不明なエラー')}"

def check_captcha_balance() -> str:
    """CAPTCHA解決クレジットの残高を確認します。"""
    result = asyncio.run(executor.execute("get_balance", {}))
    if result["success"]:
        return f"CapSolverの残高: ${result['balance']:.2f}"
    return "残高を取得できませんでした"

register_function(solve_captcha_with_retry, caller=assistant, executor=user_proxy,
    name="solve_captcha", description="リトライロジック付きでCAPTCHAを解決")
register_function(check_captcha_balance, caller=assistant, executor=user_proxy,
    name="check_balance", description="CAPTCHA解決クレジット残高を確認")