Guia de integração Scrapy + CaptchaAI

Quando um spider Scrapy passa a devolver páginas de verificação em vez de HTML útil, a correção não é reescrever o crawler: é interceptar a resposta antes que ela chegue ao callback. Um downloader middleware identifica o widget de reCAPTCHA v2, envia a sitekey e a URL da página à API da CaptchaAI, aguarda o token e devolve o controle ao spider no ciclo normal do framework.

Nada depende de navegador: é requisição HTTP pura, como o Scrapy já trabalha. O caminho tem cinco passos:

1. o módulo que conversa com a API;
2. o middleware que detecta o desafio;
3. o registro em settings.py;
4. o reenvio do formulário com o token;
5. a retentativa e o dimensionamento de threads.

Por que resolver no middleware e não dentro do spider

O process_response é o único ponto por onde toda resposta baixada passa antes de virar item. Resolver ali rende três ganhos:

• Reúso — dez spiders compartilham uma implementação só.
• Testabilidade — a detecção é verificável sem subir um crawl inteiro.
• Foco — o spider continua cuidando apenas de seletores e paginação.

A prioridade 560 no DOWNLOADER_MIDDLEWARES põe o componente perto do fim da cadeia de download, onde o desafio precisa ser interceptado.

O que você precisa antes de começar

pip install scrapy requests

Passo 1: o módulo que conversa com a API

Crie captcha_solver.py na raiz do projeto. Ele encapsula o fluxo de duas etapas da API: in.php recebe a tarefa (method=userrecaptcha, googlekey e pageurl) e devolve um ID; res.php é consultado até o token ficar pronto, aguardando cinco segundos a cada CAPCHA_NOT_READY.

O reCAPTCHA v2 é resolvido em menos de 60 s, então o timeout de 300 s dá folga para picos de fila. O solve_image cobre desafios de imagem em base64.

import requests
import time


class CaptchaAISolver:
    def __init__(self, api_key):
        self.api_key = api_key
        self.base_url = "https://ocr.captchaai.com"

    def solve_recaptcha(self, site_key, page_url, timeout=300):
        resp = requests.get(f"{self.base_url}/in.php", params={
            "key": self.api_key,
            "method": "userrecaptcha",
            "googlekey": site_key,
            "pageurl": page_url,
        })

        if not resp.text.startswith("OK|"):
            raise Exception(f"Submit failed: {resp.text}")

        task_id = resp.text.split("|")[1]
        deadline = time.time() + timeout

        while time.time() < deadline:
            time.sleep(5)
            result = requests.get(f"{self.base_url}/res.php", params={
                "key": self.api_key,
                "action": "get",
                "id": task_id,
            })

            if result.text == "CAPCHA_NOT_READY":
                continue
            if result.text.startswith("OK|"):
                return result.text.split("|", 1)[1]
            raise Exception(f"Solve failed: {result.text}")

        raise TimeoutError(f"Task {task_id} timed out")

    def solve_image(self, image_base64, timeout=120):
        resp = requests.get(f"{self.base_url}/in.php", params={
            "key": self.api_key,
            "method": "base64",
            "body": image_base64,
        })

        if not resp.text.startswith("OK|"):
            raise Exception(f"Submit failed: {resp.text}")

        task_id = resp.text.split("|")[1]
        deadline = time.time() + timeout

        while time.time() < deadline:
            time.sleep(5)
            result = requests.get(f"{self.base_url}/res.php", params={
                "key": self.api_key,
                "action": "get",
                "id": task_id,
            })

            if result.text == "CAPCHA_NOT_READY":
                continue
            if result.text.startswith("OK|"):
                return result.text.split("|", 1)[1]
            raise Exception(f"Solve failed: {result.text}")

        raise TimeoutError(f"Task {task_id} timed out")

Passo 2: o middleware que detecta o desafio

Crie middlewares.py. O from_crawler lê a chave de API das settings e falha alto se ela não existir — melhor um erro na inicialização do que um crawl inteiro devolvendo lixo.

A detecção é simples de propósito: um regex procura data-sitekey e um seletor CSS procura a imagem do desafio. O token vai para request.meta, de onde o spider o recupera.

import base64
import re
from scrapy import signals
from scrapy.http import HtmlResponse
from captcha_solver import CaptchaAISolver


class CaptchaAIMiddleware:
    """Scrapy downloader middleware that detects and solves CAPTCHAs."""

    def __init__(self, api_key):
        self.solver = CaptchaAISolver(api_key)

    @classmethod
    def from_crawler(cls, crawler):
        api_key = crawler.settings.get("CAPTCHAAI_API_KEY")
        if not api_key:
            raise ValueError("CAPTCHAAI_API_KEY setting is required")
        return cls(api_key)

    def process_response(self, request, response, spider):
        # Check for reCAPTCHA on the page
        site_key = self._find_recaptcha_key(response.text)
        if site_key:
            spider.logger.info(f"reCAPTCHA detected on {response.url}")
            token = self.solver.solve_recaptcha(site_key, response.url)
            request.meta["captcha_token"] = token
            spider.logger.info("CAPTCHA solved successfully")

        # Check for image CAPTCHA
        captcha_img = self._find_image_captcha(response)
        if captcha_img:
            spider.logger.info(f"Image CAPTCHA detected on {response.url}")
            text = self.solver.solve_image(captcha_img)
            request.meta["captcha_text"] = text
            spider.logger.info(f"Image CAPTCHA solved: {text}")

        return response

    def _find_recaptcha_key(self, html):
        match = re.search(
            r'data-sitekey=["\']([A-Za-z0-9_-]+)["\']', html
        )
        return match.group(1) if match else None

    def _find_image_captcha(self, response):
        img = response.css("img#captcha-image::attr(src)").get()
        if img and img.startswith("data:image"):
            return img.split(",", 1)[1]
        return None

Passo 3: registre o middleware em settings.py

A chave de API nunca entra no repositório:

• leia sempre de variável de ambiente;
• injete o valor pelo gerenciador de segredos do orquestrador (Docker, Kubernetes ou o runner de CI).

import os

CAPTCHAAI_API_KEY = os.environ.get("CAPTCHAAI_API_KEY")

DOWNLOADER_MIDDLEWARES = {
    "myproject.middlewares.CaptchaAIMiddleware": 560,
}

Passo 4: o spider que reenvia o formulário com o token

Resolver o desafio é metade do trabalho; a outra metade é devolver o token ao site, no campo g-recaptcha-response do formulário.

Se o middleware colocou algo em response.meta, o spider reenvia a página com um FormRequest; sem token, segue o caminho normal. O CAPTCHA vira um detalhe do transporte.

import scrapy


class ProductSpider(scrapy.Spider):
    name = "products"
    start_urls = ["https://example.com/products"]

    def parse(self, response):
        # If CAPTCHA was solved, the token is in meta
        token = response.meta.get("captcha_token")
        if token:
            # Resubmit the page with the token
            yield scrapy.FormRequest(
                url=response.url,
                formdata={"g-recaptcha-response": token},
                callback=self.parse_products,
            )
        else:
            yield from self.parse_products(response)

    def parse_products(self, response):
        for product in response.css(".product-item"):
            yield {
                "name": product.css("h2::text").get(),
                "price": product.css(".price::text").get(),
                "url": response.urljoin(
                    product.css("a::attr(href)").get()
                ),
            }

        next_page = response.css("a.next-page::attr(href)").get()
        if next_page:
            yield scrapy.Request(response.urljoin(next_page))

Passo 5: nova tentativa automática nas páginas de desafio

Nem toda página de verificação traz um widget resolvível: às vezes o servidor devolve uma tela intersticial genérica.

Um segundo middleware procura marcadores conhecidos na resposta e reagenda a requisição até três vezes. Limite as retentativas — repetir sem teto só queima tempo de crawl.

class CaptchaRetryMiddleware:
    """Retry requests that return CAPTCHA challenge pages."""

    max_retries = 3

    def process_response(self, request, response, spider):
        if self._is_captcha_page(response):
            retries = request.meta.get("captcha_retries", 0)
            if retries < self.max_retries:
                request.meta["captcha_retries"] = retries + 1
                spider.logger.info(
                    f"CAPTCHA page detected, retry {retries + 1}"
                )
                return request.copy()

        return response

    def _is_captcha_page(self, response):
        indicators = [
            "g-recaptcha",
            "cf-turnstile",
            "captcha-image",
            "Please verify you are human",
        ]
        return any(ind in response.text for ind in indicators)

Rodando o crawl

Exporte a chave no shell e rode o spider.

As linhas reCAPTCHA detected no log dizem quantas páginas do domínio realmente exigem resolução — é esse número que dimensiona o plano.

export CAPTCHAAI_API_KEY="YOUR_API_KEY"
scrapy crawl products -o products.json

Threads, concorrência e o plano adequado

CONCURRENT_REQUESTS controla quantas requisições o Scrapy mantém abertas; as threads do plano controlam quantas resoluções correm em paralelo. O menor dos dois limites define o ritmo real do crawl. A cobrança da CaptchaAI é por thread simultânea, com resoluções ilimitadas por thread no mês. Na prática:

• BASIC (US$ 15/mês, 5 threads) — crawl noturno, poucos desafios por domínio.
• STANDARD (US$ 30/mês, 15 threads) — vários spiders disparados na mesma janela curta.
• ADVANCE (US$ 90/mês, 50 threads) — desafio em boa parte das páginas e horário fixo para terminar.

Cenário: monitoramento autorizado de catálogo com workers em sa-east-1

Uma equipe brasileira de dados roda, toda madrugada, um crawler que confere os preços do catálogo do próprio cliente nos ambientes de homologação: um spider por ambiente, saída em JSON, comparação com a base interna. Quando o formulário passou a exibir reCAPTCHA v2, o crawl parou de fechar na janela combinada.

Com os workers em sa-east-1, em São Paulo, o RTT fica em dezenas de milissegundos e o tempo dominante vira a resolução, não a rede: subir de 5 para 15 threads devolveu a janela sem alterar uma linha do spider. Três regras de escopo valem sempre:

• Colete apenas dados que você tem permissão para acessar.
• Considere as obrigações da LGPD quando houver dados pessoais.
• Nos testes, use https://staging.example.com/... e dados fictícios.

Solução de problemas

Perguntas frequentes

E se o site usar hCaptcha ou FunCaptcha?

Nenhum dos dois é suportado: nem hCaptcha, nem FunCaptcha (Arkose Labs). O GeeTest v4 aparece como "em breve" e também não está disponível. O padrão deste guia cobre reCAPTCHA v2 e se estende a reCAPTCHA v3, Cloudflare Turnstile, Cloudflare Challenge, GeeTest v3 e CAPTCHAs de imagem e de grade.

Por quanto tempo o token continua válido depois de resolvido?

O token do reCAPTCHA v2 tem vida curta — cerca de dois minutos —, então o FormRequest precisa sair logo depois da resolução. Usado várias páginas adiante, ele será recusado: no log isso aparece como falha de validação do formulário, não como erro de resolução.

Quantas threads eu preciso contratar?

Conte no log quantas páginas por minuto exibem desafio e compare com o tempo de resolução. Como o reCAPTCHA v2 sai em menos de 60 s, cinco threads sustentam um fluxo contínuo modesto. Cada thread traz resoluções ilimitadas no mês: dimensionar é decidir paralelismo, não volume.

O middleware derruba a velocidade do crawl inteiro?

Não. Só as respostas com desafio pagam a espera; as demais seguem no ritmo normal. Mantenha CONCURRENT_REQUESTS num valor que deixe o Scrapy baixar outras páginas durante a resolução e evite DOWNLOAD_DELAY alto com muitas retentativas.

Guias relacionados

• Como resolver CAPTCHA em web scraping com Python
• Integração da CaptchaAI com HTTPX
• Resolução assíncrona de CAPTCHA com aiohttp