How to Solve CAPTCHA in BrowserStack with CapSolver API

import requests
import time

API_KEY = "YOUR_CAPSOLVER_API_KEY"

def solve_captcha(site_key, url):
    # Create a Capsolver task
    create_task = {
        "clientKey": API_KEY,
        "task": {
            "type": "ReCaptchaV2TaskProxyless",
            "websiteURL": url,
            "websiteKey": site_key,
        }
    }
    
    resp = requests.post("https://api.capsolver.com/createTask", json=create_task)
    task_id = resp.json().get("taskId")
    
    if not task_id:
        raise RuntimeError("Failed to create CapSolver task")
    
    # Poll for the result
    while True:
        result = requests.post(
            "https://api.capsolver.com/getTaskResult",
            json={"clientKey": API_KEY, "taskId": task_id}
        )
        payload = result.json()
        if payload.get("status") == "ready":
            return payload["solution"]["gRecaptchaResponse"]
        
        time.sleep(2)

pip install selenium playwright pyppeteer requests python-dotenv

playwright install

CAPSOLVER_API_KEY
BROWSERSTACK_USERNAME
BROWSERSTACK_ACCESS_KEY

import time
import os
import requests
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.chrome.options import Options
from dotenv import load_dotenv

# Load environment variables
load_dotenv()
API_KEY = os.getenv("CAPSOLVER_API_KEY")
BROWSERSTACK_USERNAME = os.getenv("BROWSERSTACK_USERNAME")
BROWSERSTACK_ACCESS_KEY = os.getenv("BROWSERSTACK_ACCESS_KEY")

def solve_captcha(site_key, url):
    """Create a CapSolver task and poll for the solution token."""
    create_task = {
        "clientKey": API_KEY,
        "task": {
            "type": "ReCaptchaV2TaskProxyless",
            "websiteURL": url,
            "websiteKey": site_key,
        }
    }

    print(f"Creating task with API_KEY: {API_KEY[:10]}...")
    print(f"Site key: {site_key}")
    print(f"URL: {url}")

    resp = requests.post("https://api.capsolver.com/createTask", json=create_task)
    print(f"Response status: {resp.status_code}")
    print(f"Response text: {resp.text}")
    resp.raise_for_status()
    task_id = resp.json().get("taskId")

    if not task_id:
        raise RuntimeError("Failed to create CapSolver task")

    while True:
        result = requests.post(
            "https://api.capsolver.com/getTaskResult",
            json={"clientKey": API_KEY, "taskId": task_id}
        )
        print(f"Task result status: {result.status_code}")
        print(f"Task result response: {result.text}")
        result.raise_for_status()
        payload = result.json()

        if payload.get("status") == "ready":
            return payload["solution"]["gRecaptchaResponse"]

        time.sleep(2)

def main():
    if not BROWSERSTACK_USERNAME or not BROWSERSTACK_ACCESS_KEY:
        raise SystemExit("Please set BROWSERSTACK_USERNAME and BROWSERSTACK_ACCESS_KEY environment variables")

    # BrowserStack capabilities
    bstack_options = {
        "os": "Windows",
        "osVersion": "11",
        "buildName": "Capsolver Test",
        "sessionName": "Captcha Solve",
        "local": "false",
        "debug": "true",
        "video": "true",
        "networkLogs": "true",
    }

    options = Options()
    options.set_capability("browserName", "Chrome")
    options.set_capability("browserVersion", "latest")
    options.set_capability("bstack:options", bstack_options)

    hub_url = f"https://{BROWSERSTACK_USERNAME}:{BROWSERSTACK_ACCESS_KEY}@hub-cloud.browserstack.com/wd/hub"
    driver = webdriver.Remote(command_executor=hub_url, options=options)

    # Print session info for live observation
    print(f"BrowserStack Session ID: {driver.session_id}")
    print("Watch live at: https://automate.browserstack.com/dashboard/v2/builds")

    try:
        url = "https://www.google.com/recaptcha/api2/demo"
        driver.get(url)

        site_key = driver.find_element(By.CLASS_NAME, "g-recaptcha").get_attribute("data-sitekey")
        token = solve_captcha(site_key, url)

        # Make response textarea visible and set the token
        driver.execute_script('document.getElementById("g-recaptcha-response").style.display = "block";')
        driver.execute_script(f'document.getElementById("g-recaptcha-response").value = "{token}";')
        driver.find_element(By.ID, "recaptcha-demo-submit").click()

        time.sleep(3)  # Wait for page to load

        # Inspect page after submission
        print("=" * 50)
        print("PAGE RESPONSE AFTER SUBMIT:")
        print("=" * 50)
        print(f"Current URL: {driver.current_url}")
        print(f"Page Title: {driver.title}")

        page_source = driver.page_source
        print(f"Page source length: {len(page_source)} characters")

        # Check for success or error indicators
        if any(word in page_source.lower() for word in ['success', 'verificado', 'genial', 'hooray']):
            print("✅ SUCCESS indicator found in page!")
        if "error" in page_source.lower():
            print("❌ ERROR indicator found in page!")
        if "captcha" in page_source.lower():
            print("🔄 CAPTCHA still present in page")

        # Preview first 500 characters of body text
        try:
            body = driver.find_element(By.TAG_NAME, "body")
            print(f"Body text preview:\n{body.text[:500]}")
        except Exception:
            print("Could not extract body text")

        print("=" * 50)
        time.sleep(2)

    finally:
        driver.quit()

if __name__ == "__main__":
    main()

import time
import os
import requests
import asyncio
import json
from playwright.async_api import async_playwright
from dotenv import load_dotenv

# Load environment variables
load_dotenv()
API_KEY = os.getenv("CAPSOLVER_API_KEY")
BROWSERSTACK_USERNAME = os.getenv("BROWSERSTACK_USERNAME")
BROWSERSTACK_ACCESS_KEY = os.getenv("BROWSERSTACK_ACCESS_KEY")

def solve_captcha(site_key, url):
    """
    Create a CapSolver task and poll for a solution token.
    Works the same as Selenium example.
    """
    create_task = {
        "clientKey": API_KEY,
        "task": {
            "type": "ReCaptchaV2TaskProxyless",
            "websiteURL": url,
            "websiteKey": site_key,
        }
    }

    print(f"Creating task with API_KEY: {API_KEY[:10]}...")
    print(f"Site key: {site_key}")
    print(f"URL: {url}")

    resp = requests.post("https://api.capsolver.com/createTask", json=create_task)
    print(f"Response status: {resp.status_code}")
    print(f"Response text: {resp.text}")
    resp.raise_for_status()
    task_id = resp.json().get("taskId")

    if not task_id:
        raise RuntimeError("Failed to create CapSolver task")

    # Poll until solution is ready
    while True:
        result = requests.post(
            "https://api.capsolver.com/getTaskResult",
            json={"clientKey": API_KEY, "taskId": task_id}
        )
        result.raise_for_status()
        payload = result.json()
        if payload.get("status") == "ready":
            return payload["solution"]["gRecaptchaResponse"]
        time.sleep(2)

async def run_captcha_test(platform_config):
    """Run captcha test on a specific platform configuration via BrowserStack."""
    print(f"\n🚀 Testing on: {platform_config['name']}")
    async with async_playwright() as playwright:
        # Connect to BrowserStack using CDP endpoint
        ws_endpoint = f"wss://cdp.browserstack.com/playwright?caps={platform_config['caps_string']}"
        try:
            browser = await playwright.chromium.connect_over_cdp(ws_endpoint)
            context = await browser.new_context(viewport={'width': 1280, 'height': 720})
            page = await context.new_page()
            
            print(f"✅ Connected to BrowserStack")
            print("🔗 Watch live at: https://automate.browserstack.com/dashboard/v2/builds")
            
            # Navigate to reCAPTCHA demo page
            url = "https://www.google.com/recaptcha/api2/demo"
            await page.goto(url, wait_until="domcontentloaded", timeout=30000)
            await page.wait_for_selector(".g-recaptcha", timeout=20000)
            site_key = await page.get_attribute(".g-recaptcha", "data-sitekey")
            
            print(f"🔑 Found site key: {site_key}")
            print("\n🧩 Solving captcha with CapSolver...")
            token = solve_captcha(site_key, url)
            print(f"✅ Got captcha token: {token[:50]}...")
            
            # Inject token and submit form
            await page.evaluate('document.getElementById("g-recaptcha-response").style.display = "block";')
            await page.evaluate(f'document.getElementById("g-recaptcha-response").value = "{token}";')
            await page.click("#recaptcha-demo-submit")
            await page.wait_for_timeout(5000)
            
            # Get page info and check success
            current_url = page.url
            page_title = await page.title()
            body_text = await page.text_content("body")
            
            success_indicators = ['success', 'verificado', 'genial', 'hooray']
            is_success = any(word in body_text.lower() for word in success_indicators)
            has_error = 'error' in body_text.lower()
            has_captcha = 'captcha' in body_text.lower()
            
            print("\n" + "=" * 60)
            print(f"🎯 RESULTS FOR {platform_config['name']}:")
            print("=" * 60)
            print(f"📍 URL: {current_url}")
            print(f"📄 Title: {page_title}")
            print(f"🎉 Success: {'✅ YES' if is_success else '❌ NO'}")
            if has_error:
                print("❌ ERROR indicator found!")
            if has_captcha:
                print("🔄 CAPTCHA still present!")
            print(f"\n📝 Page Preview (first 500 chars):\n{'-'*40}\n{body_text[:500]}")
            print("=" * 60)
            
            await page.wait_for_timeout(3000)
            return {'platform': platform_config['name'], 'success': is_success, 'url': current_url, 'title': page_title}
        except Exception as e:
            print(f"❌ Error on {platform_config['name']}: {str(e)}")
            return {'platform': platform_config['name'], 'success': False, 'error': str(e)}
        finally:
            try:
                await browser.close()
            except Exception as e:
                print(f"Warning: Error closing browser: {e}")
                pass

def create_browserstack_caps(platform):
    """Create BrowserStack capabilities for Playwright and return URL-encoded string."""
    import urllib.parse
    caps = {
        'browserstack.username': BROWSERSTACK_USERNAME,
        'browserstack.accessKey': BROWSERSTACK_ACCESS_KEY,
        'project': 'Playwright Capsolver Test',
        'build': 'Captcha Solve Demo',
        'name': f"Captcha Test - {platform.get('browserName', 'chrome')}",
        'browserstack.debug': 'true',
        'browserstack.console': 'info',
        'browserstack.networkLogs': 'true',
        'browserstack.timezone': 'UTC'
    }
    if 'os' in platform:
        caps['os'] = platform['os']
        caps['os_version'] = platform['osVersion']
        caps['browser'] = platform['browserName']
        caps['browser_version'] = platform['browserVersion']
    elif 'deviceName' in platform:
        caps['device'] = platform['deviceName']
        caps['os_version'] = platform['osVersion']
        caps['browser'] = platform['browserName']
        caps['real_mobile'] = 'true'
    caps_json = json.dumps(caps)
    caps_string = urllib.parse.quote(caps_json)
    return caps, caps_string

async def main():
    """Main function to run tests on multiple platforms."""
    if not API_KEY or API_KEY == "YOUR_CAPSOLVER_API_KEY":
        raise SystemExit("Please set CAPSOLVER_API_KEY environment variable")

    platforms = [
        {'name': 'Windows 11 - Chrome', 'os': 'Windows', 'osVersion': '11', 'browserName': 'chrome', 'browserVersion': 'latest'},
        {'name': 'macOS Ventura - Chrome', 'os': 'OS X', 'osVersion': 'Ventura', 'browserName': 'chrome', 'browserVersion': 'latest'}
    ]

    print("🚀 Starting Playwright + BrowserStack + CapSolver Demo...")
    results = []
    for platform in platforms:
        try:
            caps, caps_string = create_browserstack_caps(platform)
            platform['caps'] = caps
            platform['caps_string'] = caps_string
            result = await run_captcha_test(platform)
            results.append(result)
            await asyncio.sleep(2)
        except Exception as e:
            print(f"❌ Failed to test {platform['name']}: {str(e)}")
            results.append({'platform': platform['name'], 'success': False, 'error': str(e)})

    # Summary
    print("\n" + "🏆"*60)
    print("FINAL SUMMARY - ALL PLATFORMS")
    print("🏆"*60)
    successful_platforms = [r['platform'] for r in results if r['success']]
    failed_platforms = [r['platform'] for r in results if not r['success']]
    for r in results:
        if r['success']:
            print(f"✅ {r['platform']}: SUCCESS")
        else:
            print(f"❌ {r['platform']}: FAILED - {r.get('error', 'Unknown error')}")
    print(f"\n📊 Success Rate: {len(successful_platforms)}/{len(results)} platforms")
    print(f"⏰ Test completed at: {time.strftime('%Y-%m-%d %H:%M:%S')}")
    if successful_platforms:
        print(f"\n🎉 Successful platforms: {', '.join(successful_platforms)}")
    if failed_platforms:
        print(f"\n⚠️  Failed platforms: {', '.join(failed_platforms)}")
    print("🏆"*60)

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

import time
import os
import requests
import asyncio
import json
import urllib.parse
from pyppeteer import connect
from dotenv import load_dotenv

# Load environment variables
load_dotenv()
API_KEY = os.getenv("CAPSOLVER_API_KEY")
BROWSERSTACK_USERNAME = os.getenv("BROWSERSTACK_USERNAME")
BROWSERSTACK_ACCESS_KEY = os.getenv("BROWSERSTACK_ACCESS_KEY")

def solve_captcha(site_key, url):
    """
    Create a CapSolver task and poll for a solution token.
    """
    create_task = {
        "clientKey": API_KEY,
        "task": {
            "type": "ReCaptchaV2TaskProxyless",
            "websiteURL": url,
            "websiteKey": site_key,
        }
    }

    print(f"Creating task with API_KEY: {API_KEY[:10]}...")
    print(f"Site key: {site_key}")
    print(f"URL: {url}")

    resp = requests.post("https://api.capsolver.com/createTask", json=create_task)
    print(f"Response status: {resp.status_code}")
    print(f"Response text: {resp.text}")
    resp.raise_for_status()
    task_id = resp.json().get("taskId")

    if not task_id:
        raise RuntimeError("Failed to create CapSolver task")

    # Poll until solution is ready
    while True:
        result = requests.post(
            "https://api.capsolver.com/getTaskResult",
            json={"clientKey": API_KEY, "taskId": task_id}
        )
        print(f"Task result status: {result.status_code}")
        print(f"Task result response: {result.text}")
        result.raise_for_status()
        payload = result.json()
        if payload.get("status") == "ready":
            return payload["solution"]["gRecaptchaResponse"]
        time.sleep(2)


async def run_captcha_test(platform_config):
    """Run captcha test on a specific platform configuration."""
    print(f"\n🚀 Testing on: {platform_config['name']}")
    print(f"Platform details: {platform_config}")

    try:
        # Create BrowserStack WebSocket endpoint for Puppeteer
        ws_endpoint = f"wss://cdp.browserstack.com/puppeteer?caps={platform_config['caps_string']}"

        # Connect to BrowserStack using Puppeteer CDP
        browser = await connect(
            browserWSEndpoint=ws_endpoint,
            ignoreHTTPSErrors=True,
            args=['--no-sandbox', '--disable-setuid-sandbox']
        )

        # Create new page
        page = await browser.newPage()
        await page.setViewport({'width': 1280, 'height': 720})

        print(f"✅ Connected to BrowserStack via Puppeteer")
        print(f"🔗 Watch live at: https://automate.browserstack.com/dashboard/v2/builds")

        # Navigate to reCAPTCHA demo
        url = "https://www.google.com/recaptcha/api2/demo"
        await page.goto(url, {'waitUntil': 'domcontentloaded', 'timeout': 30000})
        print(f"📍 Navigated to: {url}")

        # Wait for page to load and find site key
        await page.waitForSelector(".g-recaptcha", {'timeout': 20000})
        site_key_element = await page.querySelector(".g-recaptcha")
        site_key = await page.evaluate('(element) => element.getAttribute("data-sitekey")', site_key_element)
        print(f"🔑 Found site key: {site_key}")

        # Solve captcha using CapSolver
        print("\n🧩 Solving captcha with CapSolver...")
        token = solve_captcha(site_key, url)
        print(f"✅ Got captcha token: {token[:50]}...")

        # Inject the captcha solution
        print("\n💉 Injecting captcha solution...")
        await page.evaluate('document.getElementById("g-recaptcha-response").style.display = "block";')
        await page.evaluate(f'document.getElementById("g-recaptcha-response").value = "{token}";')

        # Submit the form
        print("📤 Submitting form...")
        await page.click("#recaptcha-demo-submit")

        # Wait for response
        print("⏳ Waiting for response...")
        await asyncio.sleep(5)

        # Get results
        current_url = page.url
        page_title = await page.title()
        body_text = await page.evaluate('() => document.body.textContent')

        # Check for success indicators
        success_indicators = ['success', 'verificado', 'genial', 'hooray']
        is_success = any(word in body_text.lower() for word in success_indicators)
        has_error = 'error' in body_text.lower()
        has_captcha = 'captcha' in body_text.lower()

        # Display results
        print("\n" + "=" * 60)
        print(f"🎯 RESULTS FOR {platform_config['name']}:")
        print("=" * 60)
        print(f"📍 Current URL: {current_url}")
        print(f"📄 Page Title: {page_title}")
        print(f"⏰ Timestamp: {time.strftime('%Y-%m-%d %H:%M:%S')}")
        print(f"🎉 Success: {'✅ YES' if is_success else '❌ NO'}")
        if has_error:
            print("❌ ERROR indicator found in page!")
        if has_captcha:
            print("🔄 CAPTCHA still present in page")
        print(f"\n📝 Page Response Preview:")
        print("-" * 40)
        print(body_text[:500])
        print("=" * 60)

        if is_success:
            print(f"\n🎊 SUCCESS! Captcha solved on {platform_config['name']}!")
        else:
            print(f"\n⚠️  Something went wrong on {platform_config['name']}.")

        # Wait a bit to observe results
        await asyncio.sleep(3)

        # Mark test status in BrowserStack
        try:
            status = "passed" if is_success else "failed"
            reason = f"Captcha solved successfully on {page_title}" if is_success else "Captcha solving failed"
            await page.evaluate(f'''
                browserstack_executor: {{
                    "action": "setSessionStatus", 
                    "arguments": {{
                        "status": "{status}", 
                        "reason": "{reason}"
                    }}
                }}
            ''')
        except Exception as e:
            print(f"Warning: Could not set BrowserStack status: {e}")

        return {'platform': platform_config['name'], 'success': is_success, 'url': current_url, 'title': page_title}

    except Exception as e:
        print(f"❌ Error on {platform_config['name']}: {str(e)}")
        return {'platform': platform_config['name'], 'success': False, 'error': str(e)}

    finally:
        try:
            if 'browser' in locals():
                await browser.close()
        except Exception as e:
            print(f"Warning: Error closing browser: {e}")
            pass


def create_browserstack_caps(platform):
    """Create BrowserStack capabilities for Puppeteer."""
    caps = {
        'browserstack.username': BROWSERSTACK_USERNAME,
        'browserstack.accessKey': BROWSERSTACK_ACCESS_KEY,
        'project': 'Puppeteer Capsolver Test',
        'build': 'Captcha Solve Demo - Puppeteer',
        'name': f"Captcha Test - {platform.get('browserName', 'chrome')} (Puppeteer)",
        'browserstack.debug': 'true',
        'browserstack.console': 'info',
        'browserstack.networkLogs': 'true',
        'browserstack.timezone': 'UTC'
    }

    # Add platform-specific capabilities
    if 'os' in platform:
        caps['os'] = platform['os']
        caps['os_version'] = platform['osVersion']
        caps['browser'] = platform['browserName']
        caps['browser_version'] = platform['browserVersion']
    elif 'deviceName' in platform:
        caps['device'] = platform['deviceName']
        caps['os_version'] = platform['osVersion']
        caps['browser'] = platform['browserName']
        caps['real_mobile'] = 'true'

    # Convert to URL-encoded JSON string
    caps_json = json.dumps(caps)
    caps_string = urllib.parse.quote(caps_json)

    return caps, caps_string


async def main():
    """Main function to run tests on multiple platforms."""
    if not API_KEY or API_KEY == "YOUR_CAPSOLVER_API_KEY":
        raise SystemExit("Please set CAPSOLVER_API_KEY environment variable")

    # Define platforms to test
    platforms = [
        {'name': 'Windows 11 - Chrome (Puppeteer)', 'os': 'Windows', 'osVersion': '11', 'browserName': 'chrome', 'browserVersion': 'latest'},
        {'name': 'macOS Ventura - Chrome (Puppeteer)', 'os': 'OS X', 'osVersion': 'Ventura', 'browserName': 'chrome', 'browserVersion': 'latest'},
        {'name': 'Samsung Galaxy S23 Ultra - Chrome (Puppeteer)', 'deviceName': 'Samsung Galaxy S23 Ultra', 'browserName': 'chrome', 'osVersion': '13.0'}
    ]

    print("🚀 Starting Puppeteer + BrowserStack + CapSolver Demo...")
    print(f"📱 Testing on {len(platforms)} different platforms:")
    for i, platform in enumerate(platforms, 1):
        print(f"  {i}. {platform['name']}")

    results = []

    # Test each platform
    for platform in platforms:
        try:
            caps, caps_string = create_browserstack_caps(platform)
            platform['caps'] = caps
            platform['caps_string'] = caps_string
            result = await run_captcha_test(platform)
            results.append(result)
            await asyncio.sleep(2)
        except Exception as e:
            print(f"❌ Failed to test {platform['name']}: {str(e)}")
            results.append({'platform': platform['name'], 'success': False, 'error': str(e)})

    # Final summary
    print("\n" + "🏆"*60)
    print("FINAL SUMMARY - ALL PLATFORMS (PUPPETEER)")
    print("🏆"*60)

    successful_platforms = [r['platform'] for r in results if r['success']]
    failed_platforms = [r['platform'] for r in results if not r['success']]

    for r in results:
        if r['success']:
            print(f"✅ {r['platform']}: SUCCESS")
        else:
            print(f"❌ {r['platform']}: FAILED - {r.get('error', 'Unknown error')}")

    print(f"\n📊 Success Rate: {len(successful_platforms)}/{len(results)} platforms")
    print(f"⏰ Test completed at: {time.strftime('%Y-%m-%d %H:%M:%S')}")

    if successful_platforms:
        print(f"\n🎉 Successful platforms: {', '.join(successful_platforms)}")
    if failed_platforms:
        print(f"\n⚠️  Failed platforms: {', '.join(failed_platforms)}")

    print("🏆"*60)
    print("\n💡 Note: This demo uses pyppeteer (Python Puppeteer) with BrowserStack")
    print("   For best results, ensure your BrowserStack plan supports CDP connections")


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