aiohttp + CaptchaAI: Pemecahan CAPTCHA Asinkron

Kalau Anda harus menyelesaikan puluhan CAPTCHA dari satu skrip Python, jawabannya bukan menambah thread OS atau menjalankan lima proses—melainkan menjalankan I/O secara asinkron. Setiap solve menghabiskan sebagian besar waktunya hanya menunggu jawaban dari server, dan di sanalah aiohttp unggul: satu event loop bisa mengelola ratusan permintaan yang sedang menunggu tanpa saling memblokir. Digabung dengan CaptchaAI, satu skrip bisa mengirim banyak task sekaligus dan mengumpulkan token begitu masing-masing selesai.

Untuk beban scraping bervolume tinggi—kasus yang umum di kalangan freelancer dan tim data di Indonesia—pola ini jauh lebih hemat sumber daya daripada menjalankan puluhan proses paralel. Anda tidak membayar overhead thread OS, dan biaya CaptchaAI dihitung per thread yang berjalan bersamaan, bukan per solve, jadi konkurensi tinggi tidak menaikkan tagihan selama masih dalam batas thread paket Anda.

Alur intinya tetap empat langkah yang sama seperti integrasi CaptchaAI lainnya: kirim task ke in.php → simpan task ID → polling res.php → pakai token. Yang berbeda hanyalah keempat langkah itu berjalan secara async, sehingga puluhan solve bisa hidup berdampingan dalam satu loop.

Persyaratan

pip install aiohttp

Klien async CaptchaAI

Kelas berikut membungkus keempat langkah tadi. Metode submit mengirim task dan mengembalikan task ID, poll menunggu hasil dengan batas waktu, sedangkan solve menyatukan keduanya. Semua permintaan memakai session.get yang non-blocking, jadi selama satu task menunggu di res.php, task lain tetap jalan.

import aiohttp
import asyncio


class AsyncCaptchaAI:
    def __init__(self, api_key):
        self.api_key = api_key
        self.base_url = "https://ocr.captchaai.com"

    async def submit(self, session, params):
        """Submit a CAPTCHA task and return the task ID."""
        params["key"] = self.api_key
        async with session.get(
            f"{self.base_url}/in.php", params=params
        ) as resp:
            text = await resp.text()

        if not text.startswith("OK|"):
            raise Exception(f"Submit failed: {text}")

        return text.split("|")[1]

    async def poll(self, session, task_id, timeout=300):
        """Poll for the result with a timeout."""
        params = {
            "key": self.api_key,
            "action": "get",
            "id": task_id,
        }
        deadline = asyncio.get_event_loop().time() + timeout

        while asyncio.get_event_loop().time() < deadline:
            await asyncio.sleep(5)

            async with session.get(
                f"{self.base_url}/res.php", params=params
            ) as resp:
                text = await resp.text()

            if text == "CAPCHA_NOT_READY":
                continue
            if text.startswith("OK|"):
                return text.split("|", 1)[1]
            raise Exception(f"Solve failed: {text}")

        raise TimeoutError(f"Task {task_id} timed out after {timeout}s")

    async def solve(self, session, params, timeout=300):
        """Submit and poll in one call."""
        task_id = await self.submit(session, params)
        return await self.poll(session, task_id, timeout)

    async def get_balance(self, session):
        """Check account balance."""
        params = {"key": self.api_key, "action": "getbalance"}
        async with session.get(
            f"{self.base_url}/res.php", params=params
        ) as resp:
            return float(await resp.text())

Menyelesaikan satu CAPTCHA

Mulai dari kasus paling sederhana: cek saldo, lalu selesaikan satu reCAPTCHA v2. Perhatikan await sebelum tiap pemanggilan—itulah yang menyerahkan kendali ke event loop saat permintaan sedang menunggu.

import asyncio
import os

async def main():
    solver = AsyncCaptchaAI(os.environ["CAPTCHAAI_API_KEY"])

    async with aiohttp.ClientSession() as session:
        # Check balance
        balance = await solver.get_balance(session)
        print(f"Balance: ${balance:.2f}")

        # Solve reCAPTCHA v2
        token = await solver.solve(session, {
            "method": "userrecaptcha",
            "googlekey": "6Le-wvkS...",
            "pageurl": "https://example.com",
        })
        print(f"Token: {token[:50]}...")

asyncio.run(main())

Menyelesaikan banyak CAPTCHA sekaligus

Di sinilah async membuahkan hasil. asyncio.gather menjalankan semua solve secara bersamaan dan mengembalikan hasilnya dalam urutan yang sama. Dengan return_exceptions=True, satu task yang gagal tidak akan menjatuhkan seluruh batch—Anda tetap mendapat hasil task yang lain.

async def solve_batch(urls, site_key):
    solver = AsyncCaptchaAI(os.environ["CAPTCHAAI_API_KEY"])

    async with aiohttp.ClientSession() as session:
        tasks = [
            solver.solve(session, {
                "method": "userrecaptcha",
                "googlekey": site_key,
                "pageurl": url,
            })
            for url in urls
        ]

        results = await asyncio.gather(*tasks, return_exceptions=True)

        for url, result in zip(urls, results):
            if isinstance(result, Exception):
                print(f"FAILED {url}: {result}")
            else:
                print(f"SOLVED {url}: {len(result)} chars")

        return results


urls = [
    "https://example.com/page1",
    "https://example.com/page2",
    "https://example.com/page3",
    "https://example.com/page4",
    "https://example.com/page5",
]
asyncio.run(solve_batch(urls, "6Le-wvkS..."))

Scraping dengan penanganan CAPTCHA

Pola yang paling sering dipakai di pekerjaan scraping: ambil halaman dulu, dan hanya panggil solver jika CAPTCHA benar-benar muncul. Kalau tidak ada penanda g-recaptcha, kembalikan konten langsung dan hemat satu solve.

async def scrape_with_captcha(url, site_key):
    solver = AsyncCaptchaAI(os.environ["CAPTCHAAI_API_KEY"])

    async with aiohttp.ClientSession() as session:
        # Fetch the page
        async with session.get(url) as resp:
            html = await resp.text()

        # Check if page has a CAPTCHA
        if "g-recaptcha" not in html:
            return html  # No CAPTCHA, return content

        # Solve the CAPTCHA
        token = await solver.solve(session, {
            "method": "userrecaptcha",
            "googlekey": site_key,
            "pageurl": url,
        })

        # Submit with solved token
        async with session.post(url, data={
            "g-recaptcha-response": token,
        }) as resp:
            return await resp.text()

Semaphore untuk mengatur konkurensi

Konkurensi Anda tidak dibatasi kode, melainkan jumlah thread pada paket CaptchaAI—satu thread menangani satu solve yang sedang berjalan. Paket ADVANCE ($90/bulan, 50 thread) memungkinkan hingga 50 solve berjalan bersamaan; STANDARD ($30/bulan, 15 thread) untuk beban lebih kecil. Selaraskan max_concurrent dengan angka itu supaya Anda tidak mengantre task lebih banyak daripada yang bisa diproses akun.

async def solve_with_limit(urls, site_key, max_concurrent=10):
    solver = AsyncCaptchaAI(os.environ["CAPTCHAAI_API_KEY"])
    semaphore = asyncio.Semaphore(max_concurrent)

    async def solve_one(session, url):
        async with semaphore:
            return await solver.solve(session, {
                "method": "userrecaptcha",
                "googlekey": site_key,
                "pageurl": url,
            })

    async with aiohttp.ClientSession() as session:
        tasks = [solve_one(session, url) for url in urls]
        results = await asyncio.gather(*tasks, return_exceptions=True)

    solved = sum(1 for r in results if not isinstance(r, Exception))
    print(f"Solved {solved}/{len(urls)} CAPTCHAs")
    return results

Contoh Turnstile

Pola yang sama berlaku untuk Cloudflare Turnstile—cukup ganti method menjadi turnstile dan gunakan sitekey. CaptchaAI juga menyelesaikan reCAPTCHA v3, GeeTest v3, serta CAPTCHA gambar/OCR lewat klien async yang sama.

async def solve_turnstile(url, sitekey):
    solver = AsyncCaptchaAI(os.environ["CAPTCHAAI_API_KEY"])

    async with aiohttp.ClientSession() as session:
        token = await solver.solve(session, {
            "method": "turnstile",
            "sitekey": sitekey,
            "pageurl": url,
        })
        return token

Mengatur timeout dan retry

Batas waktu default 300 detik sudah aman untuk sebagian besar tipe, tetapi latensi jaringan ikut berpengaruh—terutama jika worker Anda deploy di ap-southeast-1 (Singapura) atau ap-southeast-3 (Jakarta) dan diakses lewat koneksi mobile. Interval polling di kelas ini disetel lima detik lewat asyncio.sleep(5); nilai itu wajar untuk kebanyakan tipe, dan memperpendeknya hanya menambah trafik res.php tanpa mempercepat hasil. Untuk batch besar, naikkan timeout sedikit dan jangan langsung mengulang task yang gagal; beri jeda yang meningkat eksponensial (exponential backoff) agar tidak menambah beban saat server sedang sibuk. Karena poll sudah menaikkan TimeoutError sendiri, cukup tangkap error itu di level batch dan jadwalkan ulang hanya URL yang benar-benar gagal.

Satu catatan kepatuhan untuk pekerjaan scraping di Indonesia: ambil hanya data yang memang boleh Anda proses dan hindari data pribadi, sejalan dengan UU Pelindungan Data Pribadi (UU 27/2022). Contoh di sini memakai domain example.com sebagai placeholder yang netral.

Pemecahan masalah

Pertanyaan umum

Apakah aiohttp memblokir event loop saat menunggu hasil?

Tidak. Selama sebuah task menunggu di await session.get(...), event loop bebas menjalankan task lain. Itulah alasan ratusan solve bisa "menunggu" bersamaan tanpa memakan satu thread OS per task.

Berapa jumlah solve paralel yang sebaiknya saya jalankan?

Sesuaikan dengan jumlah thread paket Anda—misalnya 50 untuk ADVANCE. Setel nilai max_concurrent pada semaphore ke angka itu; menaikkannya lebih jauh hanya menambah task yang mengantre, bukan mempercepat.

Apakah CaptchaAI mendukung Turnstile dan reCAPTCHA v3 lewat cara yang sama?

Ya. Cukup ganti nilai method (turnstile, userrecaptcha, geetest, dan seterusnya); struktur submit/poll tetap sama. Perlu diketahui, hCaptcha dan FunCaptcha belum didukung.

Bagaimana menangani satu task yang timeout tanpa menggagalkan seluruh batch?

Pakai return_exceptions=True pada asyncio.gather, lalu periksa hasilnya satu per satu. Task yang mengembalikan Exception bisa dijadwalkan ulang, sementara token yang sudah berhasil tetap Anda simpan.

Panduan terkait

• Integrasi HTTPX + CaptchaAI
• Solve paralel vs berurutan: mana yang lebih efisien
• Integrasi Scrapy + CaptchaAI