कैसे reCAPTCHA v3 को LlamaIndex एजेंट्स में हल करें

LlamaIndex एजेंट
    ↓ एक टाइप किए गए टूल का चयन करता है
FunctionTool वॉल्यूम
    ↓ विश्वसनीय संदर्भों की पुष्टि करता है
CapSolver निष्पादक
    ↓ छोटे जीवनकाल के समाधान वापस करता है
ब्राउज़र सेवा
    ↓ जमा और पुष्टि करता है
LlamaIndex वर्कफ़्लो जारी रहता है

from dataclasses import dataclass
from urllib.parse import urlparse

@dataclass(frozen=True)
class CaptchaContext:
    context_id: str
    website_url: str
    website_key: str
    page_action: str
    enterprise: bool = False
    proxy_profile: str | None = None
    session_mode: bool = False

TRUSTED_CONTEXTS: dict[str, CaptchaContext] = {}
ALLOWED_HOSTS = {"staging.example.com", "portal.example.org"}


def get_trusted_context(context_id: str) -> CaptchaContext:
    context = TRUSTED_CONTEXTS.get(context_id)
    if context is None:
        raise ValueError("अज्ञात CAPTCHA संदर्भ")

    host = urlparse(context.website_url).hostname
    if host not in ALLOWED_HOSTS:
        raise PermissionError("लक्ष्य अनुमोदित होस्ट के बाहर है")

    if not context.website_key or not context.page_action:
        raise ValueError("विश्वसनीय संदर्भ में आवश्यक v3 पैरामीटर कमी है")

    return context

pip install git+https://github.com/capsolver-ai/capsolver-core.git
pip install git+https://github.com/capsolver-ai/capsolver-agent.git
pip install llama-index-core

export CAPSOLVER_API_KEY="आपकी-capsolver-api-key"

import os
from capsolver_agent.schema import create_executor

executor = create_executor(
    api_key=os.environ["CAPSOLVER_API_KEY"],
    default_timeout=120,
)

from typing import Annotated

async def solve_recaptcha_v3(
    context_id: Annotated[
        str,
        "विश्वसनीय, वर्तमान ब्राउज़र CAPTCHA संदर्भ के लिए अपार्थिक पहचानकर्ता"
    ],
) -> dict:
    """अनुमोदित ब्राउज़र संदर्भ के लिए reCAPTCHA v3 हल करें।

    जब वर्तमान वर्कफ़्लो एक समर्थित reCAPTCHA v3 चेकपॉइंट रिपोर्ट करता है, तभी इसका उपयोग करें। कभी-कभी लक्ष्य URL, साइट कुंजी या कार्य को अनुमान न लगाएं या संशोधित न करें।
    """
    context = get_trusted_context(context_id)

    captcha_type = (
        "reCaptchaV3Enterprise"
        if context.enterprise
        else "reCaptchaV3"
    )

    args = {
        "captcha_type": captcha_type,
        "website_url": context.website_url,
        "website_key": context.website_key,
        "page_action": context.page_action,
    }

    if context.proxy_profile:
        args["proxy"] = resolve_proxy(context.proxy_profile)

    result = await executor.execute("solve_captcha", args)
    if not result.get("success"):
        return {
            "success": False,
            "context_id": context_id,
            "error": normalize_error(result.get("error")),
        }

    solution = result.get("solution") or {}
    token = solution.get("token")
    if not token:
        return {
            "success": False,
            "context_id": context_id,
            "error": "समाधान में टोकन नहीं है",
        }

    receipt = await submit_solution_and_verify(
        context_id=context_id,
        token=token,
        session_cookie=extract_session_cookie(solution),
    )

    return {
        "success": receipt["verified"],
        "context_id": context_id,
        "verified": receipt["verified"],
        "next_state": receipt["next_state"],
    }

from llama_index.core.tools import FunctionTool

tool = FunctionTool.from_defaults(
    async_fn=solve_recaptcha_v3,
    name="solve_recaptcha_v3",
    description=(
        "अनुमोदित वर्तमान ब्राउज़र संदर्भ के लिए reCAPTCHA v3 हल करें। "
        "इनपुट केवल ब्राउज़र सेवा द्वारा प्रदान किया गया अपार्थिक context_id होना चाहिए। "
        "असमर्थित पृष्ठों या अनुमोदित होस्ट के लिए कभी न कॉल करें।"
    ),
)

schema = tool.metadata.get_parameters_dict()
print(schema)

from llama_index.core.agent.workflow import FunctionAgent

agent = FunctionAgent(
    llm=llm,
    tools=[tool],
    system_prompt=(
        "केवल अनुमोदित ब्राउज़र वर्कफ़्लो के साथ काम करें। जब ब्राउज़र सेवा एक समर्थित reCAPTCHA v3 चेकपॉइंट रिपोर्ट करती है, तो "
        "प्रदान किए गए context_id के साथ solve_recaptcha_v3 कॉल करें। एक बार कॉल करें। "
        "केवल जब verified=true हो, तभी आगे बढ़ें; अन्यथा समीक्षा मांगें।"
    ),
)

response = await agent.run(
    "अनुमोदित स्टेजिंग वर्कफ़्लो एक reCAPTCHA v3 चेकपॉइंट पर प्रतीक्षा कर रहा है। context_id ctx_7f19 का उपयोग करें और केवल जब verified हो तभी आगे बढ़ें।"
)

async def collect_v3_context(page, context_id: str) -> CaptchaContext:
    website_url = page.url
    host = urlparse(website_url).hostname
    if host not in ALLOWED_HOSTS:
        raise PermissionError("अनुमोदित लक्ष्य नहीं")

    values = await page.evaluate("""
    () => {
      const scripts = Array.from(document.scripts)
        .map(s => s.textContent || '')
        .join('\n');

      const siteKey =
        document.querySelector('[data-sitekey]')?.getAttribute('data-sitekey')
        || null;

      const actionMatch = scripts.match(
        /grecaptcha(?:\.enterprise)?\.execute\([^,]+,\s*\{\s*action:\s*['\"]([^'\"]+)/
      );

      return {
        siteKey,
        pageAction: actionMatch ? actionMatch[1] : null,
        enterprise: scripts.includes('grecaptcha.enterprise')
      };
    }
    """)

    if not values["siteKey"] or not values["pageAction"]:
        raise RuntimeError("आवश्यक v3 पैरामीटर पढ़ने में असमर्थ")

    return CaptchaContext(
        context_id=context_id,
        website_url=website_url,
        website_key=values["siteKey"],
        page_action=values["pageAction"],
        enterprise=values["enterprise"],
    )

SESSION_KEYS = {
    "recaptcha-ca-t",
    "recaptcha_ca_t",
}


def extract_session_cookie(solution: dict) -> str | None:
    raw = solution.get("raw") or {}
    for key in SESSION_KEYS:
        value = solution.get(key) or raw.get(key)
        if value:
            return value
    return None

async def submit_solution_and_verify(
    context_id: str,
    token: str,
    session_cookie: str | None,
) -> dict:
    browser_state = BROWSER_CONTEXTS[context_id]
    page = browser_state.page

    if session_cookie:
        await browser_state.context.add_cookies([{
            "name": "recaptcha-ca-t",
            "value": session_cookie,
            "domain": urlparse(page.url).hostname,
            "path": "/",
            "secure": True,
        }])

    await page.evaluate(
        """({ token }) => {
          let input = document.querySelector(
            'textarea[name="g-recaptcha-response"]'
          );
          if (!input) {
            input = document.createElement('textarea');
            input.name = 'g-recaptcha-response';
            input.style.display = 'none';
            document.body.appendChild(input);
          }
          input.value = token;
          input.dispatchEvent(new Event('change', { bubbles: true }));
        }""",
        {"token": token},
    )

    await trigger_trusted_callback(page, browser_state.callback_name)

    try:
        await page.locator(browser_state.success_selector).wait_for(
            state="visible",
            timeout=15000,
        )
        return {"verified": True, "next_state": "continue"}
    except Exception:
        return {"verified": False, "next_state": "operator_review"}

from enum import Enum

class RecoveryState(str, Enum):
    DETECTED = "detected"
    SOLVING = "solving"
    VERIFIED = "verified"
    REVIEW_REQUIRED = "review_required"

ATTEMPTS: dict[str, int] = {}

async def guarded_solve(context_id: str) -> dict:
    attempts = ATTEMPTS.get(context_id, 0)
    if attempts >= 1:
        return {
            "success": False,
            "context_id": context_id,
            "next_state": RecoveryState.REVIEW_REQUIRED,
            "error": "पुनर्जीवन बजट समाप्त हो गया",
        }

    ATTEMPTS[context_id] = attempts + 1
    return await solve_recaptcha_v3(context_id)

from datetime import datetime, timezone


def recovery_event(context: CaptchaContext, result: dict) -> dict:
    return {
        "event": "recaptcha_v3_recovery",
"context_id": context.context_id,
        "host": urlparse(context.website_url).hostname,
        "page_action": context.page_action,
        "enterprise": context.enterprise,
        "session_mode": context.session_mode,
        "success": result.get("success", False),
        "next_state": str(result.get("next_state")),
        "observed_at": datetime.now(timezone.utc).isoformat(),
    }