用于验证码令牌 TTL 管理和缓存的 Redis

先给结论：验证码 token 的寿命只有几十秒到几分钟，缓存它时 TTL 必须比真实寿命短 30–40 秒，且每个 token 只允许被取走一次。

典型的翻车场景是：CaptchaAI 返回了 status: 1，token 也拿到了，提交表单时却被判定校验失败。问题多半不在识别环节，而在 token 从拿到手到被用掉之间——排队、重试、慢请求，多耗 40 秒它就作废了。

token 在 Redis 里的生命周期

链路很短：先查缓存，命中直接返回；没命中才调用 CaptchaAI，拿到结果后带 TTL 写回。

Solve Request → Check Redis → Cache Hit?
                    │               │
                    │ No            │ Yes → Return cached token
                    ▼               
              CaptchaAI API
                    │
                    ▼
              Store in Redis (TTL = token_lifetime - safety_margin)
                    │
                    ▼
              Return token

TTL 取的是“真实寿命减去安全裕度”，差值用来覆盖你自己的队列延迟。

Python 实现

连接 Redis 并定义各类型 TTL

把每种验证码的缓存时长集中成一张表，避免魔法数字散落各处。

import os
import time
import json
import redis
import requests

r = redis.Redis(
    host=os.environ.get("REDIS_HOST", "localhost"),
    port=int(os.environ.get("REDIS_PORT", 6379)),
    db=0,
    decode_responses=True
)

API_KEY = os.environ["CAPTCHAAI_API_KEY"]

# TTLs with safety margin (seconds before actual expiration)
TOKEN_TTLS = {
    "recaptcha_v2": 80,     # Actual: ~120s, cache for 80s
    "recaptcha_v3": 80,
    "hcaptcha": 80,
    "turnstile": 250,       # Actual: ~300s, cache for 250s
}

TOKEN_TTLS 里的 hcaptcha 只是键名占位，CaptchaAI 不支持 hCaptcha，实际项目里删掉即可。

读写缓存的两个基础函数

用 list 而不是字符串存 token，是为了让“取出即消费”成为原子操作。

def cache_key(sitekey, pageurl):
    """Generate Redis key for a specific CAPTCHA target."""
    return f"captcha:token:{sitekey}:{pageurl}"


def get_cached_token(sitekey, pageurl):
    """Pop a cached token from the queue."""
    key = cache_key(sitekey, pageurl)
    token = r.lpop(key)
    if token:
        # Verify TTL still valid on the list
        ttl = r.ttl(key)
        if ttl > 10:  # At least 10 seconds remaining
            return token
    return None


def cache_token(sitekey, pageurl, token, captcha_type="recaptcha_v2"):
    """Push a solved token to the cache with appropriate TTL."""
    key = cache_key(sitekey, pageurl)
    ttl = TOKEN_TTLS.get(captcha_type, 80)
    r.rpush(key, token)
    r.expire(key, ttl)

ttl > 10 是最后一道保险：剩余不足 10 秒的 token 宁可丢掉重识。

带去重锁的识别流程

多个 worker 抓同一页面时，最浪费线程的是对着同一个 sitekey 各提交一次。用 SET NX EX 加锁：第一个去识别，其余原地等结果。

def solve_recaptcha(sitekey, pageurl, captcha_type="recaptcha_v2"):
    """Solve reCAPTCHA with Redis cache check."""
    # 1. Check cache
    cached = get_cached_token(sitekey, pageurl)
    if cached:
        return {"solution": cached, "source": "cache"}

    # 2. Check if solve is already in progress (dedup)
    lock_key = f"captcha:lock:{sitekey}:{pageurl}"
    if not r.set(lock_key, "1", nx=True, ex=120):
        # Another worker is solving — wait for result
        for _ in range(60):
            time.sleep(2)
            cached = get_cached_token(sitekey, pageurl)
            if cached:
                return {"solution": cached, "source": "cache_wait"}
        return {"error": "TIMEOUT_WAITING_FOR_OTHER_WORKER"}

    try:
        # 3. Solve via CaptchaAI
        resp = requests.post("https://ocr.captchaai.com/in.php", data={
            "key": API_KEY,
            "method": "userrecaptcha",
            "googlekey": sitekey,
            "pageurl": pageurl,
            "json": 1
        })
        data = resp.json()
        if data.get("status") != 1:
            return {"error": data.get("request")}

        captcha_id = data["request"]

        for _ in range(60):
            time.sleep(5)
            result = requests.get("https://ocr.captchaai.com/res.php", params={
                "key": API_KEY, "action": "get",
                "id": captcha_id, "json": 1
            }).json()

            if result.get("status") == 1:
                token = result["request"]
                cache_token(sitekey, pageurl, token, captcha_type)
                return {"solution": token, "source": "api"}

            if result.get("request") != "CAPCHA_NOT_READY":
                return {"error": result.get("request")}

        return {"error": "TIMEOUT"}
    finally:
        r.delete(lock_key)

锁设 120 秒过期，是为了在 worker 崩溃时自动释放。

为高频目标维护测试 token 池

QA 流程需要亚秒级拿到 token 时，让后台线程持续补池，业务代码只负责 lpop。

import threading


class TokenPool:
    def __init__(self, sitekey, pageurl, pool_size=5, captcha_type="recaptcha_v2"):
        self.sitekey = sitekey
        self.pageurl = pageurl
        self.pool_size = pool_size
        self.captcha_type = captcha_type
        self.pool_key = f"captcha:pool:{sitekey}:{pageurl}"
        self._running = False

    def start(self):
        self._running = True
        thread = threading.Thread(target=self._refill_loop, daemon=True)
        thread.start()

    def stop(self):
        self._running = False

    def _refill_loop(self):
        while self._running:
            current = r.llen(self.pool_key)
            if current < self.pool_size:
                self._solve_and_add()
            time.sleep(2)

    def _solve_and_add(self):
        resp = requests.post("https://ocr.captchaai.com/in.php", data={
            "key": API_KEY,
            "method": "userrecaptcha",
            "googlekey": self.sitekey,
            "pageurl": self.pageurl,
            "json": 1
        })
        data = resp.json()
        if data.get("status") != 1:
            return

        captcha_id = data["request"]
        for _ in range(60):
            time.sleep(5)
            result = requests.get("https://ocr.captchaai.com/res.php", params={
                "key": API_KEY, "action": "get",
                "id": captcha_id, "json": 1
            }).json()

            if result.get("status") == 1:
                ttl = TOKEN_TTLS.get(self.captcha_type, 80)
                r.rpush(self.pool_key, result["request"])
                r.expire(self.pool_key, ttl)
                return
            if result.get("request") != "CAPCHA_NOT_READY":
                return

    def get_token(self):
        return r.lpop(self.pool_key)


# Usage
pool = TokenPool("6Le-wvkSAAAAAPBMRTvw0Q4Muexq9bi0DJwx_mJ-", "https://example.com")
pool.start()

# When you need a token:
token = pool.get_token()

池中 token 同样带 TTL，过期后整个列表自动消失。

Node.js 实现

同一套逻辑的 ioredis 版本，键名与 TTL 表保持一致，方便两端共用一个 Redis 实例。

const Redis = require("ioredis");
const axios = require("axios");

const redis = new Redis(process.env.REDIS_URL || "redis://localhost:6379");
const API_KEY = process.env.CAPTCHAAI_API_KEY;

const TOKEN_TTLS = { recaptcha_v2: 80, recaptcha_v3: 80, hcaptcha: 80, turnstile: 250 };

function cacheKey(sitekey, pageurl) {
  return `captcha:token:${sitekey}:${pageurl}`;
}

async function getCachedToken(sitekey, pageurl) {
  const key = cacheKey(sitekey, pageurl);
  const token = await redis.lpop(key);
  if (token) {
    const ttl = await redis.ttl(key);
    if (ttl > 10) return token;
  }
  return null;
}

async function solveWithCache(sitekey, pageurl, type = "recaptcha_v2") {
  // Check cache
  const cached = await getCachedToken(sitekey, pageurl);
  if (cached) return { solution: cached, source: "cache" };

  // Dedup lock
  const lockKey = `captcha:lock:${sitekey}:${pageurl}`;
  const locked = await redis.set(lockKey, "1", "NX", "EX", 120);
  if (!locked) {
    for (let i = 0; i < 60; i++) {
      await new Promise((r) => setTimeout(r, 2000));
      const waitCached = await getCachedToken(sitekey, pageurl);
      if (waitCached) return { solution: waitCached, source: "cache_wait" };
    }
    return { error: "TIMEOUT_WAITING" };
  }

  try {
    const submit = await axios.post("https://ocr.captchaai.com/in.php", null, {
      params: { key: API_KEY, method: "userrecaptcha", googlekey: sitekey, pageurl, json: 1 },
    });
    if (submit.data.status !== 1) return { error: submit.data.request };

    const captchaId = submit.data.request;
    for (let i = 0; i < 60; i++) {
      await new Promise((r) => setTimeout(r, 5000));
      const poll = await axios.get("https://ocr.captchaai.com/res.php", {
        params: { key: API_KEY, action: "get", id: captchaId, json: 1 },
      });

      if (poll.data.status === 1) {
        const key = cacheKey(sitekey, pageurl);
        const ttl = TOKEN_TTLS[type] || 80;
        await redis.rpush(key, poll.data.request);
        await redis.expire(key, ttl);
        return { solution: poll.data.request, source: "api" };
      }
      if (poll.data.request !== "CAPCHA_NOT_READY") return { error: poll.data.request };
    }
    return { error: "TIMEOUT" };
  } finally {
    await redis.del(lockKey);
  }
}

注意 ioredis 的可选参数是平铺传入的，"NX", "EX", 120 顺序写错会让锁静默失效。

Redis 键设计

监控缓存命中率

命中率长期低于 20%，说明目标太分散，直接识别更简单。

def cache_stats():
    info = r.info("stats")
    hits = info.get("keyspace_hits", 0)
    misses = info.get("keyspace_misses", 0)
    total = hits + misses
    return {
        "hit_rate": f"{hits / total * 100:.1f}%" if total else "0%",
        "hits": hits,
        "misses": misses,
        "active_keys": r.dbsize()
    }

故障排查

国内环境的两点提醒

reCAPTCHA 依赖 Google 托管的脚本，境内加载并不稳定，但这只影响本地复现：sitekey 与 pageurl 都是纯文本参数，调用接口不需要浏览器加载脚本。

境内站点更常见的是 GeeTest（极验）滑块，reCAPTCHA 与 Turnstile 多见于出海业务。CaptchaAI 支持 GeeTest v3，它返回 challenge、validate、seccode 三个值而非单个 token，建议序列化成 JSON 整体存入。

常见问题

同一个 token 会被两个 worker 重复取到吗？

不会，前提是用 list 加 lpop 消费：一个值只返回给一个客户端。改用 GET 读字符串就会重复，而目标站点通常只认第一次提交。

代码里的 hcaptcha 键名要保留吗？

不用保留，CaptchaAI 不支持 hCaptcha，那一行只是键名示例。CaptchaAI 支持 reCAPTCHA v2/v3、Turnstile 与 Cloudflare Challenge、GeeTest v3、图片/OCR、九宫格和 BLS，另有 CaptchaFox、Friendly Captcha、Lemin 三种测试版类型；GeeTest v4 为即将支持。

跑 token 池该选哪个套餐？

CaptchaAI 按并发线程计费，各套餐识别次数不限，补池并发上限就是线程数。BASIC（$15/月，5 线程）够单目标测试，STANDARD（$30/月，15 线程）适合多目标并行，长期维持池子从 ADVANCE（$90/月，50 线程）起步。

Redis 临时不可用时怎么降级？

把缓存层写成可选路径：捕获连接异常后直接走一次正常识别即可。缓存是加速手段，不是必需依赖。

下一步

拿到 API Key，把 cache_token 与 get_cached_token 接进现有代码，观察一天命中率再决定要不要上 token 池。获取你的 CaptchaAI API Key。

相关阅读：

• 用 MongoDB 记录识别历史
• 分布式 worker 的会话状态
• 用数据库锁为识别请求去重