SQLite untuk Caching dan Tracking Solve CAPTCHA Lokal

Kalau skrip Anda mengirim task baru untuk sitekey yang sama dua kali dalam sepuluh detik, solve kedua itu hampir selalu mubazir: token dari solve pertama masih hidup. Jawaban paling murah untuk pemborosan itu bukan Redis dan bukan PostgreSQL, melainkan satu file SQLite yang duduk di sebelah skrip Anda. Artikel ini memberi skema tabelnya, logika cache token dengan batas waktu, dan cara membaca statistik harian dari data yang sama.

SQLite sudah ikut terpasang bersama Python, tidak butuh proses server, dan sanggup menerima ribuan penulisan per detik. Untuk satu mesin — laptop developer, satu VPS, atau satu worker scraping — itu lebih dari cukup.

Dua kebocoran yang ditutup penyimpanan lokal

Alur kerja penyelesaian CAPTCHA tanpa database lokal hampir selalu punya dua lubang yang sama:

1. Token diminta ulang padahal masih berlaku. Token reCAPTCHA umumnya hidup 90–120 detik, dan selama jendela itu token yang belum dipakai tetap sah. Tanpa cache, setiap percobaan ulang menembak in.php lagi.
2. Tidak ada jejak untuk debugging. Saat sesuatu gagal, yang tersisa hanya log teks. Tidak ada jawaban cepat untuk "berapa persen solve yang berhasil semalam" atau "tipe mana yang paling lambat minggu ini".

Satu file database menutup keduanya sekaligus: satu tabel untuk riwayat solve, satu tabel untuk cache token.

Skema: tabel riwayat dan tabel cache

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

Perhatikan PRIMARY KEY (sitekey, pageurl, token) pada token_cache. Kombinasi itulah yang membuat pencarian cache murah sekaligus mencegah entri ganda. Kolom used memastikan satu token hanya dipakai sekali — token CAPTCHA bukan barang yang boleh dipakai ulang di dua request berbeda.

Implementasi Python

Buka koneksi dan siapkan tabel

Dua pragma di bawah ini yang membuat SQLite tahan dipakai beberapa worker sekaligus: journal_mode=WAL memisahkan pembaca dari penulis, dan busy_timeout membuat penulis kedua menunggu alih-alih langsung melempar database is locked.

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

Kirim task, polling, lalu simpan hasilnya

Urutannya mengikuti pola standar CaptchaAI — cek cache → kirim ke in.php → simpan task ID → polling res.php → pakai token. Bedanya, setiap perpindahan status ikut ditulis ke tabel, jadi baris yang berhenti di status polling langsung terlihat sebagai kasus yang perlu diselidiki.

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

Simpan dan ambil token selama masa berlaku

TTL 90 detik pada contoh berikut sengaja dipasang konservatif: lebih pendek dari umur token sebenarnya, supaya token tidak kedaluwarsa di tengah pengiriman form. Fungsi get_cached_token sekaligus menandai token sebagai terpakai, jadi tidak ada dua worker yang mengambil token yang sama.

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

Query statistik dan pembersihan berkala

Karena semuanya sudah berbentuk baris, laporan harian cukup satu query agregat: total task, jumlah yang selesai, dan rata-rata waktu penyelesaian. Jalankan cleanup_old_records() lewat cron harian agar file database tidak tumbuh tanpa batas.

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

Implementasi JavaScript dengan better-sqlite3

Untuk worker Node.js, better-sqlite3 memberi API sinkron yang justru lebih sederhana di dalam loop polling. Skema dan alur statusnya identik dengan versi Python, jadi worker Python dan Node.js bisa berbagi satu file database yang sama.

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

Operasi harian: kunci, ukuran file, dan indeks

Satu catatan tambahan yang sering terlewat: taruh file .db di disk lokal, bukan di volume jaringan (NFS atau SMB). Penguncian file SQLite tidak dapat diandalkan di atas filesystem jaringan, dan itulah penyebab paling umum dari kerusakan database yang seolah muncul tanpa sebab.

Kapan SQLite tidak lagi cukup

Batas sebenarnya bukan jumlah data, melainkan jumlah penulis. Begitu ada dua mesin yang perlu membaca cache yang sama, pindahkan lapisan cache ke manajemen masa berlaku token dengan Redis dan sisakan SQLite untuk riwayat lokal saja.

Catatan untuk tim automation di Indonesia

Pola ini cocok dengan cara kerja yang umum di sini: proyek scraping dan automation lepas (Upwork, Fastwork), agensi price monitoring, serta tim data startup yang menjalankan satu worker di VPS Jakarta atau di region ap-southeast-3. Anda tidak perlu menambah satu layanan berbayar lagi hanya untuk menyimpan riwayat solve — file .db ikut serta bersama repo deployment Anda, dan bisa diunduh kapan pun klien meminta laporan.

Sisi biayanya juga relevan. CaptchaAI menagih per thread yang berjalan bersamaan, bukan per solve; paket BASIC ($15/bulan, 5 thread) sudah mencakup solve tanpa batas selama thread tersedia. Artinya cache tidak menurunkan "harga per CAPTCHA", tetapi membebaskan thread lebih cepat — dan thread yang bebas berarti antrean pekerjaan Anda terus jalan tanpa perlu naik paket. Untuk worker yang menembak sitekey yang sama berulang kali sepanjang hari, efeknya terasa langsung pada throughput.

Terakhir, soal kepatuhan: kalau data yang Anda kumpulkan menyangkut data pribadi, UU Pelindungan Data Pribadi (UU 27/2022) berlaku penuh, jadi simpan di tabel hanya kolom teknis seperti sitekey, pageurl, status, dan durasi — bukan isi halaman atau identitas orang.

Pertanyaan umum

Apakah cache token benar-benar menghemat biaya CaptchaAI?

Tidak secara langsung, karena penagihan berbasis thread, bukan per solve. Yang Anda hemat adalah waktu dan kapasitas: solve yang tidak jadi dikirim berarti satu thread tetap bebas untuk pekerjaan lain, sehingga paket yang sama menampung volume lebih besar.

Bagaimana mencegah database is locked saat beberapa worker menulis bersamaan?

Aktifkan PRAGMA journal_mode=WAL dan PRAGMA busy_timeout=5000 di setiap koneksi, persis seperti pada fungsi get_db() di atas. WAL mengizinkan banyak pembaca berbarengan dengan satu penulis, dan busy_timeout membuat penulis kedua menunggu beberapa detik alih-alih gagal seketika.

Apakah SQLite cocok dipakai di AWS Lambda atau fungsi serverless lain?

Tidak untuk cache bersama. Setiap instance Lambda punya filesystem sementara sendiri, jadi cache tidak akan dibagi antar-invocation. Untuk pola serverless, gunakan pelacakan solve berbasis DynamoDB; SQLite tetap berguna di worker yang berumur panjang.

Bisakah skema yang sama dipakai untuk Turnstile dan GeeTest v3?

Bisa. Kolom type memang disiapkan untuk itu — isi dengan turnstile atau geetest_v3, lalu simpan parameter kunci tiap tipe di kolom sitekey. CaptchaAI mendukung reCAPTCHA v2/v3, Cloudflare Turnstile, dan GeeTest v3; hCaptcha dan FunCaptcha belum didukung, sedangkan GeeTest v4 masih berstatus segera hadir, jadi jangan menyiapkan alur produksi untuk ketiganya.

Berapa lama riwayat solve sebaiknya disimpan?

Tiga puluh hari cukup untuk analisis tren dan investigasi kegagalan. Token solusinya sendiri sudah tidak berguna setelah beberapa menit, jadi biarkan cleanup_old_records() menghapusnya bersama baris lama, lalu jalankan VACUUM supaya ukuran file benar-benar turun.

Artikel Terkait

• Pelacakan solve CAPTCHA serverless dengan DynamoDB

Langkah Selanjutnya

Mulai dari satu tabel dan satu file .db — ambil API key CaptchaAI Anda, catat solve pertama, lalu tambahkan cache token setelah pola pemakaian Anda terlihat.

Panduan terkait:

• Riwayat dan analitik solve dengan MongoDB
• Manajemen masa berlaku token di Redis
• Menghindari solve ganda dengan penguncian database