Cara Memecahkan CAPTCHA Secara Paralel dengan Python asyncio

Jika Anda perlu memecahkan captcha secara paralel dari Python, gunakan AsyncCapSkip dengan asyncio.gather. Python adalah satu-satunya SDK CapSkip yang klien async-nya merupakan implementasi async sungguhan, bukan sekadar alias, sehingga satu batch berisi sepuluh reCAPTCHA selesai kira-kira dalam waktu yang paling lambat saja alih-alih total kesepuluhnya. Panduan ini membahas kode yang berfungsi, cara membatasi konkurensi agar Anda tidak membanjiri pemecah, dan cara mencegah satu kegagalan mematikan seluruh batch.

Mengapa klien Python adalah yang paling menarik

Keempat SDK mengekspor sesuatu bernama AsyncCapSkip. Hanya salah satunya yang merupakan implementasi terpisah.

SDK	Apa AsyncCapSkip adalah	Cara Anda menjalankan pekerjaan secara bersamaan
Python	Klien async yang sesungguhnya	await asyncio.gather(...)
Node.js	Alias dari CapSkip	await Promise.all([...])
.NET	Alias dari CapSkipClient	await Task.WhenAll(...)
PHP	Hanya alias, demi keselarasan kode sumber	Sinkron, tanpa konkurensi

Node dan .NET sudah non-blocking, jadi alias itu tidak berbiaya di sana. PHP bersifat sinkron dan alias itu sama sekali tidak berguna. Di Python perbedaannya berarti: client biasa CapSkip memblokir event loop saat melakukan polling, jadi menaruhnya di dalam coroutine memberi Anda konkurensi di atas kertas dan waktu berurutan dalam praktik.

Apa yang Anda butuhkan

• Python 3.10 atau lebih baru
• pip install capskip
• CapSkip berjalan dengan API server-nya aktif, mendengarkan di 127.0.0.1:8080
• Daftar sitekey dan URL halaman yang akan diproses

Tidak ada yang keluar dari komputer Anda, jadi tidak ada batas laju yang perlu dinegosiasikan dan tidak ada meteran per pemecahan yang berjalan saat Anda bereksperimen. Tanda tangan metode lengkap untuk setiap tipe ada di SDK pemecahan CAPTCHA halaman.

Versi sekuensial, dan berapa biayanya

Inilah bentuk yang biasanya dipakai orang untuk memulai. Kodenya benar, dan lambat.

# pip install capskip
from capskip import CapSkip

solver = CapSkip(host="127.0.0.1", port=8080)

targets = [
    ("SITEKEY_A", "https://example.com/page-a"),
    ("SITEKEY_B", "https://example.com/page-b"),
    ("SITEKEY_C", "https://example.com/page-c"),
]

# Each call blocks until that one CAPTCHA comes back.
for sitekey, url in targets:
    result = solver.recaptcha(sitekey=sitekey, url=url)
    print(result["code"][:40])   # the token, truncated for the log

Pemecahan reCAPTCHA sebagian besar adalah menunggu. Proses Anda menganggur sementara pemecah bekerja, lalu berpindah ke berikutnya dan menganggur lagi. Tiga pemecahan memakan waktu nyata setara tiga pemecahan. Tiga puluh memakan tiga puluh.

Pecahkan beberapa CAPTCHA sekaligus dengan asyncio.gather

Ganti kliennya, await panggilannya, lalu serahkan semuanya ke gather. Jenisnya boleh dicampur: reCAPTCHA, Turnstile, dan GeeTest dalam satu batch tidak masalah.

# pip install capskip
import asyncio
from capskip import AsyncCapSkip

async def main():
    solver = AsyncCapSkip(host="127.0.0.1", port=8080)

    # gather starts all three now and waits for the slowest.
    results = await asyncio.gather(
        solver.recaptcha(sitekey="SITEKEY_A", url="https://example.com/page-a"),
        solver.turnstile(sitekey="SITEKEY_B", url="https://example.com/page-b"),
        solver.normal("captcha.png"),
    )

    for r in results:
        print(r["code"][:40])   # token for widgets, text for images

asyncio.run(main())

Setiap metode mengembalikan dict dengan bidang inti yang sama: captchaId dan code. Turnstile juga mengembalikan userAgent, yang harus Anda kirim kembali bersama token ketika Anda mengirimkan pemecahan halaman challenge. GeeTest menambahkan challenge, validate dan seccode, dan menaruh JSON mentah di code.

Batasi konkurensi dengan semaphore

Jangan menembakkan dua ratus pemecahan ke daemon lokal lalu berharap semuanya beres. Pemecahan adalah pekerjaan CPU yang terjadi di komputer Anda sendiri, jadi melewati lebar tertentu Anda hanya mengantre melawan diri sendiri dan setiap pemecahan justru makin lambat. Semaphore menjaga jumlah tetap yang berjalan bersamaan.

import asyncio
from capskip import AsyncCapSkip

# Start at 4 or 5, then measure. More is not automatically faster.
sem = asyncio.Semaphore(5)

async def solve_one(solver, sitekey, url):
    async with sem:
        return await solver.recaptcha(sitekey=sitekey, url=url)

async def run(targets):
    solver = AsyncCapSkip()
    tasks = [solve_one(solver, k, u) for k, u in targets]
    return await asyncio.gather(*tasks)

Tentukan angkanya dengan mengukur waktu, bukan dengan menebak. Jalankan 20 target yang sama pada 2, 5, dan 10 lalu pertahankan mana pun yang selesai paling cepat di perangkat keras Anda. Jawaban yang tepat bergantung pada CPU Anda, bukan pada SDK.

Satu klien, bukan satu per tugas

Buat satu AsyncCapSkip dan bagikan ke seluruh coroutine, seperti di atas. Membuat satu per task itu boros dan tidak memberi keuntungan apa pun: client menyimpan konfigurasi, bukan state per pemecahan.

Cegah satu kegagalan menghancurkan seluruh batch

Secara default gather meneruskan exception pertama dan Anda kehilangan hasil dari semua hal lain yang sedang berjalan. Teruskan return_exceptions=True dan exception akan datang sebagai item biasa dalam daftar hasil, sehingga Anda dapat memisahkan yang berhasil dari yang gagal.

from capskip import (
    AsyncCapSkip, ApiException, NetworkException,
    TimeoutException, ValidationException,
)

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

for target, r in zip(targets, results):
    if isinstance(r, TimeoutException):
        print("timed out, worth retrying:", target)
    elif isinstance(r, ApiException):
        print("api rejected this one:", target, r)
    elif isinstance(r, NetworkException):
        print("solver unreachable, stop the run:", target)
    elif isinstance(r, Exception):
        raise r
    else:
        print("ok:", r["code"][:40])

Keempat tipe exception ini sama di setiap CapSkip SDK, dan semuanya diturunkan dari kelas dasar CapSkipError jika Anda lebih suka menangkap satu jenis saja. ValidationException berarti parameter Anda salah dan percobaan ulang akan gagal dengan cara yang sama persis. NetworkException biasanya berarti aplikasinya tidak berjalan, yang merupakan masalah seluruh proses, bukan masalah per target.

Timeout dan polling, yang berperilaku berbeda untuk tiap jenis

Dua timeout terpisah berlaku, dan batch yang mencampur beberapa jenis tunduk pada keduanya.

Opsi	Bawaan	Berlaku untuk
defaultTimeout	120 detik	CAPTCHA gambar
recaptchaTimeout	300 detik	reCAPTCHA, Turnstile, GeeTest
pollingInterval	5 detik	The maksimum jeda antar polling

pollingInterval perlu dipahami sebelum Anda menyetelnya. SDK tidak melakukan polling pada interval tetap. Ia mulai dari 250ms dan melambat menuju nilai yang Anda tetapkan, itulah sebabnya pemecahan lewat SDK biasanya kembali lebih cepat daripada perulangan buatan sendiri yang disusun dari saran API mentah “tunggu, lalu polling setiap lima detik”. Menaikkannya membuat pemecahan yang cepat tiba lebih lambat. Menurunkannya menambah volume permintaan tanpa manfaat.

Challenge GeeTest kedaluwarsa, jadi jangan menyiapkan batch di awal

Yang ini terasa khususnya ketika Anda beralih ke mode paralel. Sebuah GeeTest gt nilainya statis per situs, tetapi challenge bersifat sekali pakai dan kedaluwarsa dalam sekitar satu menit. Jika Anda mengumpulkan lima puluh tantangan terlebih dahulu lalu mulai memecahkannya, yang berada di belakang antrean sudah mati sebelum sempat dikirimkan. Ambil setiap tantangan tepat sebelum pemecahan yang menggunakannya.

Contoh lengkap yang berfungsi

# pip install capskip
import asyncio
from capskip import AsyncCapSkip, ApiException, TimeoutException

TARGETS = [
    ("SITEKEY_A", "https://example.com/page-a"),
    ("SITEKEY_B", "https://example.com/page-b"),
    ("SITEKEY_C", "https://example.com/page-c"),
]

async def solve_one(solver, sem, sitekey, url):
    async with sem:
        return await solver.recaptcha(sitekey=sitekey, url=url)

async def main():
    solver = AsyncCapSkip(host="127.0.0.1", port=8080)
    sem = asyncio.Semaphore(5)

    tasks = [solve_one(solver, sem, k, u) for k, u in TARGETS]
    results = await asyncio.gather(*tasks, return_exceptions=True)

    tokens = {}
    for (sitekey, url), r in zip(TARGETS, results):
        if isinstance(r, (ApiException, TimeoutException)):
            print("failed:", url, r)
        else:
            tokens[url] = r["code"]

    print(len(tokens), "of", len(TARGETS), "solved")
    return tokens

asyncio.run(main())

Itulah keseluruhan polanya: satu klien bersama, sebuah semaphore, return_exceptions=True, dan sebuah dict berisi token di akhir. Masukkan ke dalam scraper dan langkah CAPTCHA berhenti menjadi hambatan. Halaman Pemecah CAPTCHA untuk web scraping menjelaskan di mana posisinya dalam pipeline yang lebih luas.

Ide yang sama di SDK lainnya

Jika Anda memindahkan ini, primitif konkurensinya berubah tetapi bentuknya tidak. Node sudah non-blocking, jadi client biasa sudah cukup.

// npm install capskip
const { CapSkip } = require('capskip');

const solver = new CapSkip({ host: '127.0.0.1', port: 8080 });

// AsyncCapSkip here is just an alias. Promise.all does the work.
const results = await Promise.all([
  solver.recaptcha('SITEKEY_A', 'https://example.com/page-a'),
  solver.turnstile('SITEKEY_B', 'https://example.com/page-b'),
]);

console.log(results.map(r => r.code));

.NET juga sama halnya dengan Task.WhenAll, dan PHP sama sekali tidak menawarkan konkurensi. Jika Anda butuh pemecahan paralel dan bebas memilih bahasa, Python adalah bahasa dengan client yang dirancang khusus. Halaman pemecah CAPTCHA Python halaman memuat sisa cakupannya.

Kesalahan umum

Kesalahan	Apa yang terjadi	Perbaiki
Menggunakan CapSkip di dalam coroutine	Memblokir event loop, sehingga waktunya tetap berurutan	Gunakan AsyncCapSkip
Tanpa semaphore	Setiap pemecahan melambat begitu antrean menumpuk	Batasi pekerjaan yang sedang berjalan, mulai dari sekitar 5
Biasa gather	Satu kegagalan membuang semua hasil lainnya	return_exceptions=True
Mengambil challenge GeeTest lebih awal	Yang belakangan kedaluwarsa sebelum dikirim	Ambil masing-masing tepat sebelum memecahkan
Menaikkan pollingInterval	Pemecahan cepat justru kembali lebih lambat, bukan lebih cepat	Biarkan pada nilai bawaan
Proxy diatur pada pemecahan gambar	Tidak didukung untuk CAPTCHA gambar	Proxy hanya berlaku untuk reCAPTCHA, Turnstile dan GeeTest

Permintaan dan respons mentah untuk setiap jenis, jika Anda ingin melihat apa yang dikirim SDK, ada di dokumentasi API. Lihat dokumentasi resmi Python: referensi task asyncio mencakup gather semantiknya secara rinci.

Pertanyaan yang sering diajukan

Berapa banyak CAPTCHA yang bisa saya pecahkan sekaligus?

Tidak ada kuota yang bisa tercapai, jadi batasnya adalah perangkat keras Anda sendiri. Pemecahan terjadi secara lokal, sehingga konkurensi dibatasi oleh CPU, bukan oleh tingkatan akun. Mulailah dengan lima yang berjalan bersamaan, ukur waktu satu batch tetap, lalu sesuaikan dari sana.

Bisakah saya mencampur jenis CAPTCHA dalam satu panggilan gather?

Ya. recaptcha, turnstile, geetest dan normal semuanya adalah coroutine pada client yang sama dan dapat di-await bersamaan. Ingat bahwa pemecahan gambar memakai timeout 120 detik sementara yang lain memakai 300.

Haruskah saya menggunakan thread saja?

Hanya jika kode di sekitarnya memang sudah menggunakan thread. Pekerjaannya adalah menunggu yang terikat I/O, dan justru itulah gunanya asyncio, lagi pula satu event loop lebih murah daripada thread pool. Jika Anda terjebak pada basis kode sinkron, thread pool di sekitar CapSkip biasa juga bisa dipakai.

Apakah AsyncCapSkip perlu ditutup?

SDK tidak mendokumentasikan metode close atau async context manager, jadi buat satu client, pakai untuk seluruh proses, dan biarkan keluar dari scope saat proses berakhir.

Ringkasan

Gunakan AsyncCapSkip, gunakan satu klien bersama, batasi pemecahan yang sedang berjalan dengan semaphore, dan teruskan return_exceptions=True sehingga satu target yang buruk tidak membatalkan seluruh batch. Itu mengubah antrean CAPTCHA dari hambatan serial menjadi satu kali menunggu.

Alasan Anda dapat memperlebar konkurensi dengan bebas adalah karena pemecahnya berjalan di komputer Anda sendiri. Tidak ada tagihan per pemecahan dan tidak ada antrean bersama yang harus dibagi dengan orang asing, jadi penskalaan adalah soal CPU Anda, bukan batas laju milik orang lain. Itulah perbedaan praktis yang dihadirkan oleh sebuah bypass captcha yang dibuat alat tersebut begitu ukuran batch Anda tidak lagi kecil.