API Tutorials

Création d'un tableau de bord et d'une surveillance de l'utilisation de CaptchaAI

Sans instrumentation, une facture CAPTCHA qui dérape se remarque trop tard, au moment du renouvellement mensuel. La parade tient en une ligne : journalisez chaque résolution (méthode, durée, statut) dans un fichier CSV, puis agrégez-le. Ce guide construit ce tableau de bord de suivi en Python, avec un collecteur de métriques, un solveur instrumenté, des rapports d'utilisation et un historique du solde.

Un point utile pour cadrer le suivi : CaptchaAI facture au thread concurrent, pas à la résolution. Une équipe d'automatisation à Lyon qui tourne sur un plan BASIC ($15/mois, 5 threads) ne surveille donc pas un compteur de résolutions qui grimpe, mais la concurrence réelle et le taux de réussite. Si le volume dépasse ce que 5 threads absorbent, la question n'est pas « combien j'ai payé de trop », mais « faut-il passer à ADVANCE ($90/mois, 50 threads) ». Un tableau de bord répond précisément à ça.

Le système se monte en quatre briques, dans cet ordre :

  • un collecteur qui journalise chaque résolution ;
  • un solveur instrumenté qui l'alimente sans effort ;
  • un générateur de rapports quotidiens et hebdomadaires ;
  • un suivi du solde pour anticiper les recharges.

Que suivre dans votre tableau de bord

Métrique Ce qu'elle révèle
Nombre de résolutions Le volume réellement consommé
Taux de réussite Une baisse de qualité ou un paramètre cassé
Temps de réponse Les ralentissements avant qu'ils ne bloquent
Rythme de dépense La dérive budgétaire au jour le jour
Répartition des erreurs Les motifs d'échec à corriger en priorité
Solde Le risque d'interruption quand il approche de zéro
Répartition par méthode Quel type de CAPTCHA pèse dans votre usage

Bonne nouvelle côté RGPD : ce journal ne stocke que des métadonnées techniques (méthode, durée, statut, code d'erreur, identifiant de tâche). Aucune donnée personnelle n'y transite, ce qui simplifie sa conservation et son archivage.


Le collecteur de métriques

Le collecteur écrit chaque tentative dans un CSV et tient des statistiques en mémoire pour la session en cours. Le verrou (threading.Lock) protège l'écriture quand plusieurs threads résolvent en parallèle.

import time
import csv
import datetime
import threading
from collections import defaultdict


class MetricsCollector:
    """Collect and store CaptchaAI solve metrics."""

    def __init__(self, log_file="captchaai_metrics.csv"):
        self.log_file = log_file
        self.lock = threading.Lock()
        self.session_stats = defaultdict(lambda: {
            "count": 0, "success": 0, "error": 0,
            "timeout": 0, "total_time": 0,
        })
        self._init_log()

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

    def record(self, method, duration, status, error_code="", task_id=""):
        """Record a solve attempt."""
        with self.lock:
            # Update in-memory stats
            stats = self.session_stats[method]
            stats["count"] += 1
            stats["total_time"] += duration
            if status == "success":
                stats["success"] += 1
            elif status == "timeout":
                stats["timeout"] += 1
            else:
                stats["error"] += 1

            # Write to CSV
            with open(self.log_file, "a", newline="") as f:
                writer = csv.writer(f)
                writer.writerow([
                    datetime.datetime.utcnow().isoformat(),
                    method, f"{duration:.2f}",
                    status, error_code, task_id,
                ])

    def get_session_summary(self):
        """Get current session statistics."""
        summary = {}
        for method, stats in self.session_stats.items():
            avg_time = (
                stats["total_time"] / stats["count"]
                if stats["count"] > 0 else 0
            )
            success_rate = (
                stats["success"] / stats["count"] * 100
                if stats["count"] > 0 else 0
            )
            summary[method] = {
                "total": stats["count"],
                "success": stats["success"],
                "errors": stats["error"],
                "timeouts": stats["timeout"],
                "success_rate": f"{success_rate:.1f}%",
                "avg_time": f"{avg_time:.1f}s",
            }
        return summary

Chaque ligne du CSV est autonome : vous pouvez l'ouvrir dans un tableur ou la rejouer dans le générateur de rapports plus bas.


Instrumenter votre solveur pour un suivi automatique

L'astuce pour ne jamais rater une mesure : placez l'appel à record() dans un bloc finally. Que la résolution réussisse, expire ou lève une exception, la métrique est écrite. C'est aussi ce qui garde vos statistiques alignées sur la facturation réelle.

import requests
import time


class MonitoredSolver:
    """Solver with automatic metric collection."""

    def __init__(self, api_key, metrics=None):
        self.api_key = api_key
        self.base = "https://ocr.captchaai.com"
        self.metrics = metrics or MetricsCollector()

    def solve(self, method, **params):
        start = time.time()
        task_id = ""
        status = "error"
        error_code = ""

        try:
            # Submit
            data = {"key": self.api_key, "method": method, "json": 1}
            data.update(params)
            resp = requests.post(
                f"{self.base}/in.php", data=data, timeout=30,
            )
            result = resp.json()

            if result.get("status") != 1:
                error_code = result.get("request", "UNKNOWN")
                raise RuntimeError(f"Submit error: {error_code}")

            task_id = result["request"]

            # Poll
            token = self._poll(task_id)
            status = "success"
            return token

        except TimeoutError:
            status = "timeout"
            raise
        except Exception as e:
            error_code = str(e)[:50]
            raise
        finally:
            duration = time.time() - start
            self.metrics.record(method, duration, status, error_code, task_id)

    def _poll(self, task_id, timeout=120):
        start = time.time()
        while time.time() - start < timeout:
            time.sleep(5)
            resp = requests.get(f"{self.base}/res.php", params={
                "key": self.api_key, "action": "get",
                "id": task_id, "json": 1,
            }, timeout=15)
            data = resp.json()
            if data["request"] != "CAPCHA_NOT_READY":
                if data.get("status") == 1:
                    return data["request"]
                raise RuntimeError(f"Solve error: {data['request']}")
        raise TimeoutError("Poll timeout")

    def print_summary(self):
        """Print current session metrics."""
        summary = self.metrics.get_session_summary()
        print("\n=== CaptchaAI Usage Summary ===")
        for method, stats in summary.items():
            print(f"\n{method}:")
            for key, value in stats.items():
                print(f"  {key}: {value}")


# Usage
metrics = MetricsCollector()
solver = MonitoredSolver("YOUR_API_KEY", metrics)

# Solve some CAPTCHAs
for i in range(10):
    try:
        token = solver.solve(
            "userrecaptcha",
            googlekey="SITE_KEY",
            pageurl="https://example.com",
        )
    except Exception as e:
        print(f"Failed: {e}")

# Print results
solver.print_summary()

Couvrir plusieurs types de CAPTCHA

L'exemple utilise userrecaptcha pour reCAPTCHA v2, mais la même enveloppe couvre Cloudflare Turnstile ou GeeTest v3 : seul le paramètre method change, l'instrumentation reste identique.


Générer des rapports d'utilisation quotidiens et hebdomadaires

Trois angles de lecture

Le collecteur produit un flux brut ; le rapport le transforme en trois synthèses lisibles :

  • par jour — repérer un pic ou une panne ponctuelle ;
  • par méthode — voir quel type de CAPTCHA pèse le plus dans la facture ;
  • par erreur — cibler les échecs les plus fréquents.

Ces vues suffisent à repérer une dérive avant qu'elle ne coûte cher.

import csv
import datetime
from collections import defaultdict


class UsageReport:
    """Generate usage reports from metrics CSV."""

    def __init__(self, log_file="captchaai_metrics.csv"):
        self.log_file = log_file

    def _load_data(self, days=None):
        """Load metrics, optionally filtered by date range."""
        cutoff = None
        if days:
            cutoff = datetime.datetime.utcnow() - datetime.timedelta(days=days)

        records = []
        with open(self.log_file, "r") as f:
            reader = csv.DictReader(f)
            for row in reader:
                ts = datetime.datetime.fromisoformat(row["timestamp"])
                if cutoff and ts < cutoff:
                    continue
                row["_ts"] = ts
                row["_duration"] = float(row["duration_s"])
                records.append(row)
        return records

    def daily_summary(self, days=7):
        """Summarize usage per day."""
        records = self._load_data(days=days)
        by_day = defaultdict(lambda: {"count": 0, "success": 0, "total_time": 0})

        for rec in records:
            day = rec["_ts"].date().isoformat()
            by_day[day]["count"] += 1
            if rec["status"] == "success":
                by_day[day]["success"] += 1
            by_day[day]["total_time"] += rec["_duration"]

        print(f"=== Daily Summary (last {days} days) ===")
        print(f"{'Date':<12} {'Total':>6} {'Success':>8} {'Rate':>7} {'Avg Time':>9}")
        for day in sorted(by_day.keys()):
            stats = by_day[day]
            rate = stats["success"] / stats["count"] * 100 if stats["count"] > 0 else 0
            avg = stats["total_time"] / stats["count"] if stats["count"] > 0 else 0
            print(f"{day:<12} {stats['count']:>6} {stats['success']:>8} {rate:>6.1f}% {avg:>8.1f}s")

    def method_breakdown(self, days=30):
        """Summarize usage by CAPTCHA type."""
        records = self._load_data(days=days)
        by_method = defaultdict(lambda: {"count": 0, "success": 0, "total_time": 0})

        for rec in records:
            method = rec["method"]
            by_method[method]["count"] += 1
            if rec["status"] == "success":
                by_method[method]["success"] += 1
            by_method[method]["total_time"] += rec["_duration"]

        print(f"\n=== Method Breakdown (last {days} days) ===")
        print(f"{'Method':<25} {'Total':>6} {'Success':>8} {'Rate':>7} {'Avg Time':>9}")
        for method in sorted(by_method.keys()):
            stats = by_method[method]
            rate = stats["success"] / stats["count"] * 100
            avg = stats["total_time"] / stats["count"]
            print(f"{method:<25} {stats['count']:>6} {stats['success']:>8} {rate:>6.1f}% {avg:>8.1f}s")

    def error_breakdown(self, days=7):
        """Show error distribution."""
        records = self._load_data(days=days)
        errors = defaultdict(int)

        for rec in records:
            if rec["status"] != "success" and rec["error_code"]:
                errors[rec["error_code"]] += 1

        if errors:
            print(f"\n=== Error Breakdown (last {days} days) ===")
            for error, count in sorted(errors.items(), key=lambda x: -x[1]):
                print(f"  {error}: {count}")


# Usage
report = UsageReport()
report.daily_summary(days=7)
report.method_breakdown(days=30)
report.error_breakdown(days=7)

Suivre le solde et le rythme de dépense

Le taux de résolution dit ce que vous consommez ; le solde dit ce qu'il vous reste. En enregistrant le solde à intervalles réguliers, vous reconstituez la courbe de dépense et vous anticipez le moment de recharger, avant l'interruption.

import requests
import time
import csv
import datetime


class BalanceDashboard:
    """Track balance over time for spending analysis."""

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

    def record(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_spending(self, hours=24):
        """Calculate spending over time period."""
        cutoff = datetime.datetime.utcnow() - datetime.timedelta(hours=hours)
        balances = []

        try:
            with open(self.log_file, "r") as f:
                reader = csv.reader(f)
                for row in reader:
                    ts = datetime.datetime.fromisoformat(row[0])
                    if ts > cutoff:
                        balances.append(float(row[1]))
        except FileNotFoundError:
            return 0

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

Deux réflexes pour ne jamais tomber à zéro en pleine campagne :

  • planifiez record() à intervalle régulier via cron ou une fonction serverless (par exemple sur un ordonnanceur Scaleway) ;
  • comparez get_spending(hours=24) à un seuil et déclenchez une alerte dès qu'il est franchi.

Dépannage

Problème Cause Correctif
Le fichier CSV devient énorme Suivi au long cours sans rotation Faites tourner les fichiers chaque jour ou chaque semaine
Des résolutions n'apparaissent pas dans les stats Un solveur non instrumenté Enveloppez tous vos solveurs avec MonitoredSolver
Les chiffres ne collent pas avec la facturation Les échecs ne sont pas journalisés Vérifiez que le bloc finally écrit toujours la métrique
Taux d'erreur anormalement haut Paramètres d'API incorrects Lancez le rapport de répartition des erreurs

Questions fréquentes

Ces métriques m'aident-elles à choisir le bon plan CaptchaAI ?

Oui. Comme la facturation est basée sur les threads concurrents, ce sont le volume et la concurrence de pointe qui guident le choix, pas un compteur de résolutions. Si 5 threads (BASIC, $15/mois) saturent régulièrement, la répartition par méthode et par jour justifie le passage à ADVANCE ($90/mois, 50 threads).

Le journal CSV contient-il des données personnelles au sens du RGPD ?

Non. Chaque ligne ne stocke que la méthode, la durée, le statut, le code d'erreur et l'identifiant de tâche, aucune donnée personnelle. Le fichier reste donc simple à conserver et à archiver sans obligation particulière.

Comment déclencher une alerte quand le solde devient trop bas ?

Enregistrez le solde périodiquement avec BalanceDashboard, puis comparez get_spending() ou le dernier solde à un seuil. Au franchissement, envoyez une notification (e-mail, webhook) ou déclenchez la recharge.

Puis-je pousser ces métriques vers Prometheus ou Grafana ?

Oui. Le MetricsCollector s'étend pour publier vers Prometheus via la bibliothèque prometheus_client, ce qui vous donne des graphes et des alertes dans Grafana par-dessus le même flux CSV.


Guides connexes


Pilotez vos coûts : ouvrez votre compte CaptchaAI et suivez chaque résolution dès la première journée.

Les commentaires sont désactivés pour cet article.