Dùng SQLite để cache và theo dõi kết quả giải CAPTCHA

Một file .db duy nhất là đủ cho phần lớn pipeline giải CAPTCHA chạy trên một máy. SQLite có sẵn trong Python, không cần dựng server, và chỉ tốn vài chục dòng code để bạn biết hôm nay gửi bao nhiêu task, bao nhiêu task giải thành công, thời gian giải trung bình ra sao, và token nào còn hạn để dùng lại.

Thứ tự triển khai bên dưới: dựng schema, ghi log vòng đời một lần giải (in.php → polling res.php → token), thêm cache token có TTL, rồi thống kê và dọn dữ liệu cũ.

Thiết kế schema: một bảng log, một bảng cache token

Hai bảng là đủ. captcha_solves là nhật ký vòng đời — mỗi lần gửi task một dòng, cập nhật dần theo trạng thái. token_cache giữ token còn hạn theo cặp sitekey + pageurl, với expires_at để loại token quá hạn và used để không dùng lại một token hai lần.

CREATE TABLE IF NOT EXISTS captcha_solves (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    captcha_id TEXT,
    type TEXT NOT NULL,
    sitekey TEXT,
    pageurl TEXT,
    status TEXT NOT NULL DEFAULT 'submitted',
    solution TEXT,
    error TEXT,
    submitted_at TEXT NOT NULL DEFAULT (datetime('now')),
    solved_at TEXT,
    elapsed_ms INTEGER,
    polls INTEGER DEFAULT 0,
    project TEXT
);

CREATE INDEX IF NOT EXISTS idx_submitted_at ON captcha_solves(submitted_at);
CREATE INDEX IF NOT EXISTS idx_type_status ON captcha_solves(type, status);
CREATE INDEX IF NOT EXISTS idx_sitekey ON captcha_solves(sitekey);

-- Token cache for reuse within TTL
CREATE TABLE IF NOT EXISTS token_cache (
    sitekey TEXT NOT NULL,
    pageurl TEXT NOT NULL,
    token TEXT NOT NULL,
    created_at TEXT NOT NULL DEFAULT (datetime('now')),
    expires_at TEXT NOT NULL,
    used INTEGER DEFAULT 0,
    PRIMARY KEY (sitekey, pageurl, token)
);

CREATE INDEX IF NOT EXISTS idx_cache_lookup
ON token_cache(sitekey, pageurl, used, expires_at);

Thiếu ba index ở submitted_at, cặp (type, status) và tổ hợp tra cứu cache, mọi truy vấn thống kê sẽ quét toàn bảng ngay khi bạn vượt vài chục nghìn dòng.

Triển khai bằng Python

Bước 1: mở kết nối và tạo bảng

Hai PRAGMA ở đây quyết định bạn có gặp lỗi database is locked hay không: journal_mode=WAL cho phép đọc song song khi đang ghi, busy_timeout=5000 bắt SQLite chờ 5 giây thay vì báo lỗi ngay.

import os
import time
import sqlite3
from datetime import datetime, timedelta, timezone
import requests

DB_PATH = os.environ.get("CAPTCHA_DB", "captcha_solves.db")
API_KEY = os.environ["CAPTCHAAI_API_KEY"]

def get_db():
    conn = sqlite3.connect(DB_PATH)
    conn.row_factory = sqlite3.Row
    conn.execute("PRAGMA journal_mode=WAL")  # Better concurrent read performance
    conn.execute("PRAGMA busy_timeout=5000")
    return conn

def init_db():
    conn = get_db()
    conn.executescript("""
        CREATE TABLE IF NOT EXISTS captcha_solves (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            captcha_id TEXT,
            type TEXT NOT NULL,
            sitekey TEXT,
            pageurl TEXT,
            status TEXT NOT NULL DEFAULT 'submitted',
            solution TEXT,
            error TEXT,
            submitted_at TEXT NOT NULL DEFAULT (datetime('now')),
            solved_at TEXT,
            elapsed_ms INTEGER,
            polls INTEGER DEFAULT 0,
            project TEXT
        );
        CREATE INDEX IF NOT EXISTS idx_submitted_at ON captcha_solves(submitted_at);
        CREATE INDEX IF NOT EXISTS idx_type_status ON captcha_solves(type, status);

        CREATE TABLE IF NOT EXISTS token_cache (
            sitekey TEXT NOT NULL,
            pageurl TEXT NOT NULL,
            token TEXT NOT NULL,
            created_at TEXT NOT NULL DEFAULT (datetime('now')),
            expires_at TEXT NOT NULL,
            used INTEGER DEFAULT 0,
            PRIMARY KEY (sitekey, pageurl, token)
        );
        CREATE INDEX IF NOT EXISTS idx_cache_lookup
        ON token_cache(sitekey, pageurl, used, expires_at);
    """)
    conn.close()

init_db()

Mọi câu lệnh đều là IF NOT EXISTS, nên cứ gọi init_db() ở đầu tiến trình, không cần bước migration riêng.

Bước 2: gửi task, polling và ghi log từng bước

Luồng gồm bốn nhịp: kiểm tra cache, gửi task tới in.php, polling res.php, rồi lưu token. Dòng log được INSERT trước khi gửi request — nhờ vậy cả task lỗi mạng cũng để lại dấu vết.

def solve_recaptcha(sitekey, pageurl, project=None):
    conn = get_db()

    # Check cache first
    cached = get_cached_token(conn, sitekey, pageurl)
    if cached:
        conn.close()
        return cached

    # Insert tracking record
    now = datetime.now(timezone.utc).isoformat()
    cursor = conn.execute(
        "INSERT INTO captcha_solves (type, sitekey, pageurl, submitted_at, project) "
        "VALUES (?, ?, ?, ?, ?)",
        ("recaptcha_v2", sitekey, pageurl, now, project)
    )
    row_id = cursor.lastrowid
    conn.commit()

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

    if data.get("status") != 1:
        conn.execute(
            "UPDATE captcha_solves SET status=?, error=? WHERE id=?",
            ("error", data.get("request"), row_id)
        )
        conn.commit()
        conn.close()
        return None

    captcha_id = data["request"]
    conn.execute(
        "UPDATE captcha_solves SET captcha_id=?, status=? WHERE id=?",
        (captcha_id, "polling", row_id)
    )
    conn.commit()

    # Poll
    polls = 0
    for _ in range(60):
        time.sleep(5)
        polls += 1
        result = requests.get("https://ocr.captchaai.com/res.php", params={
            "key": API_KEY, "action": "get",
            "id": captcha_id, "json": 1
        }).json()

        if result.get("status") == 1:
            solved_at = datetime.now(timezone.utc).isoformat()
            submitted = datetime.fromisoformat(now)
            elapsed = int((datetime.now(timezone.utc) - submitted).total_seconds() * 1000)

            conn.execute(
                "UPDATE captcha_solves SET status=?, solution=?, solved_at=?, "
                "elapsed_ms=?, polls=? WHERE id=?",
                ("solved", result["request"], solved_at, elapsed, polls, row_id)
            )
            # Cache the token
            cache_token(conn, sitekey, pageurl, result["request"])
            conn.commit()
            conn.close()
            return result["request"]

        if result.get("request") != "CAPCHA_NOT_READY":
            conn.execute(
                "UPDATE captcha_solves SET status=?, error=?, polls=? WHERE id=?",
                ("error", result.get("request"), polls, row_id)
            )
            conn.commit()
            conn.close()
            return None

    conn.execute(
        "UPDATE captcha_solves SET status=?, polls=? WHERE id=?",
        ("timeout", polls, row_id)
    )
    conn.commit()
    conn.close()
    return None

elapsed_ms và polls là hai cột bạn dùng nhiều nhất khi tinh chỉnh chu kỳ polling: đa số task xong ở lần poll thứ hai thì 5 giây là hợp lý; thường tới lần thứ mười thì nên giãn chu kỳ.

Bước 3: cache token theo TTL

Token reCAPTCHA v2 chỉ sống khoảng 90–120 giây, nên TTL 90 giây là mức an toàn. Cache phục vụ đúng một mục đích: khi cùng một form được thử lại trong vài giây (retry sau lỗi mạng), bạn dùng lại token còn hạn thay vì gửi thêm task.

def cache_token(conn, sitekey, pageurl, token, ttl_seconds=90):
    expires_at = (datetime.now(timezone.utc) + timedelta(seconds=ttl_seconds)).isoformat()
    conn.execute(
        "INSERT OR REPLACE INTO token_cache (sitekey, pageurl, token, expires_at) "
        "VALUES (?, ?, ?, ?)",
        (sitekey, pageurl, token, expires_at)
    )

def get_cached_token(conn, sitekey, pageurl):
    now = datetime.now(timezone.utc).isoformat()
    row = conn.execute(
        "SELECT token FROM token_cache "
        "WHERE sitekey=? AND pageurl=? AND used=0 AND expires_at > ? "
        "ORDER BY expires_at ASC LIMIT 1",
        (sitekey, pageurl, now)
    ).fetchone()

    if row:
        conn.execute(
            "UPDATE token_cache SET used=1 WHERE token=?",
            (row["token"],)
        )
        conn.commit()
        return row["token"]
    return None

get_cached_token() đánh dấu used=1 ngay khi trả token, nên hai worker không thể nhận cùng một token. Đừng nâng TTL vượt tuổi thọ thật của token — token hết hạn nằm lại trong cache chỉ sinh ra lỗi xác thực khó truy vết.

Bước 4: truy vấn thống kê và dọn dữ liệu cũ

def get_stats(hours=24):
    conn = get_db()
    cutoff = (datetime.now(timezone.utc) - timedelta(hours=hours)).isoformat()

    total = conn.execute(
        "SELECT COUNT(*) FROM captcha_solves WHERE submitted_at >= ?", (cutoff,)
    ).fetchone()[0]

    solved = conn.execute(
        "SELECT COUNT(*) FROM captcha_solves WHERE submitted_at >= ? AND status='solved'",
        (cutoff,)
    ).fetchone()[0]

    avg_time = conn.execute(
        "SELECT AVG(elapsed_ms) FROM captcha_solves "
        "WHERE submitted_at >= ? AND status='solved'",
        (cutoff,)
    ).fetchone()[0]

    conn.close()
    return {
        "total": total,
        "solved": solved,
        "success_rate": (solved / total * 100) if total else 0,
        "avg_time_ms": round(avg_time) if avg_time else 0
    }

def cleanup_old_records(days=30):
    conn = get_db()
    cutoff = (datetime.now(timezone.utc) - timedelta(days=days)).isoformat()
    conn.execute("DELETE FROM captcha_solves WHERE submitted_at < ?", (cutoff,))
    conn.execute("DELETE FROM token_cache WHERE expires_at < ?",
                 (datetime.now(timezone.utc).isoformat(),))
    conn.execute("VACUUM")
    conn.commit()
    conn.close()

get_stats() trả về bốn con số đủ cho một bảng điều khiển nội bộ. Chạy cleanup_old_records() hằng ngày qua cron: xoá log quá 30 ngày, xoá token hết hạn và VACUUM.

Bản Node.js cho pipeline JavaScript

Với crawler viết bằng Node.js, better-sqlite3 cho API đồng bộ và không cần callback lồng nhau. Logic giữ nguyên; lưu ý db.pragma("journal_mode = WAL") phải chạy trước mọi lệnh ghi.

const Database = require("better-sqlite3");
const axios = require("axios");

const db = new Database(process.env.CAPTCHA_DB || "captcha_solves.db");
const API_KEY = process.env.CAPTCHAAI_API_KEY;

db.pragma("journal_mode = WAL");
db.exec(`
  CREATE TABLE IF NOT EXISTS captcha_solves (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    captcha_id TEXT, type TEXT NOT NULL, sitekey TEXT, pageurl TEXT,
    status TEXT DEFAULT 'submitted', solution TEXT, error TEXT,
    submitted_at TEXT DEFAULT (datetime('now')),
    solved_at TEXT, elapsed_ms INTEGER, polls INTEGER DEFAULT 0
  );
  CREATE INDEX IF NOT EXISTS idx_submitted ON captcha_solves(submitted_at);
`);

async function solveAndStore(sitekey, pageurl) {
  const submittedAt = new Date().toISOString();
  const insert = db.prepare(
    "INSERT INTO captcha_solves (type, sitekey, pageurl, submitted_at) VALUES (?, ?, ?, ?)"
  );
  const { lastInsertRowid } = insert.run("recaptcha_v2", sitekey, pageurl, submittedAt);

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

  if (submit.data.status !== 1) {
    db.prepare("UPDATE captcha_solves SET status=?, error=? WHERE id=?")
      .run("error", submit.data.request, lastInsertRowid);
    return null;
  }

  const captchaId = submit.data.request;
  db.prepare("UPDATE captcha_solves SET captcha_id=?, status=? WHERE id=?")
    .run(captchaId, "polling", lastInsertRowid);

  let polls = 0;
  for (let i = 0; i < 60; i++) {
    await new Promise((r) => setTimeout(r, 5000));
    polls++;
    const poll = await axios.get("https://ocr.captchaai.com/res.php", {
      params: { key: API_KEY, action: "get", id: captchaId, json: 1 },
    });

    if (poll.data.status === 1) {
      const elapsed = Date.now() - new Date(submittedAt).getTime();
      db.prepare(
        "UPDATE captcha_solves SET status=?, solution=?, solved_at=?, elapsed_ms=?, polls=? WHERE id=?"
      ).run("solved", poll.data.request, new Date().toISOString(), elapsed, polls, lastInsertRowid);
      return poll.data.request;
    }
    if (poll.data.request !== "CAPCHA_NOT_READY") {
      db.prepare("UPDATE captcha_solves SET status=?, error=?, polls=? WHERE id=?")
        .run("error", poll.data.request, polls, lastInsertRowid);
      return null;
    }
  }

  db.prepare("UPDATE captcha_solves SET status=?, polls=? WHERE id=?")
    .run("timeout", polls, lastInsertRowid);
  return null;
}

Một kịch bản thực tế

Hình dung nhóm QA outsourcing ở TP.HCM kiểm thử cho vài chục website khách hàng, chạy bộ test đăng nhập hằng đêm trên staging.example.com/qa-login. Với log dạng text, không ai trả lời được câu hỏi "đêm qua hỏng vì đâu". Sau khi thêm cột project, mỗi sáng một truy vấn GROUP BY project, status cho biết ngay dự án nào có tỷ lệ lỗi tăng.

Cột đó cũng giúp phân bổ chi phí nội bộ. CaptchaAI tính tiền theo thread (luồng giải đồng thời), không theo từng lượt giải, nên số liệu trong SQLite cho biết bạn đã dùng hết số thread đã mua hay chưa. Với gói BASIC ($15/tháng, 5 thread), hiếm khi có quá hai task chạy song song thì chưa cần nâng gói; hàng đợi liên tục xếp chồng thì STANDARD ($30/tháng, 15 thread) là bước tiếp theo. Giá niêm yết bằng USD.

Khi nào SQLite đủ dùng, khi nào nên đổi

Ranh giới thực tế nằm ở số tiến trình ghi: có server thứ hai ghi vào cùng dataset là lúc chuyển sang PostgreSQL.

Xử lý sự cố thường gặp

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

Nhiều worker cùng ghi vào một file SQLite có an toàn không?

Có, nếu các worker chạy trên cùng một máy và bạn đã bật WAL kèm busy_timeout: SQLite cho phép đọc song song và tuần tự hoá các lần ghi. Với file trên ổ mạng thì không — hãy dùng PostgreSQL.

Nên đặt TTL cho cache token là bao nhiêu?

90 giây cho reCAPTCHA v2 và Turnstile là mức hợp lý, thấp hơn tuổi thọ thật của token một khoảng an toàn. Cache chỉ phục vụ retry trong vài giây kế tiếp, không phải để tích trữ token dùng dần.

Có dùng được cùng schema này cho Turnstile hay GeeTest v3 không?

Được, cột type sinh ra chính là để phân biệt. Chỉ cần đổi tham số method khi gửi task tới in.php và ghi giá trị type tương ứng, các truy vấn thống kê giữ nguyên.

Chuyển từ SQLite sang database production có mất dữ liệu không?

Không. Xuất bằng sqlite3 captcha_solves.db ".dump" rồi nạp vào PostgreSQL; TEXT và INTEGER ánh xạ thẳng, chỉ cần chỉnh cú pháp AUTOINCREMENT.

Bài viết liên quan

• Theo dõi lượt giải CAPTCHA serverless bằng DynamoDB

Các bước tiếp theo

Chạy init_db(), gọi solve_recaptcha() một lần rồi mở bảng captcha_solves để xem dòng log đầu tiên. Lấy API key CaptchaAI và thay YOUR_API_KEY trong ví dụ.

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

• Lưu lịch sử và phân tích lượt giải bằng MongoDB
• Quản lý TTL của token bằng Redis
• Chống trùng lặp yêu cầu giải CAPTCHA