这个月账单突然涨了不少，却说不清是哪种验证码类型、哪个时段用量暴增——说明你还没给 CaptchaAI 接入用量监控。本指南用 Python 搭建一套轻量监控系统：记录成功率、响应耗时和余额消耗，自动生成日报和周报。
不需要 Prometheus 或 Grafana，一个 CSV 文件加几个 Python 类就够用。
先明确要监控哪些指标
先跟踪这几个指标：
七个核心指标
- 解决次数：掌握真实用量
- 成功率：及时发现识别质量下滑
- 响应时间：定位变慢的环节
- 消耗速度：提前控制预算
- 错误分布：定位具体失败模式
- 余额：避免任务中途因欠费中断
- 方法分布：了解各验证码类型的使用占比
第一步：搭建指标采集器
MetricsCollector 只做两件事：把每次识别结果汇总进内存字典，同时逐条写入本地 CSV。加了一把线程锁，多线程并发调用时不会互相覆盖计数。
线程锁为什么必须加
不加锁，并发写会导致计数错乱、CSV 行错位。threading.Lock() 代价很小，换来数字可信。
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
get_session_summary 只是当次会话的内存统计，重启即清零；跨天数据靠的是同一份 CSV，下一步的报表类会直接读它。
第二步：让求解器自动上报指标
不想在每处业务代码里手动打点，就用 MonitoredSolver 把求解逻辑包一层：无论成功、超时还是报错，finally 块都会把耗时和错误码写进 MetricsCollector。
finally 兜底的意义
try/except 只覆盖预判到的失败路径，finally 不管走哪条分支都执行，耗时状态照样落盘。
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()
第三步：把 CSV 明细变成可读报表
有了逐条明细，UsageReport 按三个维度切片：按天看趋势、按类型看占比、按错误码看故障集中在哪，不用另外接 BI 工具。
三个切片各自解决什么问题
daily_summary 盯当天波动；method_breakdown 看哪种类型占大头，方便估算账单；error_breakdown 直接定位故障集中在哪个错误码。
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)
第四步：跟踪余额，提前预警
BalanceDashboard 定期调用 getbalance 记录余额，两次采样相减就是这段时间的实际消耗速度。接进定时任务后，余额异常下降能在耗尽前被发现，而不是等任务批量失败才排查。
怎么用这段代码报警
单独调用 record() 没什么用，要配合定时任务定期跑，再用 get_spending 算区间消耗，超阈值就通知。
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]
把 get_spending 的结果接一个 webhook，推到企业微信或飞书群机器人，异常时直接在群里提醒，比人工刷仪表板可靠。
常见故障排查
先按下面的思路排查，不用一上来就怀疑是 CaptchaAI 出了问题：
快速自查清单
- CSV 文件越来越大 —— 后台长期运行，按天或按周轮换日志文件
- 解决记录缺失 —— 有调用没接入 MonitoredSolver，统一用它包装所有求解调用
- 统计数字和账单对不上 —— 部分错误没被记录，确认
finally块每次都执行了写入 - 仪表板错误率偏高 —— API 参数传错了，查看错误分布报表定位错误码
常见问题
监控数据要留多久？
明细数据留 30 天，用来排查近期问题；汇总数据留 90 天看长期趋势，更早的建议归档。
能不能把余额异常接到企业微信或飞书机器人？
可以，get_spending 返回具体数值，超过阈值时调用对应机器人的 webhook 接口推送消息即可。
监控脚本会不会拖慢识别速度？
不会，每次写入 CSV 的开销在 1 毫秒以内，相比一次识别请求本身可以忽略不计。
生产环境有必要跑这套监控吗？
有必要，记录成本几乎为零，却能在账单异常或成功率下滑造成实际影响之前提前发现。
多进程同时跑监控，会不会把 CSV 写坏？
threading.Lock() 只在同一进程内有效。多进程各跑一份 MonitoredSolver 时，给每个进程用独立日志文件名，报表阶段再合并。
相关指南
延伸阅读
先把这些数字摸清楚，账单异常才不会等到月底才发现。
