Scarnox · Blog

验证码求解提速：如何降低流水线中的延迟

快速的验证码求解器，是指你能在恰当的时刻调用它、并行执行、且不浪费请求的求解器。Scarnox 的求解调用会在 token 准备好后返回，因此耗时取决于验证码类型、目标网站以及当前负载。本指南不承诺固定的耗时。它展示的是你能在自己的代码中控制什么、如何测量，以及如何同时运行多次求解，使你花费的每一段时间都得到充分利用。

决定求解耗时的因素

有三件事很重要，其中只有一件掌握在你手中：

验证码本身。 某些挑战类型就是比其他类型耗时更长。这是网站验证码的特性，而不是你的代码造成的。
求解服务的负载。 繁忙时段可能会让单次调用变慢。重试逻辑应当应对这种情况，而不是过早放弃调用。
你自己的流水线。 过早求解、串行处理每个请求，以及等待无关的工作，都会增加你可以消除的耗时。

请把精力集中在第三项上。它是你可以测量并加以改变的部分。

先测量，再优化

用计时器包装调用，并记录结果。你需要知道求解耗时，以及你的代码在调用前后所花的时间：

import time
import requests

API_KEY = "scarnox_YOURAPIKEY"
URL = "https://api.scarnox.com/api/tasks/create"

def timed_solve(sitekey: str, page_url: str) -> dict:
    started = time.perf_counter()
    r = requests.post(
        URL,
        headers={"Authorization": f"Bearer {API_KEY}"},
        json={
            "captcha_type": "PopularCaptchaTaskProxyless",
            "site_key": sitekey,
            "site_url": page_url,
        },
        timeout=185,
    )
    elapsed = time.perf_counter() - started
    data = r.json()
    if data.get("errorId") != 0:
        return {"ok": False, "error": data.get("errorCode"), "seconds": elapsed}
    return {"ok": True, "token": data["solution"]["token"], "seconds": elapsed}

result = timed_solve("10000000-ffff-ffff-ffff-000000000001", "https://example.com/signup")
print(result["ok"], round(result["seconds"], 2), "s")

请在你关心的页面上、在你运行任务的时间段内运行它。保留这些数字。它们能告诉你某项改动是否真的有效。

并行求解

各次求解之间互不依赖。如果你要处理十个 sitekey，逐个等待会浪费本可用于处理其他任务的时间。对于大多数脚本来说，线程池就足够了，因为每次求解只是一个 HTTP 请求，而它的大部分时间都在等待：

from concurrent.futures import ThreadPoolExecutor

jobs = [
    ("10000000-ffff-ffff-ffff-000000000001", "https://example.com/signup"),
    ("10000000-ffff-ffff-ffff-000000000001", "https://example.com/register"),
    ("10000000-ffff-ffff-ffff-000000000001", "https://example.com/contact"),
]

def run(job):
    sitekey, page_url = job
    return timed_solve(sitekey, page_url)

# 从小规模开始。只有在错误率保持较低时才提高并发数。
with ThreadPoolExecutor(max_workers=5) as pool:
    results = list(pool.map(run, jobs))

for r in results:
    status = "ok" if r["ok"] else r["error"]
    print(f"{r['seconds']:.2f} s  {status}")

如果你提高工作线程数后出现看似速率限制或容量不足的错误，请把数量调低。并行只有在服务器跟得上的情况下才有帮助。

不要提前求解

最快的代码，是不去等待眼下用不上的 token 的代码。提前请求 token 并将其保存起来，是常见的错误。token 大约在签发两分钟后过期，因此存起来的 token 等到需要用时往往已经失效。应当在表单即将提交时再求解。详细的原因请参阅验证码 token 过期：原因与解决方法。

保持其余路径简短

每个网站保持一个 HTTP 会话，避免每次请求都重新建立连接。
在求解与提交之间，不要执行缓慢的解析或数据库操作。
为你自己的请求设置合理的超时。求解调用的超时约为 185 秒，因为它会阻塞直到 token 准备好。你的其他调用应当短得多。
只有在任务需要时才发送代理。代理类任务类型接受格式为 http://user:pass@host:port 的 proxy 字段，但仅仅添加代理本身并不能提速。

复用不会过期的内容

有些工作可以安全地在多次求解之间共享。请把它们移出热路径：

Sitekey。 每种页面模板读取一次 sitekey 并保存下来。它不会像 token 那样过期。只有在表单开始失败时才需要重新读取。
HTTP 连接。 带有 keep-alive 的会话可以避免每次调用都进行一次新的 TLS 握手。每个线程或每个工作进程创建一个会话，而不是每次请求创建一个。
已解析的配置。 在启动时加载目标列表、代理列表和设置，而不是在求解函数内部读取文件。

这些改动成本很低，却能消除每次调用中都会出现的开销。

关注长尾，而不仅是平均值

平均求解耗时会掩盖那些拖慢整批任务的慢速调用。请记录每次求解的耗时，然后查看中位数和第 95 百分位数。一个流水线的完成时间取决于其最慢的任务，因此，调整慢速调用的重试方式，可能比改变中位数更有效：

import statistics

def summarise(seconds: list[float]) -> str:
    ordered = sorted(seconds)
    p95 = ordered[int(0.95 * (len(ordered) - 1))]
    return f"median {statistics.median(ordered):.2f} s, p95 {p95:.2f} s, n={len(ordered)}"

print(summarise([r["seconds"] for r in results]))

在每次改动前后对比这份摘要。如果第 95 百分位数几乎没有变化，说明耗时其实花在了你代码的其他地方。

检查结果，而不只是速度

快速失败仍然是失败。请统计返回错误的求解次数，并把错误率的上升视为放慢速度的信号。失败的求解不收费，但花在它们上面的时间仍然是流水线损失的时间。完整的重试模式请参阅验证码求解器错误处理。