CaptchaAI Pingback и шаблоны уведомлений о задачах

Опрашивать res.php каждые несколько секунд на десятках параллельных задач — рабочий, но затратный по трафику способ забрать токен. Параметр pingback в CaptchaAI решает это иначе: как только CAPTCHA решена, сервер сам отправляет GET-запрос на ваш URL с результатом, и опрос вообще не нужен. Ниже — рабочие схемы для продакшена: от простого хранилища результатов до маршрутизации уведомлений между обработчиками.

Из статьи вы получите:

• три готовых Python-паттерна приёма callback — от одиночной задачи до маршрутизации по обработчикам;
• чек-лист защиты pingback-эндпоинта от чужого трафика;
• критерий выбора между pingback и опросом res.php для конкретно вашей нагрузки.

Как CaptchaAI отправляет уведомление

Прежде чем переходить к схемам, разберём базовый поток запросов:

1. Submit task with pingback=YOUR_CALLBACK_URL
2. CaptchaAI solves the CAPTCHA
3. CaptchaAI sends GET request to your callback:
   YOUR_CALLBACK_URL?id=TASK_ID&code=TOKEN

4. Your server processes the result

На шаге 3 CaptchaAI сам обращается к вашему серверу — значит, эндпоинт должен быть доступен из интернета уже в момент отправки задачи, а не только к моменту готовности токена.

Прежде чем подключать pingback к продакшену, проверьте три вещи:

• эндпоинт слушает публичный порт (не 127.0.0.1, не адрес за NAT без проброса);
• обработчик отвечает быстро (в идеале до 1–2 секунд) — долгая обработка внутри самого HTTP-хендлера задерживает подтверждение CaptchaAI;
• логика идемпотентна: одна и та же задача может прийти повторно, и повтор не должен ломать состояние.

Приём 1: fire-and-forget с потокобезопасным хранилищем

Самый простой вариант — отправить задачу и сохранить результат в словаре с блокировкой, как только придёт callback:

import requests
import threading
import time
from flask import Flask, request


class PingbackStore:
    """Store for results received via pingback."""

    def __init__(self):
        self.results = {}
        self.events = {}
        self.lock = threading.Lock()

    def register(self, task_id):
        """Register a task ID we expect results for."""
        with self.lock:
            self.events[task_id] = threading.Event()

    def store(self, task_id, token):
        """Store result from pingback callback."""
        with self.lock:
            self.results[task_id] = token
            if task_id in self.events:
                self.events[task_id].set()

    def wait(self, task_id, timeout=120):
        """Wait for a specific result."""
        event = self.events.get(task_id)
        if not event:
            return None
        event.wait(timeout=timeout)
        return self.results.get(task_id)

    def get(self, task_id):
        """Get result without waiting (non-blocking)."""
        return self.results.get(task_id)


# Global store
store = PingbackStore()

# Flask app for receiving callbacks
app = Flask(__name__)


@app.route("/pingback")
def receive_pingback():
    """Handle CaptchaAI pingback callback."""
    task_id = request.args.get("id")
    code = request.args.get("code")

    if not task_id or not code:
        return "Bad request", 400

    store.store(task_id, code)
    return "OK", 200


def submit_with_pingback(api_key, method, callback_url, **params):
    """Submit a task with pingback enabled."""
    data = {
        "key": api_key,
        "method": method,
        "pingback": callback_url,
        "json": 1,
    }
    data.update(params)

    resp = requests.post(
        "https://ocr.captchaai.com/in.php",
        data=data,
        timeout=30,
    )
    result = resp.json()

    if result.get("status") != 1:
        raise RuntimeError(f"Submit error: {result.get('request')}")

    task_id = result["request"]
    store.register(task_id)
    return task_id


# Usage
# Start Flask server in background thread
server = threading.Thread(
    target=lambda: app.run(port=8080, debug=False),
    daemon=True,
)
server.start()

# Submit task
task_id = submit_with_pingback(
    "YOUR_API_KEY",
    "userrecaptcha",
    "https://yourserver.com/pingback",
    googlekey="SITE_KEY",
    pageurl="https://example.com",
)

# Wait for result via pingback
token = store.wait(task_id, timeout=120)
print(f"Token: {token[:50]}...")

Когда достаточно этой схемы

Такой схемы хватает, когда сервис решает задачи по одной — например, авторизация в веб-приложении по клику пользователя. Event из threading блокирует только тот запрос, который ждёт конкретный task_id, остальной процесс продолжает работать. Для десятков параллельных задач переходите к следующему приёму.

Приём 2: параллельная отправка задач (fan-out)

Когда нужно решить сразу пачку CAPTCHA, отправьте все задачи и собирайте токены по мере готовности — ждать очереди не придётся:

import requests
import threading
import time


class FanOutSolver:
    """Submit many tasks, collect results via pingback."""

    def __init__(self, api_key, callback_url):
        self.api_key = api_key
        self.callback_url = callback_url
        self.store = PingbackStore()
        self.pending = []

    def submit(self, method, **params):
        """Submit a task and track it."""
        data = {
            "key": self.api_key,
            "method": method,
            "pingback": self.callback_url,
            "json": 1,
        }
        data.update(params)

        resp = requests.post(
            "https://ocr.captchaai.com/in.php",
            data=data,
            timeout=30,
        )
        result = resp.json()

        if result.get("status") != 1:
            raise RuntimeError(f"Submit error: {result.get('request')}")

        task_id = result["request"]
        self.store.register(task_id)
        self.pending.append(task_id)
        return task_id

    def submit_batch(self, tasks):
        """Submit multiple tasks.

        tasks: list of dicts with 'method' and params
        """
        task_ids = []
        for task in tasks:
            method = task.pop("method")
            task_id = self.submit(method, **task)
            task_ids.append(task_id)
            time.sleep(0.1)  # Avoid rate limits
        return task_ids

    def collect_all(self, timeout=180):
        """Wait for all pending results."""
        results = {}
        deadline = time.time() + timeout

        for task_id in self.pending:
            remaining = max(1, deadline - time.time())
            token = self.store.wait(task_id, timeout=remaining)
            results[task_id] = token

        self.pending.clear()
        return results


# Usage
solver = FanOutSolver("YOUR_API_KEY", "https://yourserver.com/pingback")

# Submit 5 tasks
tasks = [
    {
        "method": "userrecaptcha",
        "googlekey": "SITE_KEY",
        "pageurl": f"https://example.com/page{i}",
    }
    for i in range(5)
]

task_ids = solver.submit_batch(tasks)
print(f"Submitted {len(task_ids)} tasks")

# Wait for all results
results = solver.collect_all(timeout=180)
for tid, token in results.items():
    status = "solved" if token else "failed"
    print(f"  {tid}: {status}")

Почему пауза между отправками

time.sleep(0.1) в submit_batch — не декоративная строка. Без паузы пачка из сотен задач уходит на in.php одним залпом и упирается в ограничение частоты запросов на вашем тарифе; с паузой отправка растягивается на доли секунды на задачу и остаётся в пределах лимита, при этом решение всё равно идёт параллельно на стороне CaptchaAI.

Приём 3: маршрутизация уведомлений по обработчикам

Если один и тот же pingback-эндпоинт обслуживает разные сценарии — логин, парсинг, регистрацию, — распределяйте результаты по нужным обработчикам, а не разбирайте их в одном большом if:

import threading
from collections import defaultdict


class NotificationRouter:
    """Route pingback results to registered handlers."""

    def __init__(self):
        self.handlers = {}
        self.default_handler = None
        self.task_routes = {}
        self.lock = threading.Lock()

    def register_handler(self, name, handler_fn):
        """Register a named handler function."""
        self.handlers[name] = handler_fn

    def set_default(self, handler_fn):
        """Set a default handler for unrouted tasks."""
        self.default_handler = handler_fn

    def route(self, task_id, handler_name):
        """Route a task ID to a specific handler."""
        with self.lock:
            self.task_routes[task_id] = handler_name

    def dispatch(self, task_id, token):
        """Dispatch a result to the correct handler."""
        handler_name = self.task_routes.get(task_id)

        if handler_name and handler_name in self.handlers:
            self.handlers[handler_name](task_id, token)
        elif self.default_handler:
            self.default_handler(task_id, token)


# Usage
router = NotificationRouter()

# Register handlers
def login_handler(task_id, token):
    print(f"Login flow got token from {task_id}")
    # Submit token to login form

def scraping_handler(task_id, token):
    print(f"Scraping pipeline got token from {task_id}")
    # Continue scraping with token

router.register_handler("login", login_handler)
router.register_handler("scraping", scraping_handler)

# When submitting
task_id = submit_with_pingback(
    "YOUR_API_KEY", "userrecaptcha",
    "https://yourserver.com/pingback",
    googlekey="KEY", pageurl="https://example.com",
)
router.route(task_id, "login")

# In pingback handler
# router.dispatch(task_id, token)

Когда нужен маршрутизатор

Если callback-эндпоинт один на всё приложение, а сценариев использования несколько (вход в аккаунт, парсинг карточек товара, регистрация), NotificationRouter избавляет от разрастающегося if task_type == "login": ... elif task_type == "scraping": ... прямо внутри обработчика Flask — обработчики регистрируются один раз при старте, а маршрутизация происходит по task_id.

Как защитить pingback-эндпоинт

Публичный URL, принимающий GET-запросы, — потенциальная цель для чужого трафика. Проверяйте формат параметров и не доверяйте телу запроса вслепую. Обязательный минимум проверок:

• параметры id и code присутствуют и не пустые;
• id соответствует ожидаемому формату (у CaptchaAI это числовой идентификатор задачи);
• запрос пришёл на известный вам task_id — то есть задача была зарегистрирована на шаге отправки, а не появилась из ниоткуда;
• по возможности — сверка IP-адреса отправителя со списком известных диапазонов CaptchaAI.

import hmac
import hashlib
from flask import Flask, request, abort

app = Flask(__name__)
API_KEY = "YOUR_API_KEY"


@app.route("/pingback")
def secure_pingback():
    """Validate pingback requests."""
    task_id = request.args.get("id")
    code = request.args.get("code")
    ip = request.remote_addr

    # Validate required parameters
    if not task_id or not code:
        abort(400)

    # Validate IP (CaptchaAI server IPs)
    # Add actual CaptchaAI IPs to allowlist
    ALLOWED_IPS = {"0.0.0.0/0"}  # Replace with real IPs

    # Validate task ID format (numeric)
    if not task_id.isdigit():
        abort(400)

    # Store result
    store.store(task_id, code)
    return "OK", 200

Если вы логируете IP-адрес и параметры запроса для отладки, учитывайте требования 152-ФЗ «О персональных данных» (для аудитории за пределами РФ — аналогичную дисциплину GDPR): храните только то, что нужно для диагностики, и не дольше срока ретенции логов.

Pingback или опрос: что выбрать

Команде на плане ADVANCE ($90/мес, 50 потоков) pingback особенно выгоден: поток не простаивает в ожидании ответа res.php, а сразу берёт следующую задачу — при том же тарифе через эндпоинт проходит больше решений в час, чем при постоянном опросе.

Коротко о выборе:

• Берите pingback, если у вас уже есть публичный сервер, задач много (десятки-сотни одновременно) и важна минимальная задержка получения токена.
• Оставайтесь на опросе res.php, если проект небольшой, деплой без публичного порта (например, серверлесс-функция без входящего HTTP) или прототип, который ещё не готов к продакшен-инфраструктуре.

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

Если проблема не находится по таблице, начните с самого дешёвого теста: отправьте curl -X POST на ваш собственный /pingback с тестовыми id и code вручную. Если обработчик не отвечает 200 OK на прямой вызов, дело не в CaptchaAI и не в сети — ошибка внутри вашего кода, и её проще поймать в изоляции, чем через реальную задачу.

Часто задаваемые вопросы

Нужен ли HTTPS для callback URL?

Строгого требования нет, но HTTPS обязателен на практике: токен передаётся в query-параметре, и по HTTP его может перехватить кто угодно между CaptchaAI и вашим сервером.

Что делать, если pingback не пришёл вовремя?

Держите резервный опрос res.php по task_id с таймаутом — если callback не пришёл за разумное время (например, 60–120 секунд), заберите результат опросом и не блокируйте пайплайн бесконечным ожиданием.

Можно ли совмещать pingback и опрос res.php в одном проекте?

Да, и это обычная практика: pingback как основной канал для скорости, опрос — как страховка на случай сетевых сбоев между CaptchaAI и вашим эндпоинтом.

Можно ли тестировать pingback на localhost?

Нет, callback URL должен быть общедоступным. Для локальной разработки поднимите туннель (ngrok или аналог) и укажите его адрес в параметре pingback.

Нужно ли что-то дополнительно оплачивать за pingback?

Нет, pingback — это обычный параметр запроса к in.php, а не отдельная услуга: он включён в тот же тариф, что и опрос res.php, и не тарифицируется отдельно по потокам.

Итог: с чего начать

• Для одиночных задач достаточно Приёма 1 (fire-and-forget) — минимум кода, максимум пользы.
• Для пачек CAPTCHA переходите к Приёму 2 (fan-out) — экономит время на 5+ одновременных задачах.
• Для приложений с несколькими сценариями подключите Приём 3 (маршрутизатор) поверх любого из первых двух.
• В любом случае — разверните проверку параметров из раздела о защите эндпоинта до того, как включите pingback в продакшене, а не после первого инцидента.

Связанные руководства

• Руководство по callback URL и вебхукам
• Пакетное решение CAPTCHA

Стройте событийно-ориентированные пайплайны — получите API-ключ CaptchaAI уже сейчас.