У агентства, которое ведёт парсинг для пяти-шести клиентов одновременно, отдельный скрипт решения CAPTCHA под каждый проект быстро превращается в технический долг: правки в одном месте не попадают в остальные, а отладка растягивается на все проекты сразу. Решение — один переиспользуемый конвейер: очередь задач и пул воркеров, а особенности клиента (прокси, лимит потоков, тип CAPTCHA) остаются настройками, а не кодом.
На практике переход происходит не сразу, а после второго-третьего похожего инцидента: у клиента A изменился sitekey, фикс внесли только в его скрипт, а через неделю клиент C падает с той же самой проблемой, потому что копия кода у него осталась старой. Один конвейер с общей кодовой базой и конфигурацией на клиента снимает этот класс ошибок целиком. Ниже — рабочая архитектура и код на Python и Node.js.
Архитектура конвейера для нескольких клиентов
┌──────────────┐ ┌───────────────┐ ┌──────────────┐
│ Client A │──▶ │ │ │ │
│ Client B │──▶ │ Task Queue │──▶ │ CaptchaAI │
│ Client C │──▶ │ │ │ API │
└──────────────┘ └───────────────┘ └──────────────┘
│ │
▼ ▼
┌───────────────┐ ┌──────────────┐
│ Result Store │◀── │ Polling │
│ (Redis/DB) │ │ Workers │
└───────────────┘ └──────────────┘
Четыре компонента отвечают каждый за свою часть конвейера — их стоит держать раздельно даже в небольшом проекте, чтобы отладка одного не задевала остальные.
Приём задач
Принимает запросы на решение от парсеров разных клиентов и сразу помечает каждую задачу идентификатором клиента — без этой метки дальнейшая маршрутизация результатов невозможна.
Очередь
Буферизует задачи и держит отдельный лимит параллелизма на каждого клиента, чтобы один проект не съел все потоки и не заблокировал остальных на время пиковой нагрузки.
Воркеры-решатели
Отправляют задачу в CaptchaAI и опрашивают res.php до получения токена. Количество воркеров масштабируется независимо от количества клиентов — это просто пул потоков, который разбирает общую очередь.
Хранилище результатов
Держит решённые токены, пока их не заберёт вызывающий сервис. При параллельной работе с несколькими клиентами это единственное место, где токены физически могут перепутаться, если ключ не включает client_id.
Когда объединять клиентов в одном конвейере, а когда разносить
Для агентства, которое ведёт клиентов из СНГ и Европы одновременно, единый тариф в долларах на весь конвейер удобнее, чем плата за каждое решение отдельно: план ADVANCE ($90/мес, 50 потоков) покрывает нагрузку сразу нескольких проектов, а стоимость не зависит от курса. Закладывайте лимит на клиента заметно ниже общего числа потоков плана. Разносить клиентов по отдельным инстансам конвейера имеет смысл в трёх случаях:
- клиент требует изоляции по контракту (свой сервер, свои логи, отдельный биллинг);
- нагрузка одного клиента настолько велика, что регулярно упирается в лимит плана;
- у клиента специфический прокси-пул, который нельзя смешивать с трафиком остальных проектов.
Во всех остальных случаях один конвейер с конфигурацией на клиента (см. ниже) обходится дешевле и проще в поддержке, чем N изолированных копий.
Реализация конвейера на Python
Базовый класс-решатель
import requests
import time
from dataclasses import dataclass
from typing import Optional
from collections import deque
from threading import Lock
SUBMIT_URL = "https://ocr.captchaai.com/in.php"
RESULT_URL = "https://ocr.captchaai.com/res.php"
@dataclass
class SolveRequest:
client_id: str
method: str
params: dict
callback: Optional[callable] = None
@dataclass
class SolveResult:
client_id: str
task_id: str
token: Optional[str] = None
error: Optional[str] = None
class CaptchaPipeline:
def __init__(self, api_key: str, max_concurrent: int = 10):
self.api_key = api_key
self.max_concurrent = max_concurrent
self.queue = deque()
self.active = {}
self.lock = Lock()
def enqueue(self, request: SolveRequest):
with self.lock:
self.queue.append(request)
def submit_task(self, request: SolveRequest) -> Optional[str]:
data = {
"key": self.api_key,
"method": request.method,
"json": 1,
**request.params
}
try:
resp = requests.post(SUBMIT_URL, data=data, timeout=15)
result = resp.json()
if result.get("status") == 1:
return result["request"]
else:
print(f"[{request.client_id}] Submit error: {result.get('error_text', result.get('request'))}")
return None
except requests.RequestException as e:
print(f"[{request.client_id}] Network error: {e}")
return None
def poll_result(self, task_id: str, max_wait: int = 120) -> Optional[str]:
elapsed = 0
interval = 5
while elapsed < max_wait:
time.sleep(interval)
elapsed += interval
try:
resp = requests.get(RESULT_URL, params={
"key": self.api_key,
"action": "get",
"id": task_id,
"json": 1
}, timeout=10)
result = resp.json()
if result.get("status") == 1:
return result["request"]
elif result.get("request") == "CAPCHA_NOT_READY":
continue
else:
print(f"Poll error for {task_id}: {result.get('error_text', result.get('request'))}")
return None
except requests.RequestException:
continue
return None
def process_queue(self):
while self.queue or self.active:
# Fill active slots
with self.lock:
while self.queue and len(self.active) < self.max_concurrent:
request = self.queue.popleft()
task_id = self.submit_task(request)
if task_id:
self.active[task_id] = request
# Poll active tasks
completed = []
for task_id, request in list(self.active.items()):
token = self.poll_result(task_id, max_wait=10)
if token:
result = SolveResult(
client_id=request.client_id,
task_id=task_id,
token=token
)
if request.callback:
request.callback(result)
completed.append(task_id)
with self.lock:
for task_id in completed:
del self.active[task_id]
Пример на нескольких клиентах
pipeline = CaptchaPipeline(api_key="YOUR_API_KEY", max_concurrent=15)
# Client A — reCAPTCHA v2
pipeline.enqueue(SolveRequest(
client_id="client_a",
method="userrecaptcha",
params={
"googlekey": "6Le-SITEKEY-A",
"pageurl": "https://client-a-https://staging.example.com/qa-form"
},
callback=lambda r: print(f"[{r.client_id}] Solved: {r.token[:40]}...")
))
# Client B — Turnstile
pipeline.enqueue(SolveRequest(
client_id="client_b",
method="turnstile",
params={
"sitekey": "0x4AAAA-SITEKEY-B",
"pageurl": "https://client-b-target.com/login"
},
callback=lambda r: print(f"[{r.client_id}] Solved: {r.token[:40]}...")
))
pipeline.process_queue()
Реализация конвейера на Node.js
const axios = require("axios");
const SUBMIT_URL = "https://ocr.captchaai.com/in.php";
const RESULT_URL = "https://ocr.captchaai.com/res.php";
class CaptchaPipeline {
constructor(apiKey, maxConcurrent = 10) {
this.apiKey = apiKey;
this.maxConcurrent = maxConcurrent;
this.queue = [];
this.activeCount = 0;
}
enqueue(clientId, method, params) {
return new Promise((resolve, reject) => {
this.queue.push({ clientId, method, params, resolve, reject });
this._processNext();
});
}
async _processNext() {
if (this.activeCount >= this.maxConcurrent || this.queue.length === 0) return;
this.activeCount++;
const task = this.queue.shift();
try {
const token = await this._solve(task);
task.resolve({ clientId: task.clientId, token });
} catch (err) {
task.reject(err);
} finally {
this.activeCount--;
this._processNext();
}
}
async _solve(task) {
const submitResp = await axios.post(SUBMIT_URL, null, {
params: {
key: this.apiKey,
method: task.method,
json: 1,
...task.params,
},
timeout: 15000,
});
if (submitResp.data.status !== 1) {
throw new Error(submitResp.data.error_text || submitResp.data.request);
}
const taskId = submitResp.data.request;
return this._poll(taskId);
}
async _poll(taskId, maxWait = 120000) {
const interval = 5000;
let elapsed = 0;
while (elapsed < maxWait) {
await new Promise((r) => setTimeout(r, interval));
elapsed += interval;
try {
const resp = await axios.get(RESULT_URL, {
params: {
key: this.apiKey,
action: "get",
id: taskId,
json: 1,
},
timeout: 10000,
});
if (resp.data.status === 1) return resp.data.request;
if (resp.data.request !== "CAPCHA_NOT_READY") {
throw new Error(resp.data.error_text || resp.data.request);
}
} catch (err) {
if (err.response) throw err;
}
}
throw new Error(`Timeout waiting for task ${taskId}`);
}
}
// Usage
(async () => {
const pipeline = new CaptchaPipeline("YOUR_API_KEY", 15);
const results = await Promise.allSettled([
pipeline.enqueue("client_a", "userrecaptcha", {
googlekey: "6Le-SITEKEY-A",
pageurl: "https://client-a-https://staging.example.com/qa-form",
}),
pipeline.enqueue("client_b", "turnstile", {
sitekey: "0x4AAAA-SITEKEY-B",
pageurl: "https://client-b-target.com/login",
}),
]);
results.forEach((r) => {
if (r.status === "fulfilled") {
console.log(`[${r.value.clientId}] Token: ${r.value.token.slice(0, 40)}...`);
} else {
console.error(`Failed: ${r.reason.message}`);
}
});
})();
Конфигурация на клиента: прокси, лимиты, тип CAPTCHA
Настройки каждого клиента — прокси, предпочтительный метод решения, лимит параллелизма — держите отдельно от бизнес-логики воркера, иначе при добавлении нового клиента придётся править код конвейера:
CLIENT_CONFIG = {
"client_a": {
"proxy": "host:port:user:pass",
"proxytype": "HTTP",
"max_concurrent": 5,
"default_method": "userrecaptcha"
},
"client_b": {
"proxy": None,
"proxytype": None,
"max_concurrent": 10,
"default_method": "turnstile"
}
}
def build_params(client_id, params):
config = CLIENT_CONFIG.get(client_id, {})
if config.get("proxy"):
params["proxy"] = config["proxy"]
params["proxytype"] = config["proxytype"]
return params
Мониторинг конвейера в production
Когда через один конвейер идёт трафик пяти-шести клиентов, точечные print() в логах быстро перестают помогать — нужны агрегированные метрики на уровне конвейера и на уровне каждого клиента отдельно:
- среднее время решения по типу CAPTCHA и отдельно по клиенту — рост у одного клиента обычно означает проблему с его прокси-пулом, а не с CaptchaAI;
- доля ошибок
submit_task/poll_resultза скользящее окно — резкий скачок раньше, чем закончится баланс, сигнализирует о проблеме с ключом или сетью; - глубина очереди на клиента — если она стабильно растёт,
max_concurrentдля этого клиента занижен относительно его реальной нагрузки; - возраст самой старой активной задачи — простой способ поймать «зависшую» задачу, которая не укладывается в
max_waitи блокирует слот.
Эти четыре метрики достаточно писать в тот же Redis или БД, где хранится очередь — отдельная система мониторинга на старте не нужна.
Как реагировать на ошибки API
| Ошибка | Реакция |
|---|---|
ERROR_ZERO_BALANCE |
Остановить очередь, уведомить всех клиентов |
ERROR_NO_SLOT_AVAILABLE |
Вернуть задачу в очередь с задержкой |
ERROR_WRONG_CAPTCHA_ID |
Отбросить задачу, записать ошибку в лог |
ERROR_CAPTCHA_UNSOLVABLE |
Повторить один раз, затем пометить как неудачную |
| Сетевой тайм-аут | Повторить с задержкой (не более 3 попыток) |
Типичные проблемы и их причины
Четыре симптома встречаются в мультиклиентских конвейерах чаще остальных — по каждому сразу ниже причина и решение.
Очередь растёт без ограничений
Все активные слоты заняты. Увеличьте max_concurrent или добавьте воркеров — если глубина очереди из раздела про мониторинг выше растёт у одного конкретного клиента, поднимайте лимит точечно для него, а не для конвейера целиком.
Callback не срабатывает
Задача завершилась ошибкой молча. Проверьте обработку ошибки в цикле опроса — самая частая причина в том, что ветка else в poll_result логирует ошибку, но не вызывает callback с признаком неудачи, и вызывающий код зависает в ожидании.
Токены путаются между клиентами
Общее хранилище результатов без разграничения. Ключуйте результаты парой client_id + task_id — использование одного task_id в качестве ключа выглядит рабочим на тесте с одним клиентом и ломается только под реальной многоклиентской нагрузкой.
Ошибки лимита частоты (429)
Слишком много одновременных отправок. Снизьте параллелизм, добавьте задержку между отправками — обычно достаточно развести отправку задач разных клиентов на 100–200 мс, не трогая общий max_concurrent.
Часто задаваемые вопросы
Сколько потоков закладывать на одного клиента при плане ADVANCE?
При 50 потоках на плане ADVANCE ($90/мес) разумный старт — 5–10 потоков на клиента с запасом под пиковую нагрузку остальных проектов. Смотрите на реальное время решения и долю ошибок в логах и перераспределяйте лимиты по факту, а не заранее.
Что делать, если один клиент внезапно занимает почти все потоки конвейера?
Держите жёсткий потолок max_concurrent на клиента внутри общего пула — конфигурация из раздела про настройки на клиента выше для этого и нужна. Если всплеск нагрузки регулярный, а не разовый, дешевле поднять лимит точечно этому клиенту, чем расширять план ADVANCE для всех. Если один клиент стабильно требует больше половины потоков плана, это уже повод вынести его в отдельный инстанс конвейера.
Можно ли гонять по одному конвейеру и боевой парсинг, и QA-тестирование?
Технически да — конвейер не различает боевые и тестовые задачи, если их не пометить явно. На практике добавьте к client_id суффикс вроде _qa и заведите для него отдельный max_concurrent в CLIENT_CONFIG, чтобы нагрузочное тестирование не съедало потоки боевых клиентов и не искажало метрики из раздела про мониторинг.
Сколько времени занимает подключение нового клиента к уже работающему конвейеру?
Обычно это конфигурация, а не разработка: добавить запись в CLIENT_CONFIG (прокси, лимит, тип CAPTCHA по умолчанию) и передать клиентский client_id парсеру. Код воркера и очереди не меняется.
Нужен ли отдельный пул воркеров под reCAPTCHA v2 и под Turnstile?
Не обязательно — оба метода проходят через один и тот же submit_task/poll_result, различается только параметр method и поля в params. Разделять воркеры стоит, только если время решения у типов сильно отличается и медленные задачи нужно изолировать от быстрых.
Подключите CaptchaAI к своему конвейеру
Начните создавать клиентские конвейеры на captchaai.com.
