批量任务跑到一半突然全部失败，排查半天，原因往往只是余额见底了。更靠谱的做法是把余额检查写进代码：跑批前查一次，跑的过程中常驻监控，再加一层消费记录。本文给出可以直接复制的 Python 实现。
查询余额：一次 GET 请求就够了
CaptchaAI 的 getbalance 接口不需要额外鉴权步骤，带上 API Key 发一个 GET 请求即可：
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"}
status 为 1 表示查询成功，余额金额在 request 字段里，单位始终是美元。接口本身很轻量，不占用求解额度，也不计入速率限制。
跑批前先做一次余额预检查
批量任务开跑前先确认余额够不够，比跑到一半失败再回滚划算：
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 建议按单次任务的实际消耗来设，不要随手写个数字——一批任务大概花 $2，阈值设成 $1 基本等于没设。
写一个常驻的余额监控器
预检查只能挡住“开局就没钱”，长时间跑的任务还需要持续盯着余额变化，跌破阈值就报警：
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 标志位是关键：没有它，余额持续低于阈值时 run() 每轮都会重复报警。加上这个开关后，只有“跌破阈值”的瞬间触发一次，回升后重置。get_spending_rate 和 estimate_remaining_hours 把历史记录换算成“还能撑多久”。
接入 Slack 通知
send_alert() 默认只打印日志，接上真实通知渠道才有用，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)
国内团队更常把告警接在钉钉或企业微信群机器人上，而不是 Slack。改法很简单：把 payload 换成对方要求的 JSON 格式（通常是 msgtype + text/markdown 字段），POST 到机器人 Webhook 地址，调用位置仍挂在 BalanceMonitor.send_alert() 里，其余逻辑不用改。
记录每日、每周、每月的消费
只看当前余额，说明不了“钱是怎么花掉的”。把每次查询的结果落盘成日志，就能按天核算消费：
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()
把这段代码接进定时任务（例如每小时跑一次 record_balance()），几天后 captchaai_spending.csv 就能看出真实的消费曲线，比事后翻账单直观，也方便按周/按月汇总。
把余额检查嵌进求解流程
前面几段都是“外挂式”的监控，下面这个例子把余额检查直接写进求解逻辑本身，任务发起前自动确认额度够用：
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}")
_should_check_balance 用“5 分钟或 50 次求解”两个条件做缓存，避免每次 solve() 都发一次网络请求。余额不够时直接抛 RuntimeError，上层代码可按需捕获，决定暂停、重试还是终止。
故障排查
| 问题 | 可能原因 | 处理方式 |
|---|---|---|
| 余额返回 0 | 新账户尚未充值，或额度已用完 | 登录 captchaai.com 添加资金 |
ERROR_WRONG_USER_KEY |
API Key 填错或已失效 | 到控制台重新核对 Key |
| 余额查询超时 | 网络波动 | 请求里加上 timeout=10 |
| 余额数字没变化 | 读到了缓存值 | 强制发起一次新请求，跳过本地缓存 |
仍未解决时，确认调用的确实是 getbalance，再看 status 字段——不为 1 时 request 里通常已给出原因。
常见问题
生产环境里，余额应该多久查一次？
每 5-10 分钟或每 50-100 次求解查一次就够，参考 _should_check_balance 的判断逻辑。不用每次求解前都查——查询不消耗解决额度，但过于频繁没有意义。
CaptchaAI 支持余额低于阈值时自动充值吗？
目前 API 不提供自动扣款功能。BalanceMonitor 只负责在余额跌破阈值时发通知（日志、Slack/钉钉消息等），实际充值仍需登录 captchaai.com 手动操作，或对接自己的计费系统。
余额突然变成 0 或读数异常，应该先查什么？
先登录后台核对额度是否确实用完，再检查 API Key 有没有填错——ERROR_WRONG_USER_KEY 是最常见的误报原因。数字忽高忽低多半是缓存值，强制刷新一次即可。
多个 worker 并发跑任务，要不要每个进程都单独查一次余额？
不需要。把检查放在共享的 BalanceMonitor 进程里，所有 worker 读同一份缓存结果；各自高频查询只会抢占限流配额，还会让告警重复触发。
相关指南
别等余额跑空才发现——从 CaptchaAI 开始，把每一次消耗都记下来。
