Cách gi?i quy?t CAPTCHA trên BrowserStack v?i API CapSolver

import requests
import time

API_KEY = "YOUR_CAPSOLVER_API_KEY"

def solve_captcha(site_key, url):
    # T?o tác v? Capsolver
    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("Kh?ng th? t?o tác v? CapSolver")
    
    # Ki?m tra k?t qu?
    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 bi?n m?i tr??ng
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):
    """T?o tác v? CapSolver và ki?m tra m? th?ng báo gi?i pháp."""
    create_task = {
        "clientKey": API_KEY,
        "task": {
            "type": "ReCaptchaV2TaskProxyless",
            "websiteURL": url,
            "websiteKey": site_key,
        }
    }

    print(f"T?o tác v? v?i API_KEY: {API_KEY[:10]}...")
    print(f"Khóa trang web: {site_key}")
    print(f"URL: {url}")

    resp = requests.post("https://api.capsolver.com/createTask", json=create_task)
    print(f"Tr?ng thái ph?n h?i: {resp.status_code}")
    print(f"V?n b?n ph?n h?i: {resp.text}")
    resp.raise_for_status()
    task_id = resp.json().get("taskId")

    if not task_id:
        raise RuntimeError("Kh?ng th? t?o tác v? CapSolver")

    while True:
        result = requests.post(
            "https://api.capsolver.com/getTaskResult",
            json={"clientKey": API_KEY, "taskId": task_id}
        )
        print(f"Tr?ng thái k?t qu? tác v?: {result.status_code}")
        print(f"Ph?n h?i k?t qu? tác v?: {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("Vui lòng ??t bi?n m?i tr??ng BROWSERSTACK_USERNAME và BROWSERSTACK_ACCESS_KEY")

    # Kh? n?ng BrowserStack
    bstack_options = {
        "os": "Windows",
        "osVersion": "11",
        "buildName": "Ki?m tra Capsolver",
        "sessionName": "Gi?i quy?t Captcha",
        "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)

    # In th?ng tin phiên ?? quan sát tr?c ti?p
    print(f"ID phiên BrowserStack: {driver.session_id}")
    print("Xem tr?c ti?p t?i: 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)

        # Làm cho vùng v?n b?n ph?n h?i hi?n th? và ??t m? th?ng báo
        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)  # Ch? trang t?i

        # Ki?m tra trang sau khi g?i
        print("=" * 50)
        print("PH?N H?I TRANG SAU KHI G?I:")
        print("=" * 50)
        print(f"URL hi?n t?i: {driver.current_url}")
        print(f"Tiêu ?? trang: {driver.title}")

        page_source = driver.page_source
        print(f"?? dài ngu?n trang: {len(page_source)} k? t?")

        # Ki?m tra xem có ch? báo thành c?ng hay l?i kh?ng
        if any(word in page_source.lower() for word in ['success', 'verificado', 'genial', 'hooray']):
            print("? Tìm th?y ch? báo TH?NH C?NG trong trang!")
        if "error" in page_source.lower():
            print("? Tìm th?y ch? báo L?I trong trang!")
        if "captcha" in page_source.lower():
            print("? CAPTCHA v?n còn trong trang")

        # Xem tr??c 500 k? t? ??u tiên c?a v?n b?n n?i dung
        try:
            body = driver.find_element(By.TAG_NAME, "body")
            print(f"Xem tr??c v?n b?n n?i dung:\n{body.text[:500]}")
        except Exception:
            print("Kh?ng th? trích xu?t v?n b?n n?i dung")

        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 bi?n m?i tr??ng
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):
    """
    T?o tác v? CapSolver và ki?m tra m? th?ng báo gi?i pháp.
    Ho?t ??ng gi?ng nh? ví d? Selenium.
    """
    create_task = {
        "clientKey": API_KEY,
        "task": {
            "type": "ReCaptchaV2TaskProxyless",
            "websiteURL": url,
            "websiteKey": site_key,
        }
    }

    print(f"T?o tác v? v?i API_KEY: {API_KEY[:10]}...")
    print(f"Khóa trang web: {site_key}")
    print(f"URL: {url}")

    resp = requests.post("https://api.capsolver.com/createTask", json=create_task)
    print(f"Tr?ng thái ph?n h?i: {resp.status_code}")
    print(f"V?n b?n ph?n h?i: {resp.text}")
    resp.raise_for_status()
    task_id = resp.json().get("taskId")

    if not task_id:
        raise RuntimeError("Kh?ng th? t?o tác v? CapSolver")

    # Ki?m tra cho ??n khi gi?i pháp s?n sàng
    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):
    """Ch?y ki?m tra captcha trên c?u hình n?n t?ng c? th? th?ng qua BrowserStack."""
    print(f"\n? Ki?m tra trên: {platform_config['name']}")
    async with async_playwright() as playwright:
        # K?t n?i v?i BrowserStack b?ng ?i?m cu?i CDP
        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"? ?? k?t n?i v?i BrowserStack")
            print("? Xem tr?c ti?p t?i: https://automate.browserstack.com/dashboard/v2/builds")
            
            # ?i?u h??ng ??n trang demo reCAPTCHA
            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"? Tìm th?y khóa trang web: {site_key}")
            print("\n? Gi?i quy?t captcha b?ng CapSolver...")
            token = solve_captcha(site_key, url)
            print(f"? ?? l?y m? th?ng báo captcha: {token[:50]}...")
            
            # Tiêm m? th?ng báo và g?i bi?u m?u
            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)

#!/usr/bin/env python3
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)

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

print(f"? Bài ki?m tra hoàn t?t lúc: {time.strftime('%Y-%m-%d %H:%M:%S')}")

    if successful_platforms:
        print(f"\n? N?n t?ng thành c?ng: {', '.join(successful_platforms)}")
    if failed_platforms:
        print(f"\n?? N?n t?ng th?t b?i: {', '.join(failed_platforms)}")

    print("?"*60)
    print("\n? L?u ?: B?n demo này s? d?ng pyppeteer (Python Puppeteer) v?i BrowserStack")
    print("   ?? có k?t qu? t?t nh?t, h?y ??m b?o gói BrowserStack c?a b?n h? tr? k?t n?i CDP")


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