Scarnox · Blog

验证码求解器错误处理:重试、退避,以及哪些错误绝不重试

验证码求解是一次对你无法控制的系统发起的网络调用,而挑战本身可能在毫无预告的情况下发生变化。处理好失败情况,是让流水线优雅降级,还是陷入重试风暴的分水岭。

使用 Scarnox 时,成本对你有利:失败的求解不收费,因此重试只会消耗时间,别无其他代价。但如果重试的是错误的失败,同样会浪费这些时间;而重试得过于猛烈,本身还会带来新的问题。本指南提供一套可以直接粘贴进 Python 任务的重试策略。

Scarnox 目前支持 PopularCaptcha 的求解。同样的错误处理适用于任何验证码求解器,因此下面的策略并不依赖于验证码类型。

始终先检查 errorId

求解响应中有一个 errorId 字段。0 表示成功,token 位于 solution.token 中。其他任何值都表示求解失败。失败的响应还会附带 errorCode 和 errorDescription。请把两者都记录下来:代码是你日后用来搜索的依据,描述则是在运行出错时你需要阅读的内容。

{"errorId": 1, "errorCode": "...", "errorDescription": "..."}

不要因为 HTTP 状态码是 200 就认为响应成功。每次都要检查 errorId。

需要重试的失败

有些失败是暂时性的。对这些情况,请使用指数退避并加入少量随机延迟进行重试:

这些失败值得重试,因为下一次尝试在你不做任何改动的情况下也可能成功。

需要先修复的失败

其他失败不会因为等待而消失。使用相同的输入重试,结果也会一样。请先修复以下问题:

当某个失败不在上述两份列表中时,请将其视为需要先修复的失败。查看错误文本,做出判断,并把对应的代码加入代码中正确的列表。请保持列表简短,并定期复核。

可直接粘贴的策略

import os
import random
import time

import requests

API_URL = "https://api.scarnox.com/api/tasks/create"
HEADERS = {"Authorization": f"Bearer {os.environ['SCARNOX_KEY']}"}
MAX_ATTEMPTS = 4

class SolveError(Exception):
    def __init__(self, code: str, description: str, retryable: bool):
        super().__init__(f"{code}: {description}")
        self.code = code
        self.retryable = retryable

def solve_once(payload: dict) -> str:
    try:
        r = requests.post(API_URL, headers=HEADERS, json=payload, timeout=185)
    except requests.RequestException as exc:
        # 本端网络错误或超时:可以安全重试。
        raise SolveError("NETWORK", str(exc), retryable=True)

    if r.status_code == 429 or r.status_code >= 500:
        raise SolveError(f"HTTP_{r.status_code}", r.text[:200], retryable=True)

    try:
        data = r.json()
    except ValueError:
        raise SolveError(f"HTTP_{r.status_code}", r.text[:200], retryable=False)

    if data.get("errorId") == 0:
        return data["solution"]["token"]

    # API 报告了失败。只重试已确认为临时性的错误码。
    code = data.get("errorCode", "UNKNOWN")
    retryable = code in TRANSIENT_CODES
    raise SolveError(code, data.get("errorDescription", ""), retryable=retryable)

# 只有在日志中确认某个 API 错误码属于临时性错误后,才把它添加到这里。
TRANSIENT_CODES: set[str] = set()

def solve(payload: dict) -> str:
    for attempt in range(MAX_ATTEMPTS):
        try:
            return solve_once(payload)
        except SolveError as err:
            last_attempt = attempt == MAX_ATTEMPTS - 1
            if not err.retryable or last_attempt:
                raise
            delay = min(30, 2 ** attempt) + random.random()  # 退避 + 抖动
            print(f"attempt {attempt + 1} failed ({err.code}), retrying in {delay:.1f}s")
            time.sleep(delay)
    raise RuntimeError("unreachable")

该策略从不重试需要先修复的失败,并且会在固定次数的尝试后停止。TRANSIENT_CODES 集合特意留空:只有当日志证明某个错误码是临时性的之后,才把它加进去。

为什么需要抖动

如果五十个工作进程同时失败,并且都睡眠恰好两秒,它们就会一起重试,然后再次一起失败。加入随机的零点几秒,可以把它们分散开。这也是大多数大型 API 服务商给自己的客户端提供的建议。

超时

将 HTTP 超时设置为远高于典型求解耗时的值,约为 185 秒。请求会阻塞直到 token 准备好,因此过短的客户端超时会把一次本应成功的慢速求解变成失败,然后你还要为此徒劳地重试。客户端超时长于求解耗时,是最安全的设置。

token 不可缓存

token 会在约两分钟内过期。不要在多次运行之间缓存 token,也不要把同一个 token 用于第二次提交。如果提交失败而 token 已经被使用,请重新求解,而不是重复提交旧 token。token 过期指南对时序做了更详细的说明。

多个工作进程之间的并发与退避

如果多个工作进程共用同一个密钥,其中一个工作进程的退避并不会减缓其他进程。请加入共享的并发求解数限制,例如信号量或队列,避免一阵重试突发转化为一阵请求突发。简单的并发上限,就能在速率限制失败出现之前解决大部分问题。

有用的日志

每次失败的尝试,都应记录 errorCode、尝试次数、sitekey、页面主机名,以及该失败是否被重试。不要记录 API 密钥或代理密码。按小时统计各错误码数量的仪表盘,可以帮你快速判断问题出在你的代码、代理还是服务本身。

下一步

先注册账户,获取免费试用额度(无需信用卡),然后用免费试用来测试你的错误处理。