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())