Шаблоны семафоров для управления параллелизмом CAPTCHA

Потолок параллелизма в CaptchaAI задаёт не ваш код, а тариф: BASIC ($15/мес, 5 потоков) удерживает пять задач одновременно, ADVANCE ($90/мес, 50 потоков) — пятьдесят. Семафор нужен ровно для того, чтобы приложение соблюдало эту границу само, а не узнавало о ней из ответов с ошибками и растущих очередей.

Ниже — шаблоны на Python (asyncio.Semaphore) и Node.js, приём с раздельными лимитами на отправку и опрос, адаптивный семафор и разбор типичных сбоев.

Потоки тарифа задают потолок, семафор его соблюдает

Тарификация CaptchaAI построена на потоках, а не на числе решений. Поток — это одна задача CAPTCHA «в полёте»: как только решение возвращается, поток берёт следующую. Число решений на поток внутри месяца не ограничено, отдельной платы за каждую CAPTCHA нет.

Отсюда правило: значение семафора выставляют по числу потоков плана, а не по числу ядер или длине списка ссылок.

Выше идут ENTERPRISE ($300/мес, 200 потоков) и VIP-3 ($7,500/мес, 5000 потоков). Небольшой запас относительно лимита полезен: часть потоков может занимать соседний воркер или cron-задача.

Пропускная способность зависит от времени решения: 50 потоков при цикле в 20 с дают около 9000 задач в час. Скорость различается по типам — Cloudflare Turnstile решается менее чем за 10 с, GeeTest v3 — менее чем за 12 с, reCAPTCHA v2 — менее чем за 60 с.

Что меняется, когда параллелизм ограничен

Механика простая: семафор — это счётчик. Задача занимает слот и уменьшает счётчик, а завершаясь, возвращает его обратно. Пока счётчик равен нулю, новые задачи ждут в очереди, а не уходят в сеть. Сложность не в счётчике, а в том, чтобы слот возвращался даже при исключении.

Python: asyncio.Semaphore

Базовый шаблон: один семафор на весь цикл задачи

Самый прямой вариант — держать семафор захваченным на протяжении всей задачи: и на отправке в in.php, и на опросе res.php. Конструкция async with sem освобождает счётчик автоматически, в том числе при исключении, поэтому «залипших» слотов не остаётся.

import asyncio
import aiohttp

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


async def solve_one(session, sem, sitekey, page_url):
    """Solve one CAPTCHA within the semaphore limit."""
    async with sem:
        # Submit
        async with session.post(SUBMIT_URL, data={
            "key": API_KEY,
            "method": "userrecaptcha",
            "googlekey": sitekey,
            "pageurl": page_url,
            "json": "1",
        }) as resp:
            data = await resp.json()

        if data["status"] != 1:
            return {"url": page_url, "error": data["request"]}

        task_id = data["request"]

        # Poll (still within semaphore)
        for _ in range(24):
            await asyncio.sleep(5)
            async with session.get(RESULT_URL, params={
                "key": API_KEY, "action": "get", "id": task_id, "json": "1"
            }) as resp:
                result = await resp.json()

            if result["status"] == 1:
                return {"url": page_url, "token": result["request"]}
            if result["request"] != "CAPCHA_NOT_READY":
                return {"url": page_url, "error": result["request"]}

        return {"url": page_url, "error": "TIMEOUT"}


async def solve_batch(tasks, max_concurrent=20):
    sem = asyncio.Semaphore(max_concurrent)

    async with aiohttp.ClientSession() as session:
        coros = [
            solve_one(session, sem, t["sitekey"], t["url"])
            for t in tasks
        ]
        results = await asyncio.gather(*coros)

    solved = sum(1 for r in results if "token" in r)
    print(f"Solved {solved}/{len(results)}")
    return results

Здесь max_concurrent — это и есть ваш потолок. Двадцать одновременных задач соответствуют примерно половине лимита ADVANCE, то есть у пайплайна остаётся запас на пиковые часы.

Раздельные семафоры на отправку и опрос

Отправка занимает доли секунды, а ожидание результата — десятки секунд, поэтому при общем семафоре слот почти всё время простаивает. Разделите лимиты: жёсткий на отправку и более свободный на опрос, поскольку GET-запросы к res.php дёшевы.

async def solve_split_sems(session, submit_sem, poll_sem, sitekey, page_url):
    # Submit phase — short, limited to 30 concurrent
    async with submit_sem:
        async with session.post(SUBMIT_URL, data={
            "key": API_KEY,
            "method": "userrecaptcha",
            "googlekey": sitekey,
            "pageurl": page_url,
            "json": "1",
        }) as resp:
            data = await resp.json()

    if data["status"] != 1:
        return {"error": data["request"]}

    task_id = data["request"]

    # Poll phase — longer, limited to 50 concurrent
    async with poll_sem:
        for _ in range(24):
            await asyncio.sleep(5)
            async with session.get(RESULT_URL, params={
                "key": API_KEY, "action": "get", "id": task_id, "json": "1"
            }) as resp:
                result = await resp.json()

            if result["status"] == 1:
                return {"token": result["request"]}
            if result["request"] != "CAPCHA_NOT_READY":
                return {"error": result["request"]}

    return {"error": "TIMEOUT"}


async def main(tasks):
    submit_sem = asyncio.Semaphore(30)  # 30 concurrent submits
    poll_sem = asyncio.Semaphore(50)     # 50 concurrent polls

    async with aiohttp.ClientSession() as session:
        coros = [
            solve_split_sems(session, submit_sem, poll_sem, t["sitekey"], t["url"])
            for t in tasks
        ]
        return await asyncio.gather(*coros)

На нестабильных каналах схема ведёт себя ровнее: медленный опрос одной задачи не блокирует отправку остальных.

Node.js: собственный семафор на промисах

Встроенного семафора в Node.js нет, но он собирается из очереди resolve-функций. Класс ниже повторяет поведение asyncio.Semaphore: acquire() возвращает промис, который выполняется сразу или встаёт в очередь, а release() передаёт слот следующему ожидающему.

class Semaphore {
  constructor(max) {
    this.max = max;
    this.current = 0;
    this.queue = [];
  }

  acquire() {
    return new Promise(resolve => {
      if (this.current < this.max) {
        this.current++;
        resolve();
      } else {
        this.queue.push(resolve);
      }
    });
  }

  release() {
    this.current--;
    if (this.queue.length > 0) {
      this.current++;
      const next = this.queue.shift();
      next();
    }
  }
}

Подключение семафора к решению задач

Ключевой момент — блок try/finally. Без него любое исключение внутри задачи навсегда съедает слот, и после нескольких ошибок пайплайн останавливается целиком, хотя лимиты API не превышены.

const axios = require('axios');

const API_KEY = 'YOUR_API_KEY';
const sem = new Semaphore(20);

async function solveOne(sitekey, pageUrl) {
  await sem.acquire();
  try {
    const submit = await axios.post('https://ocr.captchaai.com/in.php', null, {
      params: {
        key: API_KEY,
        method: 'userrecaptcha',
        googlekey: sitekey,
        pageurl: pageUrl,
        json: 1,
      },
    });

    if (submit.data.status !== 1) {
      return { url: pageUrl, error: submit.data.request };
    }

    const taskId = submit.data.request;

    for (let i = 0; i < 24; 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: taskId, json: 1 },
      });

      if (poll.data.status === 1) {
        return { url: pageUrl, token: poll.data.request };
      }
      if (poll.data.request !== 'CAPCHA_NOT_READY') {
        return { url: pageUrl, error: poll.data.request };
      }
    }
    return { url: pageUrl, error: 'TIMEOUT' };
  } finally {
    sem.release();
  }
}

// Solve 100 tasks with max 20 concurrent
async function solveBatch(tasks) {
  const results = await Promise.all(
    tasks.map(t => solveOne(t.sitekey, t.url))
  );
  const solved = results.filter(r => r.token).length;
  console.log(`Solved: ${solved}/${results.length}`);
  return results;
}

Адаптивный семафор: подстройка под долю ошибок

Фиксированное значение хорошо работает на стабильной нагрузке. Если целевые страницы, сеть или время суток меняются, полезнее семафор, который сам двигает планку: снижает параллелизм при росте доли ошибок и осторожно поднимает его, когда пакет проходит чисто.

class AdaptiveSemaphore:
    def __init__(self, initial=20, min_val=5, max_val=50):
        self.value = initial
        self.min_val = min_val
        self.max_val = max_val
        self.sem = asyncio.Semaphore(initial)
        self.success_count = 0
        self.error_count = 0

    async def acquire(self):
        await self.sem.acquire()

    def release(self, success=True):
        self.sem.release()
        if success:
            self.success_count += 1
        else:
            self.error_count += 1

        total = self.success_count + self.error_count
        if total % 20 == 0:
            self._adjust()

    def _adjust(self):
        error_rate = self.error_count / (self.success_count + self.error_count)

        if error_rate > 0.2 and self.value > self.min_val:
            self.value = max(self.min_val, self.value - 5)
            self.sem = asyncio.Semaphore(self.value)
            print(f"Reduced concurrency to {self.value}")
        elif error_rate < 0.05 and self.value < self.max_val:
            self.value = min(self.max_val, self.value + 5)
            self.sem = asyncio.Semaphore(self.value)
            print(f"Increased concurrency to {self.value}")

        self.success_count = 0
        self.error_count = 0

Пересчёт раз в двадцать задач намеренно инертен: слишком частая подстройка сама становится источником колебаний. Логи Reduced concurrency и Increased concurrency стоит писать в метрики — по ним видно, во что упирается пайплайн.

Сценарий: ночной парсинг у команды из Алматы

Команда собирает каталог из 40 000 страниц в окно с 01:00 до 05:00; reCAPTCHA v2 встречается примерно на каждой десятой странице — около 1000 задач в час. При цикле решения в 40 с один поток закрывает 90 задач в час, значит нужно порядка 12 потоков: тариф STANDARD ($30/мес, 15 потоков) подходит с запасом.

Семафор выставляется на 12, отправка ограничивается отдельно, значение параллелизма пишется в метрики. Для команд, получающих выручку в тенге, рублях или гривне, важно и другое: месячная стоимость плана фиксирована в USD и не зависит от числа задач за ночь.

И отдельная оговорка про данные: собирайте только те сведения, которые вы вправе обрабатывать. Для проектов с российскими пользователями ориентиром служит 152-ФЗ «О персональных данных», для трансграничных — практики уровня GDPR. Семафор ограничивает нагрузку, но не отвечает за правомерность сбора.

Что чаще всего ломается

Чек-лист перед боевым запуском

• Значение семафора не превышает число потоков вашего плана.
• Слот освобождается в finally, а не в конце «счастливого пути».
• Отправка и опрос ограничены раздельно, если время решения превышает 10 с.
• Значение параллелизма и доля ошибок пишутся в метрики.
• Все воркеры делят один общий лимит.

Часто задаваемые вопросы

Как связать число потоков тарифа со значением семафора?

Один поток — одна задача в работе, поэтому значение семафора не должно превышать число потоков плана. На STANDARD ($30/мес, 15 потоков) разумно ставить 12–15, оставляя запас на отладочные запуски.

Что произойдёт, если задач окажется больше, чем потоков?

Параллельно они всё равно не выполнятся: часть запросов упрётся в ограничение частоты, и их придётся повторять. Семафор переносит эту очередь с API внутрь вашего процесса, где ею можно управлять.

Как ограничить параллелизм, если воркеров несколько?

Локальный семафор действует только внутри процесса. При нескольких воркерах общий лимит выносят во внешний счётчик, например в Redis, а локальные значения делают долей от общего.

Работают ли эти шаблоны с другими типами CAPTCHA?

Да, схема от типа не зависит: меняются только параметр method и ожидаемое время решения. Она применима к reCAPTCHA v2 и v3, Cloudflare Turnstile, Cloudflare Challenge, GeeTest v3 и задачам распознавания изображений. Учтите, что hCaptcha, FunCaptcha и GeeTest v4 сервисом не поддерживаются, а CaptchaFox (beta), Friendly Captcha (beta) и Lemin (beta) доступны только в бета-режиме.

Подключите решение CAPTCHA к своему пайплайну

Получите API-ключ и сверьте число потоков в тарифах на сайте CaptchaAI, затем выставьте семафор по этому числу.

Похожие материалы

• Параллельное решение задач на asyncio в Python
• Пакетная обработка CAPTCHA в несколько потоков
• Лимиты частоты запросов и их регулирование