Тестирование CaptchaAI перед полной миграцией: руководство по параллельному запуску

Единственный честный способ сравнить два сервиса решения CAPTCHA — отправить им одну и ту же задачу в одну и ту же секунду и посмотреть, что вернулось. Это и есть параллельный запуск: текущий провайдер остаётся основным, CaptchaAI подключается вторым каналом, а решение о миграции принимается по собственным цифрам, а не по чужому лендингу. Ниже — схема стенда, код на Python и Node.js, набор метрик и план перехода, который не ломает продакшен.

Почему чужие бенчмарки не отвечают на ваш вопрос

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

Вторая причина чисто инженерная. При миграции меняется не только качество решений: другие коды ошибок, другое поведение при тайм-ауте, другой формат ответа. Параллельный запуск ловит эти различия на живом трафике до того, как они станут ночным инцидентом.

Локальный пример: агентство в Алматы

Типичная ситуация: агентство держит несколько десятков парсеров для клиентов и хочет предсказуемый месячный счёт вместо оплаты по факту решений. У CaptchaAI тарификация идёт по потокам — BASIC ($15/мес, 5 потоков), STANDARD ($30/мес, 15 потоков), ADVANCE ($90/мес, 50 потоков), — и число решений внутри потока не ограничено. Значит, считать надо не капчи за месяц, а задачи, висящие одновременно в пике; параллельный тест эту величину и показывает. Цены указаны в USD.

Схема стенда

Между приложением и провайдерами ставится тонкий роутер: он дублирует задачу в оба канала и пишет результаты в общий сборщик метрик. Приложение при этом продолжает использовать ответ основного провайдера — ответ CaptchaAI на первом этапе только логируется.

                    ┌──────────────┐
                    │ Your App     │
                    └──────┬───────┘
                           │
                    ┌──────▼───────┐
                    │ CAPTCHA      │
                    │ Router       │
                    └──┬───────┬───┘
                       │       │
              ┌────────▼──┐ ┌──▼────────┐
              │ Current   │ │ CaptchaAI │
              │ Provider  │ │           │
              └────────┬──┘ └──┬────────┘
                       │       │
                    ┌──▼───────▼──┐
                    │ Metrics     │
                    │ Collector   │
                    └─────────────┘

Важная деталь: роутер замеряет время сам по обеим веткам и не считает успехом просто HTTP 200.

Шаг 1. Обёртка над провайдером на Python

Оба сервиса работают по схеме in.php / res.php, поэтому достаточно одного класса с разными базовыми URL и ключами. Метод отправляет задачу, получает ID задачи и опрашивает результат до готовности, фиксируя затраченное время.

import os
import time
import requests
from dataclasses import dataclass, field
from typing import Optional
from concurrent.futures import ThreadPoolExecutor


@dataclass
class SolveResult:
    provider: str
    success: bool
    solution: Optional[str] = None
    error: Optional[str] = None
    elapsed: float = 0.0
    cost: float = 0.0


class CaptchaProvider:
    def __init__(self, name, submit_url, result_url, api_key):
        self.name = name
        self.submit_url = submit_url
        self.result_url = result_url
        self.api_key = api_key
        self.session = requests.Session()

    def solve_recaptcha(self, sitekey, pageurl):
        start = time.time()

        resp = self.session.post(self.submit_url, data={
            "key": self.api_key,
            "method": "userrecaptcha",
            "googlekey": sitekey,
            "pageurl": pageurl,
            "json": 1
        })
        data = resp.json()
        if data.get("status") != 1:
            return SolveResult(
                provider=self.name, success=False,
                error=data.get("request"), elapsed=time.time() - start
            )

        captcha_id = data["request"]

        for _ in range(60):
            time.sleep(5)
            result = self.session.get(self.result_url, params={
                "key": self.api_key, "action": "get",
                "id": captcha_id, "json": 1
            }).json()

            if result.get("status") == 1:
                return SolveResult(
                    provider=self.name, success=True,
                    solution=result["request"], elapsed=time.time() - start
                )
            if result.get("request") != "CAPCHA_NOT_READY":
                return SolveResult(
                    provider=self.name, success=False,
                    error=result.get("request"), elapsed=time.time() - start
                )

        return SolveResult(
            provider=self.name, success=False,
            error="TIMEOUT", elapsed=time.time() - start
        )

Здесь важны две вещи. elapsed считается от момента отправки, а не от первого успешного опроса — иначе вы измеряете собственный интервал опроса, а не сервис. И ошибки с тайм-аутами возвращаются тем же объектом, что и успех: неудачные попытки нужно считать, а не проглатывать.

Шаг 2. Параллельный прогон и сводный отчёт

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

class ParallelTestRunner:
    def __init__(self, primary, challenger):
        self.primary = primary
        self.challenger = challenger
        self.results = {"primary": [], "challenger": []}

    def run_test(self, sitekey, pageurl, num_runs=20):
        print(f"Running {num_runs} parallel solves...")

        for i in range(num_runs):
            with ThreadPoolExecutor(max_workers=2) as executor:
                primary_future = executor.submit(
                    self.primary.solve_recaptcha, sitekey, pageurl
                )
                challenger_future = executor.submit(
                    self.challenger.solve_recaptcha, sitekey, pageurl
                )

                primary_result = primary_future.result()
                challenger_result = challenger_future.result()

            self.results["primary"].append(primary_result)
            self.results["challenger"].append(challenger_result)

            print(f"  Run {i+1}/{num_runs}: "
                  f"{self.primary.name}={'OK' if primary_result.success else 'FAIL'} "
                  f"({primary_result.elapsed:.1f}s) | "
                  f"{self.challenger.name}={'OK' if challenger_result.success else 'FAIL'} "
                  f"({challenger_result.elapsed:.1f}s)")

        return self.generate_report()

    def generate_report(self):
        report = {}
        for label, results in self.results.items():
            total = len(results)
            successes = sum(1 for r in results if r.success)
            times = [r.elapsed for r in results if r.success]
            errors = [r.error for r in results if not r.success]

            report[label] = {
                "provider": results[0].provider if results else "unknown",
                "total": total,
                "successes": successes,
                "success_rate": (successes / total * 100) if total else 0,
                "avg_time": sum(times) / len(times) if times else 0,
                "min_time": min(times) if times else 0,
                "max_time": max(times) if times else 0,
                "errors": errors
            }

        return report


# Usage
current = CaptchaProvider(
    name="CurrentProvider",
    submit_url="https://current-provider.com/in.php",
    result_url="https://current-provider.com/res.php",
    api_key="current_key"
)

captchaai = CaptchaProvider(
    name="CaptchaAI",
    submit_url="https://ocr.captchaai.com/in.php",
    result_url="https://ocr.captchaai.com/res.php",
    api_key=os.environ["CAPTCHAAI_API_KEY"]
)

runner = ParallelTestRunner(primary=current, challenger=captchaai)
report = runner.run_test(
    sitekey="6Le-wvkSAAAAAPBMRTvw0Q4Muexq9bi0DJwx_mJ-",
    pageurl="https://example.com/form",
    num_runs=20
)

for label, stats in report.items():
    print(f"\n{stats['provider']}:")
    print(f"  Success rate: {stats['success_rate']:.1f}%")
    print(f"  Avg time: {stats['avg_time']:.1f}s")
    print(f"  Min/Max: {stats['min_time']:.1f}s / {stats['max_time']:.1f}s")
    if stats['errors']:
        print(f"  Errors: {stats['errors']}")

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

Шаг 3. Разделение трафика в продакшене

Когда совместимость подтверждена, переходите от зеркалирования к реальной доле трафика. Начните с 10 % и оставьте возврат к основному провайдеру при неудаче — тогда эксперимент не задевает пользователей.

import random


class TrafficSplitter:
    def __init__(self, primary, challenger, challenger_pct=10):
        self.primary = primary
        self.challenger = challenger
        self.challenger_pct = challenger_pct

    def solve(self, sitekey, pageurl):
        if random.randint(1, 100) <= self.challenger_pct:
            result = self.challenger.solve_recaptcha(sitekey, pageurl)
            if not result.success:
                # Fall back to primary on failure
                return self.primary.solve_recaptcha(sitekey, pageurl)
            return result
        return self.primary.solve_recaptcha(sitekey, pageurl)


# Start with 10%, increase as confidence builds
splitter = TrafficSplitter(current, captchaai, challenger_pct=10)
result = splitter.solve(sitekey="...", pageurl="...")

Долю поднимайте только после суток стабильных метрик на текущем уровне.

Шаг 4. То же самое на Node.js

Если стек на JavaScript, логика та же: те же in.php и res.php, тот же подсчёт времени, Promise.all вместо пула потоков.

const axios = require("axios");

class CaptchaProvider {
  constructor(name, submitUrl, resultUrl, apiKey) {
    this.name = name;
    this.submitUrl = submitUrl;
    this.resultUrl = resultUrl;
    this.apiKey = apiKey;
  }

  async solveRecaptcha(sitekey, pageurl) {
    const start = Date.now();
    try {
      const submit = await axios.post(this.submitUrl, null, {
        params: { key: this.apiKey, method: "userrecaptcha", googlekey: sitekey, pageurl, json: 1 },
      });
      if (submit.data.status !== 1) {
        return { provider: this.name, success: false, error: submit.data.request, elapsed: (Date.now() - start) / 1000 };
      }

      const captchaId = submit.data.request;
      for (let i = 0; i < 60; i++) {
        await new Promise((r) => setTimeout(r, 5000));
        const poll = await axios.get(this.resultUrl, {
          params: { key: this.apiKey, action: "get", id: captchaId, json: 1 },
        });
        if (poll.data.status === 1) {
          return { provider: this.name, success: true, solution: poll.data.request, elapsed: (Date.now() - start) / 1000 };
        }
        if (poll.data.request !== "CAPCHA_NOT_READY") {
          return { provider: this.name, success: false, error: poll.data.request, elapsed: (Date.now() - start) / 1000 };
        }
      }
      return { provider: this.name, success: false, error: "TIMEOUT", elapsed: (Date.now() - start) / 1000 };
    } catch (err) {
      return { provider: this.name, success: false, error: err.message, elapsed: (Date.now() - start) / 1000 };
    }
  }
}

async function parallelTest(current, captchaai, sitekey, pageurl, runs = 20) {
  const results = { current: [], captchaai: [] };

  for (let i = 0; i < runs; i++) {
    const [currentResult, captchaaiResult] = await Promise.all([
      current.solveRecaptcha(sitekey, pageurl),
      captchaai.solveRecaptcha(sitekey, pageurl),
    ]);

    results.current.push(currentResult);
    results.captchaai.push(captchaaiResult);

    console.log(`Run ${i + 1}/${runs}: ${current.name}=${currentResult.success ? "OK" : "FAIL"} ` +
      `(${currentResult.elapsed.toFixed(1)}s) | ${captchaai.name}=${captchaaiResult.success ? "OK" : "FAIL"} ` +
      `(${captchaaiResult.elapsed.toFixed(1)}s)`);
  }

  for (const [label, data] of Object.entries(results)) {
    const successes = data.filter((r) => r.success).length;
    const times = data.filter((r) => r.success).map((r) => r.elapsed);
    const avgTime = times.length ? times.reduce((a, b) => a + b, 0) / times.length : 0;
    console.log(`\n${label}: ${successes}/${runs} success (${((successes / runs) * 100).toFixed(1)}%), avg ${avgTime.toFixed(1)}s`);
  }
}

// Run
const currentProvider = new CaptchaProvider("CurrentProvider", "https://current-provider.com/in.php", "https://current-provider.com/res.php", "current_key");
const captchaai = new CaptchaProvider("CaptchaAI", "https://ocr.captchaai.com/in.php", "https://ocr.captchaai.com/res.php", process.env.CAPTCHAAI_API_KEY);

parallelTest(currentProvider, captchaai, "SITE_KEY", "https://example.com", 20);

План перехода по фазам

Между фазами держите паузу минимум в сутки: суточный цикл трафика показывает то, чего не видно за два часа.

Какие метрики сравнивать

Последняя строка — самая важная и самая забываемая. Провайдер может вернуть корректный по формату токен, который целевая страница затем отклонит; в статистику должны попадать только решения, прошедшие до конца.

Типичные ошибки при параллельном тесте

Отдельный пункт — поддерживаемые типы. Планируя тест, сверьтесь с тем, что CaptchaAI решает: reCAPTCHA v2 и v3, включая Enterprise-варианты, Cloudflare Turnstile и Cloudflare Challenge, GeeTest v3, image/OCR, grid-капчи и BLS CAPTCHA; CaptchaFox (beta), Friendly Captcha (beta) и Lemin (beta) доступны в бета-режиме. hCaptcha и FunCaptcha не поддерживаются, GeeTest v4 — в разработке. Если ваш трафик состоит в основном из неподдерживаемых типов, сравнивать будет нечего.

И если тест идёт на парсинге, помните про требования к персональным данным (для РФ — 152-ФЗ): собирайте только то, что вправе обрабатывать.

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

Сколько прогонов нужно, чтобы верить результату?

От 50 попыток на каждого провайдера как минимум. Лучше 100 и больше, распределённых по разным часам суток и дням недели — иначе вы измеряете один срез нагрузки, а не поведение сервиса.

Можно ли обойтись без удвоения расходов на время теста?

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

Как выбрать тариф по результатам теста?

Смотрите на пиковое число одновременных задач, а не на суммарный объём за месяц: тарификация у CaptchaAI идёт по потокам, а решения внутри потока не ограничены. Если в пике висит около 12 задач, подойдёт STANDARD ($30/мес, 15 потоков); если около 40 — ADVANCE ($90/мес, 50 потоков).

Что делать, если результаты расходятся только по одному типу CAPTCHA?

Разделите статистику по типам и взвесьте по реальной доле каждого типа в вашем трафике. Общая средняя цифра такие расхождения скрывает, и решение, принятое по ней, окажется неверным.

Что считать поводом откатиться на предыдущую фазу?

Заранее задайте пороги: доля успешных решений ниже базовой линии основного провайдера или P95 выше вашего тайм-аута. Откат — это возврат challenger_pct к прошлому значению, старый провайдер всё это время остаётся подключённым.

Что почитать дальше

• Быстрый старт CaptchaAI: первое решённое CAPTCHA за пять минут
• Как решить reCAPTCHA v2 через API — пошагово
• Решение Cloudflare Turnstile через API
• Решение GeeTest v3 через API