Notion API + CaptchaAI: автоматический ввод данных с обработкой CAPTCHA

Если команда уже ведёт список сайтов для парсинга в Notion, отдельный сервис под CAPTCHA не нужен. Саму базу можно превратить в очередь задач: страница хранит URL и sitekey, свойство Status отражает прогресс, а Token — готовый результат.

Ниже — рабочая схема на Python и Node.js: CaptchaAI решает reCAPTCHA v2, а Notion API отвечает за хранение и синхронизацию записей. Код одинаков для обоих языков по логике: воркер вычитывает очередь, отправляет данные в CaptchaAI и пишет результат обратно в страницу.

Что понадобится перед началом

• интеграция Notion (внутренняя, создаётся на developers.notion.com);
• база Notion, к которой открыт доступ этой интеграции;
• действующий API-ключ CaptchaAI;
• Python 3.8+ или Node.js 18+ и умение работать с переменными окружения.

Сценарий: очередь CAPTCHA-задач в Notion

Типичный кейс — в базе Notion хранится список URL для периодического парсинга, и часть из них закрыта reCAPTCHA v2. Воркер выполняет три шага по каждой записи:

1. читает записи со статусом Pending;
2. отправляет sitekey и URL в CaptchaAI и получает токен;
3. записывает токен и статус обратно в страницу Notion.

Такая схема особенно удобна для небольших QA- и парсинг-команд в России, Казахстане и Беларуси: Notion и так используется как общий трекер задач, а тарификация CaptchaAI по потокам (от BASIC — $15/мес, 5 потоков, тариф выше — STANDARD, $30/мес, 15 потоков) не зависит от того, сколько записей скопилось в очереди — платите за параллелизм решений, а не за их количество. Если в базе хранятся URL сторонних сайтов и результаты их обработки, заранее продумайте, какие данные вы вправе собирать и хранить: для читателей из РФ это 152-ФЗ, для остальных — аналогичная логика due diligence по GDPR.

Если очередь обслуживает сайты с разными типами защиты, добавьте в базу свойство Provider со значением типа CAPTCHA — это избавит от отдельной таблицы соответствий и упростит переключение метода в solve_captcha().

Структура базы данных Notion

Создайте базу со следующими свойствами. Названия важны буквально — воркер обращается к ним по имени, и опечатка выглядит как «база пустая», хотя на деле сравнение просто не находит совпадений.

Откройте базу → Share → добавьте интеграцию в список подключений, иначе первый же запрос вернёт 401.

Реализация на Python

Полный воркер: забирает задачи со статусом Pending, решает CAPTCHA через in.php/res.php и обновляет запись результатом.

# 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()

Реализация на Node.js

Тот же поток на Node.js (@notionhq/client + 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);

Типичные ошибки и их решение

Когда эта схема не подходит

Для десятков и сотен задач в час Notion как очередь работает отлично. Но у Notion API есть собственные rate-limit'ы, и он не рассчитан на высокочастотные обновления: если счёт идёт на тысячи задач в час или нужен субсекундный SLA, лучше вынести очередь в Redis или очередь сообщений (RabbitMQ, SQS), а Notion оставить только как панель для просмотра статусов.

Частые вопросы

Сколько потоков CaptchaAI нужно для очереди в Notion?

Зависит от параллельности, а не от размера очереди в Notion: потоки CaptchaAI считаются по одновременно решаемым капчам. Тариф BASIC ($15/мес, 5 потоков) хватает на десятки задач в час; для сотен параллельных проверок стоит взять STANDARD ($30/мес, 15 потоков) или выше.

Как запускать воркер по расписанию?

Через cron в Linux, Task Scheduler в Windows или облачный планировщик (AWS EventBridge, Google Cloud Scheduler). Скрипт из статьи делает один проход по очереди, поэтому регулярные повторные запуски берёт на себя планировщик, а не сам скрипт.

Что делать, если задача застряла в статусе Solving?

Обычно это значит, что процесс упал до того, как успел записать результат. Добавьте в логику проверку по времени: если запись висит в Solving дольше нескольких минут, сбрасывайте статус обратно в Pending и обрабатывайте её заново при следующем проходе.

Можно ли обрабатывать не только reCAPTCHA v2?

Да. Добавьте в базу свойство «CAPTCHA Type» и меняйте параметр method при отправке в CaptchaAI в зависимости от значения — например, на Turnstile или GeeTest v3. За пределы поддерживаемых типов выходить не стоит: hCaptcha и FunCaptcha CaptchaAI не решает.

Как не хранить API-ключи прямо в коде?

Держите NOTION_TOKEN и CAPTCHAAI_KEY в переменных окружения, как в примерах выше, а не в самом файле, и не коммитьте .env в репозиторий — это особенно важно, если воркер запускается на общем сервере.

Куда двигаться дальше

• быстрый старт CaptchaAI за 5 минут
• решение reCAPTCHA v2 через API
• решение Cloudflare Turnstile через API
• решение GeeTest v3 через API