Если доля успешных решений и расход потоков не видны в моменте, о проблеме вы узнаёте только из счёта в конце месяца — когда её уже поздно чинить. Ниже — рабочая схема на Python, которая закрывает три задачи разом:
- логирование каждого решения в CSV через потокобезопасный сборщик;
- обёртка над решателем, которая пишет метрики автоматически, без ручных вызовов;
- ежедневные/недельные отчёты и отслеживание баланса, чтобы не упереться в ноль потоков.
Никаких внешних сервисов не требуется — только csv и threading из стандартной библиотеки.
Какие метрики снимать и зачем
Прежде чем писать код, определитесь, что именно попадёт в лог. На практике достаточно семи полей:
- количество решений — видеть реальный объём нагрузки на API;
- доля успеха — ловить деградацию качества по конкретному типу CAPTCHA;
- время ответа — замечать замедления до того, как они сорвут SLA у вас в пайплайне;
- скорость расхода — планировать бюджет и не уйти в овердрафт;
- распределение ошибок — быстро находить, какой параметр API отправляется неверно;
- баланс — не допустить простоя из-за нулевого баланса;
- разбивка по методам — понимать, какие типы CAPTCHA реально нагружают тариф.
Пример из практики: команда на тарифе ADVANCE ($90/мес, 50 потоков) парсит листинги и держит в среднем 30–35 активных потоков reCAPTCHA v2. Без метрик рост доли ошибок с 2% до 11% заметили только когда конверсия парсера просела — с логированием той же гранулярности это видно в отчёте за первый час. Если потоков стабильно не хватает, следующий шаг вверх по сетке — PREMIUM ($170/мес, 100 потоков) или CORPORATE ($240/мес, 150 потоков); актуальные тарифы всегда сверяйте на странице captchaai.com/pricing.
Класс для сбора метрик
MetricsCollector держит статистику сессии в памяти и параллельно дописывает каждую запись в CSV — так отчёты переживают перезапуск процесса, а сводка за сессию доступна без чтения файла с диска.
Из чего он состоит
Ниже — минимальный набор, которого хватает большинству пайплайнов:
- потокобезопасная запись через
threading.Lock; - накопление статистики в памяти по каждому методу;
- параллельная построчная дозапись в CSV на каждый вызов
record().
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
lock защищает от гонки между потоками, которые пишут в CSV одновременно, — при 30+ параллельных потоках это не редкость, а норма.
Обёртка над решателем: метрики без ручного логирования
Чтобы не расставлять metrics.record(...) в каждом месте, где вызывается решатель, оберните сам вызов. MonitoredSolver замеряет время в finally, поэтому запись попадает в лог при любом исходе — успех, таймаут или ошибка API.
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()
Дальше MonitoredSolver подставляется на место обычного вызова API — остальной код пайплайна не меняется.
Отчёты по дням и по методам
CSV-лог сам по себе бесполезен, пока по нему не построены агрегаты. UsageReport читает файл и группирует записи по дате и по методу решения — этого достаточно, чтобы за секунды понять, просел ли reCAPTCHA v3 именно сегодня или проблема системная.
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)
Запускайте daily_summary в начале смены, а error_breakdown — сразу после того, как алерт сообщил о росте ошибок: обычно этого достаточно, чтобы указать на конкретный параметр (sitekey, pageurl) или конкретный тип CAPTCHA.
Баланс и расход потоков во времени
Метрики решений отвечают на вопрос «что происходит», а история баланса — на вопрос «сколько до нуля». BalanceDashboard опрашивает res.php с action=getbalance и пишет снимки в отдельный CSV, откуда легко посчитать расход за последние сутки.
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]
Свяжите record() с планировщиком (cron, APScheduler — на ваш выбор) и снимайте баланс раз в 15–30 минут: этого достаточно, чтобы отправить алерт задолго до того, как поток встанет из-за нулевого баланса.
Частые проблемы и как их решать
Четыре симптома встречаются чаще всего:
- CSV разрастается до неприличных размеров. Причина — мониторинг работает месяцами без ротации. Ротируйте файл ежедневно или еженедельно; детальные записи храните 30 дней, агрегаты по дням — 90 дней, а дальше имеет смысл оставлять только сжатые сводки.
- В логе не хватает части решений. Обычно решатель вызывается напрямую, в обход
MonitoredSolver. Оберните все точки вызова решателя, а не только основную. - Статистика расходится с биллингом. Как правило, блок
finallyне срабатывает при определённых исключениях — проверьте, что он действительно выполняется на каждом пути. - В отчёте резко выросла доля ошибок. Чаще всего дело в неверных параметрах API (
sitekey,pageurl,googlekey) — смотритеerror_breakdown, там виден конкретныйerror_code.
Часто задаваемые вопросы
Можно ли слать алерт о балансе в Slack или Telegram, а не просто печатать его в консоль?
Да — record() в BalanceDashboard ничего не знает о канале уведомления, он только сохраняет снимок баланса. Добавьте в вызывающий код сравнение текущего значения с порогом и отправку сообщения через webhook Slack или Telegram Bot API, если баланс ниже порога; сам сборщик менять для этого не нужно.
Можно ли завести отдельный дашборд под каждый тип CAPTCHA?
Да — method_breakdown уже группирует статистику по методу (userrecaptcha, turnstile, geetest и так далее), поэтому отдельная панель на тип строится из тех же данных без изменения сборщика.
Как выставить порог для алерта по доле ошибок?
Единого правильного числа нет: возьмите средний success_rate за последние 7 дней спокойной работы, вычтите 5–10 п.п. и используйте это как порог для оповещения — так алерт сработает раньше, чем проблема дойдёт до продакшна.
Можно ли отправлять эти метрики в Prometheus или Grafana?
Да. MetricsCollector легко расширяется библиотекой prometheus_client — вместо (или вместе с) записью в CSV достаточно инкрементировать Counter/Histogram внутри того же метода record().
Нужно ли включать мониторинг в продакшне?
Да. Накладные расходы на запись строки в CSV — доли миллисекунды на решение, а видимость, которую это даёт, окупается уже на первом же инциденте с ростом ошибок.
Что читать дальше
Считайте свои цифры — начните мониторинг CaptchaAI уже сегодня.
