API Tutorials

Проверка баланса CaptchaAI и интеграция автоматического пополнения

У большинства ночных сбоев автоматизации простая причина: баланс CaptchaAI ушёл в ноль, а скрипт узнаёт об этом только тогда, когда res.php вместо токена возвращает пустой ответ посреди прогона. Ниже — рабочая связка: как получить баланс через getbalance, поставить порог для оповещения и вести учёт расходов по потокам, чтобы конвейер не вставал, пока вы спите.


Как получить баланс через res.php

Баланс запрашивается тем же res.php, что и результат решения, — отдельного эндпоинта для этого нет. Передайте action=getbalance вместе с ключом, и API вернёт текущий остаток в долларах:

import requests

API_KEY = "YOUR_API_KEY"

resp = requests.get("https://ocr.captchaai.com/res.php", params={
    "key": API_KEY,
    "action": "getbalance",
    "json": 1,
})

data = resp.json()
balance = float(data["request"])
print(f"Balance: ${balance:.2f}")

Формат ответа:

{"status": 1, "request": "12.345"}

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


Проверка баланса перед запуском конвейера

Если парсер или воркер стартует по расписанию — например, ночной прогон перед началом рабочего дня в GMT+3–GMT+6, — разумно проверить баланс до первой задачи, а не после первого же провала. Функция ниже прерывает запуск, если остаток ниже минимально допустимого:

import requests
import sys


def check_balance(api_key, min_required=1.0):
    """Check balance and abort if too low."""
    resp = requests.get("https://ocr.captchaai.com/res.php", params={
        "key": api_key,
        "action": "getbalance",
        "json": 1,
    })
    data = resp.json()

    if data.get("status") != 1:
        print(f"Balance check failed: {data.get('request')}")
        return None

    balance = float(data["request"])
    print(f"Current balance: ${balance:.2f}")

    if balance < min_required:
        print(f"WARNING: Balance ${balance:.2f} below minimum ${min_required:.2f}")
        return None

    return balance


# Usage
API_KEY = "YOUR_API_KEY"
balance = check_balance(API_KEY, min_required=5.0)

if balance is None:
    print("Insufficient balance. Add funds before running pipeline.")
    sys.exit(1)

print(f"Balance OK (${balance:.2f}). Starting pipeline...")

min_required стоит выставлять с запасом на один полный прогон, а не впритык к нулю — так вы узнаёте о проблеме до, а не во время сбора данных.


Класс мониторинга баланса с оповещениями

Разовой проверки перед запуском достаточно для коротких job'ов, но для долгоживущего воркера нужен постоянный мониторинг с порогом и историей расходов. Класс BalanceMonitor опрашивает баланс с заданным интервалом, шлёт оповещение один раз при пересечении порога (а не на каждой итерации) и умеет прикинуть, сколько часов пайплайн ещё продержится при текущей скорости расхода:

import requests
import time
import smtplib
from email.message import EmailMessage


class BalanceMonitor:
    """Monitor CaptchaAI balance and send alerts."""

    def __init__(self, api_key, alert_threshold=5.0, check_interval=300):
        self.api_key = api_key
        self.alert_threshold = alert_threshold
        self.check_interval = check_interval  # seconds
        self.base_url = "https://ocr.captchaai.com"
        self.history = []
        self.alerted = False

    def get_balance(self):
        resp = requests.get(f"{self.base_url}/res.php", params={
            "key": self.api_key,
            "action": "getbalance",
            "json": 1,
        }, timeout=10)
        data = resp.json()
        return float(data["request"])

    def check_and_alert(self):
        balance = self.get_balance()
        self.history.append({
            "time": time.time(),
            "balance": balance,
        })

        print(f"Balance: ${balance:.2f}")

        if balance < self.alert_threshold and not self.alerted:
            self.send_alert(balance)
            self.alerted = True
        elif balance >= self.alert_threshold:
            self.alerted = False

        return balance

    def send_alert(self, balance):
        """Send low-balance alert. Override for your notification system."""
        print(f"ALERT: Balance low! ${balance:.2f} < ${self.alert_threshold:.2f}")
        # Add your notification logic:
        # - Email, Slack webhook, SMS, etc.

    def get_spending_rate(self, hours=1):
        """Calculate spending rate over the last N hours."""
        cutoff = time.time() - (hours * 3600)
        recent = [h for h in self.history if h["time"] > cutoff]

        if len(recent) < 2:
            return 0.0

        spent = recent[0]["balance"] - recent[-1]["balance"]
        return max(0.0, spent)

    def estimate_remaining_hours(self):
        """Estimate how many hours until balance runs out."""
        rate = self.get_spending_rate(hours=1)
        if rate <= 0:
            return float("inf")

        balance = self.history[-1]["balance"] if self.history else 0
        return balance / rate

    def run(self):
        """Run continuous monitoring."""
        print(f"Monitoring balance (alert at ${self.alert_threshold:.2f})")
        while True:
            try:
                self.check_and_alert()
                rate = self.get_spending_rate()
                remaining = self.estimate_remaining_hours()
                print(f"  Spending: ${rate:.2f}/hr, ~{remaining:.1f}hrs remaining")
            except Exception as e:
                print(f"Monitor error: {e}")
            time.sleep(self.check_interval)


# Usage
monitor = BalanceMonitor(
    api_key="YOUR_API_KEY",
    alert_threshold=5.0,
    check_interval=300,  # Check every 5 minutes
)
monitor.run()

alerted — простой флаг, чтобы не слать одно и то же сообщение на каждой итерации, пока баланс остаётся низким: оповещение уходит один раз при пересечении порога и снова становится активным только после пополнения. estimate_remaining_hours() полезен командам, которые ведут ночные пакетные задачи без дежурного — по нему видно, доживёт ли баланс до утра.


Оповещения о низком балансе в Slack

Email легко пропустить, если инженер не проверяет почту каждые пять минут. Для операционных команд, у которых уже есть рабочий канал в Slack, проще завести туда отдельный вебхук под алерты CaptchaAI:

import requests


def send_slack_alert(webhook_url, balance, threshold):
    """Send balance alert to Slack channel."""
    payload = {
        "text": f":warning: CaptchaAI balance low!",
        "blocks": [
            {
                "type": "section",
                "text": {
                    "type": "mrkdwn",
                    "text": (
                        f"*CaptchaAI Balance Alert*\n"
                        f"Current balance: *${balance:.2f}*\n"
                        f"Alert threshold: ${threshold:.2f}\n"
                        f"Action: Add funds at captchaai.com"
                    ),
                },
            },
        ],
    }
    requests.post(webhook_url, json=payload)


# Add to BalanceMonitor.send_alert():
# send_slack_alert(SLACK_WEBHOOK, balance, self.alert_threshold)

URL вебхука — секрет того же уровня, что и API-ключ: храните его в переменных окружения, а не в коде, и не выводите в общий лог.


Учёт расходов по дням, неделям и месяцам

Мониторинг в реальном времени отвечает на вопрос «баланс низкий прямо сейчас?», а для планирования нужна история: сколько уходит в день и когда покупать следующий пакет потоков. SpendingTracker пишет каждое измерение баланса в CSV и считает разницу за сутки:

import csv
import datetime


class SpendingTracker:
    """Track CaptchaAI spending over time."""

    def __init__(self, api_key, log_file="captchaai_spending.csv"):
        self.api_key = api_key
        self.log_file = log_file
        self._init_log()

    def _init_log(self):
        try:
            with open(self.log_file, "r") as f:
                pass
        except FileNotFoundError:
            with open(self.log_file, "w", newline="") as f:
                writer = csv.writer(f)
                writer.writerow(["timestamp", "balance"])

    def record_balance(self):
        resp = requests.get("https://ocr.captchaai.com/res.php", params={
            "key": self.api_key,
            "action": "getbalance",
            "json": 1,
        })
        balance = float(resp.json()["request"])

        with open(self.log_file, "a", newline="") as f:
            writer = csv.writer(f)
            writer.writerow([
                datetime.datetime.utcnow().isoformat(),
                f"{balance:.4f}",
            ])
        return balance

    def get_daily_spending(self):
        """Calculate today's spending from log."""
        today = datetime.date.today().isoformat()
        balances = []

        with open(self.log_file, "r") as f:
            reader = csv.DictReader(f)
            for row in reader:
                if row["timestamp"].startswith(today):
                    balances.append(float(row["balance"]))

        if len(balances) < 2:
            return 0.0
        return balances[0] - balances[-1]

    def summary(self):
        """Print spending summary."""
        balance = self.record_balance()
        daily = self.get_daily_spending()
        print(f"Current balance: ${balance:.2f}")
        print(f"Spent today: ${daily:.2f}")
        if daily > 0:
            print(f"Daily rate: ${daily:.2f}/day")
            print(f"Days remaining: {balance / daily:.1f}")


# Usage
tracker = SpendingTracker("YOUR_API_KEY")
tracker.summary()

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


Проверка баланса внутри воркера решения

Проверять баланс перед каждым отдельным решением избыточно — это удваивает число запросов к API без реальной пользы. BalanceAwareSolver проверяет баланс раз в 50 решений или раз в 5 минут (что наступит раньше) и останавливает воркер исключением, если средств не хватает на следующую задачу:

import requests
import time


class BalanceAwareSolver:
    """Solver that checks balance before solving."""

    def __init__(self, api_key, min_balance=1.0):
        self.api_key = api_key
        self.base_url = "https://ocr.captchaai.com"
        self.min_balance = min_balance
        self.last_balance_check = 0
        self.cached_balance = None
        self.solves_since_check = 0

    def solve(self, method, **params):
        """Solve with balance pre-check."""
        # Check balance every 50 solves or every 5 minutes
        if self._should_check_balance():
            balance = self._get_balance()
            if balance < self.min_balance:
                raise RuntimeError(
                    f"Balance too low: ${balance:.2f} "
                    f"(minimum: ${self.min_balance:.2f})"
                )

        return self._do_solve(method, **params)

    def _should_check_balance(self):
        elapsed = time.time() - self.last_balance_check
        return elapsed > 300 or self.solves_since_check >= 50

    def _get_balance(self):
        resp = requests.get(f"{self.base_url}/res.php", params={
            "key": self.api_key,
            "action": "getbalance",
            "json": 1,
        })
        self.cached_balance = float(resp.json()["request"])
        self.last_balance_check = time.time()
        self.solves_since_check = 0
        return self.cached_balance

    def _do_solve(self, method, **params):
        data = {"key": self.api_key, "method": method, "json": 1}
        data.update(params)
        resp = requests.post(f"{self.base_url}/in.php", data=data)
        task_id = resp.json()["request"]

        for _ in range(60):
            time.sleep(5)
            result = requests.get(f"{self.base_url}/res.php", params={
                "key": self.api_key, "action": "get",
                "id": task_id, "json": 1,
            })
            data = result.json()
            if data["request"] != "CAPCHA_NOT_READY":
                self.solves_since_check += 1
                return data["request"]

        raise TimeoutError("Solve timeout")


# Usage
solver = BalanceAwareSolver("YOUR_API_KEY", min_balance=2.0)

try:
    token = solver.solve("userrecaptcha", googlekey="KEY", pageurl="https://example.com")
except RuntimeError as e:
    print(f"Balance issue: {e}")

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


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

Проблема Причина Что делать
Баланс возвращается 0 Новый аккаунт или средства полностью потрачены Пополните баланс на captchaai.com
ERROR_WRONG_USER_KEY Неверный или отозванный API-ключ Сверьте ключ с панелью управления
Тайм-аут при проверке баланса Проблема с сетью или прокси Добавьте timeout=10 в запрос
Баланс не меняется после пополнения Показано кэшированное значение Сделайте новый запрос напрямую, минуя кэш

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

Как часто стоит проверять баланс, не перегружая API?

Для продакшн-конвейеров — раз в 5–10 минут или раз в 50–100 решений, как в примере с BalanceAwareSolver. Проверка перед каждым отдельным решением не нужна: она просто удваивает число запросов, не добавляя пользы.

В API есть автоматическое пополнение баланса?

Нет, res.php не поддерживает автоплатежи. Рабочая схема — BalanceMonitor присылает оповещение о низком балансе, а пополнение вы делаете вручную через личный кабинет на captchaai.com или через свою систему биллинга.

Что делать, если баланс обнулился прямо во время прогона парсера?

in.php начнёт возвращать ошибку по новым задачам, а уже отправленные в очередь могут не дойти до res.php. Ставьте min_balance в BalanceAwareSolver с запасом на текущий батч, чтобы воркер останавливался заранее, а не посреди партии.

Можно ли посмотреть баланс без кода, через личный кабинет?

Да, текущий баланс виден в панели управления captchaai.com. Код из этого руководства нужен, когда баланс должен проверяться и логироваться автоматически, без захода в интерфейс.

Как понять, что пора переходить на тарифный план с большим числом потоков?

Смотрите на дневной расход из SpendingTracker.summary() и на то, сколько потоков реально заняты одновременно в пиковые часы. Если очередь задач регулярно упирается в лимит потоков текущего плана, а не в баланс, — это сигнал повышать план, а не просто пополнять счёт.


Похожие материалы


Держите расходы под контролем — начните с CaptchaAI и настройте оповещения раньше, чем закончатся потоки.

Комментарии для этой статьи отключены.