Un déséquilibre en cours de pipeline provoque des échecs. Ce guide montre comment vérifier votre solde CaptchaAI, configurer des alertes de solde faible et suivre automatiquement vos dépenses.
Vérifier le solde via l'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}")
Format de réponse :
{"status": 1, "request": "12.345"}
Vérification du solde avant l'exécution
Vérifiez toujours le solde avant de démarrer un pipeline :
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...")
Moniteur de balance avec alertes
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()
Intégration des alertes 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)
Suivi des dépenses
Suivez vos dépenses quotidiennes, hebdomadaires et mensuelles :
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()
Intégration des pipelines
Intégrez la vérification du solde dans votre flux de travail de résolution :
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}")
Dépannage
| Problème | Parce que | Corriger |
|---|---|---|
| Le solde renvoie 0 | Nouveau compte ou dépensé tous les fonds | Ajoutez des fonds sur captchaai.com |
ERROR_WRONG_USER_KEY |
Clé API invalide | Vérifier la clé à partir du tableau de bord |
| Délai d'expiration de la vérification du solde | Problème de réseau | Ajouter timeout=10 à la demande |
| Le solde ne se met pas à jour | Ancienne valeur mise en cache | Forcer une nouvelle vérification, vider le cache |
FAQ
À quelle fréquence dois-je vérifier mon solde ?
Toutes les 5 à 10 minutes pour les pipelines de production, ou toutes les 50 à 100 résolutions. Évitez de vérifier avant chaque résolution – cela double vos appels API.
Puis-je configurer la recharge automatique ?
L'API ne prend pas en charge le paiement automatique. Utilisez la classe BalanceMonitor pour recevoir des alertes lorsque le solde est faible, puis ajoutez des fonds manuellement ou via votre système de facturation.
La vérification du solde compte-t-elle comme un appel API ?
Les contrôles de solde sont légers et ne sont pas pris en compte dans les limites de taux. Ils ne consomment pas de crédits de résolution.
Guides connexes
- Arbre de décision de gestion des erreurs
- Tableau de bord et surveillance de l'utilisation
Surveillez vos dépenses -commencer par CaptchaAIet suivez chaque résolution.
