Python asyncio ile CAPTCHA'lar Paralel Olarak Nasıl Çözülür

Python'dan paralel olarak CAPTCHA çözmeniz gerekiyorsa şunu kullanın: AsyncCapSkip ile asyncio.gather. Python, async istemcinin bir takma ad değil gerçek bir asenkron uygulama olduğu tek CapSkip SDK'sıdır; bu yüzden on reCAPTCHA'lık bir küme, onunun toplamı kadar değil, yaklaşık en yavaş olanı kadar sürede biter. Bu rehber çalışan kodu, çözücüyü boğmamak için eşzamanlılığı nasıl sınırlayacağınızı ve tek bir hatanın tüm kümeyi öldürmesini nasıl engelleyeceğinizi anlatıyor.

Python istemcisi neden ilginç olan

Dört SDK da şu adla bir şey dışa aktarır: AsyncCapSkip. Ancak yalnızca birinde bu ayrı bir uygulamadır.

SDK	Nedir: AsyncCapSkip şudur	İşleri eşzamanlı olarak nasıl çalıştırırsınız
Python	Gerçek bir asenkron istemci	await asyncio.gather(...)
Node.js	Şunun takma adı: CapSkip	await Promise.all([...])
.NET	Şunun takma adı: CapSkipClient	await Task.WhenAll(...)
PHP	Yalnızca takma ad, kaynak kodu tutarlılığı için	Senkron, eşzamanlılık yok

Node ve .NET zaten bloklamıyor, dolayısıyla takma adın orada bir maliyeti yok. PHP senkrondur ve takma ad orada hiçbir şey kazandırmaz. Python'da ise bu ayrım önemlidir: sade CapSkip istemci yoklama yaparken olay döngüsünü bloklar, bu yüzden onu bir coroutine içine koymak size kâğıt üzerinde eşzamanlılık, pratikte ise ardışık süreler verir.

Neye ihtiyacınız var

• Python 3.10 veya üzeri
• pip install capskip
• API sunucusu açık şekilde çalışan ve şurayı dinleyen CapSkip: 127.0.0.1:8080
• İşlenecek sitekey ve sayfa URL'lerinden oluşan bir liste

Hiçbir şey makinenizden çıkmaz; dolayısıyla pazarlık edilecek bir hız sınırı ve siz denemeler yaparken işleyen bir çözüm başına sayaç yoktur. Her tür için tam metot imzaları şu sayfada yer alır: CAPTCHA çözme SDK'sı sayfası.

Ardışık sürüm ve maliyeti

Çoğu kişinin başladığı biçim şudur. Doğrudur, ama yavaştır.

# 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

Bir reCAPTCHA çözümü büyük ölçüde beklemedir. Çözücü çalışırken süreciniz boşta oturur, sonra sıradakine geçer ve yine boşta oturur. Üç çözüm, üç çözümlük gerçek zaman alır. Otuz çözüm de otuz çözümlük.

asyncio.gather ile aynı anda birden fazla CAPTCHA çözün

İstemciyi değiştirin, çağrıları await edin ve hepsini şuna verin: gather. Türler karışık olabilir: aynı kümede reCAPTCHA, Turnstile ve GeeTest bulunması sorun değildir.

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

Her metot aynı temel alanları içeren bir dict döndürür: captchaId ve code. Turnstile ayrıca şunu döndürür: userAgent, bir challenge sayfası çözümünü gönderirken token ile birlikte geri yollamanız gerekir. GeeTest şunu ekler: challenge, validate ve seccode, ve ham JSON'u şuraya koyar: code.

Eşzamanlılığı bir semafor ile sınırlayın

Yerel bir arka plan servisine iki yüz çözüm fırlatıp en iyisini ummayın. Çözme işi kendi makinenizde gerçekleşen bir CPU işidir; belirli bir genişliğin ötesinde yalnızca kendi kendinize kuyruk oluşturursunuz ve her bir çözüm yavaşlar. Semafor, aynı anda çalışan iş sayısını sabit tutar.

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)

Sayıyı tahmin ederek değil, ölçerek seçin. Aynı 20 hedefi 2, 5 ve 10 eşzamanlılıkla çalıştırın ve donanımınızda en hızlı biteni tutun. Doğru cevap SDK'ya değil, CPU'nuza bağlıdır.

Görev başına değil, tek bir istemci

Tek bir AsyncCapSkip oluşturun ve yukarıdaki gibi coroutine'ler arasında paylaşın. Görev başına bir tane oluşturmak israftır ve hiçbir kazanç sağlamaz: istemci, çözüm başına durumu değil yapılandırmayı tutar.

Tek bir hatanın tüm kümeyi öldürmesini engelleyin

Varsayılan olarak gather ilk istisnayı yukarı fırlatır ve o sırada işlemde olan diğer her şeyin sonuçlarını kaybedersiniz. Şunu geçin: return_exceptions=True , böylece istisnalar sonuç listesinde sıradan öğeler olarak gelir ve kazananları kaybedenlerden ayırabilirsiniz.

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

Dört istisna türü her CapSkip SDK'sında aynıdır ve hepsi şu temel sınıftan türer: CapSkipError , tek bir türü yakalamayı tercih ederseniz. ValidationException parametrelerinizin hatalı olduğu ve yeniden denemenin aynı şekilde başarısız olacağı anlamına gelir. NetworkException genellikle uygulamanın çalışmadığı anlamına gelir; bu, hedefe özgü değil tüm çalışmayı ilgilendiren bir sorundur.

Türe göre farklı davranan zaman aşımları ve yoklama

İki ayrı zaman aşımı geçerlidir ve türleri karıştıran bir küme ikisine birden tabidir.

Seçenek	Varsayılan	Şunlar için geçerli
defaultTimeout	120 saniye	Resim CAPTCHA'ları
recaptchaTimeout	300 saniye	reCAPTCHA, Turnstile, GeeTest
pollingInterval	5 saniye	The en fazla yoklamalar arasındaki aralık

pollingInterval ayarına dokunmadan önce anlamakta fayda var. SDK sabit bir aralıkla yoklama yapmaz. 250 ms'den başlar ve belirlediğiniz değere doğru kademeli olarak açılır; bir SDK çözümünün, ham API'nin “bekleyin, sonra beş saniyede bir yoklayın” önerisiyle elle yazılmış bir döngüden genellikle daha erken dönmesinin nedeni budur. Değeri yükseltmek hızlı çözümlerin daha geç gelmesine yol açar. Düşürmek ise hiçbir kazanç sağlamadan istek hacmini artırır.

GeeTest challenge'ları süresi dolar, bu yüzden kümeyi önceden hazırlamayın

Bu, özellikle paralele geçtiğinizde canınızı yakar. GeeTest'te gt değeri site başına sabittir, ancak challenge tek kullanımlıktır ve yaklaşık bir dakikada süresi dolar. Önce elli challenge toplayıp sonra çözmeye başlarsanız, kuyruğun sonundakiler gönderilmeden ölmüş olur. Her challenge'ı, onu kullanacak çözümden hemen önce alın.

Tam çalışan örnek

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

Bütün kalıp bu: tek bir paylaşılan istemci, bir semafor, return_exceptions=True, ve sonunda bir token sözlüğü. Bunu bir kazıyıcıya yerleştirin, CAPTCHA adımı darboğaz olmaktan çıksın. web scraping için CAPTCHA çözücü sayfası, bunun daha geniş bir hattın neresine oturduğunu anlatıyor.

Diğer SDK'larda aynı fikir

Bunu taşıyorsanız eşzamanlılık ilkesi değişir ama yapı değişmez. Node zaten bloklamıyor, dolayısıyla sade istemci yeterlidir.

// 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'te de durum aynı, şununla: Task.WhenAll, PHP'nin ise sunacak hiçbir eşzamanlılığı yok. Paralel çözüm gerekiyorsa ve dili siz seçebiliyorsanız, amaca özel istemciye sahip olan Python'dur. Python CAPTCHA çözücü sayfası arayüzün geri kalanını anlatıyor.

Sık yapılan hatalar

Hata	Ne olur	Düzeltme
Şunun kullanımı: CapSkip coroutine'ler içinde	Olay döngüsünü bloklar, dolayısıyla süreler ardışık kalır	Kullan AsyncCapSkip
Semafor yok	Kuyruk derinleştiğinde her çözüm yavaşlar	İşlemdeki iş sayısını sınırlayın, 5 civarında başlayın
Sade gather	Tek bir hata diğer tüm sonuçları çöpe atar	return_exceptions=True
GeeTest challenge'larını önceden almak	Sonrakilerin süresi gönderilmeden dolar	Her birini çözmeden hemen önce alın
Şunu yükseltmek: pollingInterval	Hızlı çözümler daha erken değil, daha geç döner	Varsayılan değerde bırakın
Resim çözümünde proxy ayarlamak	Resim CAPTCHA'ları için desteklenmez	Proxy'ler yalnızca reCAPTCHA, Turnstile ve GeeTest için geçerlidir

SDK'nın tam olarak ne gönderdiğini görmek isterseniz, her tür için ham istek ve yanıt şurada: API dokümantasyonu. Python'un kendi asyncio görev referansı kapsar gather semantiği ayrıntılı olarak anlatıyor.

Sık sorulan sorular

Aynı anda kaç CAPTCHA çözebilirim?

Takılacağınız bir kota yok, dolayısıyla sınır kendi donanımınız. Çözme yerel olarak gerçekleşir, bu yüzden eşzamanlılık hesap seviyesiyle değil CPU ile sınırlıdır. Aynı anda beş ile başlayın, sabit bir kümeyi ölçün ve oradan ayarlayın.

Tek bir gather çağrısında CAPTCHA türlerini karıştırabilir miyim?

Evet. recaptcha, turnstile, geetest ve normal hepsi aynı istemci üzerindeki coroutine'lerdir ve birlikte await edilebilir. Resim çözümlerinin 120 saniyelik, diğerlerinin 300 saniyelik zaman aşımı kullandığını unutmayın.

Bunun yerine thread kullanmalı mıyım?

Yalnızca çevreleyen kodunuz zaten thread'liyse. Buradaki iş G/Ç'ye bağlı bekleme, yani tam da asyncio'nun var oluş nedeni; ayrıca tek bir olay döngüsü bir thread havuzundan daha ucuzdur. Senkron bir kod tabanına sıkışmışsanız, sade CapSkip istemcisi de işe yarar.

AsyncCapSkip kapatılmalı mı?

SDK ne bir close metodu ne de asenkron bir bağlam yöneticisi belgeliyor; bu yüzden tek bir istemci oluşturun, çalışma boyunca kullanın ve süreç bittiğinde kapsam dışına çıkmasına izin verin.

Özet

Kullan AsyncCapSkip, tek bir istemciyi paylaşın, işlemdeki çözümleri bir semafor ile sınırlayın ve şunu geçin: return_exceptions=True , böylece tek bir kötü hedef kümeyi çöpe atmasın. Bu, bir CAPTCHA kuyruğunu ardışık bir darboğazdan tek bir beklemeye dönüştürür.

Eşzamanlılığı rahatça genişletebilmenizin nedeni, çözücünün kendi makinenizde çalışmasıdır. Çözüm başına fatura ve yabancılarla paylaşılan bir kuyruk yoktur; dolayısıyla ölçeklenme, başkasının hız sınırı değil sizin CPU'nuzla ilgili bir sorudur. Küme boyutlarınız küçük olmaktan çıktığında yerel bir captcha atlatma aracın yarattığı pratik fark tam olarak budur.