CAPTCHA 审计日志实现：识别记录与合规追踪

合规团队问一句"这次验证码是谁触发的、花了多少钱"，日志里往往翻不出答案。本文给出可落地的审计方案：该记哪些字段、CaptchaAI 的 Python/JavaScript 实现，以及日志怎么查、存多久。

审计记录该包含哪些字段

每次验证码解决都应落一条结构化记录，至少覆盖这些字段：

timestamp —— 请求发起时间
request_id —— 本次解决的唯一标识，uuid4()
captcha_type —— 使用的验证码方法，如 userrecaptcha
target_site —— 被识别的页面 URL
task_id —— CaptchaAI 任务 ID
status —— 结果：solved、failed、timeout
solve_time_ms —— 提交到出结果的耗时
error_code —— 失败时的错误码
initiator —— 谁或哪个任务触发了这次解决
cost —— 预估费用

不要记录的三类内容：

API 密钥
CAPTCHA token（几分钟内失效，无审计价值）
目标站点返回的个人身份信息

Python 实现：给每次解决自动写审计记录

# audit_solver.py
import os
import uuid
import time
import json
import logging
from datetime import datetime, timezone
import requests

API_KEY = os.environ.get("CAPTCHAAI_KEY", "YOUR_API_KEY")

# Configure audit logger — separate from application logs
audit_logger = logging.getLogger("captcha_audit")
audit_logger.setLevel(logging.INFO)

# File handler with rotation
from logging.handlers import RotatingFileHandler
handler = RotatingFileHandler(
    "captcha_audit.jsonl",
    maxBytes=50_000_000,  # 50 MB per file
    backupCount=10,
)
handler.setFormatter(logging.Formatter("%(message)s"))
audit_logger.addHandler(handler)

def log_audit(record):
    """Write a structured audit record."""
    audit_logger.info(json.dumps(record, default=str))

def solve_with_audit(sitekey, pageurl, captcha_type="userrecaptcha",
                      initiator="unknown"):
    """Solve a CAPTCHA with full audit logging."""
    request_id = str(uuid.uuid4())
    start = time.time()

    audit_record = {
        "request_id": request_id,
        "timestamp": datetime.now(timezone.utc).isoformat(),
        "captcha_type": captcha_type,
        "target_site": pageurl,
        "initiator": initiator,
        "status": "submitted",
    }

    session = requests.Session()

    try:
        # Submit
        resp = session.get("https://ocr.captchaai.com/in.php", params={
            "key": API_KEY,
            "method": captcha_type,
            "googlekey": sitekey,
            "pageurl": pageurl,
            "json": "1",
        })
        result = resp.json()

        if result.get("status") != 1:
            audit_record.update({
                "status": "submit_failed",
                "error_code": result.get("request"),
                "solve_time_ms": int((time.time() - start) * 1000),
            })
            log_audit(audit_record)
            return None

        task_id = result["request"]
        audit_record["task_id"] = task_id

        # Poll
        time.sleep(15)
        for _ in range(25):
            poll = session.get("https://ocr.captchaai.com/res.php", params={
                "key": API_KEY, "action": "get",
                "id": task_id, "json": "1",
            })
            poll_result = poll.json()

            if poll_result.get("status") == 1:
                solve_time = int((time.time() - start) * 1000)
                audit_record.update({
                    "status": "solved",
                    "solve_time_ms": solve_time,
                    "cost_estimate": 0.003,  # Adjust per your rate
                })
                log_audit(audit_record)
                return poll_result["request"]

            if poll_result.get("request") != "CAPCHA_NOT_READY":
                audit_record.update({
                    "status": "failed",
                    "error_code": poll_result.get("request"),
                    "solve_time_ms": int((time.time() - start) * 1000),
                })
                log_audit(audit_record)
                return None

            time.sleep(5)

        audit_record.update({
            "status": "timeout",
            "solve_time_ms": int((time.time() - start) * 1000),
        })
        log_audit(audit_record)
        return None

    except Exception as e:
        audit_record.update({
            "status": "error",
            "error_code": str(e)[:200],
            "solve_time_ms": int((time.time() - start) * 1000),
        })
        log_audit(audit_record)
        raise

# Usage
token = solve_with_audit(
    sitekey="6Le-wvkSAAAAAPBMRTvw0Q4Muexq9bi0DJwx_mJ-",
    pageurl="https://www.google.com/recaptcha/api2/demo",
    initiator="price-scraper-v2",
)

审计日志输出（JSONL 格式）

{"request_id":"a1b2c3d4-...","timestamp":"2026-04-04T14:30:00+00:00","captcha_type":"userrecaptcha","target_site":"https://www.google.com/recaptcha/api2/demo","initiator":"price-scraper-v2","status":"solved","task_id":"73829451","solve_time_ms":18432,"cost_estimate":0.003}

JavaScript 实现：Node.js 场景下的等效写法

// audit_solver.js
const fs = require('fs');
const { v4: uuidv4 } = require('uuid');
const axios = require('axios');

const API_KEY = process.env.CAPTCHAAI_KEY || 'YOUR_API_KEY';
const AUDIT_FILE = 'captcha_audit.jsonl';

function logAudit(record) {
  fs.appendFileSync(AUDIT_FILE, JSON.stringify(record) + '\n');
}

async function solveWithAudit(sitekey, pageurl, initiator = 'unknown') {
  const requestId = uuidv4();
  const start = Date.now();
  const record = {
    request_id: requestId,
    timestamp: new Date().toISOString(),
    captcha_type: 'userrecaptcha',
    target_site: pageurl,
    initiator,
    status: 'submitted',
  };

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

    if (submit.data.status !== 1) {
      record.status = 'submit_failed';
      record.error_code = submit.data.request;
      record.solve_time_ms = Date.now() - start;
      logAudit(record);
      return null;
    }

    record.task_id = submit.data.request;
    await new Promise(r => setTimeout(r, 15000));

    for (let i = 0; i < 25; i++) {
      const poll = await axios.get('https://ocr.captchaai.com/res.php', {
        params: { key: API_KEY, action: 'get', id: submit.data.request, json: '1' },
      });

      if (poll.data.status === 1) {
        record.status = 'solved';
        record.solve_time_ms = Date.now() - start;
        record.cost_estimate = 0.003;
        logAudit(record);
        return poll.data.request;
      }
      if (poll.data.request !== 'CAPCHA_NOT_READY') {
        record.status = 'failed';
        record.error_code = poll.data.request;
        record.solve_time_ms = Date.now() - start;
        logAudit(record);
        return null;
      }
      await new Promise(r => setTimeout(r, 5000));
    }

    record.status = 'timeout';
    record.solve_time_ms = Date.now() - start;
    logAudit(record);
    return null;
  } catch (e) {
    record.status = 'error';
    record.error_code = e.message.slice(0, 200);
    record.solve_time_ms = Date.now() - start;
    logAudit(record);
    throw e;
  }
}

查询审计日志：按天生成汇总

每日总结

import json
from collections import Counter
from datetime import date

def daily_summary(log_file, target_date=None):
    """Generate a daily summary from audit logs."""
    target = target_date or date.today().isoformat()
    statuses = Counter()
    total_cost = 0
    solve_times = []

    with open(log_file) as f:
        for line in f:
            record = json.loads(line)
            if record["timestamp"].startswith(target):
                statuses[record["status"]] += 1
                total_cost += record.get("cost_estimate", 0)
                if record.get("solve_time_ms"):
                    solve_times.append(record["solve_time_ms"])

    print(f"Date: {target}")
    print(f"Total requests: {sum(statuses.values())}")
    print(f"Statuses: {dict(statuses)}")
    print(f"Estimated cost: ${total_cost:.2f}")
    if solve_times:
        print(f"Median solve time: {sorted(solve_times)[len(solve_times)//2]}ms")

daily_summary("captcha_audit.jsonl")

常见故障怎么排查

日志越滚越大 —— 未配置轮转，用 RotatingFileHandler 或 logrotate。
审计记录缺失 —— 异常发生在写日志之前，放进 finally 区块。
高并发写入变慢 —— 同步 I/O 拖慢主流程，改异步写入或先缓冲。
时间戳对不上 —— 时钟漂移，统一用 NTP 校时、只记 UTC。

日志该存多久、放在哪

按每天的解决量选保留方案：

100 次/天：约 30 KB/日 —— 本地文件即可
1,000 次/天：约 300 KB/日 —— 本地文件 + 自动轮转
10,000 次/天：约 3 MB/日 —— 推送到日志聚合平台
100,000 次/天：约 30 MB/日 —— 集中式日志系统（ELK、Datadog）

跨境电商、票务类团队常在万级量之前就切到集中式方案。

常见问题

审计日志和应用日志需要分开记录吗？

需要，应用日志记业务流程，审计日志记"谁、何时、用了多少资源"。

二者保留策略不同，分开存放才好查。

目标页面返回的个人信息要不要也记进审计字段？

不要。只留时间、状态、耗时、成本；关联业务数据时用 request_id 做外键去查。

审计日志的保留期该设多久？

分两种场景：

日常运营：90 天常见
合规驱动：按行业标准来，如 SOC 2、GDPR，国内团队还需对照《网络安全法》《数据安全法》和 PIPL

测试环境和生产环境的审计日志要分开存吗？

建议分开，或至少在 target_site 里标注环境，方便按环境排查，也避免测试流量污染生产成本统计。

下一步

给每一次验证码解决都加上可追溯的责任人——获取你的 CaptchaAI API 密钥。

CAPTCHA 解决审计日志：跟踪解决合规性请求

审计记录该包含哪些字段

Python 实现：给每次解决自动写审计记录

审计日志输出（JSONL 格式）

JavaScript 实现：Node.js 场景下的等效写法

查询审计日志：按天生成汇总

每日总结

常见故障怎么排查

日志该存多久、放在哪

常见问题

审计日志和应用日志需要分开记录吗？

目标页面返回的个人信息要不要也记进审计字段？

审计日志的保留期该设多久？

测试环境和生产环境的审计日志要分开存吗？

下一步

相关指南

相关文章

CaptchaAI IP白名单和API密钥安全

CaptchaAI Webhook 安全：验证回调签名

AWS Lambda + CaptchaAI：无服务器验证码解决

用于 CaptchaAI Worker 部署的 Ansible Playbook

使用 AWS SNS 和 CaptchaAI 构建事件驱动的验证码解决方案

验证码解决基础设施的蓝绿部署