SQLite для локального решения CAPTCHA, кэширование и отслеживание

Самый дешёвый способ понять, сколько задач CAPTCHA ушло впустую, — записывать каждое решение в файл SQLite рядом со скриптом. Серверную СУБД поднимать не нужно, модуль sqlite3 уже входит в стандартную библиотеку Python, а в режиме WAL база спокойно принимает сотни записей в секунду. Этого достаточно, чтобы вести журнал решений, кэшировать токены на время их жизни и не отправлять в API одну и ту же задачу дважды.

Ниже — рабочая схема таблиц, код на Python и Node.js для интеграции с быстрым стартом CaptchaAI, запросы для отчётов и процедура очистки.

Зачем хранить решения локально

Без журнала любой вопрос об интеграции упирается в догадки: парсер работает неделю, а какая доля запросов ушла в тайм-аут — уже не выяснить, логи перетёрлись. Локальная база закрывает четыре задачи сразу:

• Учёт. Сколько задач отправлено, сколько решено, сколько завершилось ошибкой — в разрезе типа CAPTCHA, sitekey и проекта.
• Кэш токенов. Токен reCAPTCHA живёт около двух минут. Если за это время скрипт снова открывает ту же страницу, повторный запрос к API не нужен.
• Дедупликация. Параллельные воркеры часто натыкаются на одну и ту же форму; уникальный ключ в таблице кэша гасит дубли.
• Диагностика. Время решения и число опросов res.php, сохранённые рядом с текстом ошибки, отвечают на вопрос «почему стало медленнее» без переписывания логики.

Тарификация CaptchaAI построена на потоках, а не на количестве решений: BASIC ($15/мес, 5 потоков) и STANDARD ($30/мес, 15 потоков) не ограничивают число задач внутри месяца. Поэтому смысл локального кэша не в экономии на счёте за решения, а в сокращении времени ожидания: пока поток занят повторной задачей, он не берёт новую.

Когда SQLite подходит, а когда пора уходить

Практическое правило: пока база лежит на локальном диске одной машины, SQLite — правильный выбор. Как только к файлу обращаются с нескольких хостов или через сетевую ФС, блокировки ломаются, и пора на клиент-серверную СУБД.

Шаг 1: спроектируйте схему

Таблиц две: журнал решений и кэш токенов. Индексы — по времени отправки, по паре «тип + статус» и по sitekey; именно эти три среза нужны в отчётах чаще всего.

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);

Первичный ключ кэша собран из трёх полей (sitekey, pageurl, token), а флаг used позволяет выдать токен ровно один раз. Составной индекс idx_cache_lookup покрывает запрос выборки целиком, поэтому обращение к кэшу не читает саму таблицу.

Шаг 2: подключитесь к базе и включите WAL

Два PRAGMA-параметра снимают почти все проблемы конкурентного доступа. journal_mode=WAL разрешает читать во время записи, а busy_timeout=5000 заставляет соединение подождать освобождения блокировки вместо немедленной ошибки database is locked.

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()

Путь к файлу вынесен в переменную окружения CAPTCHA_DB, API-ключ — в CAPTCHAAI_API_KEY. Держать ключ в коде не стоит ни на локальной машине, ни тем более в репозитории.

Шаг 3: решайте и пишите в журнал

Порядок действий: проверить кэш, вставить строку журнала, отправить задачу в in.php, опрашивать res.php, обновить строку итогом. Каждый переход статуса — отдельный UPDATE, поэтому даже прерванный процесс оставляет в базе понятный след: submitted, polling, solved, error или timeout.

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

Пример написан под reCAPTCHA v2 (method=userrecaptcha, поле ответа g-recaptcha-response) — разбор самого потока есть в руководстве как решить reCAPTCHA v2 через API. Для Cloudflare Turnstile меняются метод и имя параметра ключа, это описано в материале про Turnstile, а для GeeTest v3 — в отдельном руководстве. Поле type в таблице хранит строку типа, так что одна база обслуживает все типы сразу.

Шаг 4: включите кэш токенов с TTL

TTL в 90 секунд взят с запасом относительно реального времени жизни токена. Выдача помечает запись использованной в той же транзакции, что и чтение, — иначе два воркера получат один токен, и второй запрос сайт отклонит.

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

Кэш имеет смысл там, где одна и та же форма открывается повторно за короткий промежуток: пагинация, повторные попытки после сетевой ошибки, прогоны QA-сценария в staging. Для одноразовых задач он только усложняет код.

Шаг 5: считайте статистику и чистите базу

Отчёт за сутки строится тремя агрегатами: всего задач, решено, среднее время. Отдельная процедура удаляет старые строки и просроченные токены, после чего 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()

Запускайте cleanup_old_records() по расписанию — например, раз в сутки через cron. Без очистки файл растёт линейно, а VACUUM на большой базе надолго блокирует запись. И помните про локальное регулирование: если в pageurl попадают адреса с персональными данными, храните только то, что вы вправе обрабатывать, — это требование 152-ФЗ «О персональных данных» для команд в РФ и та же логика GDPR для распределённых команд.

Тот же журнал на Node.js

Для Node.js нужен better-sqlite3: синхронный API драйвера удобно ложится на пошаговый сценарий и на реальных объёмах не проигрывает асинхронным обёрткам.

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;
}

Схема совпадает с Python-версией, поэтому один файл базы можно читать и писать из обоих окружений — удобно, когда парсер написан на Node.js, а отчёты собирает Python-скрипт.

Частые проблемы

FAQ

Какой объём выдержит SQLite на одной машине?

В режиме WAL — порядка тысячи операций записи в секунду на обычном SSD. Для сценария «до 1000 решений в час» это запас в несколько порядков: узким местом станет не база, а число потоков вашего тарифа и время решения CAPTCHA.

Что делать, если база лежит в Docker-контейнере?

Файл базы выносите на именованный том, а не в слой образа: иначе журнал исчезнет вместе с контейнером. Для WAL нужен обычный том, а не bind-mount на сетевой каталог.

Как перенести данные в PostgreSQL или MongoDB?

Выгрузите дамп командой sqlite3 captcha_solves.db ".dump" и импортируйте в целевую СУБД. Типы столбцов переносятся почти без правок — поменять придётся AUTOINCREMENT на SERIAL/IDENTITY и способ генерации datetime('now').

Можно ли писать в одну базу из нескольких контейнеров?

Нет, если файл лежит на сетевом томе или в разных подах: блокировки SQLite на NFS работают ненадёжно. Внутри одной машины несколько процессов с WAL и busy_timeout сосуществуют нормально. Для нескольких хостов берите PostgreSQL или Redis.

Как связать журнал с расходом потоков?

Считайте одновременность: сгруппируйте строки со статусом polling по минутам и посмотрите пиковое значение. Если пик упирается в лимит потоков вашего тарифа, задачи начинают ждать в очереди — это сигнал перейти на следующий уровень, например со STANDARD ($30/мес, 15 потоков) на ADVANCE ($90/мес, 50 потоков).

Следующие шаги

• CaptchaAI Quickstart: ваше первое решение CAPTCHA за 5 минут
• Как решить reCAPTCHA v2 через API: пошаговое руководство
• Как решить Cloudflare Turnstile через API
• Как решить GeeTest v3 с помощью API