Nhật ký kiểm tra giải quyết CAPTCHA: Theo dõi giải quyết các yêu cầu tuân thủ

Đội scraping hay QA giải hàng nghìn CAPTCHA mỗi ngày. Khi cần đối soát chi phí hoặc điều tra sự cố, câu hỏi luôn là: request nào, lúc nào, tốn bao nhiêu, ai kích hoạt — và không có audit log thì không trả lời được. Nghị định 13/2023/NĐ-CP về bảo vệ dữ liệu cá nhân cũng yêu cầu chứng minh dữ liệu được xử lý có kiểm soát; audit log là cách rẻ nhất để có bằng chứng đó. Luồng chuẩn cho một lần giải có audit log:

• Gửi request tới in.php, ghi request_id
• Nhận task_id từ CaptchaAI
• Poll res.php (hoặc dùng callback) đến khi có kết quả
• Ghi status, solve_time_ms và chi phí vào log

Field cần ghi trong audit log CAPTCHA

Không ghi vào audit log:

• API key
• Token CAPTCHA — chỉ tồn tại tạm thời
• Thông tin định danh cá nhân từ trang mục tiêu

Thời gian lưu trữ theo khối lượng giải

Sự cố thường gặp khi ghi audit log

Triển khai audit log bằng Python

Code dưới đây ghi từng bước — submit, poll, kết quả — vào file JSONL, tách biệt khỏi log ứng dụng:

# 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",
)

Output audit log ở định dạng JSONL

Mỗi dòng log là một JSON object độc lập:

{"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}

Triển khai audit log bằng Node.js

Ưu tiên stack Node.js? Code dưới đây cùng logic với bản Python phía trên:

// 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;
  }
}

Truy vấn và tổng hợp audit log

Báo cáo tổng hợp theo ngày

Script sau đọc file JSONL và in tổng request, trạng thái, chi phí ước tính và thời gian giải trung vị trong ngày:

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")

Ví dụ: nhóm theo dõi giá trên Shopee, Lazada, Tiki tại TP.HCM chạy daily_summary() mỗi sáng để đối chiếu chi phí với usage dashboard của CaptchaAI. Với gói BASIC ($15/tháng, 5 thread), log JSONL này đủ dùng ngay, không cần công cụ riêng.

Câu hỏi thường gặp

Có nên ghi token CAPTCHA vào audit log không?

Không. Token chỉ tồn tại tạm thời (60–300 giây) và không có giá trị kiểm tra về sau.

Đối soát hóa đơn CaptchaAI bằng audit log như thế nào?

So sánh tổng chi phí ước tính trong log với usage dashboard của CaptchaAI cùng khoảng thời gian; chênh lệch lớn thường do lệch múi giờ.

Audit log CAPTCHA có giúp đáp ứng Nghị định 13/2023/NĐ-CP không?

Một phần. Log ghi request, thời gian và initiator đáp ứng yêu cầu "truy vết được", nhưng bạn vẫn cần chính sách lưu trữ dữ liệu riêng — đây không phải tư vấn pháp lý.

Nên lưu audit log tập trung khi chạy nhiều worker không?

Có, từ vài worker trở lên. Gửi thẳng vào một log aggregator dùng chung (ELK, Datadog…) tránh log bị phân mảnh giữa các máy.

Bài viết liên quan

• Theo dõi giải CAPTCHA trên DynamoDB serverless

Bước tiếp theo

Thêm audit log cho mọi lần giải CAPTCHA — lấy API key CaptchaAI và áp dụng code Python hoặc Node.js ở trên ngay hôm nay.

Hướng dẫn liên quan:

• Xây dựng dashboard theo dõi usage
• Structured logging cho CAPTCHA
• Kiểm tra số dư và tự động nạp tiền