SQLite 缓存验证码 token：本地识别记录落库方案

单机脚本最常见的浪费不是识别失败，而是同一个 sitekey 在几十秒内被重复提交好几次。不必先搭 Redis：本地放一个 SQLite 文件，提交前查缓存、结果落库，重复请求当场消失。

顺带把账算清楚：CaptchaAI 按并发线程计费（BASIC $15/月，5 线程），套餐内识别次数不限。缓存省下的不是单次费用，而是被占住的线程——5 个线程有 2 个在重复识别同一页面，并发就只剩六成。

SQLite 够用的场景与该换库的信号

场景	SQLite	更合适的选择
本机开发调试	✅	——
小规模生产（<1000 次/小时）	✅	——
QA 结果留档	✅	——
多机共享状态	❌	PostgreSQL、MongoDB
高吞吐分布式	❌	Redis、DynamoDB
实时分析看板	❌	TimescaleDB、InfluxDB

一句话判断：只被一台机器访问就够用，要跨机器共享就换网络数据库。

两张表：识别记录与 token 缓存

captcha_solves 记录每次任务的全生命周期，是事后分析的流水；token_cache 只存有效期内的 token，是热路径上的查表。

CREATE TABLE IF NOT EXISTS captcha_solves (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    captcha_id TEXT,
    type TEXT NOT NULL,
    sitekey TEXT,
    pageurl TEXT,
    status TEXT NOT NULL DEFAULT 'submitted',
    solution TEXT,
    error TEXT,
    submitted_at TEXT NOT NULL DEFAULT (datetime('now')),
    solved_at TEXT,
    elapsed_ms INTEGER,
    polls INTEGER DEFAULT 0,
    project TEXT
);

CREATE INDEX IF NOT EXISTS idx_submitted_at ON captcha_solves(submitted_at);
CREATE INDEX IF NOT EXISTS idx_type_status ON captcha_solves(type, status);
CREATE INDEX IF NOT EXISTS idx_sitekey ON captcha_solves(sitekey);

-- Token cache for reuse within TTL
CREATE TABLE IF NOT EXISTS token_cache (
    sitekey TEXT NOT NULL,
    pageurl TEXT NOT NULL,
    token TEXT NOT NULL,
    created_at TEXT NOT NULL DEFAULT (datetime('now')),
    expires_at TEXT NOT NULL,
    used INTEGER DEFAULT 0,
    PRIMARY KEY (sitekey, pageurl, token)
);

CREATE INDEX IF NOT EXISTS idx_cache_lookup
ON token_cache(sitekey, pageurl, used, expires_at);

主键加 used 标志位，保证同一个 token 不被两个协程同时取走。

Python 实现：从建库到落库

建库与连接：WAL 和 busy_timeout

两个 PRAGMA 决定并发上限：journal_mode=WAL 让读写不再互相阻塞，busy_timeout=5000 让锁冲突自动重试 5 秒。国内装依赖可走镜像源：pip install -i https://pypi.tuna.tsinghua.edu.cn/simple requests。

import os
import time
import sqlite3
from datetime import datetime, timedelta, timezone
import requests

DB_PATH = os.environ.get("CAPTCHA_DB", "captcha_solves.db")
API_KEY = os.environ["CAPTCHAAI_API_KEY"]


def get_db():
    conn = sqlite3.connect(DB_PATH)
    conn.row_factory = sqlite3.Row
    conn.execute("PRAGMA journal_mode=WAL")  # Better concurrent read performance
    conn.execute("PRAGMA busy_timeout=5000")
    return conn


def init_db():
    conn = get_db()
    conn.executescript("""
        CREATE TABLE IF NOT EXISTS captcha_solves (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            captcha_id TEXT,
            type TEXT NOT NULL,
            sitekey TEXT,
            pageurl TEXT,
            status TEXT NOT NULL DEFAULT 'submitted',
            solution TEXT,
            error TEXT,
            submitted_at TEXT NOT NULL DEFAULT (datetime('now')),
            solved_at TEXT,
            elapsed_ms INTEGER,
            polls INTEGER DEFAULT 0,
            project TEXT
        );
        CREATE INDEX IF NOT EXISTS idx_submitted_at ON captcha_solves(submitted_at);
        CREATE INDEX IF NOT EXISTS idx_type_status ON captcha_solves(type, status);

        CREATE TABLE IF NOT EXISTS token_cache (
            sitekey TEXT NOT NULL,
            pageurl TEXT NOT NULL,
            token TEXT NOT NULL,
            created_at TEXT NOT NULL DEFAULT (datetime('now')),
            expires_at TEXT NOT NULL,
            used INTEGER DEFAULT 0,
            PRIMARY KEY (sitekey, pageurl, token)
        );
        CREATE INDEX IF NOT EXISTS idx_cache_lookup
        ON token_cache(sitekey, pageurl, used, expires_at);
    """)
    conn.close()

init_db()

先查缓存，再提交任务，全程留痕

顺序不能颠倒：先查 token_cache，命中直接返回；没命中才写 submitted 记录，再提交到 in.php，每 5 秒轮询 res.php。状态变化都落库，脚本挂了也能看出卡在哪。

def solve_recaptcha(sitekey, pageurl, project=None):
    conn = get_db()

    # Check cache first
    cached = get_cached_token(conn, sitekey, pageurl)
    if cached:
        conn.close()
        return cached

    # Insert tracking record
    now = datetime.now(timezone.utc).isoformat()
    cursor = conn.execute(
        "INSERT INTO captcha_solves (type, sitekey, pageurl, submitted_at, project) "
        "VALUES (?, ?, ?, ?, ?)",
        ("recaptcha_v2", sitekey, pageurl, now, project)
    )
    row_id = cursor.lastrowid
    conn.commit()

    # Submit to 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:
        conn.execute(
            "UPDATE captcha_solves SET status=?, error=? WHERE id=?",
            ("error", data.get("request"), row_id)
        )
        conn.commit()
        conn.close()
        return None

    captcha_id = data["request"]
    conn.execute(
        "UPDATE captcha_solves SET captcha_id=?, status=? WHERE id=?",
        (captcha_id, "polling", row_id)
    )
    conn.commit()

    # Poll
    polls = 0
    for _ in range(60):
        time.sleep(5)
        polls += 1
        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:
            solved_at = datetime.now(timezone.utc).isoformat()
            submitted = datetime.fromisoformat(now)
            elapsed = int((datetime.now(timezone.utc) - submitted).total_seconds() * 1000)

            conn.execute(
                "UPDATE captcha_solves SET status=?, solution=?, solved_at=?, "
                "elapsed_ms=?, polls=? WHERE id=?",
                ("solved", result["request"], solved_at, elapsed, polls, row_id)
            )
            # Cache the token
            cache_token(conn, sitekey, pageurl, result["request"])
            conn.commit()
            conn.close()
            return result["request"]

        if result.get("request") != "CAPCHA_NOT_READY":
            conn.execute(
                "UPDATE captcha_solves SET status=?, error=?, polls=? WHERE id=?",
                ("error", result.get("request"), polls, row_id)
            )
            conn.commit()
            conn.close()
            return None

    conn.execute(
        "UPDATE captcha_solves SET status=?, polls=? WHERE id=?",
        ("timeout", polls, row_id)
    )
    conn.commit()
    conn.close()
    return None

token 的写入与一次性取用

TTL 默认 90 秒，处在 reCAPTCHA token 90–120 秒有效期的下沿，留出余量。取用时按 used=0 过滤并立即置 1，缓存于是成了一次性队列。

def cache_token(conn, sitekey, pageurl, token, ttl_seconds=90):
    expires_at = (datetime.now(timezone.utc) + timedelta(seconds=ttl_seconds)).isoformat()
    conn.execute(
        "INSERT OR REPLACE INTO token_cache (sitekey, pageurl, token, expires_at) "
        "VALUES (?, ?, ?, ?)",
        (sitekey, pageurl, token, expires_at)
    )


def get_cached_token(conn, sitekey, pageurl):
    now = datetime.now(timezone.utc).isoformat()
    row = conn.execute(
        "SELECT token FROM token_cache "
        "WHERE sitekey=? AND pageurl=? AND used=0 AND expires_at > ? "
        "ORDER BY expires_at ASC LIMIT 1",
        (sitekey, pageurl, now)
    ).fetchone()

    if row:
        conn.execute(
            "UPDATE token_cache SET used=1 WHERE token=?",
            (row["token"],)
        )
        conn.commit()
        return row["token"]
    return None

成功率、识别耗时与定期清理

成功率和平均耗时就是两条 SQL 的事。看两个信号：成功率骤降，通常是 sitekey 变了；耗时变长而成功率不变，多半是线程不够、任务在排队。清理任务每天跑一次，删完过期记录再 VACUUM。

def get_stats(hours=24):
    conn = get_db()
    cutoff = (datetime.now(timezone.utc) - timedelta(hours=hours)).isoformat()

    total = conn.execute(
        "SELECT COUNT(*) FROM captcha_solves WHERE submitted_at >= ?", (cutoff,)
    ).fetchone()[0]

    solved = conn.execute(
        "SELECT COUNT(*) FROM captcha_solves WHERE submitted_at >= ? AND status='solved'",
        (cutoff,)
    ).fetchone()[0]

    avg_time = conn.execute(
        "SELECT AVG(elapsed_ms) FROM captcha_solves "
        "WHERE submitted_at >= ? AND status='solved'",
        (cutoff,)
    ).fetchone()[0]

    conn.close()
    return {
        "total": total,
        "solved": solved,
        "success_rate": (solved / total * 100) if total else 0,
        "avg_time_ms": round(avg_time) if avg_time else 0
    }


def cleanup_old_records(days=30):
    conn = get_db()
    cutoff = (datetime.now(timezone.utc) - timedelta(days=days)).isoformat()
    conn.execute("DELETE FROM captcha_solves WHERE submitted_at < ?", (cutoff,))
    conn.execute("DELETE FROM token_cache WHERE expires_at < ?",
                 (datetime.now(timezone.utc).isoformat(),))
    conn.execute("VACUUM")
    conn.commit()
    conn.close()

Node.js 实现：better-sqlite3 同步接口

Node.js 用 better-sqlite3，同步接口比回调版清爽。同样先开 WAL 再建表；表结构与 Python 版一致，可共用同一个数据库文件。

const Database = require("better-sqlite3");
const axios = require("axios");

const db = new Database(process.env.CAPTCHA_DB || "captcha_solves.db");
const API_KEY = process.env.CAPTCHAAI_API_KEY;

db.pragma("journal_mode = WAL");
db.exec(`
  CREATE TABLE IF NOT EXISTS captcha_solves (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    captcha_id TEXT, type TEXT NOT NULL, sitekey TEXT, pageurl TEXT,
    status TEXT DEFAULT 'submitted', solution TEXT, error TEXT,
    submitted_at TEXT DEFAULT (datetime('now')),
    solved_at TEXT, elapsed_ms INTEGER, polls INTEGER DEFAULT 0
  );
  CREATE INDEX IF NOT EXISTS idx_submitted ON captcha_solves(submitted_at);
`);

async function solveAndStore(sitekey, pageurl) {
  const submittedAt = new Date().toISOString();
  const insert = db.prepare(
    "INSERT INTO captcha_solves (type, sitekey, pageurl, submitted_at) VALUES (?, ?, ?, ?)"
  );
  const { lastInsertRowid } = insert.run("recaptcha_v2", sitekey, pageurl, submittedAt);

  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) {
    db.prepare("UPDATE captcha_solves SET status=?, error=? WHERE id=?")
      .run("error", submit.data.request, lastInsertRowid);
    return null;
  }

  const captchaId = submit.data.request;
  db.prepare("UPDATE captcha_solves SET captcha_id=?, status=? WHERE id=?")
    .run(captchaId, "polling", lastInsertRowid);

  let polls = 0;
  for (let i = 0; i < 60; i++) {
    await new Promise((r) => setTimeout(r, 5000));
    polls++;
    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 elapsed = Date.now() - new Date(submittedAt).getTime();
      db.prepare(
        "UPDATE captcha_solves SET status=?, solution=?, solved_at=?, elapsed_ms=?, polls=? WHERE id=?"
      ).run("solved", poll.data.request, new Date().toISOString(), elapsed, polls, lastInsertRowid);
      return poll.data.request;
    }
    if (poll.data.request !== "CAPCHA_NOT_READY") {
      db.prepare("UPDATE captcha_solves SET status=?, error=?, polls=? WHERE id=?")
        .run("error", poll.data.request, polls, lastInsertRowid);
      return null;
    }
  }

  db.prepare("UPDATE captcha_solves SET status=?, polls=? WHERE id=?")
    .run("timeout", polls, lastInsertRowid);
  return null;
}

一个本地化场景：国内站与海外站混跑

国内站点多是 GeeTest（极验）滑块、网易易盾、腾讯防水墙，海外站点以 reCAPTCHA v2 和 Turnstile 为主。CaptchaAI 支持 reCAPTCHA、Turnstile 与 GeeTest v3（v4 即将支持），hCaptcha 和 FunCaptcha 暂不支持。类型混在一起统计会失真，提交时写好 type 并按它分组。reCAPTCHA 脚本由 Google 域名托管，国内加载不稳，容易把页面加载耗时算进识别耗时，两段时间分开记。采集脚本也只抓自有或已授权的页面，遵守 robots 协议与 PIPL。

常见故障与处理

问题	原因	处理方式
`database is locked`	未开 WAL 就并发写	加 `PRAGMA journal_mode=WAL`
文件持续变大	没有清理任务	每天跑 `cleanup_old_records()`
查询变慢	缺索引	给 `submitted_at`、`type` 建索引
缓存全是过期 token	过期条目没清	查询前按 `expires_at` 过滤

常见问题

多个进程同时写，会不会一直报 database is locked？

WAL 下读写不互斥，但同一时刻仍只允许一个写事务。把 busy_timeout 设成 5000 毫秒，冲突时自动重试，单机场景基本不再报错；重试也顶不住，就该换 PostgreSQL。

token 缓存的 TTL 设成多少合适？

比 token 实际有效期短 20%–30%。reCAPTCHA token 通常 90–120 秒过期，示例取 90 秒；Turnstile token 约 300 秒，TTL 设到 250 秒仍有余量。

这套表结构能同时记录 Turnstile 和 GeeTest v3 吗？

可以，type 字段就是为此留的。GeeTest v3 返回 challenge、validate、seccode 三个值，序列化成 JSON 写进 solution 列即可。

容器重启后记录丢失怎么办？

把数据库文件挂到宿主机 volume，别留在容器可写层。WAL 还会生成 -wal 与 -shm，备份要三个一起拷。

用 DynamoDB 做无服务器识别记录追踪

下一步

先把识别记录落进库，再谈优化——领取你的 CaptchaAI API Key，今天就让每一次识别都留下数据。

延伸阅读：

SQLite 验证码 token 缓存与识别记录追踪

SQLite 够用的场景与该换库的信号

两张表：识别记录与 token 缓存

Python 实现：从建库到落库

建库与连接：WAL 和 busy_timeout

先查缓存，再提交任务，全程留痕

token 的写入与一次性取用

成功率、识别耗时与定期清理

Node.js 实现：better-sqlite3 同步接口

一个本地化场景：国内站与海外站混跑

常见故障与处理

常见问题

多个进程同时写，会不会一直报 database is locked？

token 缓存的 TTL 设成多少合适？

这套表结构能同时记录 Turnstile 和 GeeTest v3 吗？

容器重启后记录丢失怎么办？

相关文章

下一步

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

用于 CaptchaAI Worker 部署的 Ansible Playbook

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

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

Azure Functions + CaptchaAI：云集成

构建 CaptchaAI 使用仪表板和监控