Création de pipelines CAPTCHA clients avec CaptchaAI

Un pipeline CAPTCHA mutualisé résout un problème simple : au lieu de recoder la résolution dans chaque projet, vous exposez un seul service qui reçoit les demandes, les soumet à CaptchaAI et renvoie les tokens. Une agence ou un freelance qui gère le scraping et l'automatisation pour plusieurs clients y gagne en cohérence, en observabilité et en maîtrise des coûts. Ce guide décrit l'architecture, puis fournit une implémentation complète en Python et en Node.js que vous pouvez reprendre telle quelle.

Architecture d'un pipeline mutualisé

┌──────────────┐    ┌───────────────┐    ┌──────────────┐
│  Client A    │──▶ │               │    │              │
│  Client B    │──▶ │  Task Queue   │──▶ │  CaptchaAI   │
│  Client C    │──▶ │               │    │  API         │
└──────────────┘    └───────────────┘    └──────────────┘
                           │                    │
                           ▼                    ▼
                    ┌───────────────┐    ┌──────────────┐
                    │  Result Store │◀── │  Polling      │
                    │  (Redis/DB)   │    │  Workers      │
                    └───────────────┘    └──────────────┘

Quatre composants suffisent, et chacun a une responsabilité unique :

• La prise en charge des demandes reçoit les tâches de résolution envoyées par les scrapers de chaque client.
• La file d'attente met les tâches en tampon et applique une limite de concurrence par client.
• Les workers de résolution soumettent le défi à CaptchaAI, puis interrogent régulièrement le résultat.
• Le magasin de résultats conserve les tokens résolus jusqu'à leur récupération par le consommateur.

Cette séparation est ce qui rend le pipeline réutilisable : un nouveau client n'ajoute qu'une configuration, jamais du code. Elle permet aussi d'isoler les incidents — un client qui sature ses proxys ne ralentit pas les autres — et de mesurer précisément la consommation par projet.

Dimensionner les threads et le plan

CaptchaAI facture au thread simultané, pas au CAPTCHA résolu : 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'une résolution se termine, le thread enchaîne la suivante. Pour une agence qui démarre avec deux ou trois clients, STANDARD ($30/mois, 15 threads) laisse de la marge ; un pipeline qui traite plusieurs projets en parallèle passe vite à ADVANCE ($90/mois, 50 threads). Règle simple : la valeur max_concurrent de votre code ne doit jamais dépasser les threads de votre plan, sinon les tâches en trop reçoivent ERROR_NO_SLOT_AVAILABLE.

Le pipeline en Python

La classe de résolution

La classe ci-dessous encapsule tout le cycle de vie d'une tâche : mise en file avec enqueue, soumission avec submit_task, interrogation du résultat avec poll_result et orchestration globale avec process_queue. Le statut CAPCHA_NOT_READY n'est pas une erreur : il signale simplement que la résolution est en cours, et la boucle continue d'interroger toutes les 5 secondes jusqu'au token ou au délai maximal.

import requests
import time
from dataclasses import dataclass
from typing import Optional
from collections import deque
from threading import Lock

SUBMIT_URL = "https://ocr.captchaai.com/in.php"
RESULT_URL = "https://ocr.captchaai.com/res.php"

@dataclass
class SolveRequest:
    client_id: str
    method: str
    params: dict
    callback: Optional[callable] = None

@dataclass
class SolveResult:
    client_id: str
    task_id: str
    token: Optional[str] = None
    error: Optional[str] = None


class CaptchaPipeline:
    def __init__(self, api_key: str, max_concurrent: int = 10):
        self.api_key = api_key
        self.max_concurrent = max_concurrent
        self.queue = deque()
        self.active = {}
        self.lock = Lock()

    def enqueue(self, request: SolveRequest):
        with self.lock:
            self.queue.append(request)

    def submit_task(self, request: SolveRequest) -> Optional[str]:
        data = {
            "key": self.api_key,
            "method": request.method,
            "json": 1,
            **request.params
        }

        try:
            resp = requests.post(SUBMIT_URL, data=data, timeout=15)
            result = resp.json()

            if result.get("status") == 1:
                return result["request"]
            else:
                print(f"[{request.client_id}] Submit error: {result.get('error_text', result.get('request'))}")
                return None
        except requests.RequestException as e:
            print(f"[{request.client_id}] Network error: {e}")
            return None

    def poll_result(self, task_id: str, max_wait: int = 120) -> Optional[str]:
        elapsed = 0
        interval = 5
        while elapsed < max_wait:
            time.sleep(interval)
            elapsed += interval

            try:
                resp = requests.get(RESULT_URL, params={
                    "key": self.api_key,
                    "action": "get",
                    "id": task_id,
                    "json": 1
                }, timeout=10)
                result = resp.json()

                if result.get("status") == 1:
                    return result["request"]
                elif result.get("request") == "CAPCHA_NOT_READY":
                    continue
                else:
                    print(f"Poll error for {task_id}: {result.get('error_text', result.get('request'))}")
                    return None
            except requests.RequestException:
                continue

        return None

    def process_queue(self):
        while self.queue or self.active:
            # Fill active slots
            with self.lock:
                while self.queue and len(self.active) < self.max_concurrent:
                    request = self.queue.popleft()
                    task_id = self.submit_task(request)
                    if task_id:
                        self.active[task_id] = request

            # Poll active tasks
            completed = []
            for task_id, request in list(self.active.items()):
                token = self.poll_result(task_id, max_wait=10)
                if token:
                    result = SolveResult(
                        client_id=request.client_id,
                        task_id=task_id,
                        token=token
                    )
                    if request.callback:
                        request.callback(result)
                    completed.append(task_id)

            with self.lock:
                for task_id in completed:
                    del self.active[task_id]

Soumettre les tâches de plusieurs clients

Chaque client déclare son propre type de CAPTCHA et sa cible. Le callback est déclenché dès qu'un token est disponible, ce qui permet à chaque projet de traiter ses résultats indépendamment sans bloquer les autres.

pipeline = CaptchaPipeline(api_key="YOUR_API_KEY", max_concurrent=15)

# Client A — reCAPTCHA v2
pipeline.enqueue(SolveRequest(
    client_id="client_a",
    method="userrecaptcha",
    params={
        "googlekey": "6Le-SITEKEY-A",
        "pageurl": "https://client-a-target.com/form"
    },
    callback=lambda r: print(f"[{r.client_id}] Solved: {r.token[:40]}...")
))

# Client B — Turnstile
pipeline.enqueue(SolveRequest(
    client_id="client_b",
    method="turnstile",
    params={
        "sitekey": "0x4AAAA-SITEKEY-B",
        "pageurl": "https://client-b-target.com/login"
    },
    callback=lambda r: print(f"[{r.client_id}] Solved: {r.token[:40]}...")
))

pipeline.process_queue()

Ici, client_a résout reCAPTCHA v2 (méthode userrecaptcha) et client_b résout Cloudflare Turnstile (méthode turnstile) — deux types pris en charge nativement par CaptchaAI, servis par le même pipeline.

Le pipeline en Node.js

La version Node.js repose sur des promesses : enqueue renvoie une promesse résolue avec le token, et activeCount sert de mécanisme de contre-pression pour ne jamais dépasser la concurrence autorisée. C'est le choix naturel si vos scrapers clients sont déjà en JavaScript.

const axios = require("axios");

const SUBMIT_URL = "https://ocr.captchaai.com/in.php";
const RESULT_URL = "https://ocr.captchaai.com/res.php";

class CaptchaPipeline {
  constructor(apiKey, maxConcurrent = 10) {
    this.apiKey = apiKey;
    this.maxConcurrent = maxConcurrent;
    this.queue = [];
    this.activeCount = 0;
  }

  enqueue(clientId, method, params) {
    return new Promise((resolve, reject) => {
      this.queue.push({ clientId, method, params, resolve, reject });
      this._processNext();
    });
  }

  async _processNext() {
    if (this.activeCount >= this.maxConcurrent || this.queue.length === 0) return;

    this.activeCount++;
    const task = this.queue.shift();

    try {
      const token = await this._solve(task);
      task.resolve({ clientId: task.clientId, token });
    } catch (err) {
      task.reject(err);
    } finally {
      this.activeCount--;
      this._processNext();
    }
  }

  async _solve(task) {
    const submitResp = await axios.post(SUBMIT_URL, null, {
      params: {
        key: this.apiKey,
        method: task.method,
        json: 1,
        ...task.params,
      },
      timeout: 15000,
    });

    if (submitResp.data.status !== 1) {
      throw new Error(submitResp.data.error_text || submitResp.data.request);
    }

    const taskId = submitResp.data.request;
    return this._poll(taskId);
  }

  async _poll(taskId, maxWait = 120000) {
    const interval = 5000;
    let elapsed = 0;

    while (elapsed < maxWait) {
      await new Promise((r) => setTimeout(r, interval));
      elapsed += interval;

      try {
        const resp = await axios.get(RESULT_URL, {
          params: {
            key: this.apiKey,
            action: "get",
            id: taskId,
            json: 1,
          },
          timeout: 10000,
        });

        if (resp.data.status === 1) return resp.data.request;
        if (resp.data.request !== "CAPCHA_NOT_READY") {
          throw new Error(resp.data.error_text || resp.data.request);
        }
      } catch (err) {
        if (err.response) throw err;
      }
    }

    throw new Error(`Timeout waiting for task ${taskId}`);
  }
}

// Usage
(async () => {
  const pipeline = new CaptchaPipeline("YOUR_API_KEY", 15);

  const results = await Promise.allSettled([
    pipeline.enqueue("client_a", "userrecaptcha", {
      googlekey: "6Le-SITEKEY-A",
      pageurl: "https://client-a-target.com/form",
    }),
    pipeline.enqueue("client_b", "turnstile", {
      sitekey: "0x4AAAA-SITEKEY-B",
      pageurl: "https://client-b-target.com/login",
    }),
  ]);

  results.forEach((r) => {
    if (r.status === "fulfilled") {
      console.log(`[${r.value.clientId}] Token: ${r.value.token.slice(0, 40)}...`);
    } else {
      console.error(`Failed: ${r.reason.message}`);
    }
  });
})();

Configuration par client

Isolez les paramètres propres à chaque client — proxy, méthode de résolution par défaut, plafond de concurrence — dans un dictionnaire de configuration plutôt que dans le code. Ajouter un client revient alors à ajouter une entrée.

CLIENT_CONFIG = {
    "client_a": {
        "proxy": "host:port:user:pass",
        "proxytype": "HTTP",
        "max_concurrent": 5,
        "default_method": "userrecaptcha"
    },
    "client_b": {
        "proxy": None,
        "proxytype": None,
        "max_concurrent": 10,
        "default_method": "turnstile"
    }
}

def build_params(client_id, params):
    config = CLIENT_CONFIG.get(client_id, {})
    if config.get("proxy"):
        params["proxy"] = config["proxy"]
        params["proxytype"] = config["proxytype"]
    return params

Isolation et conformité RGPD

Un pipeline multi-clients traite des données pour le compte de tiers, donc le cloisonnement n'est pas optionnel. Préfixez les clés du magasin de résultats par client_id, séparez les logs, et attribuez des proxys distincts quand le contrat l'exige. Côté journalisation, appliquez le principe de minimisation : un identifiant de tâche et un statut suffisent au débogage. Évitez de stocker les URL cibles complètes ou des données personnelles ; cette hygiène vous aligne sur vos obligations RGPD et simplifie vos audits, un point sensible pour les clients européens.

Gérer les erreurs de l'API

La robustesse d'un pipeline tient à la façon dont il réagit aux codes d'erreur renvoyés par CaptchaAI. Traitez-les au niveau du worker, pas du scraper client :

Dépannage

FAQ

Comment empêcher que le token d'un client se retrouve chez un autre ?

Indexez chaque résultat par la combinaison client_id + task_id et ne partagez jamais une clé de stockage entre projets. Avec un magasin commun (Redis ou base de données), préfixez les clés par le client_id : c'est la source la plus fréquente de fuites de tokens dans les pipelines mutualisés.

Quel plan CaptchaAI choisir pour un pipeline multi-clients ?

Cela dépend du nombre de résolutions simultanées, pas du volume total. Comme la facturation est au thread, additionnez la concurrence de tous vos clients : STANDARD ($30/mois, 15 threads) convient à un démarrage, ADVANCE ($90/mois, 50 threads) à plusieurs projets actifs en parallèle. Gardez max_concurrent sous le nombre de threads de votre plan.

Comment reprendre une file d'attente après un redémarrage ?

Persistez la file dans Redis ou une base de données plutôt qu'en mémoire. Au redémarrage, rechargez les tâches en attente et relancez process_queue. Sans persistance, un simple redéploiement fait perdre toutes les résolutions en cours et oblige les scrapers clients à resoumettre.

CaptchaAI gère-t-il bien une forte concurrence ?

Oui, l'API accepte une concurrence élevée dans la limite des threads de votre plan. Dans la pratique, le goulot d'étranglement vient plutôt de votre pool de proxys ou de la vitesse de vos workers de polling. Commencez avec une concurrence de 5 à 10 par client, mesurez les temps de résolution et le taux de réussite, puis ajustez.

Lancez votre pipeline CAPTCHA avec CaptchaAI

Créez votre compte et commencez à assembler vos pipelines clients sur la plateforme CaptchaAI.

Guides associés

• La résolution de CAPTCHA en parallèle
• Mettre en place une logique de nouvelle tentative
• Le traitement distribué avec une file Redis
• Le script de surveillance de l'état de santé