CaptchaAI余额检查和自动充值集成

管道中途失去平衡会导致故障。本指南介绍如何检查您的 CaptchaAI 余额、设置低余额警报以及自动跟踪支出。

通过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"}

运行前平衡检查

在启动管道之前始终验证平衡：

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...")

带警报的平衡监视器

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()

Slack 警报集成

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)

支出追踪器

跟踪每日、每周和每月的支出：

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()

管道整合

将平衡检查集成到您的求解工作流程中：

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}")

故障排除

常问问题

我应该多久检查一次余额？

生产管道每 5-10 分钟一次，或每 50-100 次解决一次。避免在每次求解之前进行检查，这会使 API 调用量增加一倍。

可以设置自动充值吗？

该API不支持自动付款。使用 BalanceMonitor 类在余额较低时获取警报，然后手动或通过计费系统添加资金。

检查余额是否算作 API 调用？

余额检查是轻量级的，并且不计入速率限制。他们不消耗解决积分。

相关指南

• 错误处理决策树
• 使用情况仪表板和监控

监控你的支出——以 CaptchaAI 开头并跟踪每个解决方案。