Notion API + CaptchaAI: entrada de datos automatizada con manejo de CAPTCHA

¿Puedes usar una base de datos de Notion como cola para resolver CAPTCHA de forma automatizada, sin construir tu propio sistema de tareas? Sí. La API de Notion lee y escribe registros mediante programación, así que funciona como un panel de control ligero para tus flujos, y CaptchaAI resuelve el reCAPTCHA v2. En esta guía montas un worker que lee las tareas pendientes de Notion, las resuelve y escribe el token y el estado de vuelta en cada fila.

Requisitos previos

• Una integración de Notion (interna, creada en developers.notion.com).
• Una base de datos de Notion compartida con esa integración.
• Tu clave API de CaptchaAI.
• Python 3.8+ o Node.js 18+.

Cómo funciona el flujo

El worker se apoya en una columna de estado dentro de la propia base de datos y da tres pasos:

1. Consulta las filas con estado Pending.
2. Envía el sitekey y la URL a CaptchaAI, sondea el resultado y recupera el token.
3. Escribe el token, la marca de tiempo y el nuevo estado (Solved o Failed) en esa fila.

Como el estado vive en Notion, la tanda es idempotente: si el script se cae a mitad, al reiniciarlo solo retoma lo que sigue en Pending. Por eso el patrón funciona bien para ejecuciones periódicas.

Escenario real

Una agencia de datos en Madrid o Ciudad de México mantiene en Notion una lista de URL de portales públicos que monitoriza para sus clientes —categorías como cita previa, trámites de SAT o consultas de AFIP, siempre sobre flujos que tiene autorización para revisar—. Algunas muestran un reCAPTCHA v2 antes de dejar leer el contenido. En lugar de que cada analista lo resuelva a mano, guardan la URL y el sitekey en una fila y dejan que el worker vacíe la cola por la noche. Respeta siempre los términos de servicio del sitio y la normativa de protección de datos aplicable (GDPR y LOPDGDD en España, LFPDPPP en México).

Configuración de la base de datos en Notion

Crea una base de datos con estas propiedades. Los nombres deben coincidir exactamente con los del código, porque Notion distingue mayúsculas y minúsculas:

Comparte la base de datos con tu integración desde el menú de conexiones. Sin ese paso, la API responde 401 Unauthorized.

Implementación en Python

# notion_captcha_worker.py
import os
import time
import requests

NOTION_TOKEN = os.environ.get("NOTION_TOKEN")
NOTION_DB_ID = os.environ.get("NOTION_DB_ID")
CAPTCHAAI_KEY = os.environ.get("CAPTCHAAI_KEY", "YOUR_API_KEY")

NOTION_HEADERS = {
    "Authorization": f"Bearer {NOTION_TOKEN}",
    "Content-Type": "application/json",
    "Notion-Version": "2022-06-28",
}

def get_pending_tasks():
    """Fetch tasks with Status = Pending from Notion."""
    url = f"https://api.notion.com/v1/databases/{NOTION_DB_ID}/query"
    payload = {
        "filter": {
            "property": "Status",
            "select": {"equals": "Pending"},
        }
    }
    resp = requests.post(url, headers=NOTION_HEADERS, json=payload)
    resp.raise_for_status()
    return resp.json()["results"]

def update_task(page_id, properties):
    """Update a Notion page with new property values."""
    url = f"https://api.notion.com/v1/pages/{page_id}"
    payload = {"properties": properties}
    resp = requests.patch(url, headers=NOTION_HEADERS, json=payload)
    resp.raise_for_status()

def set_status(page_id, status, token=None, error=None):
    """Update task status in Notion."""
    props = {"Status": {"select": {"name": status}}}

    if token:
        props["Token"] = {"rich_text": [{"text": {"content": token[:2000]}}]}
        props["Solved At"] = {"date": {"start": time.strftime("%Y-%m-%dT%H:%M:%S")}}

    if error:
        props["Error"] = {"rich_text": [{"text": {"content": error[:200]}}]}

    update_task(page_id, props)

def solve_captcha(sitekey, pageurl):
    """Submit to CaptchaAI and poll for result."""
    # Submit
    resp = requests.get("https://ocr.captchaai.com/in.php", params={
        "key": CAPTCHAAI_KEY,
        "method": "userrecaptcha",
        "googlekey": sitekey,
        "pageurl": pageurl,
        "json": "1",
    })
    result = resp.json()

    if result.get("status") != 1:
        raise Exception(f"Submit failed: {result.get('request')}")

    task_id = result["request"]

    # Poll
    time.sleep(15)
    for _ in range(25):
        poll = requests.get("https://ocr.captchaai.com/res.php", params={
            "key": CAPTCHAAI_KEY,
            "action": "get",
            "id": task_id,
            "json": "1",
        })
        poll_result = poll.json()

        if poll_result.get("status") == 1:
            return poll_result["request"]
        if poll_result.get("request") != "CAPCHA_NOT_READY":
            raise Exception(f"Solve failed: {poll_result.get('request')}")

        time.sleep(5)

    raise Exception("Polling timeout")

def extract_property(page, prop_name, prop_type="rich_text"):
    """Extract a property value from a Notion page."""
    prop = page["properties"].get(prop_name, {})
    if prop_type == "rich_text":
        texts = prop.get("rich_text", [])
        return texts[0]["plain_text"] if texts else ""
    elif prop_type == "url":
        return prop.get("url", "")
    return ""

def main():
    tasks = get_pending_tasks()
    print(f"Found {len(tasks)} pending tasks")

    for task in tasks:
        page_id = task["id"]
        sitekey = extract_property(task, "Sitekey")
        pageurl = extract_property(task, "URL", "url")

        if not sitekey or not pageurl:
            set_status(page_id, "Failed", error="Missing sitekey or URL")
            continue

        print(f"Solving: {pageurl}")
        set_status(page_id, "Solving")

        try:
            token = solve_captcha(sitekey, pageurl)
            set_status(page_id, "Solved", token=token)
            print(f"  Solved successfully")
        except Exception as e:
            set_status(page_id, "Failed", error=str(e))
            print(f"  Failed: {e}")

        time.sleep(1)  # Rate limit for Notion API

    print("All tasks processed")

if __name__ == "__main__":
    main()

El envío usa el método userrecaptcha contra in.php y luego sondea res.php. Mientras el token no está listo, la API responde CAPCHA_NOT_READY y el bucle espera 5 segundos antes de volver a consultar.

Implementación en JavaScript

Si tu stack es Node.js, esta versión hace lo mismo con el cliente oficial @notionhq/client y axios:

// notion_captcha_worker.js
const { Client } = require('@notionhq/client');
const axios = require('axios');

const notion = new Client({ auth: process.env.NOTION_TOKEN });
const DB_ID = process.env.NOTION_DB_ID;
const API_KEY = process.env.CAPTCHAAI_KEY || 'YOUR_API_KEY';

async function getPendingTasks() {
  const response = await notion.databases.query({
    database_id: DB_ID,
    filter: { property: 'Status', select: { equals: 'Pending' } },
  });
  return response.results;
}

async function updateTask(pageId, status, token, error) {
  const properties = {
    Status: { select: { name: status } },
  };
  if (token) {
    properties.Token = { rich_text: [{ text: { content: token.slice(0, 2000) } }] };
    properties['Solved At'] = { date: { start: new Date().toISOString() } };
  }
  if (error) {
    properties.Error = { rich_text: [{ text: { content: error.slice(0, 200) } }] };
  }
  await notion.pages.update({ page_id: pageId, properties });
}

async function solveCaptcha(sitekey, pageurl) {
  const submit = await axios.get('https://ocr.captchaai.com/in.php', {
    params: {
      key: API_KEY, method: 'userrecaptcha',
      googlekey: sitekey, pageurl, json: '1',
    },
  });
  if (submit.data.status !== 1) throw new Error(submit.data.request);

  await new Promise(r => setTimeout(r, 15000));

  for (let i = 0; i < 25; i++) {
    const poll = await axios.get('https://ocr.captchaai.com/res.php', {
      params: { key: API_KEY, action: 'get', id: submit.data.request, json: '1' },
    });
    if (poll.data.status === 1) return poll.data.request;
    if (poll.data.request !== 'CAPCHA_NOT_READY') throw new Error(poll.data.request);
    await new Promise(r => setTimeout(r, 5000));
  }
  throw new Error('Timeout');
}

async function main() {
  const tasks = await getPendingTasks();
  console.log(`Found ${tasks.length} pending tasks`);

  for (const task of tasks) {
    const sitekey = task.properties.Sitekey?.rich_text?.[0]?.plain_text;
    const pageurl = task.properties.URL?.url;

    if (!sitekey || !pageurl) {
      await updateTask(task.id, 'Failed', null, 'Missing sitekey or URL');
      continue;
    }

    console.log(`Solving: ${pageurl}`);
    await updateTask(task.id, 'Solving');

    try {
      const token = await solveCaptcha(sitekey, pageurl);
      await updateTask(task.id, 'Solved', token);
      console.log('  Solved');
    } catch (e) {
      await updateTask(task.id, 'Failed', null, e.message);
      console.log(`  Failed: ${e.message}`);
    }

    await new Promise(r => setTimeout(r, 1000));
  }
}

main().catch(console.error);

Escalar el worker cuando crece el volumen

El ejemplo procesa las tareas una a una, lo cual basta para colas pequeñas. Con cientos de filas por tanda, el cuello de botella pasa a ser cuántas resoluciones tienes en curso a la vez, y ahí entra el modelo de threads de CaptchaAI: pagas por thread concurrente —no por resolución— con resoluciones ilimitadas por thread durante el mes.

• BASIC ($15/mes, 5 threads): tandas ligeras y pruebas.
• ADVANCE ($90/mes, 50 threads) o PREMIUM ($170/mes, 100 threads): cuando quieres paralelizar de verdad.

Para una agencia que factura en una moneda local volátil, ese costo mensual fijo en USD es más fácil de presupuestar que un pago por resolución. Al pasar al modelo concurrente, lanza varias llamadas a solve_captcha en paralelo y deja el resto del flujo —lectura y escritura en Notion— tal cual.

Solución de problemas

Preguntas frecuentes

¿Qué plan de CaptchaAI necesito para vaciar una cola grande de Notion?

Depende de cuántas resoluciones quieras tener en curso a la vez, no del total de filas. Como cada plan trae resoluciones ilimitadas por thread, la decisión es cuánta concurrencia necesitas: BASIC ($15/mes, 5 threads) para tandas ligeras y ADVANCE ($90/mes, 50 threads) cuando quieras paralelizar.

¿Cómo mantengo el worker dentro del límite de solicitudes de Notion?

La API de Notion limita las solicitudes por integración, así que el time.sleep(1) entre actualizaciones no es opcional en tandas grandes. Si aun así ves errores 429, sube el retraso o agrupa las escrituras y actualiza el estado con menos frecuencia.

¿Puedo usar Airtable o Google Sheets en lugar de Notion?

Sí. El patrón —una tabla como cola de tareas y un worker que resuelve y escribe el resultado— es agnóstico de la herramienta. Solo cambia la capa de lectura y escritura; la parte de resolución con CaptchaAI se queda igual. Tienes un ejemplo equivalente con Airtable al final de esta guía.

¿Qué pasa si un CAPTCHA no se resuelve dentro del tiempo de espera?

El bucle de sondeo se detiene tras 25 intentos y lanza un timeout. El worker marca esa fila como Failed y guarda el mensaje en la columna Error, así que la tanda no se corta. En la siguiente ejecución reintentas solo las filas fallidas cambiándolas de nuevo a Pending.

Artículos relacionados

• Resolver el callback de reCAPTCHA v2 con la API
• Manejar reCAPTCHA v2 y Turnstile en el mismo sitio
• Cómo funciona el mecanismo de callback de reCAPTCHA v2

Próximos pasos

Convierte tus bases de datos de Notion en colas automatizadas de resolución de CAPTCHA. Obtén tu clave API de CaptchaAI y ejecuta tu primera tanda.

Guías relacionadas:

• Airtable + CaptchaAI
• Guardar resultados de CAPTCHA en PostgreSQL
• Zapier + CaptchaAI