Membangun Pipeline CAPTCHA Klien dengan CaptchaAI

Agensi yang menangani scraping atau otomatisasi untuk lebih dari satu klien cuma butuh empat komponen untuk berhenti menulis ulang kode solver di setiap proyek baru: intake per klien, antrean bersama, worker yang submit dan polling ke CaptchaAI, serta result store yang memisahkan token per klien. Pola berikut sering terjadi di agensi otomasi yang mengontrak lewat Upwork atau Fastwork: proyek pertama solve reCAPTCHA v2 dengan cara ad-hoc, proyek kedua menambal kode yang nyaris sama untuk Turnstile, dan dalam beberapa bulan ada lima versi solver yang mirip tapi tidak konsisten satu sama lain. Panduan ini membongkar arsitekturnya dan menunjukkan implementasi lengkap di Python dan Node.js.

Empat Komponen Wajib dalam Pipeline CAPTCHA Multi-Klien

┌──────────────┐    ┌───────────────┐    ┌──────────────┐
│  Client A    │──▶ │               │    │              │
│  Client B    │──▶ │  Task Queue   │──▶ │  CaptchaAI   │
│  Client C    │──▶ │               │    │  API         │
└──────────────┘    └───────────────┘    └──────────────┘
                           │                    │
                           ▼                    ▼
                    ┌───────────────┐    ┌──────────────┐
                    │  Result Store │◀── │  Polling      │
                    │  (Redis/DB)   │    │  Workers      │
                    └───────────────┘    └──────────────┘

Empat bagian ini menentukan apakah pipeline Anda tahan dipakai di produksi atau cuma tahan demo:

1. Intake — titik masuk tempat scraper tiap klien mengirim permintaan solve, selalu disertai client_id supaya token tidak pernah tertukar antar klien.
2. Antrean — menampung tugas dan membatasi concurrency per klien, jadi satu klien bervolume besar tidak menghabiskan seluruh slot worker.
3. Worker solver — submit task ke CaptchaAI, lalu polling sampai token keluar.
4. Result store (Redis atau database) — menyimpan token berdasarkan client_id + task_id, siap diambil consumer masing-masing klien.

Kenapa ini penting untuk agensi

Tanpa pemisahan ini, bug klasik agensi multi-klien muncul: task klien A menunggu di antrean yang sama dengan task klien B, satu klien dengan volume besar menghabiskan semua slot worker, dan klien lain jadi telat dapat token padahal tidak ada masalah di sisi mereka. Memisahkan intake, antrean, dan result store per client_id sejak awal jauh lebih murah daripada menambalnya setelah klien pertama komplain.

Implementasi Python: Kelas Pipeline dan Worker

Kelas inti: submit dan polling

Kelas CaptchaPipeline di bawah menggabungkan antrean, submit, dan polling dalam satu objek. max_concurrent mengunci berapa banyak task yang boleh berjalan bersamaan lintas semua klien — angka ini yang nanti Anda samakan dengan jumlah thread pada paket CaptchaAI yang dipakai.

import requests
import time
from dataclasses import dataclass
from typing import Optional
from collections import deque
from threading import Lock

SUBMIT_URL = "https://ocr.captchaai.com/in.php"
RESULT_URL = "https://ocr.captchaai.com/res.php"

@dataclass
class SolveRequest:
    client_id: str
    method: str
    params: dict
    callback: Optional[callable] = None

@dataclass
class SolveResult:
    client_id: str
    task_id: str
    token: Optional[str] = None
    error: Optional[str] = None


class CaptchaPipeline:
    def __init__(self, api_key: str, max_concurrent: int = 10):
        self.api_key = api_key
        self.max_concurrent = max_concurrent
        self.queue = deque()
        self.active = {}
        self.lock = Lock()

    def enqueue(self, request: SolveRequest):
        with self.lock:
            self.queue.append(request)

    def submit_task(self, request: SolveRequest) -> Optional[str]:
        data = {
            "key": self.api_key,
            "method": request.method,
            "json": 1,
            **request.params
        }

        try:
            resp = requests.post(SUBMIT_URL, data=data, timeout=15)
            result = resp.json()

            if result.get("status") == 1:
                return result["request"]
            else:
                print(f"[{request.client_id}] Submit error: {result.get('error_text', result.get('request'))}")
                return None
        except requests.RequestException as e:
            print(f"[{request.client_id}] Network error: {e}")
            return None

    def poll_result(self, task_id: str, max_wait: int = 120) -> Optional[str]:
        elapsed = 0
        interval = 5
        while elapsed < max_wait:
            time.sleep(interval)
            elapsed += interval

            try:
                resp = requests.get(RESULT_URL, params={
                    "key": self.api_key,
                    "action": "get",
                    "id": task_id,
                    "json": 1
                }, timeout=10)
                result = resp.json()

                if result.get("status") == 1:
                    return result["request"]
                elif result.get("request") == "CAPCHA_NOT_READY":
                    continue
                else:
                    print(f"Poll error for {task_id}: {result.get('error_text', result.get('request'))}")
                    return None
            except requests.RequestException:
                continue

        return None

    def process_queue(self):
        while self.queue or self.active:
            # Fill active slots
            with self.lock:
                while self.queue and len(self.active) < self.max_concurrent:
                    request = self.queue.popleft()
                    task_id = self.submit_task(request)
                    if task_id:
                        self.active[task_id] = request

            # Poll active tasks
            completed = []
            for task_id, request in list(self.active.items()):
                token = self.poll_result(task_id, max_wait=10)
                if token:
                    result = SolveResult(
                        client_id=request.client_id,
                        task_id=task_id,
                        token=token
                    )
                    if request.callback:
                        request.callback(result)
                    completed.append(task_id)

            with self.lock:
                for task_id in completed:
                    del self.active[task_id]

Menjalankan beberapa klien dalam satu proses

Contoh berikut mendaftarkan dua klien berbeda — satu pakai reCAPTCHA v2, satu pakai Turnstile — lalu memprosesnya lewat antrean yang sama:

pipeline = CaptchaPipeline(api_key="YOUR_API_KEY", max_concurrent=15)

# Client A — reCAPTCHA v2
pipeline.enqueue(SolveRequest(
    client_id="client_a",
    method="userrecaptcha",
    params={
        "googlekey": "6Le-SITEKEY-A",
        "pageurl": "https://client-a-staging.example.com/qa-form"
    },
    callback=lambda r: print(f"[{r.client_id}] Solved: {r.token[:40]}...")
))

# Client B — Turnstile
pipeline.enqueue(SolveRequest(
    client_id="client_b",
    method="turnstile",
    params={
        "sitekey": "0x4AAAA-SITEKEY-B",
        "pageurl": "https://client-b-target.com/login"
    },
    callback=lambda r: print(f"[{r.client_id}] Solved: {r.token[:40]}...")
))

pipeline.process_queue()

Versi Node.js untuk Stack JavaScript

Logikanya identik dengan versi Python: antrean in-memory, batas concurrency, submit lalu polling. Berikut versi Node.js-nya kalau stack Anda sudah berbasis JavaScript atau TypeScript:

const axios = require("axios");

const SUBMIT_URL = "https://ocr.captchaai.com/in.php";
const RESULT_URL = "https://ocr.captchaai.com/res.php";

class CaptchaPipeline {
  constructor(apiKey, maxConcurrent = 10) {
    this.apiKey = apiKey;
    this.maxConcurrent = maxConcurrent;
    this.queue = [];
    this.activeCount = 0;
  }

  enqueue(clientId, method, params) {
    return new Promise((resolve, reject) => {
      this.queue.push({ clientId, method, params, resolve, reject });
      this._processNext();
    });
  }

  async _processNext() {
    if (this.activeCount >= this.maxConcurrent || this.queue.length === 0) return;

    this.activeCount++;
    const task = this.queue.shift();

    try {
      const token = await this._solve(task);
      task.resolve({ clientId: task.clientId, token });
    } catch (err) {
      task.reject(err);
    } finally {
      this.activeCount--;
      this._processNext();
    }
  }

  async _solve(task) {
    const submitResp = await axios.post(SUBMIT_URL, null, {
      params: {
        key: this.apiKey,
        method: task.method,
        json: 1,
        ...task.params,
      },
      timeout: 15000,
    });

    if (submitResp.data.status !== 1) {
      throw new Error(submitResp.data.error_text || submitResp.data.request);
    }

    const taskId = submitResp.data.request;
    return this._poll(taskId);
  }

  async _poll(taskId, maxWait = 120000) {
    const interval = 5000;
    let elapsed = 0;

    while (elapsed < maxWait) {
      await new Promise((r) => setTimeout(r, interval));
      elapsed += interval;

      try {
        const resp = await axios.get(RESULT_URL, {
          params: {
            key: this.apiKey,
            action: "get",
            id: taskId,
            json: 1,
          },
          timeout: 10000,
        });

        if (resp.data.status === 1) return resp.data.request;
        if (resp.data.request !== "CAPCHA_NOT_READY") {
          throw new Error(resp.data.error_text || resp.data.request);
        }
      } catch (err) {
        if (err.response) throw err;
      }
    }

    throw new Error(`Timeout waiting for task ${taskId}`);
  }
}

// Usage
(async () => {
  const pipeline = new CaptchaPipeline("YOUR_API_KEY", 15);

  const results = await Promise.allSettled([
    pipeline.enqueue("client_a", "userrecaptcha", {
      googlekey: "6Le-SITEKEY-A",
      pageurl: "https://client-a-staging.example.com/qa-form",
    }),
    pipeline.enqueue("client_b", "turnstile", {
      sitekey: "0x4AAAA-SITEKEY-B",
      pageurl: "https://client-b-target.com/login",
    }),
  ]);

  results.forEach((r) => {
    if (r.status === "fulfilled") {
      console.log(`[${r.value.clientId}] Token: ${r.value.token.slice(0, 40)}...`);
    } else {
      console.error(`Failed: ${r.reason.message}`);
    }
  });
})();

Menyimpan Konfigurasi per Klien

Setiap klien biasanya punya proxy sendiri, preferensi solver, dan batas concurrency berbeda. Simpan semua itu di satu tempat, bukan hardcode di banyak file:

CLIENT_CONFIG = {
    "client_a": {
        "proxy": "host:port:user:pass",
        "proxytype": "HTTP",
        "max_concurrent": 5,
        "default_method": "userrecaptcha"
    },
    "client_b": {
        "proxy": None,
        "proxytype": None,
        "max_concurrent": 10,
        "default_method": "turnstile"
    }
}

def build_params(client_id, params):
    config = CLIENT_CONFIG.get(client_id, {})
    if config.get("proxy"):
        params["proxy"] = config["proxy"]
        params["proxytype"] = config["proxytype"]
    return params

Catatan kepatuhan: pastikan proxy dan target URL per klien hanya menjangkau data yang memang berwenang diproses klien tersebut — relevan kalau Anda tunduk pada UU Pelindungan Data Pribadi (UU 27/2022) saat menangani data pihak ketiga.

Berapa Thread CaptchaAI yang Dibutuhkan untuk Banyak Klien?

CaptchaAI menagih per thread concurrent, bukan per solve — jadi max_concurrent di kelas pipeline Anda idealnya tidak melebihi jumlah thread pada paket yang dipakai. Untuk agensi yang menangani 3–5 klien kecil sekaligus, paket BASIC ($15/bulan, 5 thread) atau STANDARD ($30/bulan, 15 thread) biasanya sudah cukup selama tiap klien cuma butuh beberapa task paralel. Begitu volume naik — misalnya ada klien tambahan dengan scraping harian dalam jumlah besar — ADVANCE ($90/bulan, 50 thread) memberi ruang tanpa menaikkan biaya per solve, karena tiap thread punya solve tak terbatas selama masa berlaku paket. Karena harga tidak berubah walau volume klien naik-turun, model ini lebih gampang dianggarkan dibanding skema per-solve saat Anda menagih klien secara flat bulanan.

Strategi Menangani Error per Klien

Error dari CaptchaAI perlu direspons berbeda-beda — sebagian harus menghentikan seluruh antrean, sebagian cukup diulang:

• ERROR_ZERO_BALANCE — hentikan seluruh antrean, kirim notifikasi ke semua klien aktif.
• ERROR_NO_SLOT_AVAILABLE — masukkan tugas kembali ke antrean dengan jeda.
• ERROR_WRONG_CAPTCHA_ID — buang task, catat ke log error.
• ERROR_CAPTCHA_UNSOLVABLE — coba ulang satu kali, setelah itu gagalkan.
• Batas waktu jaringan (timeout) — coba ulang dengan backoff eksponensial, maksimal 3 percobaan.

Debugging Masalah yang Sering Muncul di Produksi

Empat masalah ini paling sering muncul begitu pipeline jalan dengan trafik klien sungguhan:

Pertanyaan Umum

Paket CaptchaAI mana yang paling pas untuk agensi dengan 3–5 klien sekaligus?

Untuk volume kecil-menengah, BASIC ($15/bulan, 5 thread) atau STANDARD ($30/bulan, 15 thread) biasanya cukup. Naikkan ke ADVANCE ($90/bulan, 50 thread) begitu jumlah task paralel lintas klien mulai mendekati batas thread paket yang sedang dipakai.

Apakah hCaptcha bisa diselesaikan lewat pipeline ini?

Belum. hCaptcha dan FunCaptcha (Arkose Labs) tidak didukung saat ini — kalau ada klien yang butuh salah satunya, sampaikan dari awal supaya ekspektasi jelas. reCAPTCHA v2/v3, Cloudflare Turnstile dan Challenge, GeeTest v3, serta CAPTCHA gambar/grid sudah didukung penuh.

Bagaimana cara memisahkan pelacakan biaya per klien dalam satu akun CaptchaAI?

Pakai satu API key untuk seluruh klien supaya penagihan lebih sederhana, lalu manfaatkan parameter soft_id dari CaptchaAI kalau Anda tetap perlu melacak konsumsi tiap klien secara terpisah.

Apa yang terjadi kalau satu klien tiba-tiba mengirim volume CAPTCHA jauh lebih besar dari klien lain?

Set max_concurrent per klien di CLIENT_CONFIG, bukan cuma di level pipeline global. Dengan begitu satu klien yang tiba-tiba lonjak tidak menghabiskan seluruh slot worker dan membuat antrean klien lain macet.

Bagaimana pipeline tetap jalan setelah server restart atau deployment ulang?

Simpan antrean di Redis atau database, bukan di memory proses. Saat proses restart, muat ulang task yang masih pending dan lanjutkan polling dari situ — token yang sudah setengah jalan tidak hilang.

Bangun Pipeline CAPTCHA Multi-Klien dengan CaptchaAI

Ambil API key CaptchaAI dan mulai proses klien pertama Anda hari ini di captchaai.com.

Panduan Terkait

• Cara menyusun retry logic untuk API CaptchaAI
• Redis queue untuk pemrosesan CaptchaAI terdistribusi
• Membangun antrean penyelesaian CAPTCHA di Python