CAPTCHA 操作的结构化日志记录

排查验证码识别问题，日志只需回答三件事：任务 ID、耗时多少毫秒、失败时的错误码。写成 JSON 并固定字段名，一条 jq 就能查；写成 Error solving captcha 这种纯文本只能靠 grep 猜。下面用 structlog 和 pino 记录整条链路。

纯文本日志和 JSON 日志的差别

采集任务凌晨跑批、早上发现成功率下滑，纯文本日志只能确认“有报错”；结构化日志按 captcha_type 一分组，就知道是 reCAPTCHA v2 变慢还是 sitekey 配错。

先定字段表，再动代码

字段名先定下来，后面接 ELK 或 Loki 都不改采集端：

三条纪律：字段名小写加下划线；时间统一毫秒整数；API Key 绝不进日志。国内做采集再加一条——按《网络安全法》《数据安全法》和 PIPL，日志只记元数据。

Python：用 structlog 输出 JSON

structlog 配置一次全局生效，bind() 的上下文自动带进后续日志。

import structlog
import time

structlog.configure(
    processors=[
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.add_log_level,
        structlog.processors.JSONRenderer(),
    ],
    logger_factory=structlog.PrintLoggerFactory(),
)

log = structlog.get_logger()

记录识别任务的完整生命周期

一次识别拆成三个节点：提交开始、提交成功、最终成功或失败。拿到任务 ID 后重新 bind 是关联查询的前提。

import requests

API_KEY = "YOUR_API_KEY"


def solve_captcha(captcha_type, sitekey, page_url, proxy=None):
    solve_log = log.bind(
        captcha_type=captcha_type,
        site_url=page_url,
        sitekey=sitekey[:12] + "...",
    )

    # Submit
    start = time.time()
    solve_log.info("captcha_submit_start")

    resp = requests.post("https://ocr.captchaai.com/in.php", data={
        "key": API_KEY,
        "method": "userrecaptcha",
        "googlekey": sitekey,
        "pageurl": page_url,
        "json": "1",
    }).json()

    if resp["status"] != 1:
        solve_log.error("captcha_submit_failed", error=resp["request"])
        return None

    task_id = resp["request"]
    submit_ms = int((time.time() - start) * 1000)
    solve_log = solve_log.bind(task_id=task_id)
    solve_log.info("captcha_submitted", submit_ms=submit_ms)

    # Poll
    for attempt in range(24):
        time.sleep(5)
        result = requests.get("https://ocr.captchaai.com/res.php", params={
            "key": API_KEY, "action": "get", "id": task_id, "json": "1"
        }).json()

        if result["status"] == 1:
            solve_ms = int((time.time() - start) * 1000)
            solve_log.info(
                "captcha_solved",
                solve_time_ms=solve_ms,
                poll_attempts=attempt + 1,
                token_length=len(result["request"]),
            )
            return result["request"]

        if result["request"] != "CAPCHA_NOT_READY":
            solve_log.error(
                "captcha_solve_failed",
                error=result["request"],
                poll_attempts=attempt + 1,
            )
            return None

    solve_log.warning("captcha_solve_timeout", poll_attempts=24)
    return None

输出：

{"event":"captcha_submit_start","captcha_type":"recaptcha_v2","site_url":"https://example.com","sitekey":"6Le-wvkSAAAA...","timestamp":"2025-07-15T10:30:00Z","level":"info"}
{"event":"captcha_submitted","task_id":"71845302","submit_ms":245,"timestamp":"2025-07-15T10:30:00Z","level":"info"}
{"event":"captcha_solved","task_id":"71845302","solve_time_ms":18230,"poll_attempts":4,"token_length":580,"timestamp":"2025-07-15T10:30:18Z","level":"info"}

三行日志共用一个 task_id，在 Kibana 里一搜即还原全过程。

Node.js：pino 输出同样的字段

pino 的 child() 对应 structlog 的 bind()，字段名务必和 Python 侧一致，否则看板得写两套查询。

const pino = require('pino');

const log = pino({
  level: 'info',
  timestamp: pino.stdTimeFunctions.isoTime,
});

记录识别任务的完整生命周期

const axios = require('axios');

const API_KEY = 'YOUR_API_KEY';

async function solveCaptcha(captchaType, sitekey, pageUrl) {
  const taskLog = log.child({
    captchaType,
    siteUrl: pageUrl,
    sitekey: sitekey.substring(0, 12) + '...',
  });

  const start = Date.now();
  taskLog.info('captcha_submit_start');

  const submit = await axios.post('https://ocr.captchaai.com/in.php', null, {
    params: {
      key: API_KEY, method: 'userrecaptcha',
      googlekey: sitekey, pageurl: pageUrl, json: 1,
    },
  });

  if (submit.data.status !== 1) {
    taskLog.error({ error: submit.data.request }, 'captcha_submit_failed');
    return null;
  }

  const taskId = submit.data.request;
  const boundLog = taskLog.child({ taskId });
  boundLog.info({ submitMs: Date.now() - start }, 'captcha_submitted');

  for (let attempt = 1; attempt <= 24; attempt++) {
    await new Promise(r => setTimeout(r, 5000));
    const poll = await axios.get('https://ocr.captchaai.com/res.php', {
      params: { key: API_KEY, action: 'get', id: taskId, json: 1 },
    });

    if (poll.data.status === 1) {
      boundLog.info({
        solveTimeMs: Date.now() - start,
        pollAttempts: attempt,
        tokenLength: poll.data.request.length,
      }, 'captcha_solved');
      return poll.data.request;
    }

    if (poll.data.request !== 'CAPCHA_NOT_READY') {
      boundLog.error({ error: poll.data.request, pollAttempts: attempt }, 'captcha_solve_failed');
      return null;
    }
  }

  boundLog.warn({ pollAttempts: 24 }, 'captcha_solve_timeout');
  return null;
}

过滤、聚合与告警

查最近一小时的失败任务

# With jq
cat captcha.log | jq 'select(.level == "error" and .event == "captcha_solve_failed")'

按滑动窗口监控失败率

识别成功率天然有波动，单次失败不值得报警。计数器只在窗口失败率超阈值才告警：

# Count errors vs successes in a rolling window
from collections import deque

class ErrorRateMonitor:
    def __init__(self, window_size=100, threshold=0.2):
        self.results = deque(maxlen=window_size)
        self.threshold = threshold

    def record(self, success):
        self.results.append(success)
        if len(self.results) >= 50:
            error_rate = 1 - sum(self.results) / len(self.results)
            if error_rate > self.threshold:
                log.warning(
                    "captcha_error_rate_high",
                    error_rate=round(error_rate, 3),
                    window=len(self.results),
                )

阈值和线程数有关。CaptchaAI 按并发线程计费，套餐从 BASIC（$15/月，5 线程）到 VIP-3（$7,500/月，5,000 线程），套餐内识别不限量。线程偏少时排队会拉长 solve_time_ms，这不算失败；线程够用仍频繁超时才该告警。

一个国内团队常见的场景

国内团队做跨境电商价格监控，国际站点多用 reCAPTCHA 和 Cloudflare Turnstile；reCAPTCHA 脚本由 Google 域名分发，国内网络下加载并不稳定。于是常有一种误判：captcha_solve_failed 不多，captcha_solve_timeout 晚间成片冒出来。

结构化日志正好把两者分开：超时集中在少数 site_url 且 submit_ms 同时抬高，问题多半在出口链路；submit_ms 正常而 solve_time_ms 整体上移，才该调线程数。国内站点更常见 GeeTest（极验）滑块，CaptchaAI 支持 GeeTest v3，captcha_type 里单列即可。

常见故障排查

常见问题

哪些字段绝对不能写进日志？

API Key、完整 token 和目标页面上的个人信息都不能写。sitekey 截断到前 12 位即可，既能区分站点又不泄露。

轮询过程要不要逐次记日志？

不要。CAPCHA_NOT_READY 是轮询期间的正常状态，逐次记录会让一次任务刷出二十几条噪声。只在成功、失败或超时时写一条，次数放进 poll_attempts。

日志保留多久合适？

分两层：原始 JSON 留 7～14 天，够覆盖一个排障周期；solve_time_ms 和成功/失败计数按分钟聚合成指标长期保存。

这套字段能接进 Prometheus 吗？

可以，但要转一层：日志查单次任务，指标看趋势。在 captcha_solved 与 captcha_solve_failed 上打点计数器和直方图，做法见用 Prometheus 和 Grafana 监控识别成功率。

下一步

需要 API Key，到 CaptchaAI 官网注册即可。

相关指南

• 把识别结果写入 PostgreSQL
• CaptchaAI 用量监控看板搭建
• 用 Prometheus 和 Grafana 监控识别成功率