Exemplo de código para resolver CAPTCHA da Amazon durante web scraping em Python:
Durante o web scraping podem surgir diversos obstáculos, como a interrupção do script devido ao aparecimento de um CAPTCHA da Amazon no site de destino. Para contornar esse problema, você pode adicionar ao seu scraper um código adicional para resolução automática do CAPTCHA, que aguardará o frame do CAPTCHA, extrairá todos os parâmetros necessários e enviará a solução para o servidor CapMonster Cloud.
Como isso pode ser implementado: suponha que exista um script de scraping de um site de clima em Python usando Selenium:
from selenium.webdriver.common.by import By
from selenium.webdriver.common.keys import Keys
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
driver = webdriver.Chrome()
try:
# Abertura da página principal de clima
driver.get('https://example.com')
# Busca do campo de entrada de cidade, inserção do nome da cidade
search_box = WebDriverWait(driver, 10).until(
EC.presence_of_element_located((By.ID, "header-location-search")) # Substitua pelo valor necessário
)
search_box.send_keys("Moscow") # Substitua pelo valor necessário
search_box.send_keys(Keys.RETURN)
# Aguarda o carregamento da página de resultados
WebDriverWait(driver, 10).until(
EC.presence_of_element_located((By.CSS_SELECTOR, 'div.locations-list') )# Substitua pelo valor necessário
)
# Seleciona e clica no primeiro resultado
first_result = driver.find_element(By.CSS_SELECTOR, 'div.locations-list a') # Substitua pelo valor necessário
first_result.click()
E nessa mesma página é necessário resolver o CAPTCHA da Amazon. Adicione o código para resolução do CAPTCHA:
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.common.keys import Keys
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
import os
import requests
import time
# API_KEY do CapMonster Cloud
API_KEY = os.getenv('CAPMONSTER_API_KEY')
CREATE_TASK_URL = 'https://api.capmonster.cloud/createTask'
GET_TASK_RESULT_URL = 'https://api.capmonster.cloud/getTaskResult'
def create_task(website_key, context, iv, challenge_script_url, captcha_script_url):
print("Criando tarefa...")
task_data = {
"clientKey": API_KEY,
"task": {
"type": "AmazonTask",
"websiteURL": 'https://example.com', # Substitua pelo valor necessário
"challengeScript": challenge_script_url,
"captchaScript": captcha_script_url,
"websiteKey": website_key,
"context": context,
"iv": iv,
"cookieSolution": False # Defina como True se precisar de cookies "aws-waf-token"
}
}
response = requests.post(CREATE_TASK_URL, json=task_data)
response_json = response.json()
if response_json['errorId'] == 0:
print(f"Tarefa criada com sucesso. Task ID: {response_json['taskId']}")
return response_json['taskId']
else:
print(f"Erro ao criar tarefa: {response_json['errorCode']}")
return None
def get_task_result(task_id):
print("Obtendo resultado da tarefa...")
result_data = {"clientKey": API_KEY, "taskId": task_id}
while True:
response = requests.post(GET_TASK_RESULT_URL, json=result_data)
response_json = response.json()
if response_json['status'] == 'ready':
print(f"Resultado da tarefa pronto: {response_json}")
return response_json
elif response_json['status'] == 'processing':
print("A tarefa ainda está sendo processada...")
time.sleep(5)
else:
print(f"Erro ao obter resultado da tarefa: {response_json['errorCode']}")
return response_json
# Inicialização do navegador
driver = webdriver.Chrome()
try:
# Abertura da página principal
print("Abrindo a página...")
driver.get('https://example.com')
# Inserção da cidade para busca
search_box = WebDriverWait(driver, 10).until(
EC.presence_of_element_located((By.ID, "header-location-search"))
)
search_box.send_keys("Moscow") # Substitua pela sua cidade
search_box.send_keys(Keys.RETURN)
# Aguarda o aparecimento do CAPTCHA
print("Aguardando iframe...")
iframe = WebDriverWait(driver, 20).until(
EC.presence_of_element_located((By.CSS_SELECTOR, 'iframe[src*="execute-api"]'))
)
driver.switch_to.frame(iframe)
print("Aguardando CAPTCHA, extraindo parâmetros...")
WebDriverWait(driver, 20).until(
EC.presence_of_element_located((By.CSS_SELECTOR, '#captcha-container'))
)
goku_props = driver.execute_script("return window.gokuProps;")
website_key = goku_props["key"]
context = goku_props["context"]
iv = goku_props["iv"]
challenge_script_url, captcha_script_url = [
driver.execute_script(f"return document.querySelector('script[src*=\"{x}\"]').src;")
for x in ("challenge.js", "captcha.js")
]
# Criação e resolução da tarefa CAPTCHA
task_id = create_task(website_key, context, iv, challenge_script_url, captcha_script_url)
if task_id:
result = get_task_result(task_id)
# Use o resultado para enviar a solução do CAPTCHA à página, se necessário
# Após resolver o CAPTCHA, continue o scraping
WebDriverWait(driver, 10).until(
EC.presence_of_element_located((By.CSS_SELECTOR, 'div.locations-list')) # Substitua pelo valor necessário
)
first_result = driver.find_element(By.CSS_SELECTOR, 'div.locations-list a') # Substitua pelo valor necessário
first_result.click()
# Demais ações
finally:
print("Fechando o navegador...")
driver.quit()
Explicação mais detalhada do código atualizado:
Importação adicional de bibliotecas requests (permite enviar requisições HTTP para interagir com a API do CapMonster Cloud) e time (usada para pausar a execução do código por um determinado tempo):
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.common.keys import Keys
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
import os
import requests
import time
Definição de variáveis que contêm a chave API para autenticação e os URLs para criação de tarefas e obtenção de resultados do CapMonster Cloud:
API_KEY = os.getenv('CAPMONSTER_API_KEY')
CREATE_TASK_URL = 'https://api.capmonster.cloud/createTask'
GET_TASK_RESULT_URL = 'https://api.capmonster.cloud/getTaskResult'
Adição da função que cria uma tarefa para resolução de CAPTCHA e envia os dados para o servidor CapMonster Cloud. Se a tarefa for criada com sucesso, ela retorna o taskId:
def create_task(website_key, context, iv, challenge_script_url, captcha_script_url):
# ... código da função
Adição da função para enviar uma requisição POST para verificar o status da tarefa. Se o status for 'ready', a função retorna o resultado; caso contrário, aguarda e verifica novamente:
def get_task_result(task_id):
# ... código da função
Após inicializar o navegador e abrir a página alvo (neste caso https://example.com como exemplo), aguarda-se o campo de busca, insere-se o nome da cidade e envia o formulário:
search_box = WebDriverWait(driver, 10).until(
EC.presence_of_element_located((By.ID, "header-location-search"))
)
search_box.send_keys("Moscow")
search_box.send_keys(Keys.RETURN)
Em seguida, aguarda-se o aparecimento do iframe do CAPTCHA, alterna-se para ele e extrai-se os parâmetros necessários para a resolução:
iframe = WebDriverWait(driver, 20).until(
EC.presence_of_element_located((By.CSS_SELECTOR, 'iframe[src*="execute-api"]'))
)
driver.switch_to.frame(iframe)
—-----
goku_props = driver.execute_script("return window.gokuProps;")
website_key = goku_props["key"]
context = goku_props["context"]
iv = goku_props["iv"]
challenge_script_url, captcha_script_url = [
driver.execute_script(f"return document.querySelector('script[src*=\"{x}\"]').src;")
for x in ("challenge.js", "captcha.js")
]
Criação da tarefa para resolução do CAPTCHA e espera pelo resultado:
task_id = create_task(website_key, context, iv, challenge_script_url, captcha_script_url)
if task_id:
result = get_task_result(task_id)
Continuação do scraping:
WebDriverWait(driver, 10).until(
EC.presence_of_element_located((By.CSS_SELECTOR, 'div.locations-list'))
)
first_result = driver.find_element(By.CSS_SELECTOR, 'div.locations-list a')
first_result.click()
# Demais ações ...
