Membangun Antrian Pemecahan CAPTCHA dengan Python dan CaptchaAI

Kalau CAPTCHA dikirim ke API satu per satu, sebagian besar waktu scraper Anda habis menunggu, bukan bekerja. Jawabannya: pisahkan pengiriman dari pengambilan hasil lewat antrian, sehingga ratusan task berjalan bersamaan. Panduan ini membahas empat pola antrian di Python — threading, asyncio, producer-consumer, dan priority queue — plus cara memantau throughput-nya.

Kenapa Loop Satu-per-Satu Tidak Cukup

Tim scraping dan price-monitoring di Indonesia — banyak lewat Upwork atau Fastwork — biasanya mulai dengan loop for sederhana: kirim CAPTCHA, tunggu hasil, lanjut ke halaman berikutnya. Cukup untuk sepuluh halaman, tapi begitu volumenya naik ke ratusan, waktu tunggu polling menumpuk dan scraper jadi I/O-bound tanpa alasan.

Sistem antrian memecah proses ini jadi dua tahap: kirim task sebanyak mungkin di awal, baru polling hasilnya secara paralel. Dengan pola ini, antrian:

• Mengirim semua CAPTCHA ke in.php tanpa menunggu satu per satu selesai
• Melakukan polling banyak task_id sekaligus ke res.php
• Retry otomatis untuk task yang gagal
• Mengontrol concurrency supaya tidak melanggar batas rate API
• Menyediakan pelacakan progress dan callback per task

Kalau scraper Anda dideploy di AWS ap-southeast-1 (Singapura) atau GCP asia-southeast2 (Jakarta), latensi ke API CaptchaAI relatif rendah — tapi jaringan mobile-first tim lapangan tetap bisa membuat request individual gagal sesekali. Karena itu, retry dan timeout yang jelas perlu masuk ke desain antrian sejak awal.

Langkah 1: Antrian Berbasis Thread — Titik Awal Paling Sederhana

Kalau kode Anda sudah synchronous, mulai dari sini. CaptchaQueue di bawah menjalankan beberapa worker thread yang mengambil task dari Queue bawaan Python, mengirimkannya ke CaptchaAI, lalu polling res.php sampai task selesai atau timeout.

import time
import threading
import requests
from queue import Queue, Empty

API_KEY = "YOUR_API_KEY"


class CaptchaQueue:
    """Thread-based CAPTCHA solving queue."""

    def __init__(self, api_key, max_workers=10):
        self.api_key = api_key
        self.task_queue = Queue()
        self.result_queue = Queue()
        self.max_workers = max_workers
        self.workers = []

    def submit(self, method, callback=None, **params):
        """Add a CAPTCHA task to the queue."""
        task = {
            "method": method,
            "params": params,
            "callback": callback,
        }
        self.task_queue.put(task)

    def start(self):
        """Start worker threads."""
        for _ in range(self.max_workers):
            t = threading.Thread(target=self._worker, daemon=True)
            t.start()
            self.workers.append(t)

    def wait(self):
        """Wait for all tasks to complete."""
        self.task_queue.join()

    def get_results(self):
        """Get all available results."""
        results = []
        while not self.result_queue.empty():
            try:
                results.append(self.result_queue.get_nowait())
            except Empty:
                break
        return results

    def _worker(self):
        while True:
            try:
                task = self.task_queue.get(timeout=1)
            except Empty:
                continue

            try:
                result = self._solve(task["method"], **task["params"])
                entry = {"status": "solved", "result": result, "task": task}
                self.result_queue.put(entry)
                if task["callback"]:
                    task["callback"](result)
            except Exception as e:
                entry = {"status": "error", "error": str(e), "task": task}
                self.result_queue.put(entry)
            finally:
                self.task_queue.task_done()

    def _solve(self, method, **params):
        submit = requests.post("https://ocr.captchaai.com/in.php", data={
            "key": self.api_key, "method": method, "json": 1, **params,
        }, timeout=30).json()

        if submit.get("status") != 1:
            raise Exception(f"Submit error: {submit.get('request')}")

        task_id = submit["request"]
        for _ in range(30):
            time.sleep(5)
            result = requests.get("https://ocr.captchaai.com/res.php", params={
                "key": self.api_key, "action": "get", "id": task_id, "json": 1,
            }, timeout=30).json()
            if result.get("status") == 1:
                return result["request"]
            if result.get("request") == "ERROR_CAPTCHA_UNSOLVABLE":
                raise Exception("CAPTCHA unsolvable")
        raise TimeoutError("Solve timed out")


# Usage
queue = CaptchaQueue(API_KEY, max_workers=5)
queue.start()

# Submit multiple CAPTCHAs
urls_and_sitekeys = [
    ("https://example.com/page1", "SITEKEY_1"),
    ("https://example.com/page2", "SITEKEY_2"),
    ("https://example.com/page3", "SITEKEY_3"),
]

for url, sitekey in urls_and_sitekeys:
    queue.submit("userrecaptcha", googlekey=sitekey, pageurl=url)

queue.wait()
results = queue.get_results()
print(f"Solved {len(results)} CAPTCHAs")
for r in results:
    print(f"  {r['status']}: {r.get('result', r.get('error', ''))[:50]}")

Empat bagian inti: submit() menambahkan task tanpa memblokir, start() menjalankan worker sebanyak max_workers, _worker() adalah loop utama tiap thread, dan _solve() menangani pengiriman plus polling ke CaptchaAI. Panggil wait() untuk menunggu semua task selesai, lalu get_results() untuk mengambil hasilnya.

Langkah 2: Naik ke asyncio untuk Concurrency yang Lebih Tinggi

Threading bekerja, tapi tiap thread Python tetap menunggu (blocking) selama request HTTP berjalan. Untuk task I/O-bound seperti solve CAPTCHA, asyncio biasanya lebih hemat resource karena satu event loop bisa menangani ratusan koneksi tanpa overhead context-switch antar thread. AsyncCaptchaQueue di bawah memakai aiohttp dan Semaphore untuk membatasi berapa banyak solve yang berjalan bersamaan.

import asyncio
import aiohttp

API_KEY = "YOUR_API_KEY"


class AsyncCaptchaQueue:
    """Async CAPTCHA solving queue with concurrency control."""

    def __init__(self, api_key, max_concurrent=10):
        self.api_key = api_key
        self.semaphore = asyncio.Semaphore(max_concurrent)
        self.results = []

    async def solve_batch(self, tasks):
        """Solve a batch of CAPTCHA tasks concurrently."""
        coros = [self._solve_task(task) for task in tasks]
        self.results = await asyncio.gather(*coros, return_exceptions=True)
        return self.results

    async def _solve_task(self, task):
        async with self.semaphore:
            return await self._solve(task["method"], **task["params"])

    async def _solve(self, method, **params):
        async with aiohttp.ClientSession() as session:
            # Submit
            async with session.post("https://ocr.captchaai.com/in.php", data={
                "key": self.api_key, "method": method, "json": 1, **params,
            }) as resp:
                data = await resp.json(content_type=None)
                if data.get("status") != 1:
                    raise Exception(f"Submit error: {data.get('request')}")
                task_id = data["request"]

            # Poll
            for _ in range(30):
                await asyncio.sleep(5)
                async with session.get("https://ocr.captchaai.com/res.php", params={
                    "key": self.api_key, "action": "get", "id": task_id, "json": 1,
                }) as resp:
                    result = await resp.json(content_type=None)
                    if result.get("status") == 1:
                        return result["request"]
                    if result.get("request") == "ERROR_CAPTCHA_UNSOLVABLE":
                        raise Exception("CAPTCHA unsolvable")

            raise TimeoutError("Solve timed out")


# Usage
async def main():
    queue = AsyncCaptchaQueue(API_KEY, max_concurrent=5)

    tasks = [
        {"method": "userrecaptcha", "params": {"googlekey": f"SITEKEY_{i}", "pageurl": f"https://example.com/page{i}"}}
        for i in range(10)
    ]

    results = await queue.solve_batch(tasks)
    for i, result in enumerate(results):
        if isinstance(result, Exception):
            print(f"Task {i}: ERROR — {result}")
        else:
            print(f"Task {i}: {result[:50]}...")


asyncio.run(main())

solve_batch() menjalankan semua task lewat asyncio.gather(), dan Semaphore(max_concurrent) memastikan Anda tidak mengirim lebih banyak request paralel daripada yang sanggup ditangani paket CaptchaAI Anda. Sesuaikan max_concurrent dengan jumlah thread di paket Anda — lihat bagian FAQ untuk pemetaannya ke tier BASIC sampai VIP-3.

Langkah 3: Producer-Consumer untuk Scraping yang Jalan Terus-Menerus

Kalau daftar CAPTCHA tidak Anda ketahui semuanya di awal — misalnya scraper yang menemukan halaman baru sambil berjalan — pola batch di atas kurang pas. Producer-consumer memisahkan "menemukan task" dari "menyelesaikan task": satu coroutine terus menambah task ke antrian (producer), beberapa coroutine consumer mengambil dan menyelesaikannya paralel.

import asyncio
import aiohttp

API_KEY = "YOUR_API_KEY"


class ProducerConsumerQueue:
    """Continuous CAPTCHA solving with producer-consumer pattern."""

    def __init__(self, api_key, queue_size=100, num_consumers=5):
        self.api_key = api_key
        self.queue = asyncio.Queue(maxsize=queue_size)
        self.num_consumers = num_consumers
        self.solved_count = 0
        self.error_count = 0
        self.running = True

    async def produce(self, tasks):
        """Producer: feed CAPTCHA tasks into the queue."""
        for task in tasks:
            await self.queue.put(task)
        # Signal consumers to stop
        for _ in range(self.num_consumers):
            await self.queue.put(None)

    async def consume(self, result_handler):
        """Consumer: solve CAPTCHAs and call result handler."""
        async with aiohttp.ClientSession() as session:
            while True:
                task = await self.queue.get()
                if task is None:
                    self.queue.task_done()
                    break

                try:
                    result = await self._solve(session, task["method"], **task["params"])
                    self.solved_count += 1
                    if result_handler:
                        await result_handler(task, result)
                except Exception as e:
                    self.error_count += 1
                    print(f"Error: {e}")
                finally:
                    self.queue.task_done()

    async def run(self, tasks, result_handler=None):
        """Run the producer-consumer pipeline."""
        # Start producer
        producer = asyncio.create_task(self.produce(tasks))

        # Start consumers
        consumers = [
            asyncio.create_task(self.consume(result_handler))
            for _ in range(self.num_consumers)
        ]

        # Wait for everything to finish
        await producer
        await asyncio.gather(*consumers)

        print(f"Complete: {self.solved_count} solved, {self.error_count} errors")

    async def _solve(self, session, method, **params):
        async with session.post("https://ocr.captchaai.com/in.php", data={
            "key": self.api_key, "method": method, "json": 1, **params,
        }) as resp:
            data = await resp.json(content_type=None)
            if data.get("status") != 1:
                raise Exception(f"Submit: {data.get('request')}")
            task_id = data["request"]

        for _ in range(30):
            await asyncio.sleep(5)
            async with session.get("https://ocr.captchaai.com/res.php", params={
                "key": self.api_key, "action": "get", "id": task_id, "json": 1,
            }) as resp:
                result = await resp.json(content_type=None)
                if result.get("status") == 1:
                    return result["request"]
        raise TimeoutError("Timed out")


# Usage
async def handle_result(task, token):
    url = task["params"]["pageurl"]
    print(f"Solved for {url}: {token[:30]}...")


async def main():
    queue = ProducerConsumerQueue(API_KEY, num_consumers=5)

    tasks = [
        {"method": "userrecaptcha", "params": {"googlekey": f"SITEKEY_{i}", "pageurl": f"https://example.com/page{i}"}}
        for i in range(20)
    ]

    await queue.run(tasks, result_handler=handle_result)


asyncio.run(main())

produce() mengisi antrian sampai habis, lalu mengirim sinyal None sebanyak jumlah consumer supaya tiap consumer tahu kapan harus berhenti — pola sentinel yang umum di antrian asyncio. consume() memanggil result_handler tiap kali satu CAPTCHA selesai, jadi Anda bisa langsung memakai token-nya tanpa menunggu seluruh batch beres.

Langkah 4: Priority Queue Saat Tidak Semua CAPTCHA Sama Pentingnya

Tidak semua CAPTCHA punya bobot yang sama — task di halaman checkout biasanya lebih mendesak dibanding halaman listing produk yang cuma untuk pemantauan harga. PriorityCaptchaQueue memakai asyncio.PriorityQueue supaya task dengan angka priority lebih kecil diproses lebih dulu.

import asyncio
from dataclasses import dataclass, field

API_KEY = "YOUR_API_KEY"


@dataclass(order=True)
class PriorityTask:
    priority: int
    task: dict = field(compare=False)


class PriorityCaptchaQueue:
    """CAPTCHA queue with priority levels."""

    def __init__(self, api_key, num_workers=5):
        self.api_key = api_key
        self.queue = asyncio.PriorityQueue()
        self.num_workers = num_workers
        self.results = {}

    async def submit(self, task_id, method, priority=5, **params):
        """Submit with priority (lower number = higher priority)."""
        await self.queue.put(PriorityTask(
            priority=priority,
            task={"id": task_id, "method": method, "params": params},
        ))

    async def process(self):
        """Process all queued tasks by priority."""
        workers = [asyncio.create_task(self._worker()) for _ in range(self.num_workers)]

        # Wait for queue to drain
        await self.queue.join()

        # Cancel workers
        for w in workers:
            w.cancel()

        return self.results

    async def _worker(self):
        import aiohttp
        async with aiohttp.ClientSession() as session:
            while True:
                item = await self.queue.get()
                task = item.task
                try:
                    result = await self._solve(session, task["method"], **task["params"])
                    self.results[task["id"]] = {"status": "solved", "token": result}
                except Exception as e:
                    self.results[task["id"]] = {"status": "error", "error": str(e)}
                finally:
                    self.queue.task_done()

    async def _solve(self, session, method, **params):
        import aiohttp
        async with session.post("https://ocr.captchaai.com/in.php", data={
            "key": self.api_key, "method": method, "json": 1, **params,
        }) as resp:
            data = await resp.json(content_type=None)
            if data.get("status") != 1:
                raise Exception(data.get("request"))
            task_id = data["request"]

        for _ in range(30):
            await asyncio.sleep(5)
            async with session.get("https://ocr.captchaai.com/res.php", params={
                "key": self.api_key, "action": "get", "id": task_id, "json": 1,
            }) as resp:
                result = await resp.json(content_type=None)
                if result.get("status") == 1:
                    return result["request"]
        raise TimeoutError()


# Usage
async def main():
    pq = PriorityCaptchaQueue(API_KEY, num_workers=3)

    # High priority — checkout pages
    await pq.submit("checkout_1", "turnstile", priority=1, sitekey="KEY", pageurl="https://shop.com/checkout")

    # Normal priority — product pages
    for i in range(5):
        await pq.submit(f"product_{i}", "userrecaptcha", priority=5, googlekey="KEY", pageurl=f"https://shop.com/p/{i}")

    # Low priority — info pages
    for i in range(3):
        await pq.submit(f"info_{i}", "userrecaptcha", priority=10, googlekey="KEY", pageurl=f"https://shop.com/info/{i}")

    results = await pq.process()
    for task_id, result in results.items():
        print(f"{task_id}: {result['status']}")


asyncio.run(main())

Contoh di atas memberi priority 1 untuk checkout, 5 untuk halaman produk, dan 10 untuk halaman info — semakin kecil angkanya, semakin dulu diproses. Berguna kalau tim Anda mengelola beberapa job scraping sekaligus dan tidak mau task penting menunggu di belakang task berprioritas rendah.

Langkah 5: Pantau Throughput dan Success Rate

Antrian yang jalan tanpa metrik itu kotak hitam — Anda baru tahu ada masalah setelah task menumpuk. QueueMetrics di bawah menghitung success rate dan throughput per menit secara real-time, supaya Anda tahu kapan menambah worker atau justru menguranginya.

import time
from dataclasses import dataclass, field


@dataclass
class QueueMetrics:
    submitted: int = 0
    solved: int = 0
    failed: int = 0
    total_solve_time: float = 0.0
    start_time: float = field(default_factory=time.time)

    @property
    def avg_solve_time(self):
        return self.total_solve_time / self.solved if self.solved else 0

    @property
    def success_rate(self):
        total = self.solved + self.failed
        return (self.solved / total * 100) if total else 0

    @property
    def throughput(self):
        elapsed = time.time() - self.start_time
        return self.solved / elapsed * 60 if elapsed > 0 else 0

    def report(self):
        return (
            f"Submitted: {self.submitted} | "
            f"Solved: {self.solved} | "
            f"Failed: {self.failed} | "
            f"Avg time: {self.avg_solve_time:.1f}s | "
            f"Success: {self.success_rate:.1f}% | "
            f"Throughput: {self.throughput:.0f}/min"
        )

Panggil report() secara berkala dan log hasilnya. Kalau success_rate turun drastis, curigai dulu perubahan sitekey atau tipe CAPTCHA di halaman target sebelum menambah retry; retry agresif untuk error permanen cuma membuang thread yang bisa dipakai task lain.

Masalah yang Sering Muncul dan Cara Mengatasinya

Pertanyaan Seputar Antrian CAPTCHA di Python

Apakah semua jenis CAPTCHA bisa masuk ke antrian yang sama?

Bisa — tiap task membawa method-nya sendiri, jadi satu antrian bisa mencampur reCAPTCHA v2, Turnstile, dan GeeTest v3 sekaligus. reCAPTCHA v2/v3 (termasuk Enterprise), Turnstile, Cloudflare Challenge, GeeTest v3, gambar/OCR, grid, dan BLS captcha didukung penuh; CaptchaFox, Friendly Captcha, dan Lemin dalam status beta. hCaptcha dan FunCaptcha (Arkose Labs) belum didukung, dan GeeTest v4 masih "segera hadir" — task untuk tipe itu akan selalu gagal berapa pun retry-nya.

Threading atau asyncio, mana yang lebih cocok untuk pemula?

Kalau proyek Anda baru, mulai dari asyncio — untuk beban I/O-bound seperti solve CAPTCHA, satu event loop lebih hemat resource. Pakai threading kalau menambahkan antrian ke codebase lama yang sudah synchronous.

Berapa banyak thread yang sebaiknya saya jalankan bersamaan?

Ini terkait ke paket CaptchaAI Anda, karena penagihannya berbasis thread concurrent, bukan per solve. Paket BASIC ($15/bulan, 5 thread) cukup untuk uji coba dengan max_workers=5. Untuk workload harian yang lebih besar, tim price-monitoring dan scraping lepas biasanya naik ke ADVANCE ($90/bulan, 50 thread) — solve per thread tidak dibatasi, jadi biaya tetap flat berapa pun volume solve-nya.

Bagaimana saya tahu antrian sedang stuck, bukan sekadar lambat?

Pantau success_rate dan throughput dari QueueMetrics. Kalau throughput mendekati nol sementara submitted terus naik, task menumpuk tanpa diproses — biasanya worker mati diam-diam atau ada await yang hilang. Cek juga log untuk ERROR_NO_SLOT_AVAILABLE yang berulang, tanda paket Anda mentok di batas thread.

Apakah aman menjalankan antrian ini untuk scraping yang menyentuh data pribadi?

Antrian sendiri hanya menangani CAPTCHA, tapi kalau task Anda ikut mengumpulkan data pribadi, tetap perhatikan UU Pelindungan Data Pribadi (UU 27/2022) dan UU ITE — scrape hanya data yang memang berwenang Anda proses.

Kesimpulan: Pola Mana yang Harus Anda Pakai

Antrian CAPTCHA memisahkan pengiriman dari polling supaya solve berjalan paralel lewat CaptchaAI. Pakai threading untuk kode synchronous, asyncio untuk proyek baru, producer-consumer untuk scraping berkelanjutan, dan priority queue saat ada task yang jelas lebih mendesak. Mulai dari max_workers=5, ukur lewat QueueMetrics, baru naikkan sesuai jumlah thread di paket Anda.

Panduan Terkait

• Antrian Pemecahan CAPTCHA di Node.js
• Membangun Pipeline CAPTCHA Klien CaptchaAI
• Panduan Lengkap Python Playwright + CaptchaAI