recaptcha_v2_callback_proxy.py

import os
import time
import sys
from pathlib import Path
from seleniumbase import Driver
from selenium.common.exceptions import JavascriptException, TimeoutException
from selenium.webdriver.common.by import By
from selenium.webdriver.support.wait import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
from twocaptcha import TwoCaptcha

# Allow running this script from any working directory by adding the project root to sys.path
PROJECT_ROOT = Path(__file__).resolve().parents[2]
if str(PROJECT_ROOT) not in sys.path:
    sys.path.insert(0, str(PROJECT_ROOT))

from utilities.proxy_extension import proxies

# CONFIGURATION

url = "https://2captcha.com/demo/recaptcha-v2-callback"
apikey = os.getenv("APIKEY_2CAPTCHA")
proxy = {
    'type': 'HTTPS',
    'uri': 'username:password@ip:port',
}

# JavaScript script to find reCAPTCHA clients and extract sitekey and callback function
script = """
    function findRecaptchaClients() {
  // eslint-disable-next-line camelcase
  if (typeof (___grecaptcha_cfg) !== 'undefined') {
    const clients = ___grecaptcha_cfg.clients || {};
    // eslint-disable-next-line camelcase, no-undef
    return Object.entries(clients).map(([cid, client]) => {
      const data = { id: cid, version: cid >= 10000 ? 'V3' : 'V2' };
      const objects = Object.entries(client).filter(([_, value]) => value && typeof value === 'object');

      objects.forEach(([toplevelKey, toplevel]) => {
        const found = Object.entries(toplevel).find(([_, value]) => (
          value && typeof value === 'object' && 'sitekey' in value && 'size' in value
        ));

        if (typeof toplevel === 'object' && toplevel instanceof HTMLElement && toplevel['tagName'] === 'DIV'){
            data.pageurl = toplevel.baseURI;
        }

        if (found) {
          const [sublevelKey, sublevel] = found;

          data.sitekey = sublevel.sitekey;
          const callbackKey = data.version === 'V2' ? 'callback' : 'promise-callback';
          const callback = sublevel[callbackKey];
          if (!callback) {
            data.callback = null;
            data.function = null;
          } else {
            data.function = callback;
            const keys = [cid, toplevelKey, sublevelKey, callbackKey].map((key) => `['${key}']`).join('');
            data.callback = `___grecaptcha_cfg.clients${keys}`;
          }
        }
      });
      return data;
    });
  }
  return [];
}
return findRecaptchaClients()
"""

# LOCATORS

success_message_locator = "//p[contains(@class,'successMessage')]"


# GETTERS

def get_element(browser, locator):
    """
    Waits for an element to be clickable and returns it.

    This helper can be copied and reused in other projects that use SeleniumBase.
    """
    return WebDriverWait(browser, 30).until(EC.element_to_be_clickable((By.XPATH, locator)))


# ACTIONS

def parse_proxy_uri(proxy):
    """
    Parses the proxy URI to extract the scheme, login, password, IP, and port.

    Args:
        proxy (dict): Dictionary containing the proxy type and URI.
    Returns:
        tuple: A tuple containing scheme, login, password, IP, and port.
    """
    scheme = proxy['type'].lower()
    auth, address = proxy['uri'].split('@')
    login, password = auth.split(':')
    ip, port = address.split(':')
    return scheme, login, password, ip, port

def setup_proxy(proxy):
    """
    Builds a Chrome extension zip for authenticated proxy usage.

    Args:
        proxy (dict): Dictionary containing the proxy type and URI.

    Returns:
        str: Path to the generated proxy extension zip.
    """
    scheme, username, password, ip, port = parse_proxy_uri(proxy)
    proxies_extension = proxies(scheme, username, password, ip, port)
    return proxies_extension

def get_captcha_params(browser, script):
    """
    Executes the given JavaScript script to extract the captcha callback function name and sitekey.

    Args:
        script (str): The JavaScript script to execute.
    Returns:
        tuple: A tuple containing the callback function name and the sitekey.
    """
    WebDriverWait(browser, 30).until(
        lambda driver: driver.execute_script("""
            return Boolean(
                window.___grecaptcha_cfg &&
                ___grecaptcha_cfg.clients &&
                Object.keys(___grecaptcha_cfg.clients).length
            );
        """)
    )

    retries = 0
    while retries < 3:
        try:
            result = browser.execute_script(script)
            if not result:
                raise IndexError("reCAPTCHA clients list is empty")

            captcha_data = next(
                (
                    item for item in result
                    if item and item.get("sitekey") and item.get("function")
                ),
                None,
            )
            if not captcha_data:
                raise IndexError("Callback function or sitekey is not ready yet")

            callback_function_name = captcha_data['function']
            sitekey = captcha_data['sitekey']
            print("Got the callback function name and site key")
            return callback_function_name, sitekey
        except (IndexError, KeyError, TypeError, JavascriptException):
            retries += 1
            time.sleep(1)  # Wait a bit before retrying

    raise TimeoutException("Timed out waiting for reCAPTCHA callback and sitekey")

def solver_captcha(apikey, sitekey, url, proxy):
    """
    Solves the reCaptcha using the 2Captcha service.

    Args:
        apikey (str): The 2Captcha API key.
        sitekey (str): The sitekey for the captcha.
        url (str): The URL where the captcha is located.
        proxy (dict): Dictionary containing the proxy settings.
    Returns:
        str: The solved captcha code, or None if an error occurred.
    """
    solver = TwoCaptcha(apikey)
    try:
        result = solver.recaptcha(sitekey=sitekey, url=url, proxy=proxy)
        print(f"Captcha solved")
        return result['code']
    except Exception as e:
        print(f"An error occurred: {e}")
        return None

def send_token_callback(browser, callback_function, token):
    """
    Executes the callback function with the given token.

    Args:
        callback_function (str): The name of the callback function.
        token (str): The solved captcha token.
    """
    script = f"{callback_function}('{token}')"
    browser.execute_script(script)
    print("The token is sent to the callback function")

def final_message(browser, locator):
    """
    Retrieves and prints the final success message.

    Args:
        locator (str): The XPath locator of the success message.
    """
    message = get_element(browser, locator).text
    print(message)


def main():
    """
    Runs the demo flow for solving reCaptcha v2 with callback + proxy using 2Captcha.

    Helper functions (`parse_proxy_uri`, `setup_proxy`, `get_captcha_params`,
    `solver_captcha`, `send_token_callback`, etc.) are designed so they can be
    copied and reused independently.
    """
    if not apikey:
        raise RuntimeError("Set APIKEY_2CAPTCHA environment variable")

    # Generate a Chrome extension that applies authenticated proxy settings
    proxy_extension_zip = setup_proxy(proxy)

    with Driver(
        browser="chrome",
        headless=False,
        extension_zip=proxy_extension_zip,
    ) as browser:
        browser.get(url)
        print("Started")

        # Extracting callback function name and sitekey using the provided script
        callback_function, sitekey = get_captcha_params(browser, script)

        # Solving the captcha and receiving the token
        token = solver_captcha(apikey, sitekey, url, proxy)

        if token:
            # Sending the solved captcha token to the callback function
            send_token_callback(browser, callback_function, token)

            # Retrieving and printing the final success message
            final_message(browser, success_message_locator)

            # Explicit pause to observe the result
            time.sleep(5)
            print("Finished")
        else:
            print("Failed to solve captcha")


if __name__ == "__main__":
    main()