Развёртывание CaptchaAI в Google Cloud Functions

Serverless-функция снимает главный операционный вопрос при редких вызовах CaptchaAI API: не нужно держать выделенный сервер ради нагрузки, которая приходит неравномерно. Google Cloud Functions масштабируется от нуля до сотен параллельных вызовов автоматически, выставляет счёт по факту использования и отдаёт API-ключ через Secret Manager, а не через переменные окружения в открытом виде.

Ниже — три рабочих сценария: синхронная HTTP-функция для одиночных задач, воркер на Pub/Sub для пакетной обработки и расчёт, когда serverless дешевле постоянной виртуальной машины, а когда — нет.

HTTP-функция для решения CAPTCHA

Функция принимает POST-запрос с полями method и params, забирает API-ключ CaptchaAI из Secret Manager и отдаёт готовый токен синхронно — клиенту не нужно самому опрашивать res.php, весь цикл «отправить → подождать → забрать результат» скрыт внутри _solve().

# main.py
import json
import time
import urllib.request
import urllib.parse
import functions_framework


@functions_framework.http
def solve_captcha(request):
    """HTTP Cloud Function for CAPTCHA solving."""
    # Parse request
    request_json = request.get_json(silent=True)
    if not request_json:
        return json.dumps({"error": "JSON body required"}), 400

    method = request_json.get("method", "userrecaptcha")
    params = request_json.get("params", {})

    # Get API key from Secret Manager
    api_key = _get_secret("captchaai-key")

    try:
        token = _solve(api_key, method, params)
        return json.dumps({"token": token})
    except Exception as e:
        return json.dumps({"error": str(e)}), 500


def _get_secret(secret_id):
    """Get secret from GCP Secret Manager."""
    from google.cloud import secretmanager
    client = secretmanager.SecretManagerServiceClient()
    name = f"projects/{_get_project_id()}/secrets/{secret_id}/versions/latest"
    response = client.access_secret_version(request={"name": name})
    return response.payload.data.decode("UTF-8")


def _get_project_id():
    """Get current GCP project ID."""
    import urllib.request
    req = urllib.request.Request(
        "http://metadata.google.internal/computeMetadata/v1/project/project-id",
        headers={"Metadata-Flavor": "Google"},
    )
    with urllib.request.urlopen(req) as resp:
        return resp.read().decode()


def _solve(api_key, method, params, timeout=90):
    """Solve CAPTCHA via CaptchaAI API."""
    # Submit
    submit_data = urllib.parse.urlencode({
        "key": api_key,
        "method": method,
        "json": 1,
        **params,
    }).encode()

    req = urllib.request.Request(
        "https://ocr.captchaai.com/in.php",
        data=submit_data,
    )
    with urllib.request.urlopen(req, timeout=30) as resp:
        result = json.loads(resp.read())

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

    task_id = result["request"]

    # Poll
    start = time.time()
    while time.time() - start < timeout:
        time.sleep(5)
        poll_url = (
            f"https://ocr.captchaai.com/res.php"
            f"?key={api_key}&action=get&id={task_id}&json=1"
        )
        with urllib.request.urlopen(poll_url, timeout=15) as resp:
            data = json.loads(resp.read())

        if data["request"] != "CAPCHA_NOT_READY":
            if data.get("status") == 1:
                return data["request"]
            raise RuntimeError(f"Solve error: {data['request']}")

    raise TimeoutError("Solve timeout")

Модуль urllib вместо requests — не случайность: меньше зависимостей в пакете функции — короче холодный старт (см. таблицу неисправностей ниже).

Зависимости

Функции нужны всего два пакета — functions-framework для HTTP- и CloudEvent-триггеров и клиент Secret Manager:

# requirements.txt
functions-framework==3.*
google-cloud-secret-manager==2.*

Разверните функцию

Сначала создайте секрет с API-ключом CaptchaAI, затем разверните функцию с таймаутом, покрывающим весь цикл опроса res.php внутри _solve():

# Create secret
echo -n "YOUR_API_KEY" | gcloud secrets create captchaai-key --data-file=-

# Deploy function
gcloud functions deploy solve-captcha \
  --gen2 \
  --runtime=python311 \
  --region=us-central1 \
  --source=. \
  --entry-point=solve_captcha \
  --trigger-http \
  --allow-unauthenticated \
  --timeout=120s \
  --memory=256MB \
  --max-instances=100

# Test
curl -X POST https://us-central1-PROJECT.cloudfunctions.net/solve-captcha \
  -H "Content-Type: application/json" \
  -d '{
    "method": "userrecaptcha",
    "params": {
      "googlekey": "SITE_KEY",
      "pageurl": "https://example.com"
    }
  }'

Если основная аудитория и инфраструктура ближе к Европе, замените --region=us-central1 на europe-west3 или europe-west1 — команды из России, Беларуси, Казахстана и Украины обычно получают меньший RTT до CaptchaAI API именно так, а флаг деплоя остаётся тем же.

Пакетная обработка через Pub/Sub

Для тысяч задач в день синхронный HTTP-вызов на каждую CAPTCHA — плохая идея: клиент будет висеть на соединении до 90 секунд, пока идёт опрос res.php. Здесь удобнее очередь — задача публикуется в топик, функция решает её в фоне и кладёт результат в отдельный топик с ответами:

import base64
import json
import functions_framework
from google.cloud import pubsub_v1


@functions_framework.cloud_event
def process_captcha_task(cloud_event):
    """Process CAPTCHA task from Pub/Sub message."""
    data = base64.b64decode(cloud_event.data["message"]["data"])
    task = json.loads(data)

    api_key = _get_secret("captchaai-key")

    try:
        token = _solve(api_key, task["method"], task["params"])
        # Publish result
        publisher = pubsub_v1.PublisherClient()
        topic = f"projects/{_get_project_id()}/topics/captcha-results"
        publisher.publish(topic, json.dumps({
            "task_id": task["id"],
            "status": "success",
            "token": token,
        }).encode())

    except Exception as e:
        print(f"Task {task.get('id')} failed: {e}")

Разверните обработчик как CloudEvent-функцию, подписанную на топик с задачами:

gcloud functions deploy process-captcha-task \
  --gen2 \
  --runtime=python311 \
  --trigger-topic=captcha-tasks \
  --timeout=120s \
  --memory=256MB

Отправка задач в Pub/Sub

Публикатор ниже кладёт пакет задач в топик — например, из скрипта, который раз в час собирает список URL для проверки:

from google.cloud import pubsub_v1
import json

publisher = pubsub_v1.PublisherClient()
topic = "projects/YOUR_PROJECT/topics/captcha-tasks"

# Submit batch
urls = ["https://site1.com", "https://site2.com", "https://site3.com"]
for i, url in enumerate(urls):
    task = {
        "id": f"task-{i}",
        "method": "userrecaptcha",
        "params": {"googlekey": "SITE_KEY", "pageurl": url},
    }
    publisher.publish(topic, json.dumps(task).encode())
    print(f"Published task-{i}")

Экономика: Cloud Functions против постоянной VM

При нерегулярной нагрузке serverless выигрывает по цене на порядок. При устойчивом потоке от 10 000 решений в день разница стирается — VM с фиксированной ценой перестаёт проигрывать:

Эта таблица про инфраструктуру GCP. Сами решения CaptchaAI тарифицируются отдельно — по потокам, а не поштучно: план BASIC ($15/мес, 5 потоков) уже даёт параллелизм, в который сложно упереться на объёмах из таблицы выше, а более высокая нагрузка закрывается STANDARD ($30/мес, 15 потоков) или ADVANCE ($90/мес, 50 потоков) без пересчёта архитектуры функции.

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

Четыре ситуации встречаются чаще остальных:

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

Сколько потоков CaptchaAI нужно для параллельных вызовов из Cloud Functions?

Один поток = одна CAPTCHA, которая решается прямо сейчас; как только решение готово, поток освобождается для следующей задачи. При --max-instances=100 реальным ограничителем параллелизма обычно становится план CaptchaAI, а не Cloud Functions — для сотен одновременных задач нужен тариф с сопоставимым числом потоков, а не только высокий max-instances.

Gen1 или Gen2 для CAPTCHA-функции?

Gen2. У него таймаут до 60 минут вместо 9, больше памяти и параллелизм на уровне инстанса — при опросе res.php в цикле это снижает риск обрыва на медленных типах CAPTCHA.

Что делать, если функция постоянно ловит CAPCHA_NOT_READY до истечения таймаута?

Это нормальное поведение при опросе, а не ошибка — важно, чтобы timeout в _solve() не был меньше --timeout деплоя. Если тайм-аут истекает регулярно на одном и том же типе CAPTCHA, увеличьте оба значения синхронно, а не только один из них.

Нужен ли прокси при вызове CaptchaAI API из Cloud Functions?

Нет — Cloud Functions обращается к ocr.captchaai.com напрямую по HTTPS без дополнительной настройки сети. Serverless VPC Access понадобится, только если сама функция должна попасть во внутреннюю VPC (например, к внутренней очереди задач), а не для вызова внешнего API CaptchaAI.

Как не платить за холодный старт при редких вызовах?

Поставьте Cloud Scheduler дёргать функцию раз в 5 минут, либо задайте --min-instances=1, чтобы один инстанс оставался тёплым постоянно (около $7/мес по прайсу Cloud Functions). Для действительно редкой нагрузки холодный старт на ~300 мс обычно дешевле, чем держать инстанс тёплым круглосуточно.

Связанные материалы

• AWS Lambda и CaptchaAI
• Azure Functions и CaptchaAI

Serverless на GCP — получите API-ключ CaptchaAI и разверните первую функцию уже сегодня.