Deduplicación de solicitudes CAPTCHA con bloqueo en base de datos

¿Tu flota de workers resuelve el mismo CAPTCHA varias veces y pagas por cada intento? La solución es una capa de deduplicación: un candado compartido —en Redis o con advisory locks de PostgreSQL— que reconoce solicitudes idénticas, deja que solo una llegue a la API y reparte el mismo token entre el resto. El ahorro es doble: menos threads en trabajo redundante y menos latencia para el worker que espera.

Piensa en una agencia que monitorea precios en marketplaces regionales (del estilo de MercadoLibre o Amazon.es) con diez procesos en paralelo. Si tres golpean la misma página protegida a la vez, sin deduplicación pagas tres resoluciones por un token que sirve para las tres. A escala, ese desperdicio se nota en la factura mensual en USD.

¿Cuándo merece la pena una capa de deduplicación?

La respuesta corta: cuando varios workers apuntan a la misma combinación de sitekey y pageurl. Incluso una tasa del 10 % ahorra créditos de forma notable a escala y elimina tiempo de resolución desperdiciado. Si tu automatización es de un solo proceso y nunca repite objetivos, aún no hace falta.

Por dónde se cuelan los duplicados

Conviene saber de dónde salen las solicitudes repetidas antes de blindar el flujo. Los cuatro patrones más habituales:

Cómo construir la clave de deduplicación

Cada worker debe llegar a la misma clave para el mismo CAPTCHA. Lo más fiable es derivar un hash SHA-256 de los parámetros que lo identifican —método, sitekey y pageurl— y usar ese valor como identificador compartido:

import hashlib


def dedup_key(method, sitekey, pageurl):
    """Generate a deduplication key for a CAPTCHA solve request."""
    raw = f"{method}:{sitekey}:{pageurl}"
    return f"captcha:dedup:{hashlib.sha256(raw.encode()).hexdigest()[:16]}"

Qué parámetros entran según el tipo

Qué entra en la clave depende del tipo de CAPTCHA. Incluye siempre lo que distingue una solicitud de otra; si falta un componente, dos CAPTCHAs distintos colisionarán en la misma clave:

Deduplicación con Redis

Redis es la opción natural cuando ya forma parte de tu stack. Escribe la clave con estado solving de forma atómica; si otro worker la encuentra, espera el resultado en vez de lanzar un envío nuevo. Al llegar el token, se cachea unos segundos para que el resto lo reutilice antes de que caduque.

El patrón en tres pasos

1. Reclamar la clave. El primer worker escribe el estado solving con SET NX y un TTL de seguridad por si el proceso se cae.
2. Resolver una sola vez. Solo quien reclamó la clave envía el CAPTCHA y sondea el resultado.
3. Compartir el token. Al resolverse, se cachea con un TTL corto y el resto lo lee sin volver a pagar.

Implementación en Python

El flujo completo comprueba la clave, marca el estado, envía a CaptchaAI, sondea el resultado y guarda el token para los demás workers:

import os
import time
import json
import hashlib
import redis
import requests

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

API_KEY = os.environ["CAPTCHAAI_API_KEY"]

# Dedup window: how long to consider a request "in progress"
DEDUP_TTL = 180  # seconds


def dedup_key(method, sitekey, pageurl, extra=""):
    raw = f"{method}:{sitekey}:{pageurl}:{extra}"
    return f"captcha:dedup:{hashlib.sha256(raw.encode()).hexdigest()[:16]}"


def solve_with_dedup(sitekey, pageurl, method="userrecaptcha"):
    key = dedup_key(method, sitekey, pageurl)

    # Check if this request is already being solved
    existing = r.get(key)
    if existing:
        state = json.loads(existing)
        if state["status"] == "solving":
            # Wait for the result
            return wait_for_result(key)
        elif state["status"] == "solved":
            return {"solution": state["solution"], "source": "dedup_cache"}
        elif state["status"] == "error":
            pass  # Allow retry on error

    # Mark as solving
    r.set(key, json.dumps({"status": "solving", "started": time.time()}), ex=DEDUP_TTL)

    # Submit to CaptchaAI
    resp = requests.post("https://ocr.captchaai.com/in.php", data={
        "key": API_KEY,
        "method": method,
        "googlekey": sitekey,
        "pageurl": pageurl,
        "json": 1
    })
    data = resp.json()

    if data.get("status") != 1:
        r.set(key, json.dumps({"status": "error", "error": data.get("request")}), ex=30)
        return {"error": data.get("request")}

    captcha_id = data["request"]

    # Poll for result
    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:
            solution = result["request"]
            # Cache the result for other workers (short TTL since tokens expire)
            r.set(key, json.dumps({
                "status": "solved",
                "solution": solution,
                "solved_at": time.time()
            }), ex=60)  # Cache result for 60 seconds
            return {"solution": solution, "source": "api"}

        if result.get("request") != "CAPCHA_NOT_READY":
            r.set(key, json.dumps({
                "status": "error", "error": result.get("request")
            }), ex=30)
            return {"error": result.get("request")}

    r.set(key, json.dumps({"status": "error", "error": "TIMEOUT"}), ex=30)
    return {"error": "TIMEOUT"}


def wait_for_result(key, timeout=120):
    """Wait for another worker to finish solving."""
    start = time.time()
    while time.time() - start < timeout:
        data = r.get(key)
        if data:
            state = json.loads(data)
            if state["status"] == "solved":
                return {"solution": state["solution"], "source": "dedup_wait"}
            if state["status"] == "error":
                return {"error": state.get("error", "UNKNOWN")}
        time.sleep(2)
    return {"error": "DEDUP_WAIT_TIMEOUT"}

Implementación en Node.js

La misma lógica en Node.js con ioredis y axios. Mantén idénticos la clave y el TTL en toda tu flota para que compartan el candado:

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

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

function dedupKey(method, sitekey, pageurl) {
  const raw = `${method}:${sitekey}:${pageurl}`;
  const hash = crypto.createHash("sha256").update(raw).digest("hex").slice(0, 16);
  return `captcha:dedup:${hash}`;
}

async function solveWithDedup(sitekey, pageurl, method = "userrecaptcha") {
  const key = dedupKey(method, sitekey, pageurl);

  // Check existing
  const existing = await redis.get(key);
  if (existing) {
    const state = JSON.parse(existing);
    if (state.status === "solving") return await waitForResult(key);
    if (state.status === "solved") return { solution: state.solution, source: "dedup_cache" };
  }

  // Mark as solving
  await redis.set(key, JSON.stringify({ status: "solving", started: Date.now() }), "EX", DEDUP_TTL);

  // Submit
  const submit = await axios.post("https://ocr.captchaai.com/in.php", null, {
    params: { key: API_KEY, method, googlekey: sitekey, pageurl, json: 1 },
  });

  if (submit.data.status !== 1) {
    await redis.set(key, JSON.stringify({ status: "error", error: submit.data.request }), "EX", 30);
    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) {
      await redis.set(key, JSON.stringify({ status: "solved", solution: poll.data.request }), "EX", 60);
      return { solution: poll.data.request, source: "api" };
    }
    if (poll.data.request !== "CAPCHA_NOT_READY") {
      await redis.set(key, JSON.stringify({ status: "error", error: poll.data.request }), "EX", 30);
      return { error: poll.data.request };
    }
  }

  await redis.set(key, JSON.stringify({ status: "error", error: "TIMEOUT" }), "EX", 30);
  return { error: "TIMEOUT" };
}

async function waitForResult(key, timeout = 120000) {
  const start = Date.now();
  while (Date.now() - start < timeout) {
    const data = await redis.get(key);
    if (data) {
      const state = JSON.parse(data);
      if (state.status === "solved") return { solution: state.solution, source: "dedup_wait" };
      if (state.status === "error") return { error: state.error };
    }
    await new Promise((r) => setTimeout(r, 2000));
  }
  return { error: "DEDUP_WAIT_TIMEOUT" };
}

Alternativa sin Redis: advisory locks de PostgreSQL

Si no quieres sumar Redis, PostgreSQL ofrece advisory locks: candados de sesión que no bloquean filas ni tablas y se liberan solos al cerrar la conexión.

Adquirir el lock y leer el caché

Deriva un identificador numérico de la clave, intenta adquirir el lock sin bloquear y, si otro worker lo tiene, espera y lee el resultado cacheado:

import psycopg2


def solve_with_pg_dedup(conn, sitekey, pageurl):
    """Use PostgreSQL advisory locks for deduplication."""
    # Generate a numeric lock key from the dedup key
    lock_id = hash(f"{sitekey}:{pageurl}") & 0x7FFFFFFF

    cursor = conn.cursor()

    # Try to acquire advisory lock (non-blocking)
    cursor.execute("SELECT pg_try_advisory_lock(%s)", (lock_id,))
    acquired = cursor.fetchone()[0]

    if not acquired:
        # Another worker is solving — wait for result
        cursor.execute("SELECT pg_advisory_lock(%s)", (lock_id,))
        # Lock acquired means other worker finished — check cache
        cursor.execute(
            "SELECT solution FROM captcha_cache "
            "WHERE sitekey = %s AND pageurl = %s "
            "AND created_at > NOW() - INTERVAL '60 seconds'",
            (sitekey, pageurl)
        )
        row = cursor.fetchone()
        cursor.execute("SELECT pg_advisory_unlock(%s)", (lock_id,))
        if row:
            return {"solution": row[0], "source": "pg_cache"}
        return {"error": "NO_CACHED_RESULT"}

    try:
        # Solve the CAPTCHA
        solution = solve_via_api(sitekey, pageurl)
        if solution:
            cursor.execute(
                "INSERT INTO captcha_cache (sitekey, pageurl, solution) "
                "VALUES (%s, %s, %s)",
                (sitekey, pageurl, solution)
            )
            conn.commit()
        return {"solution": solution} if solution else {"error": "SOLVE_FAILED"}
    finally:
        cursor.execute("SELECT pg_advisory_unlock(%s)", (lock_id,))

Cómo medir el ahorro real

Sin números, la deduplicación es una corazonada. Cuenta cuántas solicitudes se sirvieron desde el caché o la espera frente a las que llegaron a la API y calcula el porcentaje ahorrado:

def track_dedup_stats(source):
    """Increment counters for dedup tracking."""
    today = time.strftime("%Y-%m-%d")
    r.hincrby(f"dedup:stats:{today}", source, 1)
    r.expire(f"dedup:stats:{today}", 7 * 86400)


def get_dedup_report():
    today = time.strftime("%Y-%m-%d")
    stats = r.hgetall(f"dedup:stats:{today}")
    total = sum(int(v) for v in stats.values())
    saved = int(stats.get("dedup_cache", 0)) + int(stats.get("dedup_wait", 0))
    return {
        "total_requests": total,
        "deduplicated": saved,
        "savings_pct": f"{saved / total * 100:.1f}%" if total else "0%",
        "breakdown": stats
    }

Diagnóstico de problemas comunes

Cuando la deduplicación no rinde lo esperado, casi siempre es una de estas cuatro causas:

Deduplicación y facturación por threads en CaptchaAI

CaptchaAI factura por threads concurrentes con resoluciones ilimitadas por thread, no por resolución individual. Eso cambia la lógica del ahorro: un duplicado no suma un cargo suelto, pero ocupa un thread que podría atender trabajo nuevo. Deduplicar te deja exprimir el plan que ya pagas: empieza con el plan BASIC ($15/mes, 5 threads) y sube de escalón solo cuando tu concurrencia real lo pida.

Preguntas frecuentes

¿Conviene usar Redis o los advisory locks de PostgreSQL?

Depende de tu infraestructura. Si ya tienes Redis, su modelo SET NX con TTL es el camino más simple. Si prefieres no añadir dependencias, los advisory locks de PostgreSQL resuelven lo mismo sin servicios extra.

¿La deduplicación reduce mi factura de CaptchaAI?

Sí, de forma indirecta. CaptchaAI factura por threads concurrentes, no por resolución, así que cada duplicado que evitas libera un thread para trabajo útil. Con mucho solapamiento, eso puede significar quedarte en el plan BASIC ($15/mes, 5 threads) en vez de saltar al siguiente escalón.

¿Qué pasa si dos workers consultan la clave a la vez?

Ese es el caso que el candado atómico previene. Con SET NX en Redis o pg_try_advisory_lock en PostgreSQL, solo un worker gana la clave y envía a la API; el resto detecta el estado solving y espera. Sin esa atomicidad tendrías una condición de carrera y pagarías duplicados.

¿Debo incluir el proxy en la clave de deduplicación?

No. El token resuelto es válido sin importar qué proxy se usó, así que sumar el proxy a la clave fragmentaría la deduplicación y resolverías lo mismo para cada salida de red. Deja la clave con method, sitekey y pageurl.

Guías relacionadas

• Gestión del TTL de tokens en Redis
• Estado de sesión en workers distribuidos
• Recuperación de errores en lotes con fallos parciales