Если скрипт время от времени падает с ERROR_ZERO_BALANCE или упирается в лимит потоков посреди рабочего дня — скорее всего, весь трафик идёт через один API-ключ CaptchaAI. Решение простое: развести нагрузку на несколько ключей и настроить автоматическое переключение при сбое, чтобы отказ одного ключа не останавливал пайплайн целиком.
Это не разовая настройка «на всякий случай», а рабочая часть инфраструктуры: чем больше потоков и чем выше объём решений в сутки, тем быстрее один ключ упирается в лимит или требует ручного вмешательства. Ниже — три стратегии ротации от простой до промышленной, с готовым кодом для каждой.
| Стратегия | Когда использовать |
|---|---|
| Циклическая (round-robin) | Ключи одного тарифа, задача — просто распределить нагрузку |
| Взвешенная по балансу | Ключи разных клиентов/тарифов с разным остатком |
| Аварийное переключение (failover) | Нужна отказоустойчивость при сбое или деактивации ключа |
Циклическая ротация (round-robin)
Самый простой вариант — раздавать ключи по очереди, равномерно. Такой схемы достаточно, если:
- все ключи принадлежат одному аккаунту или тарифу и имеют сопоставимый лимит потоков;
- баланс на ключах пополняется централизованно, а не по клиентам;
- задача — распределить нагрузку, а не изолировать биллинг между заказчиками.
Если хотя бы одно из условий не выполняется — например, ключи с разным балансом или разными тарифами — переходите сразу к взвешенной ротации ниже, иначе часть ключей будет простаивать, а другие упрутся в лимит раньше времени.
Python
Базовая реализация на встроенном itertools.cycle:
import itertools
import requests
API_KEYS = [
"KEY_ACCOUNT_1",
"KEY_ACCOUNT_2",
"KEY_ACCOUNT_3",
]
key_cycle = itertools.cycle(API_KEYS)
def get_next_key():
return next(key_cycle)
def solve_captcha(sitekey, page_url):
api_key = get_next_key()
resp = requests.post("https://ocr.captchaai.com/in.php", data={
"key": api_key,
"method": "userrecaptcha",
"googlekey": sitekey,
"pageurl": page_url,
"json": "1",
})
data = resp.json()
if data["status"] != 1:
raise Exception(f"[{api_key[:8]}...] {data['request']}")
print(f"Submitted with key {api_key[:8]}...")
return data["request"], api_key
task_id, used_key = solve_captcha("6Le-SITEKEY", "https://example.com")
Взвешенная ротация с учётом баланса ключей
Направляйте больше запросов на ключи с бóльшим балансом. Это уместно, когда ключи выданы разным клиентам или подразделениям и пополняются независимо друг от друга: ключ с бóльшим остатком должен получать больше трафика, а ключ на грани нуля — постепенно выводиться из ротации, а не отваливаться резко посреди пакетной задачи.
import random
import requests
import threading
SUBMIT_URL = "https://ocr.captchaai.com/in.php"
RESULT_URL = "https://ocr.captchaai.com/res.php"
class KeyRotator:
def __init__(self, keys):
self.keys = {k: {"balance": 0, "failures": 0, "disabled": False} for k in keys}
self._lock = threading.Lock()
self.refresh_balances()
def refresh_balances(self):
for key in self.keys:
try:
resp = requests.get(RESULT_URL, params={
"key": key, "action": "getbalance", "json": "1"
}, timeout=10).json()
if resp["status"] == 1:
self.keys[key]["balance"] = float(resp["request"])
self.keys[key]["disabled"] = False
else:
self.keys[key]["disabled"] = True
except Exception:
self.keys[key]["disabled"] = True
def get_key(self):
with self._lock:
available = {
k: v for k, v in self.keys.items()
if not v["disabled"] and v["balance"] > 0.01
}
if not available:
raise Exception("No API keys with balance available")
# Weighted random by balance
keys = list(available.keys())
weights = [available[k]["balance"] for k in keys]
return random.choices(keys, weights=weights, k=1)[0]
def report_failure(self, key, error_code):
with self._lock:
self.keys[key]["failures"] += 1
if error_code in ("ERROR_WRONG_USER_KEY", "ERROR_KEY_DOES_NOT_EXIST",
"ERROR_ZERO_BALANCE", "ERROR_IP_NOT_ALLOWED"):
self.keys[key]["disabled"] = True
print(f"[rotator] Disabled key {key[:8]}...: {error_code}")
def report_success(self, key, cost=0.003):
with self._lock:
self.keys[key]["balance"] -= cost
self.keys[key]["failures"] = 0
rotator = KeyRotator(["KEY_1", "KEY_2", "KEY_3"])
# Usage
api_key = rotator.get_key()
# ... solve captcha ...
rotator.report_success(api_key)
Ключ с нулевым балансом просто выпадает из выборки. Обратите внимание на блокировку (threading.Lock) вокруг выбора ключа и обновления баланса — без неё два потока могут одновременно прочитать один и тот же остаток и «перерасходовать» баланс ключа до того, как счётчик обновится.
Аварийное переключение между ключами при сбое
Отказавший ключ не должен ронять задачу — она уходит следующему доступному. Для пользователя или скрипта, который ждёт решение CAPTCHA, разница между «упал один ключ» и «упал весь пайплайн» — это разница между незаметным повтором и инцидентом, который придётся разбирать вручную.
Python
Оборачиваем вызов в цикл попыток — каждая неудача переводит запрос на следующий ключ:
def solve_with_failover(sitekey, page_url, max_attempts=3):
for attempt in range(max_attempts):
api_key = rotator.get_key()
try:
resp = requests.post(SUBMIT_URL, data={
"key": api_key,
"method": "userrecaptcha",
"googlekey": sitekey,
"pageurl": page_url,
"json": "1",
}, timeout=15)
data = resp.json()
if data["status"] != 1:
rotator.report_failure(api_key, data["request"])
continue
rotator.report_success(api_key)
return data["request"], api_key
except requests.RequestException:
rotator.report_failure(api_key, "NETWORK_ERROR")
continue
raise Exception(f"All {max_attempts} keys failed")
JavaScript
Тот же принцип на Node.js с axios:
const axios = require('axios');
class KeyRotator {
constructor(keys) {
this.keys = keys.map(k => ({ key: k, disabled: false, failures: 0 }));
this.index = 0;
}
getKey() {
const available = this.keys.filter(k => !k.disabled);
if (available.length === 0) throw new Error('No API keys available');
const entry = available[this.index % available.length];
this.index++;
return entry.key;
}
disable(key, reason) {
const entry = this.keys.find(k => k.key === key);
if (entry) {
entry.disabled = true;
console.log(`[rotator] Disabled ${key.substring(0, 8)}...: ${reason}`);
}
}
}
const rotator = new KeyRotator(['KEY_1', 'KEY_2', 'KEY_3']);
async function solveWithFailover(sitekey, pageurl, maxAttempts = 3) {
for (let i = 0; i < maxAttempts; i++) {
const apiKey = rotator.getKey();
try {
const resp = await axios.post('https://ocr.captchaai.com/in.php', null, {
params: { key: apiKey, method: 'userrecaptcha', googlekey: sitekey, pageurl, json: 1 }
});
if (resp.data.status !== 1) {
rotator.disable(apiKey, resp.data.request);
continue;
}
return { taskId: resp.data.request, apiKey };
} catch (err) {
rotator.disable(apiKey, 'NETWORK_ERROR');
}
}
throw new Error('All keys failed');
}
Загрузка ключей из переменных окружения
Ключи в коде рано или поздно попадут в git-историю. Загружайте их из окружения, как показано ниже.
Для одиночного сервиса переменной окружения достаточно. Когда ключей становится больше и они начинают относиться к разным окружениям (staging, продакшн, разные клиенты), удобнее вынести их в отдельное хранилище:
- секрет-менеджер облачной платформы (AWS Secrets Manager, Google Secret Manager, Azure Key Vault);
- Vault для self-hosted инфраструктуры;
- секреты CI/CD-системы для конвейеров, которые запускаются по расписанию.
import os
API_KEYS = os.environ["CAPTCHAAI_KEYS"].split(",")
# Set: CAPTCHAAI_KEYS=key1,key2,key3
rotator = KeyRotator(API_KEYS)
const API_KEYS = process.env.CAPTCHAAI_KEYS.split(',');
const rotator = new KeyRotator(API_KEYS);
Если ключ всё же попал в публичный репозиторий или лог — деактивируйте его в личном кабинете CaptchaAI и выпустите новый, а не полагайтесь на то, что «никто не заметит».
Плановое обновление баланса по расписанию
Для долгоживущих процессов обновляйте балансы периодически, а не только при ошибке. Это дешевле, чем ждать первого сбоя: фоновая задача видит, что баланс ключа приближается к нулю, и вы успеваете пополнить его до того, как ротатор начнёт отбрасывать запросы этому ключу в проде.
import threading
def periodic_refresh(rotator, interval=300):
def refresh():
while True:
rotator.refresh_balances()
for key, info in rotator.keys.items():
print(f" {key[:8]}...: ${info['balance']:.2f} "
f"{'(disabled)' if info['disabled'] else '(active)'}")
threading.Event().wait(interval)
t = threading.Thread(target=refresh, daemon=True)
t.start()
periodic_refresh(rotator, interval=300) # every 5 minutes
Когда ротация ключей реально нужна: пример из практики
Агентство ведёт QA и парсинг для клиентов из России, Казахстана и Беларуси: один на BASIC ($15/мес, 5 потоков), другой — на STANDARD ($30/мес, 15 потоков). Ключи изолированы по клиентам для честного биллинга. При нестабильных региональных сетях ключ клиента может упереться в лимит потоков раньше остальных — ротатор с failover передаёт задачу следующему доступному, и клиент не видит лишних ошибок.
Такая изоляция по ключам решает и второй вопрос — прозрачность биллинга: агентство видит расход каждого клиента отдельно по балансу его ключа, а не оценивает его на глаз по общему счёту. Если клиент вырастает в объёме, для него просто заводится ключ на тариф выше (например, переход с BASIC на STANDARD) — остальная ротация не меняется.
Диагностика типичных проблем ротации
| Проблема | Причина | Решение |
|---|---|---|
| Все ключи отключены | Баланс обнулился везде | Пополните баланс, проверьте ERROR_ZERO_BALANCE |
| Ротатор выбирает один и тот же ключ | Индекс не растёт из-за гонки потоков | Инкрементируйте индекс под блокировкой (Lock) |
| Ключ отключается ошибочно | Временный сбой принят за постоянный | Отключайте только по ERROR_WRONG_USER_KEY, ERROR_ZERO_BALANCE, ERROR_IP_NOT_ALLOWED |
Большинство проблем ротации сводятся к одному из двух источников: гонка потоков вокруг общего состояния (индекс, баланс) или неверная классификация ошибки как постоянной. Оба чинятся кодом из разделов выше — блокировкой при выборе ключа и точным списком кодов ошибок для отключения.
Частые вопросы
Сколько ключей API нужно для стабильной ротации?
Двух ключей достаточно для базового аварийного переключения. Три и больше дают реальное распределение нагрузки — для объёма от 1000 решений в день разумно держать 3–5 ключей.
Ротация ключей — это то же самое, что увеличить число потоков в одном плане?
Нет. Потоки — лимит одновременных решений внутри аккаунта (ADVANCE — 50 потоков за $90/мес). Ротация ключей — несколько независимых аккаунтов со своим балансом, нужна для изоляции по клиентам или отказоустойчивости.
По каким ошибкам ключ стоит отключать навсегда, а не повторять попытку?
Только по ERROR_WRONG_USER_KEY, ERROR_KEY_DOES_NOT_EXIST, ERROR_ZERO_BALANCE и ERROR_IP_NOT_ALLOWED. Сетевые тайм-ауты временны — для них нужен повтор, а не отключение ключа.
Как хранить несколько ключей, чтобы не хардкодить их в коде?
Читайте их из переменной окружения (CAPTCHAAI_KEYS) через запятую — см. раздел выше. Для команды лучше подключить секрет-хранилище, а не .env-файл рядом с кодом.
Как узнать, что ключ вот-вот отключится, до того как он начнёт ронять задачи?
Опрашивайте getbalance по расписанию (см. раздел про плановое обновление) и логируйте баланс каждого ключа. Настройте алерт на пороговое значение — например, когда остаток падает ниже стоимости 50–100 решений, — чтобы пополнить ключ заранее, а не разбирать инцидент по ERROR_ZERO_BALANCE в проде.
Итоги
- Один ключ — точка отказа; несколько ключей с ротацией убирают её.
- Round-robin подходит для однородных ключей одного тарифа; взвешенная ротация — когда балансы и лимиты у ключей разные.
- Failover должен отключать ключ только по
ERROR_WRONG_USER_KEY,ERROR_KEY_DOES_NOT_EXIST,ERROR_ZERO_BALANCE,ERROR_IP_NOT_ALLOWED— сетевые тайм-ауты не повод для постоянного отключения. - Ключи храните в переменных окружения или секрет-хранилище, не в коде.
Масштабируйте решение CAPTCHA с ротацией API-ключей
Получите API-ключ на captchaai.com.
Похожие руководства
- Белый список IP-адресов и безопасность API-ключей CaptchaAI
- Как защитить учётные данные CaptchaAI в переменных окружения
- Проверка баланса CaptchaAI и автоматическое пополнение
