Реализация надежной логики повторных попыток с помощью CaptchaAI API

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

Какие ошибки повторять

Базовая повторная попытка с экспоненциальной задержкой

import requests
import time
import random

API_KEY = "YOUR_API_KEY"
BASE_URL = "https://ocr.captchaai.com"

# Errors that should NOT be retried
PERMANENT_ERRORS = {
    "ERROR_WRONG_USER_KEY",
    "ERROR_KEY_DOES_NOT_EXIST",
    "ERROR_ZERO_BALANCE",
    "ERROR_BAD_PARAMETERS",
    "ERROR_WRONG_CAPTCHA_ID",
}

# Errors that should be retried
TRANSIENT_ERRORS = {
    "ERROR_NO_SLOT_AVAILABLE",
    "ERROR_TOO_MUCH_REQUESTS",
}


def submit_with_retry(method, max_retries=5, **params):
    """Submit task with retry on transient errors."""
    data = {"key": API_KEY, "method": method, "json": 1}
    data.update(params)

    for attempt in range(max_retries):
        try:
            resp = requests.post(
                f"{BASE_URL}/in.php", data=data, timeout=30,
            )

            # HTTP-level errors
            if resp.status_code in (429, 500, 502, 503):
                wait = _backoff(attempt)
                print(f"HTTP {resp.status_code}, retry in {wait:.1f}s")
                time.sleep(wait)
                continue

            result = resp.json()

            # Permanent errors — don't retry
            if result.get("request") in PERMANENT_ERRORS:
                raise RuntimeError(f"Permanent error: {result['request']}")

            # Transient errors — retry
            if result.get("request") in TRANSIENT_ERRORS:
                wait = _backoff(attempt)
                print(f"{result['request']}, retry in {wait:.1f}s")
                time.sleep(wait)
                continue

            # Success
            if result.get("status") == 1:
                return result["request"]

            # Unknown error
            raise RuntimeError(f"Unknown error: {result.get('request')}")

        except requests.ConnectionError:
            wait = _backoff(attempt)
            print(f"Connection error, retry in {wait:.1f}s")
            time.sleep(wait)

        except requests.Timeout:
            wait = _backoff(attempt)
            print(f"Timeout, retry in {wait:.1f}s")
            time.sleep(wait)

    raise RuntimeError(f"Failed after {max_retries} retries")


def _backoff(attempt, base=2, max_wait=60):
    """Exponential backoff with jitter."""
    wait = min(base ** attempt, max_wait)
    jitter = random.uniform(0, wait * 0.5)
    return wait + jitter

Опрос с повтором

def poll_with_retry(task_id, timeout=120, max_poll_errors=3):
    """Poll for result with error retry."""
    start = time.time()
    consecutive_errors = 0

    while time.time() - start < timeout:
        time.sleep(5)

        try:
            resp = requests.get(f"{BASE_URL}/res.php", params={
                "key": API_KEY, "action": "get",
                "id": task_id, "json": 1,
            }, timeout=15)

            if resp.status_code in (429, 500, 502, 503):
                consecutive_errors += 1
                if consecutive_errors >= max_poll_errors:
                    raise RuntimeError("Too many poll errors")
                time.sleep(_backoff(consecutive_errors))
                continue

            data = resp.json()
            consecutive_errors = 0  # Reset on success

            if data["request"] == "CAPCHA_NOT_READY":
                continue

            if data["request"] in PERMANENT_ERRORS:
                raise RuntimeError(f"Solve error: {data['request']}")

            return data["request"]

        except (requests.ConnectionError, requests.Timeout):
            consecutive_errors += 1
            if consecutive_errors >= max_poll_errors:
                raise RuntimeError("Too many poll connection errors")
            time.sleep(_backoff(consecutive_errors))

    raise TimeoutError(f"Task {task_id} timeout after {timeout}s")

Полный решатель с поддержкой повторных попыток

class RetrySolver:
    """Production-grade solver with comprehensive retry logic."""

    def __init__(self, api_key, max_submit_retries=5, max_poll_retries=3,
                 poll_timeout=120):
        self.api_key = api_key
        self.base = "https://ocr.captchaai.com"
        self.max_submit_retries = max_submit_retries
        self.max_poll_retries = max_poll_retries
        self.poll_timeout = poll_timeout
        self.stats = {
            "total": 0, "success": 0, "retry": 0,
            "permanent_error": 0, "timeout": 0,
        }

    def solve(self, method, **params):
        self.stats["total"] += 1

        # Submit with retry
        task_id = self._submit(method, **params)

        # Poll with retry
        try:
            token = self._poll(task_id)
            self.stats["success"] += 1
            return token
        except TimeoutError:
            self.stats["timeout"] += 1
            raise

    def _submit(self, method, **params):
        data = {"key": self.api_key, "method": method, "json": 1}
        data.update(params)

        for attempt in range(self.max_submit_retries):
            try:
                resp = requests.post(
                    f"{self.base}/in.php", data=data, timeout=30,
                )

                if resp.status_code in (429, 500, 502, 503):
                    self.stats["retry"] += 1
                    time.sleep(_backoff(attempt))
                    continue

                result = resp.json()

                if result.get("request") in PERMANENT_ERRORS:
                    self.stats["permanent_error"] += 1
                    raise RuntimeError(f"Permanent: {result['request']}")

                if result.get("request") in TRANSIENT_ERRORS:
                    self.stats["retry"] += 1
                    time.sleep(_backoff(attempt))
                    continue

                if result.get("status") == 1:
                    return result["request"]

            except (requests.ConnectionError, requests.Timeout):
                self.stats["retry"] += 1
                time.sleep(_backoff(attempt))

        raise RuntimeError("Submit failed after retries")

    def _poll(self, task_id):
        start = time.time()
        errors = 0

        while time.time() - start < self.poll_timeout:
            time.sleep(5)
            try:
                resp = requests.get(f"{self.base}/res.php", params={
                    "key": self.api_key, "action": "get",
                    "id": task_id, "json": 1,
                }, timeout=15)

                if resp.status_code in (429, 500, 502, 503):
                    errors += 1
                    if errors >= self.max_poll_retries:
                        raise RuntimeError("Poll errors exceeded limit")
                    time.sleep(_backoff(errors))
                    continue

                data = resp.json()
                errors = 0

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

            except (requests.ConnectionError, requests.Timeout):
                errors += 1
                if errors >= self.max_poll_retries:
                    raise

        raise TimeoutError("Poll timeout")

    def get_stats(self):
        return self.stats


# Usage
solver = RetrySolver("YOUR_API_KEY")

token = solver.solve(
    "userrecaptcha",
    googlekey="SITE_KEY",
    pageurl="https://example.com",
)

print(solver.get_stats())

Схема автоматического выключателя

Прекратить отправку запросов после слишком большого количества последовательных ошибок:

class CircuitBreaker:
    """Stop requests when the service appears down."""

    def __init__(self, failure_threshold=5, recovery_time=60):
        self.failure_threshold = failure_threshold
        self.recovery_time = recovery_time
        self.failures = 0
        self.last_failure = 0
        self.state = "closed"  # closed=normal, open=blocking

    def can_proceed(self):
        if self.state == "closed":
            return True
        # Check if recovery time has passed
        if time.time() - self.last_failure > self.recovery_time:
            self.state = "half-open"
            return True
        return False

    def record_success(self):
        self.failures = 0
        self.state = "closed"

    def record_failure(self):
        self.failures += 1
        self.last_failure = time.time()
        if self.failures >= self.failure_threshold:
            self.state = "open"
            print(f"Circuit OPEN — pausing for {self.recovery_time}s")


# Integrate with solver
breaker = CircuitBreaker(failure_threshold=5, recovery_time=60)


def solve_with_breaker(method, **params):
    if not breaker.can_proceed():
        raise RuntimeError("Circuit open — API appears unavailable")

    try:
        token = solver.solve(method, **params)
        breaker.record_success()
        return token
    except RuntimeError:
        breaker.record_failure()
        raise

Поиск неисправностей

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

Сколько повторов мне следует использовать?

3-5 попыток отправки, 2-3 ошибки опроса. Более 5 повторных попыток редко помогают — проблема, скорее всего, не временная.

Стоит ли мне повторить ERROR_CAPTCHA_UNSOLVABLE?

Вы можете повторно отправить задачу (новый запрос). Тот же идентификатор задачи не даст другого результата.

Какова оптимальная стратегия отсрочки?

Экспоненциальная задержка (2 секунды попытки) со случайным джиттером 0–50%. Это предотвращает громоподобные проблемы стада, когда многие клиенты повторяют попытку одновременно.

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

• Ограничение скорости и 429 ответов
• Дерево решений обработки ошибок

Создание устойчивой автоматизации —попробуй CaptchaAIс надежностью производственного уровня.