Guide d'intégration Scrapy + CaptchaAI

Dans Scrapy, un CAPTCHA ne se traite pas dans le spider : il se traite dans un downloader middleware, entre la réponse HTTP et votre méthode parse. Ce guide construit ce middleware ligne par ligne — détection du sitekey, appel à l'API CaptchaAI, réinjection du token g-recaptcha-response — pour qu'un défi CAPTCHA au milieu d'un crawl ne fasse plus tomber la moitié de vos éléments extraits.

Le code couvre les deux types les plus courants en scraping web : reCAPTCHA v2 et les CAPTCHA image résolus par OCR. Comptez une trentaine de minutes d'intégration.

Où la résolution s'insère dans le pipeline Scrapy

Scrapy fait passer chaque réponse dans la chaîne des downloader middlewares avant de la remettre au spider. C'est le point d'accroche idéal : le middleware inspecte le HTML, et s'il y trouve un défi, il déclenche la résolution avant que votre logique d'extraction ne voie une page vide.

Le trajet complet tient en cinq temps :

1. Le middleware reçoit la réponse et cherche un attribut data-sitekey ou une image de défi.
2. S'il en trouve un, il envoie la tâche à l'endpoint in.php de CaptchaAI et récupère un identifiant.
3. Il interroge res.php toutes les 5 secondes tant que la réponse vaut CAPCHA_NOT_READY.
4. Le token obtenu est déposé dans request.meta, où le spider viendra le lire.
5. Le spider renvoie la page avec le token dans un FormRequest, puis reprend son extraction normale.

Prérequis

pip install scrapy requests

Étape 1 : le module de résolution

Créez captcha_solver.py à la racine de votre projet. Cette classe ne connaît rien à Scrapy : elle envoie une tâche, interroge le résultat et lève une exception explicite en cas d'échec — réutilisable telle quelle hors framework.

Deux méthodes suffisent : solve_recaptcha pour reCAPTCHA v2, avec method=userrecaptcha, le sitekey et l'URL de la page ; solve_image pour un CAPTCHA image envoyé en base64. Le timeout par défaut de 300 secondes laisse une marge confortable au-dessus du plafond annoncé pour reCAPTCHA v2, qui est de moins de 60 secondes.

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

Étape 2 : le middleware Scrapy

Créez ensuite middlewares.py. La méthode from_crawler récupère la clé API depuis les settings et échoue immédiatement si elle manque — mieux vaut une erreur au démarrage qu'un crawl de quatre heures qui ne ramène que des pages de défi.

process_response fait le travail : détection du data-sitekey par expression régulière, résolution, puis stockage du token dans request.meta["captcha_token"]. Les CAPTCHA image suivent le même schéma, via un sélecteur CSS.

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

Étape 3 : déclarer le middleware dans settings.py

La priorité 560 place le middleware juste après RetryMiddleware (550) : les erreurs réseau sont donc déjà traitées quand votre code reçoit la réponse. Gardez la clé API dans une variable d'environnement, jamais en dur dans le dépôt.

import os

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

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

Étape 4 : consommer le token dans le spider

Le spider reste presque inchangé. Il regarde si un token attend dans meta, renvoie la page avec ce token dans le champ g-recaptcha-response, puis enchaîne sur l'extraction et la pagination habituelles.

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

Étape 5 : relancer les pages de défi

Certains sites renvoient un interstitiel complet plutôt qu'un formulaire porteur d'un sitekey. Un second middleware repère ces pages via des marqueurs connus et rejoue la requête, plafonnée à trois tentatives pour éviter les boucles.

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)

Étape 6 : lancer le crawl

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

Dimensionner les threads pour un crawl réel

Prenons un cas concret : le catalogue d'un marchand francophone, environ 40 000 fiches produit, crawlé depuis un worker hébergé chez OVHcloud ou Scaleway. Sur ce type de site, seules les pages de recherche déclenchent un défi — mettons 3 % des URL, soit environ 1 200 résolutions pour le crawl complet.

CaptchaAI facture au thread concurrent, pas à la résolution : chaque plan inclut un nombre de threads et des résolutions illimitées par thread sur le mois. Un thread correspond à un CAPTCHA en cours ; dès qu'il aboutit, il reprend le suivant. Le dimensionnement est donc un calcul de débit.

Avec le plafond de moins de 60 secondes pour reCAPTCHA v2, ces 1 200 résolutions représentent au pire 72 000 secondes de temps de résolution cumulé. Réparties sur BASIC ($15/mois, 5 threads), elles ajoutent environ 4 heures au crawl ; sur STANDARD ($30/mois, 15 threads), un peu moins de 1 h 30. Si votre fenêtre de crawl nocturne est courte, c'est le nombre de threads qu'il faut augmenter, pas le CONCURRENT_REQUESTS de Scrapy. La facturation reste en dollars US.

Côté conformité, un crawl de catalogue ne devrait ramener que des données produit. Si vos sélecteurs captent au passage des avis nominatifs ou des identifiants clients, retirez-les dans l'item pipeline : vos obligations RGPD portent sur ce que vous stockez, pas seulement sur ce que vous affichez.

Dépannage

FAQ

Combien de threads faut-il pour un crawl Scrapy ?

Autant que de CAPTCHA à résoudre en parallèle. Le CONCURRENT_REQUESTS de Scrapy gouverne les requêtes HTTP, pas les résolutions : si dix défis tombent en même temps sur un plan à 5 threads, cinq attendent. Partez de BASIC ($15/mois, 5 threads) et montez d'un cran si la file d'attente devient visible dans vos logs.

Où stocker la clé API dans un projet Scrapy ?

Dans une variable d'environnement lue par settings.py, comme à l'étape 3. Ne la committez jamais, et utilisez une clé distincte pour vos environnements de test si vous voulez isoler la consommation.

CaptchaAI prend-il en charge hCaptcha dans ce middleware ?

Non — hCaptcha n'est pas pris en charge, pas plus que FunCaptcha. Le code ci-dessus traite reCAPTCHA v2 et les CAPTCHA image ; l'API accepte en plus reCAPTCHA v3, Cloudflare Turnstile et GeeTest v3, qu'il suffit de brancher sur le même process_response. GeeTest v4 est annoncé comme à venir ; CaptchaFox (bêta), Friendly Captcha (bêta) et Lemin (bêta) restent en phase bêta.

Comment vérifier qu'un token a réellement été accepté ?

Journalisez un marqueur de la page suivante après le FormRequest. Si un défi revient, le problème vient rarement du token : vérifiez la réputation de l'IP sortante, puis le délai écoulé entre la résolution et l'envoi.

Guides connexes

• Le scraping web en Python face aux CAPTCHA
• Brancher CaptchaAI sur un client HTTPX
• La résolution asynchrone avec aiohttp