Streaming des résultats par lots : traitement des solutions CAPTCHA dès leur arrivée

Le premier token résolu d'un lot est utilisable tout de suite, sans attendre les 499 autres. Dès que l'API renvoie une solution, votre code peut soumettre le formulaire, écrire en base ou rafraîchir un tableau de bord. C'est tout l'intérêt du streaming : le premier résultat exploitable arrive après la tâche la plus rapide, pas après la plus lente.

Sur un lot réel, l'écart entre les deux se creuse vite.

Streaming ou collecte complète : comment trancher

Tout dépend d'une chose : le traitement en aval peut-il démarrer sur un résultat isolé ?

Ce que chaque approche coûte

Le micro-lot reste le compromis quand l'écriture en aval a un coût fixe.

Python : un générateur asynchrone qui produit chaque token résolu

La brique centrale est asyncio.wait(..., return_when=FIRST_COMPLETED) : le générateur rend la main à chaque résolution. Un sémaphore borne les tâches en vol.

import asyncio
import aiohttp
import time

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


async def submit_task(session, task_data):
    """Submit a single CAPTCHA task."""
    params = {
        "key": API_KEY,
        "method": task_data.get("method", "userrecaptcha"),
        "json": 1,
    }
    if params["method"] == "userrecaptcha":
        params["googlekey"] = task_data["sitekey"]
        params["pageurl"] = task_data["pageurl"]
    elif params["method"] == "turnstile":
        params["sitekey"] = task_data["sitekey"]
        params["pageurl"] = task_data["pageurl"]

    async with session.post(SUBMIT_URL, data=params) as resp:
        result = await resp.json(content_type=None)
        if result.get("status") != 1:
            return None, result.get("request", "unknown")
        return result["request"], None


async def poll_task(session, task_id, timeout=300):
    """Poll until solved or timeout."""
    start = time.monotonic()
    while time.monotonic() - start < timeout:
        await asyncio.sleep(5)
        params = {"key": API_KEY, "action": "get", "id": task_id, "json": 1}
        async with session.get(RESULT_URL, params=params) as resp:
            result = await resp.json(content_type=None)

        if result.get("request") == "CAPCHA_NOT_READY":
            continue
        if result.get("status") == 1:
            return result["request"], None
        return None, result.get("request", "unknown")

    return None, "TIMEOUT"


async def solve_one(session, index, task_data, semaphore):
    """Solve a single task within concurrency limits."""
    async with semaphore:
        start = time.monotonic()
        task_id, error = await submit_task(session, task_data)
        if error:
            return {"index": index, "status": "failed", "error": error, "time": 0}

        token, error = await poll_task(session, task_id)
        elapsed = time.monotonic() - start

        if token:
            return {"index": index, "status": "solved", "token": token, "time": round(elapsed, 1)}
        return {"index": index, "status": "failed", "error": error, "time": round(elapsed, 1)}


async def stream_results(tasks, max_concurrent=20):
    """
    Async generator that yields each result as it completes.
    Results arrive in completion order, not submission order.
    """
    semaphore = asyncio.Semaphore(max_concurrent)

    async with aiohttp.ClientSession() as session:
        pending = set()
        for i, task in enumerate(tasks):
            coro = solve_one(session, i, task, semaphore)
            pending.add(asyncio.ensure_future(coro))

        while pending:
            done, pending = await asyncio.wait(pending, return_when=asyncio.FIRST_COMPLETED)
            for future in done:
                yield future.result()


async def main():
    tasks = [
        {"sitekey": "SITE_KEY", "pageurl": f"https://example.com/page{i}"}
        for i in range(50)
    ]

    solved = 0
    failed = 0

    async for result in stream_results(tasks, max_concurrent=15):
        # Process each result immediately
        if result["status"] == "solved":
            solved += 1
            print(f"  [{solved + failed}/{len(tasks)}] Task {result['index']} SOLVED in {result['time']}s")

            # Use token immediately — don't wait for batch
            # await submit_form(result["token"])
            # await save_to_database(result)
        else:
            failed += 1
            print(f"  [{solved + failed}/{len(tasks)}] Task {result['index']} FAILED: {result['error']}")

    print(f"\nDone: {solved} solved, {failed} failed")


asyncio.run(main())

Installer la dépendance

pip install aiohttp

Le timeout de poll_task est essentiel : sans lui, une tâche jamais résolue bloquerait le générateur.

Node.js : diffuser chaque résultat avec un EventEmitter

L'équivalent idiomatique en Node.js est l'événement : la classe ci-dessous gère sa file, respecte la limite de concurrence et émet un result par tâche.

const { EventEmitter } = require("events");

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

class CaptchaStream extends EventEmitter {
  constructor(maxConcurrent = 15) {
    super();
    this.maxConcurrent = maxConcurrent;
    this.active = 0;
    this.queue = [];
    this.total = 0;
    this.completed = 0;
  }

  async submitAndPoll(index, taskData) {
    const params = new URLSearchParams({
      key: API_KEY,
      method: taskData.method || "userrecaptcha",
      googlekey: taskData.sitekey,
      pageurl: taskData.pageurl,
      json: "1",
    });

    const start = Date.now();
    const submitResp = await (await fetch(SUBMIT_URL, { method: "POST", body: params })).json();

    if (submitResp.status !== 1) {
      return { index, status: "failed", error: submitResp.request, time: 0 };
    }

    const taskId = submitResp.request;
    for (let i = 0; i < 60; i++) {
      await new Promise((r) => setTimeout(r, 5000));
      const url = `${RESULT_URL}?key=${API_KEY}&action=get&id=${taskId}&json=1`;
      const poll = await (await fetch(url)).json();

      if (poll.request === "CAPCHA_NOT_READY") continue;
      const elapsed = ((Date.now() - start) / 1000).toFixed(1);
      if (poll.status === 1) return { index, status: "solved", token: poll.request, time: elapsed };
      return { index, status: "failed", error: poll.request, time: elapsed };
    }
    return { index, status: "failed", error: "TIMEOUT", time: ((Date.now() - start) / 1000).toFixed(1) };
  }

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

    const { index, taskData } = this.queue.shift();
    this.active++;

    try {
      const result = await this.submitAndPoll(index, taskData);
      this.emit("result", result);
    } catch (err) {
      this.emit("result", { index, status: "failed", error: err.message });
    } finally {
      this.active--;
      this.completed++;

      if (this.completed === this.total) {
        this.emit("done");
      } else {
        this.processNext();
      }
    }
  }

  start(tasks) {
    this.total = tasks.length;
    this.queue = tasks.map((taskData, index) => ({ index, taskData }));

    // Launch initial batch
    const initial = Math.min(this.maxConcurrent, tasks.length);
    for (let i = 0; i < initial; i++) {
      this.processNext();
    }
    return this;
  }
}

// Usage
const tasks = Array.from({ length: 50 }, (_, i) => ({
  sitekey: "SITE_KEY",
  pageurl: `https://example.com/page${i}`,
}));

const stream = new CaptchaStream(15);
let solved = 0, failed = 0;

stream.on("result", (result) => {
  if (result.status === "solved") {
    solved++;
    console.log(`[${solved + failed}/${tasks.length}] Task ${result.index} SOLVED (${result.time}s)`);
    // Use token immediately
    // submitForm(result.token);
  } else {
    failed++;
    console.log(`[${solved + failed}/${tasks.length}] Task ${result.index} FAILED: ${result.error}`);
  }
});

stream.on("done", () => {
  console.log(`\nComplete: ${solved} solved, ${failed} failed`);
});

stream.start(tasks);

Caler la concurrence sur les threads de votre plan

CaptchaAI facture des threads, pas des résolutions : un thread correspond à un CAPTCHA en cours. Votre max_concurrent ne doit donc jamais dépasser les threads de votre plan.

• BASIC ($15/mois, 5 threads) : 5 tâches en vol.
• STANDARD ($30/mois, 15 threads) : la valeur des exemples.
• ADVANCE ($90/mois, 50 threads) : les gros lots nocturnes.

Cas concret : une agence lyonnaise vérifie chaque nuit 400 formulaires de recette (reCAPTCHA v2 et Cloudflare Turnstile) depuis un worker Scaleway à Paris ; avec ADVANCE, l'aval démarre avant la fin du lot. Côté journalisation, l'index, l'horodatage et le statut suffisent : les tokens sont éphémères, et en écrire le moins possible reste le bon réflexe RGPD.

Dépannage

FAQ

Combien de tâches puis-je lancer en parallèle sans saturer mon plan ?

Autant que votre plan compte de threads : 5 avec BASIC, 15 avec STANDARD, 50 avec ADVANCE. Les tâches en trop attendent côté API.

Comment retrouver l'ordre initial des tâches ?

Chaque résultat porte son index de soumission. Si l'aval exige l'ordre, tamponnez dans une structure triée et ne libérez que les séquences contiguës, comme un réassemblage TCP.

Que faire d'une tâche qui échoue au milieu du flux ?

Rien ne s'arrête : le résultat en échec est produit comme un autre, avec son code d'erreur. Reprogrammez la tâche dans un second lot plutôt qu'un retry bloquant.

Peut-on mélanger plusieurs types de CAPTCHA dans le même lot ?

Oui : method est porté par chaque tâche, donc un flux peut mêler reCAPTCHA v2 et Cloudflare Turnstile. Leurs temps de résolution diffèrent : le streaming y gagne d'autant plus.

Articles connexes

• Traiter les CAPTCHA en streaming avec Kafka
• Files d'attente et priorités pour les lots CAPTCHA

Prochaines étapes

Passez du lot bloquant au flux continu : récupérez votre clé API CaptchaAI et testez le générateur.

Guides associés :

• Résoudre un lot de CAPTCHA depuis un CSV
• Organiser un lot autour d'une file d'attente
• Résoudre des CAPTCHA en parallèle