Пакетное решение CAPTCHA для изображений: обработка более 1000 изображений

Если скрипт отправляет и опрашивает капчи по одной, обработка 1000 изображений растягивается на десятки минут — большая часть времени уходит на ожидание ответа сети между запросами, а не на распознавание. Решение простое: разделить отправку и опрос на два пула воркеров и вести их параллельно через API CaptchaAI.

Ориентир: 20 одновременных отправок и отдельный пул опроса на 30 задач обычно дают лучший баланс скорости и стабильности для пакета в 1000 изображений — без риска упереться в лимит запросов.

Когда пакетная обработка вообще нужна

Последовательный цикл «отправил → дождался → отправил следующую» подходит для одиночных капч в интерактивном сценарии, но на объёме он превращается в узкое место:

• Регистрационные формы или сканы документов, где счёт идёт на сотни и тысячи изображений за прогон.
• Периодический сбор данных (парсинг), где капчи приходят пакетами, а не по одной.
• QA-тестирование интеграции, где нужно прогнать заранее заготовленный набор изображений и сравнить результат с эталоном.

Если ваш случай — единичные капчи по требованию пользователя, последовательный вызов in.php/res.php проще и без пакетной обвязки.

Схема конвейера

[Image Queue] → [Submit Workers] → [Poll Workers] → [Results Store]
     ↓                ↓                  ↓                ↓
  1000 images    20 concurrent      Adaptive poll     CSV/JSON output
                   submits           intervals

Очередь разбирают воркеры отправки, а отдельный пул опроса забирает готовые ответы через res.php и пишет их в CSV или JSON — так отправка и ожидание результата не блокируют друг друга.

Прежде чем запускать пакет на 1000+ изображений

1. Получите API-ключ CaptchaAI и держите его вне репозитория (переменная окружения, секрет-хранилище).
2. Проверьте баланс — пакет на тысячу изображений не должен остановиться на середине из-за ERROR_ZERO_BALANCE.
3. Валидируйте файлы на входе: формат, размер, отсутствие повреждённых изображений в очереди.
4. Выставьте MAX_CONCURRENT_SUBMITS в разумных пределах (20–30) — это отправная точка, а не жёсткий лимит API.

Python: асинхронный пакетный процессор

import asyncio
import aiohttp
import base64
import json
import time
import csv
from pathlib import Path

API_KEY = "YOUR_API_KEY"
SUBMIT_URL = "https://ocr.captchaai.com/in.php"
RESULT_URL = "https://ocr.captchaai.com/res.php"
MAX_CONCURRENT_SUBMITS = 20
MAX_CONCURRENT_POLLS = 30
POLL_INTERVAL = 5


async def submit_image(session, sem, image_path):
    """Submit a single image CAPTCHA."""
    async with sem:
        with open(image_path, "rb") as f:
            img_b64 = base64.b64encode(f.read()).decode()

        data = {
            "key": API_KEY,
            "method": "base64",
            "body": img_b64,
            "json": "1",
        }

        async with session.post(SUBMIT_URL, data=data) as resp:
            result = await resp.json()

        if result["status"] != 1:
            return {"file": str(image_path), "error": result["request"]}

        return {
            "file": str(image_path),
            "task_id": result["request"],
            "submitted_at": time.time(),
        }


async def poll_result(session, sem, task):
    """Poll for a single task result."""
    async with sem:
        for attempt in range(24):
            await asyncio.sleep(POLL_INTERVAL)

            params = {
                "key": API_KEY,
                "action": "get",
                "id": task["task_id"],
                "json": "1",
            }

            async with session.get(RESULT_URL, params=params) as resp:
                result = await resp.json()

            if result["status"] == 1:
                return {
                    "file": task["file"],
                    "task_id": task["task_id"],
                    "answer": result["request"],
                    "solve_time": time.time() - task["submitted_at"],
                }
            if result["request"] != "CAPCHA_NOT_READY":
                return {
                    "file": task["file"],
                    "task_id": task["task_id"],
                    "error": result["request"],
                }

        return {
            "file": task["file"],
            "task_id": task["task_id"],
            "error": "TIMEOUT",
        }


async def process_batch(image_dir, output_file="results.csv"):
    """Process all images in a directory."""
    image_paths = sorted(Path(image_dir).glob("*.png")) + \
                  sorted(Path(image_dir).glob("*.jpg"))

    print(f"Found {len(image_paths)} images")

    submit_sem = asyncio.Semaphore(MAX_CONCURRENT_SUBMITS)
    poll_sem = asyncio.Semaphore(MAX_CONCURRENT_POLLS)

    async with aiohttp.ClientSession() as session:
        # Phase 1: Submit all images
        print("Submitting...")
        submit_tasks = [
            submit_image(session, submit_sem, path)
            for path in image_paths
        ]
        submissions = await asyncio.gather(*submit_tasks)

        # Separate successes and errors
        pending = [s for s in submissions if "task_id" in s]
        errors = [s for s in submissions if "error" in s]
        print(f"Submitted: {len(pending)}, Errors: {len(errors)}")

        # Phase 2: Poll all pending tasks
        print("Polling for results...")
        poll_tasks = [
            poll_result(session, poll_sem, task)
            for task in pending
        ]
        results = await asyncio.gather(*poll_tasks)

    # Combine results
    all_results = results + errors

    # Write to CSV
    with open(output_file, "w", newline="") as f:
        writer = csv.DictWriter(f, fieldnames=[
            "file", "task_id", "answer", "solve_time", "error"
        ])
        writer.writeheader()
        for r in all_results:
            writer.writerow({
                "file": r.get("file", ""),
                "task_id": r.get("task_id", ""),
                "answer": r.get("answer", ""),
                "solve_time": round(r.get("solve_time", 0), 2),
                "error": r.get("error", ""),
            })

    solved = sum(1 for r in results if "answer" in r)
    failed = sum(1 for r in results if "error" in r)
    print(f"Done: {solved} solved, {failed} failed, {len(errors)} submit errors")
    print(f"Results saved to {output_file}")


# Run
asyncio.run(process_batch("./captcha_images"))

Пример вывода в терминале:

Found 1000 images
Submitting...
Submitted: 997, Errors: 3
Polling for results...
Done: 985 solved, 12 failed, 3 submit errors
Results saved to results.csv

Node.js: пул воркеров для пакетной обработки

const axios = require('axios');
const fs = require('fs');
const path = require('path');
const { createObjectCsvWriter } = require('csv-writer');

const API_KEY = 'YOUR_API_KEY';
const SUBMIT_URL = 'https://ocr.captchaai.com/in.php';
const RESULT_URL = 'https://ocr.captchaai.com/res.php';
const MAX_CONCURRENT = 20;
const POLL_INTERVAL_MS = 5000;

class BatchProcessor {
  constructor(concurrency = MAX_CONCURRENT) {
    this.concurrency = concurrency;
    this.results = [];
    this.processed = 0;
    this.total = 0;
  }

  async submitImage(imagePath) {
    const imgBase64 = fs.readFileSync(imagePath, { encoding: 'base64' });
    const resp = await axios.post(SUBMIT_URL, null, {
      params: {
        key: API_KEY,
        method: 'base64',
        body: imgBase64,
        json: 1,
      },
    });

    if (resp.data.status !== 1) {
      throw new Error(resp.data.request);
    }
    return resp.data.request;
  }

  async pollResult(taskId) {
    for (let i = 0; i < 24; i++) {
      await new Promise(r => setTimeout(r, POLL_INTERVAL_MS));
      const resp = await axios.get(RESULT_URL, {
        params: { key: API_KEY, action: 'get', id: taskId, json: 1 },
      });

      if (resp.data.status === 1) return resp.data.request;
      if (resp.data.request !== 'CAPCHA_NOT_READY') {
        throw new Error(resp.data.request);
      }
    }
    throw new Error('TIMEOUT');
  }

  async processOne(imagePath) {
    const startTime = Date.now();
    try {
      const taskId = await this.submitImage(imagePath);
      const answer = await this.pollResult(taskId);
      this.processed++;
      const elapsed = ((Date.now() - startTime) / 1000).toFixed(1);
      console.log(`[${this.processed}/${this.total}] ${path.basename(imagePath)}: ${answer} (${elapsed}s)`);
      return { file: imagePath, answer, solveTime: elapsed, error: '' };
    } catch (err) {
      this.processed++;
      return { file: imagePath, answer: '', solveTime: 0, error: err.message };
    }
  }

  async run(imageDir, outputFile = 'results.csv') {
    const files = fs.readdirSync(imageDir)
      .filter(f => /\.(png|jpg|jpeg|gif)$/i.test(f))
      .map(f => path.join(imageDir, f));

    this.total = files.length;
    console.log(`Processing ${this.total} images with ${this.concurrency} workers`);

    // Process in chunks
    for (let i = 0; i < files.length; i += this.concurrency) {
      const chunk = files.slice(i, i + this.concurrency);
      const chunkResults = await Promise.all(
        chunk.map(f => this.processOne(f))
      );
      this.results.push(...chunkResults);
    }

    // Write CSV
    const csvWriter = createObjectCsvWriter({
      path: outputFile,
      header: [
        { id: 'file', title: 'File' },
        { id: 'answer', title: 'Answer' },
        { id: 'solveTime', title: 'Solve Time (s)' },
        { id: 'error', title: 'Error' },
      ],
    });
    await csvWriter.writeRecords(this.results);

    const solved = this.results.filter(r => r.answer).length;
    console.log(`Done: ${solved}/${this.total} solved. Results: ${outputFile}`);
  }
}

const processor = new BatchProcessor(20);
processor.run('./captcha_images');

Этот вариант обрабатывает файлы чанками по concurrency штук и логирует каждый решённый файл в момент готовности — удобно, когда нужно видеть прогресс в консоли без отдельного трекера.

Ограничение частоты запросов при пакетной отправке

Ошибки 429 почти всегда означают одно: воркеры бьют по in.php быстрее, чем допускает лимит. Простой ограничитель на скользящем окне в секунду решает проблему:

class RateLimiter:
    def __init__(self, max_per_second=10):
        self.max_per_second = max_per_second
        self.timestamps = []

    async def acquire(self):
        now = time.time()
        self.timestamps = [t for t in self.timestamps if now - t < 1.0]

        if len(self.timestamps) >= self.max_per_second:
            wait = 1.0 - (now - self.timestamps[0])
            if wait > 0:
                await asyncio.sleep(wait)

        self.timestamps.append(time.time())

# Use in submit loop
rate_limiter = RateLimiter(max_per_second=10)

async def submit_with_rate_limit(session, image_path):
    await rate_limiter.acquire()
    # ... submit as before

Мониторинг прогресса пакета

Для пакета в тысячу изображений полезно видеть скорость и ETA в реальном времени, а не ждать финального отчёта вслепую:

import sys

class ProgressTracker:
    def __init__(self, total):
        self.total = total
        self.completed = 0
        self.solved = 0
        self.failed = 0
        self.start_time = time.time()

    def update(self, success=True):
        self.completed += 1
        if success:
            self.solved += 1
        else:
            self.failed += 1

        elapsed = time.time() - self.start_time
        rate = self.completed / elapsed if elapsed > 0 else 0
        eta = (self.total - self.completed) / rate if rate > 0 else 0

        sys.stdout.write(
            f"\r[{self.completed}/{self.total}] "
            f"Solved: {self.solved} | Failed: {self.failed} | "
            f"Rate: {rate:.1f}/s | ETA: {eta:.0f}s"
        )
        sys.stdout.flush()

Сколько потоков нужно для 1000+ изображений

Скорость пакета в первую очередь упирается не в количество картинок, а в лимит одновременных запросов вашего плана — CaptchaAI тарифицирует по потокам, а не по числу решений, и внутри плана решения не ограничены. Агентство, которое сканирует формы для нескольких клиентов и выставляет счета в долларах, получает на ADVANCE предсказуемую стоимость независимо от того, приходит 200 изображений в день или 2000.

Типичные проблемы при пакетной обработке

Большинство сбоев в пакете сводится к четырём причинам — ниже они собраны вместе с быстрым исправлением:

О персональных данных в изображениях

Если пакет включает сканы форм или документов, а не абстрактные капчи-картинки, до отправки стоит убедиться, что вы вправе их обрабатывать — это касается любой автоматизации, работающей с чужими данными.

Собирайте и передавайте только те данные, которые вам разрешено обрабатывать, и учитывайте требования 152-ФЗ «О персональных данных» для аудитории в РФ или сопоставимую осмотрительность в духе GDPR для трансграничных сценариев. Это не юридическая консультация, а напоминание для чек-листа перед запуском пакета.

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

Как не получить ошибку 429 при пакетной отправке?

Держите MAX_CONCURRENT_SUBMITS в диапазоне 20–30 и используйте ограничитель скорости поверх семафора — так вы не упрётесь в лимит, даже если запросы на секунду «сгрудятся».

Что делать, если ERROR_ZERO_BALANCE останавливает пакет на середине?

Пакет прервётся на первом же исчерпании баланса, а часть задач останется без ответа. Проверяйте баланс перед стартом и заранее прикидывайте стоимость пакета — Image/OCR стоит недорого, но на 1000+ изображениях разница в тарифе всё равно заметна.

Сколько стоит распознать 1000 CAPTCHA-изображений?

Точная сумма зависит от плана — актуальные тарифы CaptchaAI смотрите на captchaai.com. Image/OCR входит в число самых дешёвых типов решения, а тарификация по потокам означает, что стоимость пакета не растёт линейно с числом изображений.

Нужно ли валидировать изображения перед отправкой в очередь?

Да — повреждённые файлы, нулевые байты или неподдерживаемые форматы (не .png/.jpg) дают всплеск ошибок и впустую тратят слот в MAX_CONCURRENT_SUBMITS. Дешевле отсеять такие файлы локальной проверкой размера и заголовка файла, чем ловить их ошибки после отправки на in.php.

Обрабатывайте тысячи CAPTCHA с помощью CaptchaAI

Получите ключ API на captchaai.com и подключите пример из этого руководства к своей очереди изображений.

Связанные руководства

• Асинхронное параллельное решение на Python
• Параллельное решение CAPTCHA
• Лимиты запросов и защита от троттлинга