验证码解析失败怎么处理？CaptchaAI 优雅降级三种方案

验证码请求失败很正常——超时、余额不足都可能发生。

关键是脚本怎么反应：崩溃丢光进度，还是优雅跳过、重试或降级？下面三种容错模式可直接套用。

方案一：跳过失败任务

批量场景下，单个失败不该拖累整体——跳过它，继续下一个：

import requests
import time

API_KEY = "YOUR_API_KEY"


def solve_or_skip(captcha_type, sitekey, page_url, max_retries=2):
    """Try to solve; return None on failure instead of crashing."""
    for attempt in range(max_retries):
        try:
            token = solve_captcha(captcha_type, sitekey, page_url)
            if token:
                return token
        except Exception as e:
            print(f"Attempt {attempt + 1} failed: {e}")

    return None  # Skip this item


def process_urls(urls):
    results = []
    skipped = []

    for url in urls:
        sitekey = extract_sitekey(url)
        if not sitekey:
            skipped.append({"url": url, "reason": "no_sitekey"})
            continue

        token = solve_or_skip("recaptcha_v2", sitekey, url)
        if token:
            data = submit_form(url, token)
            results.append({"url": url, "data": data})
        else:
            skipped.append({"url": url, "reason": "solve_failed"})

    print(f"Processed: {len(results)}, Skipped: {len(skipped)}")
    return results, skipped

方案二：失败任务放入重试队列

失败任务不必立刻放弃，放进重试队列，指数退避延后再试：

from collections import deque
import json

class RetryQueue:
    def __init__(self, max_retries=3, backoff_base=60):
        self.queue = deque()
        self.max_retries = max_retries
        self.backoff_base = backoff_base

    def add(self, task):
        task["retry_count"] = task.get("retry_count", 0) + 1
        if task["retry_count"] <= self.max_retries:
            task["retry_after"] = time.time() + (
                self.backoff_base * task["retry_count"]
            )
            self.queue.append(task)
            return True
        return False  # Exceeded max retries

    def get_ready(self):
        """Get tasks ready for retry."""
        ready = []
        remaining = deque()
        now = time.time()

        while self.queue:
            task = self.queue.popleft()
            if task["retry_after"] <= now:
                ready.append(task)
            else:
                remaining.append(task)

        self.queue = remaining
        return ready

    def save(self, filepath="retry_queue.json"):
        with open(filepath, "w") as f:
            json.dump(list(self.queue), f)

    def load(self, filepath="retry_queue.json"):
        try:
            with open(filepath) as f:
                self.queue = deque(json.load(f))
        except FileNotFoundError:
            pass


# Usage
retry_q = RetryQueue()

def process_with_retry(task):
    try:
        token = solve_captcha(task["type"], task["sitekey"], task["url"])
        if token:
            return submit_form(task["url"], token)
        else:
            retry_q.add(task)
    except Exception:
        retry_q.add(task)

# Process retry queue periodically
def drain_retry_queue():
    ready = retry_q.get_ready()
    for task in ready:
        process_with_retry(task)

方案三：进入降级模式

识别服务持续不可用时，与其让每个请求都报错一次，不如主动切到受限模式：

class CaptchaSolver:
    def __init__(self, api_key):
        self.api_key = api_key
        self.degraded = False
        self.failure_count = 0
        self.failure_threshold = 5
        self.recovery_time = None

    def solve(self, captcha_type, sitekey, page_url):
        if self.degraded:
            if time.time() < self.recovery_time:
                return self._degraded_action(page_url)
            else:
                self.degraded = False
                self.failure_count = 0

        try:
            token = self._solve_api(captcha_type, sitekey, page_url)
            self.failure_count = 0
            return token
        except Exception as e:
            self.failure_count += 1
            if self.failure_count >= self.failure_threshold:
                self._enter_degraded_mode()
            raise

    def _enter_degraded_mode(self):
        self.degraded = True
        self.recovery_time = time.time() + 300  # 5 min
        print("Entering degraded mode for 5 minutes")
        # Send alert

    def _degraded_action(self, url):
        """What to do when solving is unavailable."""
        # Option A: Skip CAPTCHA pages entirely
        return None

        # Option B: Queue for later
        # retry_queue.add({"url": url, ...})
        # return None

        # Option C: Try alternative solver
        # return self._solve_with_backup_api(...)

    def _solve_api(self, captcha_type, sitekey, page_url):
        # Normal CaptchaAI API call
        resp = requests.post("https://ocr.captchaai.com/in.php", data={
            "key": self.api_key,
            "method": "userrecaptcha",
            "googlekey": sitekey,
            "pageurl": page_url,
            "json": "1",
        }).json()

        if resp["status"] != 1:
            raise Exception(resp["request"])

        task_id = resp["request"]
        for _ in range(24):
            time.sleep(5)
            result = requests.get("https://ocr.captchaai.com/res.php", params={
                "key": self.api_key, "action": "get",
                "id": task_id, "json": "1"
            }).json()
            if result["status"] == 1:
                return result["request"]
            if result["request"] != "CAPCHA_NOT_READY":
                raise Exception(result["request"])

        raise Exception("TIMEOUT")

Node.js 实现：三种模式合一

Node.js 技术栈可以把跳过、重试队列和降级模式合并进同一个类：

class ResilientSolver {
  constructor(apiKey) {
    this.apiKey = apiKey;
    this.retryQueue = [];
    this.failureCount = 0;
    this.degraded = false;
  }

  async solve(type, sitekey, pageUrl) {
    if (this.degraded) {
      this.retryQueue.push({ type, sitekey, pageUrl, addedAt: Date.now() });
      return null;
    }

    try {
      const token = await this._callApi(type, sitekey, pageUrl);
      this.failureCount = 0;
      return token;
    } catch (err) {
      this.failureCount++;

      if (err.message === 'ERROR_ZERO_BALANCE') {
        this._enterDegraded(600000); // 10 min
        return null;
      }

      if (this.failureCount >= 5) {
        this._enterDegraded(300000); // 5 min
      }

      this.retryQueue.push({ type, sitekey, pageUrl, addedAt: Date.now() });
      return null;
    }
  }

  _enterDegraded(durationMs) {
    this.degraded = true;
    console.warn(`Degraded mode for ${durationMs / 1000}s`);
    setTimeout(() => {
      this.degraded = false;
      this.failureCount = 0;
      this.drainRetryQueue();
    }, durationMs);
  }

  async drainRetryQueue() {
    const tasks = this.retryQueue.splice(0);
    for (const task of tasks) {
      await this.solve(task.type, task.sitekey, task.pageUrl);
    }
  }

  async _callApi(type, sitekey, pageUrl) {
    // Standard submit + poll
    const axios = require('axios');
    const submit = await axios.post('https://ocr.captchaai.com/in.php', null, {
      params: { key: this.apiKey, method: 'userrecaptcha', googlekey: sitekey, pageurl: pageUrl, json: 1 },
    });
    if (submit.data.status !== 1) throw new Error(submit.data.request);

    const taskId = submit.data.request;
    for (let i = 0; i < 24; i++) {
      await new Promise(r => setTimeout(r, 5000));
      const poll = await axios.get('https://ocr.captchaai.com/res.php', {
        params: { key: this.apiKey, action: 'get', id: taskId, json: 1 },
      });
      if (poll.data.status === 1) return poll.data.request;
      if (poll.data.request !== 'CAPCHA_NOT_READY') throw new Error(poll.data.request);
    }
    throw new Error('TIMEOUT');
  }
}

实战场景：跨境电商爬虫的组合用法

价格监控爬虫覆盖 GeeTest（极验）v3 和 Turnstile 站点，账号是 ADVANCE（$90/月，50 线程）套餐。

超时进重试队列，ERROR_ZERO_BALANCE 直接降级并告警，其余错误跳过，爬虫整夜自己跑。

常见错误码对照表

失败类型	错误码	恢复策略
超时	`CAPCHA_NOT_READY`（轮询次数耗尽）	用新挑战重试
参数错误	`ERROR_BAD_PARAMETERS`	记录并跳过——需要修复参数提取逻辑
sitekey 错误	`ERROR_WRONG_GOOGLEKEY`	重新提取 sitekey
余额耗尽	`ERROR_ZERO_BALANCE`	暂停任务、发送告警、等待充值
触发限流	`ERROR_TOO_MUCH_REQUESTS`	指数退避
API 不可达	连接错误	断路器 + 重试

重试决策矩阵

场景	判断
故障是暂时性的，且会话状态仍有效	值得重试
能无副作用重建同一请求	直接重放这一步
再重试只会浪费余额或撞限流阈值	果断跳过

常见故障排查

问题	原因	处理方式
所有任务都被跳过	降级阈值太低	提高 `failure_threshold`
重试队列一直变大	任务永远不会成功	设最大重试次数，转入死信队列
恢复速度太慢	降级超时设得太长	缩短恢复时间，加健康检查探针
重启后排队任务全丢	队列只在内存里	队列持久化到文件或数据库

常见问题

优雅降级和断路器有什么区别？

断路器检测到故障后直接切断调用。

优雅降级范围更大，包含跳过、重试队列、降级模式等回退行为，两者搭配使用最好。

被跳过的任务会不会永久丢失？

不会。把跳过原因和 URL 记到日志，复盘后多数可以人工补跑或修复逻辑重跑。

降级模式的恢复时间怎么定？

连接类故障 5 分钟起步。

ERROR_ZERO_BALANCE 需人工充值，建议窗口设长些（如 10 分钟）并同步告警。

用 CaptchaAI 打造扛得住失败的验证码自动化

获取你的 API Key：CaptchaAI 官网。

验证码解析失败时的优雅降级

方案一：跳过失败任务

方案二：失败任务放入重试队列

方案三：进入降级模式

Node.js 实现：三种模式合一

实战场景：跨境电商爬虫的组合用法

常见错误码对照表

重试决策矩阵

常见故障排查

常见问题

优雅降级和断路器有什么区别？

被跳过的任务会不会永久丢失？

降级模式的恢复时间怎么定？

用 CaptchaAI 打造扛得住失败的验证码自动化

相关阅读

AWS Lambda + CaptchaAI：无服务器验证码解决

使用 AWS SNS 和 CaptchaAI 构建事件驱动的验证码解决方案

用于 CaptchaAI Worker 部署的 Ansible Playbook

验证码解决基础设施的蓝绿部署

Azure Functions + CaptchaAI：云集成

构建 CaptchaAI 使用仪表板和监控

方案一：跳过失败任务

方案二：失败任务放入重试队列

方案三：进入降级模式

Node.js 实现：三种模式合一

实战场景：跨境电商爬虫的组合用法

常见错误码对照表

重试决策矩阵

常见故障排查

常见问题

优雅降级和断路器有什么区别？

被跳过的任务会不会永久丢失？

降级模式的恢复时间怎么定？

用 CaptchaAI 打造扛得住失败的验证码自动化

相关阅读

相关文章

AWS Lambda + CaptchaAI：无服务器验证码解决

使用 AWS SNS 和 CaptchaAI 构建事件驱动的验证码解决方案

用于 CaptchaAI Worker 部署的 Ansible Playbook

验证码解决基础设施的蓝绿部署

Azure Functions + CaptchaAI：云集成

构建 CaptchaAI 使用仪表板和监控