MongoDB для истории и аналитики решения CAPTCHA

История решений CAPTCHA — это данные с непостоянной структурой: у reCAPTCHA один набор полей запроса, у Cloudflare Turnstile другой, у image-CAPTCHA третий, а объём растёт вместе с трафиком. Жёсткая реляционная схема на это отвечает ALTER TABLE и кучей nullable-колонок, а MongoDB — просто ещё одним документом.

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

В этом руководстве вы найдёте:

• схему документа под записи любого типа CAPTCHA;
• функцию приёма и опроса с записью каждого статуса в MongoDB;
• готовые запросы: доля успешных решений, время решения по типу, почасовой объём, разбивка ошибок;
• политику хранения с TTL-индексом и разбором частых проблем;
• ответы на вопросы, которые обычно возникают при эксплуатации такой коллекции.

Почему схема MongoDB подходит для логов CAPTCHA

Набор полей у записи решения меняется от типа к типу: reCAPTCHA требует googlekey, hCaptcha — sitekey, image-CAPTCHA — body. В реляционной БД это означает либо одну таблицу с десятком nullable-колонок под каждый возможный тип, либо отдельную таблицу на тип с последующими JOIN при аналитике по всем типам сразу. MongoDB хранит их как обычные документы: один тип — одна форма документа, без миграций схемы при добавлении нового типа CAPTCHA в проект.

Это же удобно и для метаданных задачи. Команда, которая парсит цены конкурентов через несколько воркеров, может держать в каждой записи metadata.project: "price-monitor" и metadata.worker_id, а затем без единой миграции построить агрегацию по проекту, воркеру или целевому домену — поле metadata.target_domain в примере ниже как раз под это.

Схема документа для коллекции solves

Документ хранит весь путь одной попытки: от отправки задачи через sitekey/pageurl до итогового solution или кода error, с отметками времени submitted_at/solved_at и посчитанным elapsed_ms. Поле metadata — свободный объект под контекст вызова (проект, воркер, домен), поэтому оно не потребует изменения схемы при росте числа сценариев использования.

{
  "_id": "ObjectId",
  "captcha_id": "12345678",
  "type": "recaptcha_v2",
  "method": "userrecaptcha",
  "sitekey": "6Le-wvkSAAAAAPBMRTvw0Q4Muexq9bi0DJwx_mJ-",
  "pageurl": "https://example.com/form",
  "status": "solved",
  "solution": "03AGdBq26...",
  "error": null,
  "submitted_at": "2026-04-04T10:15:30.000Z",
  "solved_at": "2026-04-04T10:15:45.000Z",
  "elapsed_ms": 15000,
  "polls": 3,
  "proxy_used": true,
  "cost": 0.00299,
  "metadata": {
    "project": "price-monitor",
    "worker_id": "worker-3",
    "target_domain": "example.com"
  }
}

Реализация на Python

Ниже — четыре части одной реализации: подключение к MongoDB, индексы под будущую аналитику, функция приёма и опроса задачи у CaptchaAI и сами аналитические запросы. Все четыре работают с одной и той же коллекцией solves из схемы выше.

Подключение к MongoDB

import os
import time
from datetime import datetime, timezone
from pymongo import MongoClient, ASCENDING, DESCENDING
import requests

MONGO_URI = os.environ.get("MONGO_URI", "mongodb://localhost:27017")
API_KEY = os.environ["CAPTCHAAI_API_KEY"]

client = MongoClient(MONGO_URI)
db = client["captcha_tracking"]
solves = db["solves"]

Индексы для быстрой аналитики

Без индексов на submitted_at и (type, status) каждый $match из раздела аналитики ниже превращается в полный скан коллекции — на десятках тысяч записей разница уже заметна. TTL-индекс ttl_cleanup здесь же берёт на себя автоматическую очистку старых записей.

def setup_indexes():
    solves.create_index([("submitted_at", DESCENDING)])
    solves.create_index([("type", ASCENDING), ("status", ASCENDING)])
    solves.create_index([("metadata.project", ASCENDING)])
    solves.create_index([("metadata.target_domain", ASCENDING)])
    solves.create_index(
        [("submitted_at", ASCENDING)],
        expireAfterSeconds=90 * 24 * 3600,  # Auto-delete after 90 days
        name="ttl_cleanup"
    )

setup_indexes()

Отправить задачу и записать результат

Функция создаёт документ со статусом submitted, отправляет задачу в CaptchaAI на in.php, затем опрашивает res.php с интервалом в 5 секунд и обновляет тот же документ на каждом шаге — так весь путь попытки, включая число опросов и итоговое время решения, остаётся в одной записи, а не размазывается по логам.

def solve_and_store(sitekey, pageurl, captcha_type="recaptcha_v2", metadata=None):
    record = {
        "type": captcha_type,
        "method": "userrecaptcha",
        "sitekey": sitekey,
        "pageurl": pageurl,
        "status": "submitted",
        "submitted_at": datetime.now(timezone.utc),
        "metadata": metadata or {}
    }

    result = solves.insert_one(record)
    doc_id = result.inserted_id

    # 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:
        solves.update_one(
            {"_id": doc_id},
            {"$set": {"status": "error", "error": data.get("request")}}
        )
        return None

    captcha_id = data["request"]
    solves.update_one(
        {"_id": doc_id},
        {"$set": {"captcha_id": captcha_id, "status": "polling"}}
    )

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

        if poll_resp.get("status") == 1:
            solved_at = datetime.now(timezone.utc)
            elapsed_ms = int(
                (solved_at - record["submitted_at"]).total_seconds() * 1000
            )
            solves.update_one({"_id": doc_id}, {"$set": {
                "status": "solved",
                "solution": poll_resp["request"],
                "solved_at": solved_at,
                "elapsed_ms": elapsed_ms,
                "polls": polls
            }})
            return poll_resp["request"]

        if poll_resp.get("request") != "CAPCHA_NOT_READY":
            solves.update_one({"_id": doc_id}, {"$set": {
                "status": "error",
                "error": poll_resp.get("request"),
                "polls": polls
            }})
            return None

    solves.update_one({"_id": doc_id}, {"$set": {
        "status": "timeout", "polls": polls
    }})
    return None

Аналитические запросы к истории решений

Четыре готовых pipeline закрывают большинство вопросов, которые обычно задают об истории решений: какая доля решений успешна за период, сколько в среднем занимает решение по каждому типу CAPTCHA, как распределён объём по часам и какие ошибки встречаются чаще всего.

def get_success_rate(hours=24):
    """Success rate for the last N hours."""
    from datetime import timedelta
    cutoff = datetime.now(timezone.utc) - timedelta(hours=hours)

    pipeline = [
        {"$match": {"submitted_at": {"$gte": cutoff}}},
        {"$group": {
            "_id": "$status",
            "count": {"$sum": 1}
        }}
    ]
    results = {r["_id"]: r["count"] for r in solves.aggregate(pipeline)}
    total = sum(results.values())
    solved = results.get("solved", 0)
    return (solved / total * 100) if total else 0


def get_avg_solve_time_by_type():
    """Average solve time grouped by CAPTCHA type."""
    pipeline = [
        {"$match": {"status": "solved"}},
        {"$group": {
            "_id": "$type",
            "avg_time_ms": {"$avg": "$elapsed_ms"},
            "min_time_ms": {"$min": "$elapsed_ms"},
            "max_time_ms": {"$max": "$elapsed_ms"},
            "count": {"$sum": 1}
        }},
        {"$sort": {"count": -1}}
    ]
    return list(solves.aggregate(pipeline))


def get_hourly_solve_volume(days=7):
    """Hourly solve volume for charting."""
    from datetime import timedelta
    cutoff = datetime.now(timezone.utc) - timedelta(days=days)

    pipeline = [
        {"$match": {"submitted_at": {"$gte": cutoff}}},
        {"$group": {
            "_id": {
                "date": {"$dateToString": {"format": "%Y-%m-%d", "date": "$submitted_at"}},
                "hour": {"$hour": "$submitted_at"}
            },
            "total": {"$sum": 1},
            "solved": {"$sum": {"$cond": [{"$eq": ["$status", "solved"]}, 1, 0]}}
        }},
        {"$sort": {"_id.date": 1, "_id.hour": 1}}
    ]
    return list(solves.aggregate(pipeline))


def get_error_breakdown(hours=24):
    """Error frequency by error code."""
    from datetime import timedelta
    cutoff = datetime.now(timezone.utc) - timedelta(hours=hours)

    pipeline = [
        {"$match": {"submitted_at": {"$gte": cutoff}, "status": "error"}},
        {"$group": {"_id": "$error", "count": {"$sum": 1}}},
        {"$sort": {"count": -1}}
    ]
    return list(solves.aggregate(pipeline))

Реализация на Node.js

Та же логика — приём, опрос, обновление статуса и расчёт доли успешных решений — на axios и официальном драйвере mongodb, если стек проекта уже на JavaScript и заводить второй язык под один сервис нет смысла.

const { MongoClient } = require("mongodb");
const axios = require("axios");

const MONGO_URI = process.env.MONGO_URI || "mongodb://localhost:27017";
const API_KEY = process.env.CAPTCHAAI_API_KEY;

let db, solves;

async function connect() {
  const client = await MongoClient.connect(MONGO_URI);
  db = client.db("captcha_tracking");
  solves = db.collection("solves");

  await solves.createIndex({ submitted_at: -1 });
  await solves.createIndex({ type: 1, status: 1 });
  await solves.createIndex({ "metadata.project": 1 });
  await solves.createIndex(
    { submitted_at: 1 },
    { expireAfterSeconds: 90 * 24 * 3600 }
  );
}

async function solveAndStore(sitekey, pageurl, type = "recaptcha_v2", metadata = {}) {
  const submittedAt = new Date();
  const { insertedId } = await solves.insertOne({
    type, method: "userrecaptcha", sitekey, pageurl,
    status: "submitted", submitted_at: submittedAt, metadata,
  });

  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) {
    await solves.updateOne({ _id: insertedId }, { $set: { status: "error", error: submit.data.request } });
    return null;
  }

  const captchaId = submit.data.request;
  await solves.updateOne({ _id: insertedId }, { $set: { captcha_id: captchaId, status: "polling" } });

  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 solvedAt = new Date();
      await solves.updateOne({ _id: insertedId }, { $set: {
        status: "solved", solution: poll.data.request,
        solved_at: solvedAt, elapsed_ms: solvedAt - submittedAt, polls,
      }});
      return poll.data.request;
    }
    if (poll.data.request !== "CAPCHA_NOT_READY") {
      await solves.updateOne({ _id: insertedId }, { $set: { status: "error", error: poll.data.request, polls } });
      return null;
    }
  }

  await solves.updateOne({ _id: insertedId }, { $set: { status: "timeout", polls } });
  return null;
}

async function getSuccessRate(hours = 24) {
  const cutoff = new Date(Date.now() - hours * 3600 * 1000);
  const pipeline = [
    { $match: { submitted_at: { $gte: cutoff } } },
    { $group: { _id: "$status", count: { $sum: 1 } } },
  ];
  const results = await solves.aggregate(pipeline).toArray();
  const total = results.reduce((s, r) => s + r.count, 0);
  const solved = results.find((r) => r._id === "solved")?.count || 0;
  return total ? ((solved / total) * 100).toFixed(1) : 0;
}

Хранение и очистка данных

Выбор TTL зависит от того, зачем вообще хранится история. При выборе стратегии учитывайте:

• как долго записи реально нужны для отладки конкретного воркера;
• нужна ли долгосрочная аналитика по типам и доменам, а не только последние сутки;
• есть ли требование хранить данные для аудита дольше, чем для повседневной работы;
• сколько метаданных пишется в каждую запись — от этого зависит темп роста коллекции.

Если в записи попадают данные реальных пользователей, а не только служебные метаданные самой CAPTCHA, учитывайте требования к персональным данным — 152-ФЗ для аудитории из РФ, GDPR-дисциплина для остальных. Храните только то, что действительно нужно обрабатывать, и не дольше срока, который можете обосновать.

Типичные проблемы и их решение

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

Частые вопросы

Что делать, если TTL-индекс не чистит коллекцию вовремя?

Сначала проверьте, что индекс ttl_cleanup вообще создан и указывает на поле с типом Date — на строке или числе expireAfterSeconds просто не сработает. Если индекс на месте, дайте фоновому TTL-монитору время: он запускается раз в 60 секунд, а на коллекциях с миллионами документов чистка идёт пакетами и может отставать от submitted_at на десятки минут.

Какие индексы обязательны, если решений уже миллионы?

Минимум три: составной (type, status) под группировки в аналитических pipeline, submitted_at под выборки по времени и отдельный индекс на metadata.project или metadata.target_domain, если вы часто фильтруете по проекту или домену. Без них каждый $match в pipeline из раздела аналитики превращается в полный скан коллекции, и время ответа растёт линейно с её размером.

Сколько воркеров можно одновременно писать в коллекцию solves?

Технического потолка со стороны самой MongoDB здесь нет — узкое место обычно в maxPoolSize клиента. Драйвер PyMongo по умолчанию открывает пул на 100 соединений, и если несколько десятков воркеров параллельно опрашивают res.php и обновляют статус в solves, пул может исчерпаться раньше, чем ответит CaptchaAI. Поднимите maxPoolSize в строке подключения и ориентируйтесь на строку «Исчерпан пул соединений» в таблице выше, если запросы начинают падать по тайм-ауту.

Можно ли использовать MongoDB Atlas вместо self-hosted кластера?

Да, схема и оба примера кода работают без изменений — Atlas поддерживает TTL-индексы и aggregation pipeline так же, как self-hosted кластер. Достаточно подставить строку подключения из панели Atlas в переменную MONGO_URI; команды из СНГ и Восточной Европы обычно выигрывают в задержке, выбирая ближайший европейский регион, а не американский по умолчанию.

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

Да, в этом и смысл схемы из начала статьи: поле type определяет, какие остальные поля заполнены, поэтому reCAPTCHA, Cloudflare Turnstile и GeeTest v3 спокойно живут в одной коллекции solves. При добавлении нового поддерживаемого типа меняется только method и набор полей самого запроса — индексы, TTL и все четыре pipeline из раздела аналитики продолжают работать без изменений.

Что дальше

Логика из этого руководства переносится на любой поддерживаемый тип CAPTCHA — меняется только method и набор полей запроса в схеме. Разверните такую же коллекцию под свой сценарий и подключите остальные нужные типы по мере необходимости:

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