Redis para la gestión y el almacenamiento en caché de TTL del token CAPTCHA

Cachear tokens CAPTCHA en Redis solo compensa cuando resuelves muchos desafíos contra el mismo sitekey y necesitas respuestas casi inmediatas. En ese escenario, el TTL nativo de Redis encaja de forma natural: cada token se guarda con una caducidad que refleja su vida útil real —los tokens de reCAPTCHA aguantan entre 90 y 120 segundos y los de Cloudflare Turnstile unos 300— y desaparece solo antes de que el sitio de destino lo rechace. Un mismo lock de deduplicación, además, impide que dos workers gasten esfuerzo resolviendo el mismo CAPTCHA a la vez.

Piensa en un caso concreto: monitorizas precios en un marketplace regional y varios workers golpean la misma página protegida. Sin caché, cada uno vuelve a resolver el mismo reCAPTCHA; con una caché compartida en Redis, el primero resuelve y el resto reutiliza el token mientras siga vivo.

¿Cuándo conviene cachear tokens CAPTCHA?

No siempre. Si resuelves un CAPTCHA aislado y lo consumes al instante, la caché solo añade piezas móviles sin ganancia real. El patrón brilla en tres situaciones muy concretas:

• Alta concurrencia sobre el mismo sitekey: varios workers atacan el mismo objetivo y no quieres pagar la misma resolución tantas veces.
• Respuestas casi inmediatas: un pool caliente evita esperar los segundos que tarda cada resolución nueva.
• Deduplicación: un lock impide que dos procesos resuelvan el mismo CAPTCHA en paralelo.

Si tu caso no es ninguno de estos, quédate con la resolución directa y ahórrate la complejidad.

Cómo viaja un token por Redis

Solve Request → Check Redis → Cache Hit?
                    │               │
                    │ No            │ Yes → Return cached token
                    ▼               
              CaptchaAI API
                    │
                    ▼
              Store in Redis (TTL = token_lifetime - safety_margin)
                    │
                    ▼
              Return token

Diseño de las claves en Redis

Antes de escribir código, conviene fijar el esquema de claves. Cada tipo de dato vive bajo un prefijo propio y siempre lleva un TTL, para que Redis limpie por su cuenta lo que caduca.

Resolución con caché en Python

Conexión y márgenes de TTL

Arranca configurando el cliente de Redis y el mapa de TTL. Cada valor deja un colchón por debajo de la vida útil real del token, para no servir nunca uno que esté a punto de caducar.

import os
import time
import json
import redis
import requests

r = redis.Redis(
    host=os.environ.get("REDIS_HOST", "localhost"),
    port=int(os.environ.get("REDIS_PORT", 6379)),
    db=0,
    decode_responses=True
)

API_KEY = os.environ["CAPTCHAAI_API_KEY"]

# TTLs with safety margin (seconds before actual expiration)
TOKEN_TTLS = {
    "recaptcha_v2": 80,     # Actual: ~120s, cache for 80s
    "recaptcha_v3": 80,
    "hcaptcha": 80,
    "turnstile": 250,       # Actual: ~300s, cache for 250s
}

Leer y escribir tokens en la caché

Estas dos funciones encapsulan el acceso a Redis: una recupera un token comprobando que aún le quede TTL útil, y la otra guarda un token resuelto con la caducidad que corresponde a su tipo.

def cache_key(sitekey, pageurl):
    """Generate Redis key for a specific CAPTCHA target."""
    return f"captcha:token:{sitekey}:{pageurl}"


def get_cached_token(sitekey, pageurl):
    """Pop a cached token from the queue."""
    key = cache_key(sitekey, pageurl)
    token = r.lpop(key)
    if token:
        # Verify TTL still valid on the list
        ttl = r.ttl(key)
        if ttl > 10:  # At least 10 seconds remaining
            return token
    return None


def cache_token(sitekey, pageurl, token, captcha_type="recaptcha_v2"):
    """Push a solved token to the cache with appropriate TTL."""
    key = cache_key(sitekey, pageurl)
    ttl = TOKEN_TTLS.get(captcha_type, 80)
    r.rpush(key, token)
    r.expire(key, ttl)

El flujo completo con lock de deduplicación

El flujo tiene tres pasos y, si otro worker ya sostiene el lock, este espera al token en vez de duplicar la resolución:

1. Mirar la caché por si el token ya existe.
2. Tomar el lock de deduplicación si nadie más está resolviendo.
3. Llamar a la API de CaptchaAI y guardar el resultado con su TTL.

def solve_recaptcha(sitekey, pageurl, captcha_type="recaptcha_v2"):
    """Solve reCAPTCHA with Redis cache check."""
    # 1. Check cache
    cached = get_cached_token(sitekey, pageurl)
    if cached:
        return {"solution": cached, "source": "cache"}

    # 2. Check if solve is already in progress (dedup)
    lock_key = f"captcha:lock:{sitekey}:{pageurl}"
    if not r.set(lock_key, "1", nx=True, ex=120):
        # Another worker is solving — wait for result
        for _ in range(60):
            time.sleep(2)
            cached = get_cached_token(sitekey, pageurl)
            if cached:
                return {"solution": cached, "source": "cache_wait"}
        return {"error": "TIMEOUT_WAITING_FOR_OTHER_WORKER"}

    try:
        # 3. Solve via CaptchaAI
        resp = requests.post("https://ocr.captchaai.com/in.php", data={
            "key": API_KEY,
            "method": "userrecaptcha",
            "googlekey": sitekey,
            "pageurl": pageurl,
            "json": 1
        })
        data = resp.json()
        if data.get("status") != 1:
            return {"error": data.get("request")}

        captcha_id = data["request"]

        for _ in range(60):
            time.sleep(5)
            result = requests.get("https://ocr.captchaai.com/res.php", params={
                "key": API_KEY, "action": "get",
                "id": captcha_id, "json": 1
            }).json()

            if result.get("status") == 1:
                token = result["request"]
                cache_token(sitekey, pageurl, token, captcha_type)
                return {"solution": token, "source": "api"}

            if result.get("request") != "CAPCHA_NOT_READY":
                return {"error": result.get("request")}

        return {"error": "TIMEOUT"}
    finally:
        r.delete(lock_key)

Pool de tokens para picos de tráfico

Cuando un objetivo recibe mucho tráfico, mantén un pool de tokens ya resueltos y listos para consumir. Un hilo en segundo plano rellena la lista en cuanto baja del tamaño deseado, de modo que casi siempre tengas un token válido a mano dentro de su TTL.

import threading


class TokenPool:
    def __init__(self, sitekey, pageurl, pool_size=5, captcha_type="recaptcha_v2"):
        self.sitekey = sitekey
        self.pageurl = pageurl
        self.pool_size = pool_size
        self.captcha_type = captcha_type
        self.pool_key = f"captcha:pool:{sitekey}:{pageurl}"
        self._running = False

    def start(self):
        self._running = True
        thread = threading.Thread(target=self._refill_loop, daemon=True)
        thread.start()

    def stop(self):
        self._running = False

    def _refill_loop(self):
        while self._running:
            current = r.llen(self.pool_key)
            if current < self.pool_size:
                self._solve_and_add()
            time.sleep(2)

    def _solve_and_add(self):
        resp = requests.post("https://ocr.captchaai.com/in.php", data={
            "key": API_KEY,
            "method": "userrecaptcha",
            "googlekey": self.sitekey,
            "pageurl": self.pageurl,
            "json": 1
        })
        data = resp.json()
        if data.get("status") != 1:
            return

        captcha_id = data["request"]
        for _ in range(60):
            time.sleep(5)
            result = requests.get("https://ocr.captchaai.com/res.php", params={
                "key": API_KEY, "action": "get",
                "id": captcha_id, "json": 1
            }).json()

            if result.get("status") == 1:
                ttl = TOKEN_TTLS.get(self.captcha_type, 80)
                r.rpush(self.pool_key, result["request"])
                r.expire(self.pool_key, ttl)
                return
            if result.get("request") != "CAPCHA_NOT_READY":
                return

    def get_token(self):
        return r.lpop(self.pool_key)


# Usage
pool = TokenPool("6Le-wvkSAAAAAPBMRTvw0Q4Muexq9bi0DJwx_mJ-", "https://example.com")
pool.start()

# When you need a token:
token = pool.get_token()

La misma lógica en Node.js

La misma lógica se traslada a Node.js con ioredis y axios: comprueba la caché, toma el lock de deduplicación y, si te toca resolver, envía la tarea y sondea el resultado antes de guardarlo con su TTL.

const Redis = require("ioredis");
const axios = require("axios");

const redis = new Redis(process.env.REDIS_URL || "redis://localhost:6379");
const API_KEY = process.env.CAPTCHAAI_API_KEY;

const TOKEN_TTLS = { recaptcha_v2: 80, recaptcha_v3: 80, hcaptcha: 80, turnstile: 250 };

function cacheKey(sitekey, pageurl) {
  return `captcha:token:${sitekey}:${pageurl}`;
}

async function getCachedToken(sitekey, pageurl) {
  const key = cacheKey(sitekey, pageurl);
  const token = await redis.lpop(key);
  if (token) {
    const ttl = await redis.ttl(key);
    if (ttl > 10) return token;
  }
  return null;
}

async function solveWithCache(sitekey, pageurl, type = "recaptcha_v2") {
  // Check cache
  const cached = await getCachedToken(sitekey, pageurl);
  if (cached) return { solution: cached, source: "cache" };

  // Dedup lock
  const lockKey = `captcha:lock:${sitekey}:${pageurl}`;
  const locked = await redis.set(lockKey, "1", "NX", "EX", 120);
  if (!locked) {
    for (let i = 0; i < 60; i++) {
      await new Promise((r) => setTimeout(r, 2000));
      const waitCached = await getCachedToken(sitekey, pageurl);
      if (waitCached) return { solution: waitCached, source: "cache_wait" };
    }
    return { error: "TIMEOUT_WAITING" };
  }

  try {
    const submit = await axios.post("https://ocr.captchaai.com/in.php", null, {
      params: { key: API_KEY, method: "userrecaptcha", googlekey: sitekey, pageurl, json: 1 },
    });
    if (submit.data.status !== 1) return { error: submit.data.request };

    const captchaId = submit.data.request;
    for (let i = 0; i < 60; i++) {
      await new Promise((r) => setTimeout(r, 5000));
      const poll = await axios.get("https://ocr.captchaai.com/res.php", {
        params: { key: API_KEY, action: "get", id: captchaId, json: 1 },
      });

      if (poll.data.status === 1) {
        const key = cacheKey(sitekey, pageurl);
        const ttl = TOKEN_TTLS[type] || 80;
        await redis.rpush(key, poll.data.request);
        await redis.expire(key, ttl);
        return { solution: poll.data.request, source: "api" };
      }
      if (poll.data.request !== "CAPCHA_NOT_READY") return { error: poll.data.request };
    }
    return { error: "TIMEOUT" };
  } finally {
    await redis.del(lockKey);
  }
}

Diagnóstico de problemas habituales

Medir la tasa de aciertos de la caché

Para saber si la caché merece la pena, mide la tasa de aciertos. Redis expone contadores de keyspace_hits y keyspace_misses que, combinados, te dan el porcentaje de solicitudes servidas sin volver a resolver.

def cache_stats():
    info = r.info("stats")
    hits = info.get("keyspace_hits", 0)
    misses = info.get("keyspace_misses", 0)
    total = hits + misses
    return {
        "hit_rate": f"{hits / total * 100:.1f}%" if total else "0%",
        "hits": hits,
        "misses": misses,
        "active_keys": r.dbsize()
    }

Buenas prácticas para el TTL en producción

• Ajusta el margen por tipo: reCAPTCHA y Cloudflare Turnstile caducan a ritmos distintos, así que evita un único valor global.
• Trata el token como de un solo uso salvo que el destino demuestre lo contrario; ante la duda, resuelve de nuevo.
• Vigila el consumo de memoria: un TTL en cada clave mantiene la caché acotada sin intervención manual.
• Respeta los términos de servicio y la normativa de protección de datos aplicable cuando trabajes con datos de terceros.

Preguntas frecuentes

¿Cachear tokens reduce el consumo de threads en mi plan de CaptchaAI?

Indirectamente, sí. CaptchaAI factura por thread concurrente con resoluciones ilimitadas, así que cada resolución que evitas con la caché libera ese thread para otra tarea. Dos matices importantes:

• No crea capacidad nueva: solo elimina trabajo redundante contra el mismo sitekey.
• Con menos resoluciones repetidas, un plan pequeño como BASIC ($15/mes, 5 threads) cubre más volumen antes de que necesites escalar.

¿Qué margen de seguridad debo dejar en el TTL?

Resta entre 30 y 40 segundos de la vida útil real del token. Un token de reCAPTCHA dura unos 120 segundos, por lo que cachearlo 80 deja 40 segundos de margen para que tu aplicación lo consuma antes de que el sitio de destino lo rechace.

¿Qué pasa con el lock si un worker se cae a mitad de la resolución?

Nada queda bloqueado para siempre. La clave captcha:lock:* se crea con un TTL de 120 segundos, así que si el worker muere sin liberarla, Redis la elimina sola y otro worker puede retomar la resolución.

¿Es seguro reutilizar el mismo token CAPTCHA varias veces?

Depende del proveedor. Muchos tratan el token como de un solo uso y lo invalidan tras la primera verificación; en ese caso, reutilizarlo falla. La regla práctica:

• Entornos de prueba que controlas: puedes reutilizar el token mientras siga vivo dentro de su TTL.
• Producción con tokens de un solo uso: no los reutilices; resuelve de nuevo cada vez.

Siguientes pasos

Acelera tu flujo de CAPTCHA con una caché de tokens respaldada por Redis: consigue tu clave API de CaptchaAI.

Guías relacionadas

• Historial de resoluciones con MongoDB
• Estado de sesión en workers distribuidos
• Deduplicar resoluciones con bloqueo en base de datos