CaptchaAI prend en charge des volumes de requêtes élevés, mais comprendre les limites de capacité et mettre en œuvre une limitation appropriée garantit la fiabilité de votre pipeline.
Limites de débit de l'API
CaptchaAI n'applique pas de limites strictes de débit par seconde. Au lieu de cela, la capacité est gérée par la disponibilité des travailleurs :
| Facteur | Comportement |
|---|---|
| Taux de soumission | Aucune limite stricte par seconde |
| Tâches simultanées | Gère plus de 100 par compte |
| Fréquence des sondages | Recommandé : toutes les 5 secondes par tâche |
| Vérification du solde | Aucune limite |
La principale contrainte est la capacité des travailleurs. Lorsque tous les travailleurs sont occupés, vous obtenez ERROR_NO_SLOT_AVAILABLE.
Gestion de ERROR_NO_SLOT_AVAILABLE
Cette erreur signifie que le système est à pleine capacité. Implémentez un recul exponentiel :
Python
import time
import requests
API_KEY = "YOUR_API_KEY"
def submit_with_backoff(params, max_retries=5):
params["key"] = API_KEY
for attempt in range(max_retries):
resp = requests.get(
"https://ocr.captchaai.com/in.php", params=params
)
if resp.text.startswith("OK|"):
return resp.text.split("|")[1]
if resp.text == "ERROR_NO_SLOT_AVAILABLE":
wait = min(2 ** attempt * 2, 60) # 2, 4, 8, 16, 32s max 60
print(f"No slots, waiting {wait}s (attempt {attempt + 1})")
time.sleep(wait)
continue
raise Exception(f"Submit error: {resp.text}")
raise Exception("Max retries exceeded — no slots available")
Noeud.js
async function submitWithBackoff(params, maxRetries = 5) {
params.key = process.env.CAPTCHAAI_API_KEY;
for (let attempt = 0; attempt < maxRetries; attempt++) {
const resp = await axios.get("https://ocr.captchaai.com/in.php", {
params,
});
const text = String(resp.data);
if (text.startsWith("OK|")) {
return text.split("|")[1];
}
if (text === "ERROR_NO_SLOT_AVAILABLE") {
const wait = Math.min(2 ** attempt * 2000, 60000);
console.log(`No slots, waiting ${wait}ms (attempt ${attempt + 1})`);
await new Promise((r) => setTimeout(r, wait));
continue;
}
throw new Error(`Submit error: ${text}`);
}
throw new Error("Max retries exceeded");
}
Limitation de débit côté client
Compartiment de jetons (Python)
import time
import threading
class RateLimiter:
def __init__(self, rate, per=1.0):
"""Allow `rate` requests per `per` seconds."""
self.rate = rate
self.per = per
self.tokens = rate
self.last_refill = time.monotonic()
self.lock = threading.Lock()
def acquire(self):
with self.lock:
now = time.monotonic()
elapsed = now - self.last_refill
self.tokens = min(self.rate, self.tokens + elapsed * (self.rate / self.per))
self.last_refill = now
if self.tokens >= 1:
self.tokens -= 1
return
else:
sleep_time = (1 - self.tokens) * (self.per / self.rate)
time.sleep(sleep_time)
self.acquire()
# Allow 10 submissions per second
limiter = RateLimiter(rate=10, per=1.0)
def submit_limited(params):
limiter.acquire()
return submit_with_backoff(params)
Sémaphore asyncio
import asyncio
# Limit to 20 concurrent tasks
semaphore = asyncio.Semaphore(20)
async def solve_limited(solver, session, params):
async with semaphore:
return await solver.solve(session, params)
Contrôle du taux de sondage
N'interrogez pas plus fréquemment que toutes les 5 secondes par tâche :
async def smart_poll(session, task_id, solver):
"""Polls with adaptive intervals."""
intervals = [5, 5, 5, 10, 10, 15, 15, 30, 30, 60]
for wait in intervals:
await asyncio.sleep(wait)
result = await solver.check(session, task_id)
if result is not None:
return result
raise TimeoutError(f"Task {task_id} timed out")
Surveillance et mesures
Suivez votre utilisation de l'API pour rester dans les limites :
import time
from collections import deque
class APIMetrics:
def __init__(self, window=60):
self.window = window
self.requests = deque()
self.errors = deque()
def record_request(self):
now = time.time()
self.requests.append(now)
self._cleanup(self.requests, now)
def record_error(self, error_code):
now = time.time()
self.errors.append((now, error_code))
self._cleanup_tuples(self.errors, now)
def get_rate(self):
now = time.time()
self._cleanup(self.requests, now)
return len(self.requests) / self.window
def get_error_rate(self):
now = time.time()
self._cleanup(self.requests, now)
self._cleanup_tuples(self.errors, now)
if not self.requests:
return 0
return len(self.errors) / len(self.requests)
def _cleanup(self, dq, now):
while dq and dq[0] < now - self.window:
dq.popleft()
def _cleanup_tuples(self, dq, now):
while dq and dq[0][0] < now - self.window:
dq.popleft()
metrics = APIMetrics()
# Use in your submit function
def submit_tracked(params):
metrics.record_request()
try:
return submit_with_backoff(params)
except Exception as e:
metrics.record_error(str(e))
raise
# Check metrics periodically
print(f"Rate: {metrics.get_rate():.1f} req/s")
print(f"Error rate: {metrics.get_error_rate():.1%}")
Planification des capacités
| Volume | Concurrence | Stratégie |
|---|---|---|
| < 100/hour | 1-5 | Séquentiel, aucun contrôle de débit nécessaire |
| 100-1K/hour | 5-20 | Concurrence basée sur les sémaphores |
| 1K-10K/hour | 20-50 | Asynchrone avec file d'attente, rappels |
| 10K+/hour | 50-100 | Pool de travailleurs, capacité dédiée |
Pour les volumes supérieurs à 10 K/hour, contactez le support CaptchaAI pour connaître la capacité dédiée.
FAQ
Y a-t-il une limite quotidienne ?
Aucune limite quotidienne stricte. Votre limite est déterminée par votre solde et la capacité de travail de l'API.
Que se passe-t-il si je dépasse la capacité ?
Vous recevrez ERROR_NO_SLOT_AVAILABLE. C'est temporaire : réessayez après une période d'attente.
Dois-je limiter le côté client ?
Oui, pour un comportement prévisible. Utilisez un sémaphore pour limiter les tâches simultanées et un compartiment de jetons pour contrôler le taux de soumission.
