एग्नो में कैप्चा हल करें कैपसॉल्वर एंटीग्रेशन के साथ

pip install agno
pip install requests

# OpenAI
from agno.models.openai import OpenAIChat

# Anthropic Claude
from agno.models.anthropic import Claude

# Google Gemini
from agno.models.google import Gemini

# और अन्य कई...

import requests
import time
from agno.tools import tool

CAPSOLVER_API_KEY = "आपका कैपसॉल्वर API कुंजी"


@tool
def solve_captcha(
    website_url: str,
    website_key: str,
    captcha_type: str = "ReCaptchaV2TaskProxyLess"
) -> str:
    """
    कैपसॉल्वर API के माध्यम से कैपचा चुनौतियां हल करता है।

    आर्ग्स:
        website_url: कैपचा वाले वेबसाइट का URL
        website_key: कैपचा का साइट कुंजी
        captcha_type: कैपचा प्रकार (ReCaptchaV2TaskProxyLess, ReCaptchaV3TaskProxyLess, AntiTurnstileTaskProxyLess)

    वापस करता है:
        कैपचा समाधान टोकन
    """
    # कार्य बनाएं
    payload = {
        "clientKey": CAPSOLVER_API_KEY,
        "task": {
            "type": captcha_type,
            "websiteURL": website_url,
            "websiteKey": website_key
        }
    }

    response = requests.post("https://api.capsolver.com/createTask", json=payload)
    result = response.json()

    if result.get("errorId") != 0:
        return f"कार्य बनाने में त्रुटि: {result.get('errorDescription')}"

    task_id = result.get("taskId")

    # परिणाम के लिए पूछताछ करें
    for _ in range(60):
        time.sleep(2)

        result = requests.post(
            "https://api.capsolver.com/getTaskResult",
            json={"clientKey": CAPSOLVER_API_KEY, "taskId": task_id}
        ).json()

        if result.get("status") == "ready":
            solution = result.get("solution", {})
            return solution.get("gRecaptchaResponse") or solution.get("token")
        elif result.get("status") == "failed":
            return f"कार्य विफल: {result.get('errorDescription')}"

    return "कैपचा समाधान के लिए प्रतीक्षा समय समाप्त हो गया"

import requests
import time
from agno.tools import tool

CAPSOLVER_API_KEY = "आपका कैपसॉल्वर API कुंजी"


@tool
def solve_recaptcha_v2(website_url: str, website_key: str) -> str:
    """
    कैपसॉल्वर के माध्यम से reCAPTCHA v2 चुनौतियां हल करता है।

    आर्ग्स:
        website_url: reCAPTCHA v2 वाले वेबसाइट का URL
        website_key: साइट कुंजी (data-sitekey लक्षण)

    वापस करता है:
        g-रिकैपचा-रिस्पॉन्स टोकन
    """
    payload = {
        "clientKey": CAPSOLVER_API_KEY,
        "task": {
            "type": "ReCaptchaV2TaskProxyLess",
            "websiteURL": website_url,
            "websiteKey": website_key
        }
    }

    # कार्य बनाएं
    response = requests.post("https://api.capsolver.com/createTask", json=payload)
    result = response.json()

    if result.get("errorId") != 0:
        return f"त्रुटि: {result.get('errorDescription')}"

    task_id = result.get("taskId")

    # परिणाम के लिए पूछताछ करें
    for attempt in range(60):
        time.sleep(2)
        result = requests.post(
            "https://api.capsolver.com/getTaskResult",
            json={"clientKey": CAPSOLVER_API_KEY, "taskId": task_id}
        ).json()

        if result.get("status") == "ready":
            return result["solution"]["gRecaptchaResponse"]
        if result.get("status") == "failed":
            return f"असफल: {result.get('errorDescription')}"

    return "समाधान के लिए प्रतीक्षा समय समाप्त हो गया"

import requests
import time
from agno.tools import tool

CAPSOLVER_API_KEY = "आपका कैपसॉल्वर API कुंजी"


@tool
def solve_recaptcha_v3(
    website_url: str,
    website_key: str,
    page_action: str = "submit",
    min_score: float = 0.7
) -> str:
    """
    अंक-आधारित सत्यापन के साथ reCAPTCHA v3 चुनौतियां हल करता है।

    आर्ग्स:
        website_url: reCAPTCHA v3 वाले वेबसाइट का URL
        website_key: reCAPTCHA की साइट कुंजी
        page_action: reCAPTCHA v3 के लिए कार्यक्रम पैरामीटर
        min_score: आवश्यक न्यूनतम स्कोर (0.1 से 0.9 तक)

    वापस करता है:
        g-रिकैपचा-रिस्पॉन्स टोकन
    """
    payload = {
        "clientKey": CAPSOLVER_API_KEY,
        "task": {
            "type": "ReCaptchaV3TaskProxyLess",
            "websiteURL": website_url,
            "websiteKey": website_key,
            "pageAction": page_action,
            "minScore": min_score
        }
    }

    # कार्य बनाएं
    response = requests.post("https://api.capsolver.com/createTask", json=payload)
    result = response.json()

    if result.get("errorId") != 0:
        return f"त्रुटि: {result.get('errorDescription')}"

    task_id = result.get("taskId")

    # परिणाम के लिए पूछताछ करें
    for attempt in range(60):
        time.sleep(2)
        result = requests.post(
            "https://api.capsolver.com/getTaskResult",
            json={"clientKey": CAPSOLVER_API_KEY, "taskId": task_id}
        ).json()

        if result.get("status") == "ready":
            return result["solution"]["gRecaptchaResponse"]
        if result.get("status") == "failed":
            return f"असफल: {result.get('errorDescription')}"

    return "समाधान के लिए प्रतीक्षा समय समाप्त हो गया"

import requests
import time
from agno.tools import tool

CAPSOLVER_API_KEY = "आपका कैपसॉल्वर API कुंजी"


@tool
def solve_turnstile(website_url: str, website_key: str) -> str:
    """
    Cloudflare Turnstile चुनौतियां हल करता है।

    आर्ग्स:
        website_url: Turnstile वाले वेबसाइट का URL
        website_key: Turnstile विजेट की साइट कुंजी

    वापस करता है:
        Turnstile टोकन
    """
    payload = {
        "clientKey": CAPSOLVER_API_KEY,
        "task": {
            "type": "AntiTurnstileTaskProxyLess",
            "websiteURL": website_url,
            "websiteKey": website_key
        }
    }

    # कार्य बनाएं
    response = requests.post("https://api.capsolver.com/createTask", json=payload)
    result = response.json()

    if result.get("errorId") != 0:
        return f"त्रुटि: {result.get('errorDescription')}"

    task_id = result.get("taskId")

    # परिणाम के लिए पूछताछ करें
    for attempt in range(60):
        time.sleep(2)
        result = requests.post(
            "https://api.capsolver.com/getTaskResult",
            json={"clientKey": CAPSOLVER_API_KEY, "taskId": task_id}
        ).json()

        if result.get("status") == "ready":
            return result["solution"]["token"]
        if result.get("status") == "failed":
            return f"असफल: {result.get('errorDescription')}"

    return "समाधान के लिए प्रतीक्षा समय समाप्त हो गया"

import requests
import time
from agno.tools import tool

CAPSOLVER_API_KEY = "आपका कैपसॉल्वर API कुंजी"


@tool
def solve_cloudflare_challenge(website_url: str, proxy: str) -> dict:
    """
    Cloudflare 5-सेकंड चुनौती पृष्ठों को हल करता है।

    आर्ग्स:
        website_url: सुरक्षित पृष्ठ का URL
        proxy: प्रॉक्सी के रूप में: http://user:pass@ip:port

    वापस करता है:
        सुरक्षित पृष्ठ तक पहुंचने के लिए कुकीज़ और user_agent के साथ शब्दकोश
    """
    payload = {
        "clientKey": CAPSOLVER_API_KEY,
        "task": {
            "type": "AntiCloudflareTask",
            "websiteURL": website_url,
            "proxy": proxy
        }
    }

    # कार्य बनाएं
    response = requests.post("https://api.capsolver.com/createTask", json=payload)
    result = response.json()

    if result.get("errorId") != 0:
        return {"error": result.get('errorDescription')}

    task_id = result.get("taskId")

    # परिणाम के लिए पूछताछ करें
    for attempt in range(60):
        time.sleep(3)
        result = requests.post(
            "https://api.capsolver.com/getTaskResult",
            json={"clientKey": CAPSOLVER_API_KEY, "taskId": task_id}
        ).json()

        if result.get("status") == "ready":
            return {
                "cookies": result["solution"]["cookies"],
                "user_agent": result["solution"]["userAgent"]
            }
        if result.get("status") == "failed":
            return {"error": result.get('errorDescription')}

    return {"error": "समाधान के लिए प्रतीक्षा समय समाप्त हो गया"}

import os
import requests
import time
from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.tools import tool

# कॉन्फ़िगरेशन
CAPSOLVER_API_KEY = "आपका कैपसॉल्वर API कुंजी"
os.environ["OPENAI_API_KEY"] = "आपका ओपनएआई API कुंजी"


@tool
def solve_recaptcha_v2(website_url: str, website_key: str) -> str:
    """
    कैपसॉल्वर के माध्यम से reCAPTCHA v2 चुनौतियां हल करता है।

    आर्ग्स:
        website_url: reCAPTCHA v2 वाले वेबसाइट का URL
        website_key: साइट कुंजी (data-sitekey लक्षण)

    वापस करता है:
        g-रिकैपचा-रिस्प
"https://api.capsolver.com/getTaskResult",
            json={"clientKey": CAPSOLVER_API_KEY, "taskId": task_id}
        ).json()

        if result.get("status") == "ready":
            return result["solution"]["token"]
        if result.get("status") == "failed":
            return f"असफल: {result.get('errorDescription')}"

    return "समय सीमा पार"


@tool
def check_capsolver_balance() -> str:
    """
    वर्तमान कैपसॉल्वर खाता बैलेंस की जांच करता है।

    लौटाता है:
        वर्तमान बैलेंस जानकारी
    """
    response = requests.post(
        "https://api.capsolver.com/getBalance",
        json={"clientKey": CAPSOLVER_API_KEY}
    )
    result = response.json()

    if result.get("errorId") != 0:
        return f"त्रुटि: {result.get('errorDescription')}"

    return f"बैलेंस: ${result.get('balance', 0):.4f}"


# वेब स्क्रैपर एजेंट बनाएं
web_scraper_agent = Agent(
    name="वेब स्क्रैपर एजेंट",
    model=OpenAIChat(id="gpt-4o"),
    tools=[solve_recaptcha_v2, solve_turnstile, check_capsolver_balance],
    description="कैप्चा चुनौतियों को स्वचालित रूप से संभालने वाला विशेषज्ञ वेब स्क्रैपर",
    instructions=[
        "आप एक वेब स्क्रैपिंग विशेषज्ञ हैं जिसके पास कैप्चा समाधान क्षमता है।",
        "कैप्चा के सामने आने पर, प्रकार की पहचान करें और उपयुक्त समाधान का उपयोग करें।",
        "रीकैप्चा v2 के लिए, वेबसाइट के URL और साइट कुंजी के साथ solve_recaptcha_v2 का उपयोग करें।",
        "टर्नस्टाइल के लिए, वेबसाइट के URL और साइट कुंजी के साथ solve_turnstile का उपयोग करें।",
        "बड़े डेटा निकालने के कार्य शुरू करने से पहले हमेशा बैलेंस की जांच करें।"
    ],
    markdown=True
)


def main():
    print("=" * 60)
    print("एग्नो + कैपसॉल्वर एकीकरण डेमो")
    print("=" * 60)

    # कार्य: एक reCAPTCHA चुनौति हल करें
    task = """
    मुझे एक reCAPTCHA v2 चुनौति हल करने की आवश्यकता है।

    वेबसाइट URL: https://www.google.com/recaptcha/api2/demo
    साइट कुंजी: 6Le-wvkSAAAAAPBMRTvw0Q4Muexq9bi0DJwx_mJ-

    कृपया इस कैप्चा को हल करें और टोकन के पहले 50 अक्षर बताएं।
    शुरू करने से पहले मेरा कैपसॉल्वर बैलेंस भी जांचें।
    """

    response = web_scraper_agent.run(task)
    print("\nएजेंट प्रतिक्रिया:")
    print(response.content)


if __name__ == "__main__":
    main()

from agno.agent import Agent
from agno.team import Team
from agno.models.openai import OpenAIChat
from agno.tools import tool
import requests
import time

CAPSOLVER_API_KEY = "YOUR_CAPSOLVER_API_KEY"


@tool
def solve_any_captcha(
    website_url: str,
    website_key: str,
    captcha_type: str = "ReCaptchaV2TaskProxyLess"
) -> str:
    """विभिन्न प्रकार के कैप्चा समाधान के लिए एकल कैप्चा सॉल्वर।"""
    payload = {
        "clientKey": CAPSOLVER_API_KEY,
        "task": {
            "type": captcha_type,
            "websiteURL": website_url,
            "websiteKey": website_key
        }
    }

    response = requests.post("https://api.capsolver.com/createTask", json=payload)
    result = response.json()

    if result.get("errorId") != 0:
        return f"त्रुटि: {result.get('errorDescription')}"

    task_id = result.get("taskId")

    for _ in range(60):
        time.sleep(2)
        result = requests.post(
            "https://api.capsolver.com/getTaskResult",
            json={"clientKey": CAPSOLVER_API_KEY, "taskId": task_id}
        ).json()

        if result.get("status") == "ready":
            solution = result.get("solution", {})
            return solution.get("gRecaptchaResponse") or solution.get("token")
        if result.get("status") == "failed":
            return f"असफल: {result.get('errorDescription')}"

    return "समय सीमा पार"


# कैप्चा विशेषज्ञ एजेंट
captcha_agent = Agent(
    name="कैप्चा विशेषज्ञ",
    model=OpenAIChat(id="gpt-4o"),
    tools=[solve_any_captcha],
    description="विभिन्न कैप्चा प्रकार की पहचान और समाधान में विशेषज्ञ",
    instructions=[
        "पृष्ठ विश्लेषण से कैप्चा प्रकार पहचानें",
        "सही पैरामीटर के साथ उपयुक्त समाधान का उपयोग करें",
        "सफलता या असफलता स्पष्ट रूप से रिपोर्ट करें"
    ]
)

# डेटा निकालने वाला एजेंट
data_agent = Agent(
    name="डेटा निकालने वाला",
    model=OpenAIChat(id="gpt-4o"),
    description="वेब पृष्ठों से डेटा निकालता है और प्रकार्य करता है",
    instructions=[
        "HTML सामग्री से संरचित डेटा निकालें",
        "आवश्यकता होने पर कैप्चा समाधान मांगें",
        "निकाला गया डेटा वैध और साफ करें"
    ]
)

# टीम बनाएं
scraping_team = Team(
    name="वेब स्क्रैपिंग टीम",
    agents=[captcha_agent, data_agent],
    description="कैप्चा हैंडलिंग के साथ वेब स्क्रैपिंग में विशेषज्ञ टीम"
)

from selenium import webdriver
from selenium.webdriver.common.by import By

def submit_recaptcha_token(driver, token: str):
    """रीकैप्चा टोकन डालें और जमा करें"""
    recaptcha_response = driver.find_element(By.ID, "g-recaptcha-response")
    driver.execute_script("arguments[0].style.display = 'block';", recaptcha_response)
    recaptcha_response.clear()
    recaptcha_response.send_keys(token)

    form = driver.find_element(By.TAG_NAME, "form")
    form.submit()

def submit_turnstile_token(driver, token: str):
    """टर्नस्टाइल टोकन डालें और जमा करें"""
    turnstile_input = driver.find_element(By.NAME, "cf-turnstile-response")
    driver.execute_script("arguments[0].value = arguments[1];", turnstile_input, token)

    form = driver.find_element(By.TAG_NAME, "form")
    form.submit()

import requests

def access_cloudflare_protected_page(url: str, cf_solution: dict):
    """क्लाउडफ़ेयर चुनौति समाधान का उपयोग करके सुरक्षित पृष्ठ तक पहुंचें।"""
    session = requests.Session()

    for cookie in cf_solution["cookies"]:
        session.cookies.set(cookie["name"], cookie["value"])

    headers = {"User-Agent": cf_solution["user_agent"]}

    response = session.get(url, headers=headers)
    return response.text

from agno.tools import tool
import time

@tool
def solve_with_retry(
    website_url: str,
    website_key: str,
    max_retries: int = 3
) -> str:
    """त्रुटि पर स्वचालित पुनर्प्रयास के साथ कैप्चा हल करें।"""
    for attempt in range(max_retries):
        try:
            result = solve_recaptcha_v2(website_url, website_key)
            if not result.startswith("त्रुटि") and not result.startswith("असफल"):
                return result
        except Exception as e:
            if attempt == max_retries - 1:
                return f"सभी पुनर्प्रयास असफल: {str(e)}"
        time.sleep(2 ** attempt)  # एक्स्पोनेंशियल बैकऑफ

    return "अधिकतम पुनर्प्रयास पार कर गए"

@tool
def check_balance() -> float:
    """कैपसॉल्वर खाता बैलेंस की जांच करें।"""
    response = requests.post(
        "https://api.capsolver.com/getBalance",
        json={"clientKey": CAPSOLVER_API_KEY}
    )
    return response.json().get("balance", 0)

import asyncio
import aiohttp
from agno.tools import tool

@tool
async def solve_captcha_async(website_url: str, website_key: str) -> str:
    """बेहतर समानांतरता के लिए एसिंक्रोनस कैप्चा सॉल्वर।"""
    async with aiohttp.ClientSession() as session:
        payload = {
            "clientKey": CAPSOLVER_API_KEY,
            "task": {
                "type": "ReCaptchaV2TaskProxyLess",
                "websiteURL": website_url,
                "websiteKey": website_key
            }
        }

        async with session.post(
            "https://api.capsolver.com/createTask",
            json=payload
        ) as response:
            result = await response.json()

        if result.get("errorId") != 0:
            return f"त्रुटि: {result.get('errorDescription')}"

        task_id = result.get("taskId")

        for _ in range(60):
            await asyncio.sleep(2)

            async with session.post(
                "https://api.capsolver.com/getTaskResult",
                json={"clientKey": CAPSOLVER_API_KEY, "taskId": task_id}
            ) as response:
                result = await response.json()

            if result.get("status") == "ready":
                solution = result.get("solution", {})
                return solution.get("gRecaptchaResponse") or solution.get("token")
            if result.get("status") == "failed":
                return f"असफल: {result.get('errorDescription')}"

        return "समय सीमा पार"