API Tutorials

CaptchaAI API 密钥轮换:多密钥管理

验证码识别量上来后,很多团队先加线程、升级套餐。但真正卡住流水线的,常常是只用了一个 API 密钥——限流、清零或禁用,任务就停摆。把流量分散到多个密钥,让轮换策略接管故障转移更稳妥。

比如 618、双11 期间,只用一个 STANDARD($30/月,15 线程)账号很容易同时打满余额和限流;把 3 个账号的密钥做加权轮换,通常就够扛住突发流量,不必升级到 ADVANCE($90/月,50 线程)。

三种轮换策略,复杂度递增:

  • 轮询轮换:顺序循环,最简单。
  • 按余额加权轮换:余额高的账号多扛流量。
  • 故障转移轮换:出错即切下一个,常与前两种搭配。

轮询轮换:最简单的负载分摊

最简单的策略是依次使用每个密钥:

Python

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)

_lock 不能省略:多线程同时读写会出现竞态条件。


故障转移轮换:一个密钥失败就切下一个

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")

达到 max_attempts 仍失败,说明密钥已耗尽,需要人工介入。

JavaScript

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');
}

Python 和 JavaScript 两个版本的重试逻辑完全对应:网络异常和业务失败都会触发换密钥重试,达到 max_attempts 才抛出异常。


用环境变量管理密钥,别硬编码

  • API 密钥不能写进代码仓库,一律从环境变量读取。
  • 只放进 .env 或 CI/CD 密钥库(如 GitHub Actions Secrets),加入 .gitignore
  • 怀疑泄露时立即停用该密钥、生成新密钥,并从轮换池中移除旧密钥。
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);

最短路径:读取环境变量 → 拆分成列表 → 传给 KeyRotator


定时刷新余额,别等到密钥用尽才发现

长期运行的服务应定期主动刷新余额:

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

5 分钟是经验值:密钥池小可缩短到 1–2 分钟,密钥多时适当拉长,避免频繁调用 getbalance


三种策略怎么选

按场景对号入座:

场景 推荐策略 密钥数量参考
单账号 STANDARD($30/月,15 线程),流量平稳 轮询轮换 2–3 个
多账号套餐,余额差异大 按余额加权轮换 3–5 个
618、双11 大促流量突增 加权轮换 + 故障转移兜底 3 个以上

故障排查

先看余额,再看索引,再核对错误码。

问题 原因 处理方式
所有密钥都被禁用 所有账户余额都是零 给账户充值,检查是否命中 ERROR_ZERO_BALANCE
一直在用同一个密钥 轮询索引没有正确推进 确认索引更新有加锁保护,排查线程安全问题
密钥被错误禁用 把临时性错误当成了永久性错误 只在 ERROR_WRONG_USER_KEYERROR_ZERO_BALANCEERROR_IP_NOT_ALLOWED 上禁用密钥

常见问题

落地时最容易卡壳的几个问题:

应该用几个 API 密钥做轮换?

两个密钥提供基本故障转移,三个以上可做负载分摊。识别量 1000+ 次/天时,建议 3–5 个密钥配合加权。

密钥轮换会增加我的账单吗?

不会。CaptchaAI 按线程数计费,不按密钥数计费。同账号轮换无额外费用;跨账号轮换,各账号按自身套餐计费。

怎么避免轮换一直卡在一个已失效的密钥上?

只对 ERROR_WRONG_USER_KEYERROR_ZERO_BALANCE 等永久性错误禁用密钥,网络超时等临时错误不应永久踢出。

可以把不同 CaptchaAI 账号的密钥放进同一个轮换池吗?

可以,每个密钥的余额和限流额度相互独立,轮换器会分别处理。

密钥被禁用后要不要设置告警?

小规模场景,日志够用;密钥多或对故障敏感,建议把禁用事件推到告警渠道(如 Slack webhook)。


用多密钥轮换稳定你的验证码识别流水线

机制可按流量规模逐步上:

  1. 先用 2–3 个密钥轮询。
  2. 流量涨了切到加权。
  3. 对故障敏感再叠加告警。

注册 CaptchaAI,拿到密钥即可接入上面的轮换代码:CaptchaAI 官网


相关指南

延伸阅读:

该文章已禁用评论。