aiohttp + CaptchaAI: Eşzamansız CAPTCHA Çözme

Yüzlerce sayfada CAPTCHA'yı tek tek, sırayla çözmek dakikalar alır — çünkü her istek bir öncekinin bitmesini bekler. aiohttp bu beklemeyi ortadan kaldırır: olay döngüsünü bloke etmeden onlarca çözüm görevini aynı anda başlatır, CaptchaAI de bunları paralel olarak çözer. Bu rehberde üretime hazır bir asenkron CaptchaAI istemcisini sıfırdan kurup toplu çözüm, hız kontrolü ve gerçek kazıma akışlarında nasıl kullanacağınızı göstereceğiz.

CAPTCHA çözümü doğası gereği bir bekleme işidir: görevi gönderirsiniz, ardından sonuç hazır olana kadar birkaç saniyede bir sorgularsınız. Senkron kodda bu bekleme boyunca iş parçacığınız boşta durur. Asenkron modelde ise aynı bekleme sırasında olay döngüsü diğer görevlerin sorgularını yürütür — tek bir çekirdekle yüzlerce çözümü aynı anda ilerletebilmenizin nedeni budur. Bu yüzden CAPTCHA çözme, aiohttp'nin gerçekten fark yarattığı klasik bir kullanım senaryosudur.

Gereksinimler

Başlamadan önce ihtiyacınız olan tek şey Python, aiohttp ve bir CaptchaAI API anahtarı:

pip install aiohttp

Asenkron CaptchaAI istemcisi

İstemci üç işi üstlenir: görevi in.php'ye gönderir, sonucu res.php'den sorgular ve bakiyeyi kontrol eder. submit ile poll ayrı tutulduğu için tek bir aiohttp.ClientSession üzerinden yüzlerce görevi eşzamanlı yürütebilirsiniz. poll, sonuç hazır olana kadar 5 saniyede bir sorgular; timeout süresi içinde yanıt gelmezse zaman aşımı verir.

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

Tek bir CAPTCHA çözün

Toplu akışa geçmeden önce tek görevle boru hattını doğrulayın: bakiyeyi okuyun, bir reCAPTCHA v2 görevi gönderin ve dönen token'ı yazdırın. Bu adım, API anahtarınızın ve googlekey/pageurl değerlerinizin doğru olduğunu kanıtlar. API anahtarını koda gömmek yerine CAPTCHAAI_API_KEY ortam değişkeninden okuyun; bu hem anahtarınızı sürüm kontrolünden uzak tutar hem de aynı kodu farklı ortamlarda yeniden yapılandırmadan çalıştırmanızı sağlar.

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

Birden fazla CAPTCHA'yı aynı anda çözün

Asıl kazanç burada. asyncio.gather tüm görevleri aynı olay döngüsünde başlatır; beş sayfalık bir yığın, sürelerin toplamı kadar değil, yalnızca en yavaş tek çözüm kadar sürede biter. return_exceptions=True sayesinde başarısız olan tek bir görev tüm yığını düşürmez.

Somut bir örnek: İstanbul'da bir e-ticaret ekibi için çalışan bir otomasyon geliştiricisi, onlarca staging sayfasındaki ödeme adımını gece boyunca test ederken her sayfadaki CAPTCHA'yı sırayla beklemek yerine hepsini tek seferde kuyruğa alabilir. Sıralı akışta saatler süren bir tarama, eşzamanlı çözümle dakikalara iner.

Üretimde geçici ağ hatalarına karşı solve çağrısını üstel geri çekilme (exponential backoff) ile birkaç kez yeniden denemek isteyebilirsiniz. return_exceptions=True bayrağı, bu yeniden deneme mantığını yığının geri kalanını durdurmadan eklemenize olanak tanır: her görev kendi başına başarılı ya da başarısız olur, tek bir hata diğerlerini iptal etmez.

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..."))

CAPTCHA çözümüyle sayfa kazıma

Gerçek dünyada CAPTCHA genelde bir kazıma akışının ortasında çıkar. Aşağıdaki desen sayfayı çeker, g-recaptcha var mı diye bakar, varsa çözer ve token'ı formla birlikte geri gönderir. Kişisel veri topluyorsanız bu verinin KVKK kapsamına girdiğini ve yalnızca yetkili olduğunuz veriler üzerinde çalışmanız gerektiğini unutmayın.

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

Semafor ile hız kontrolü

Sınırsız eşzamanlılık her zaman daha hızlı değildir: plan thread sayınızın üzerine çıkan istekler kuyruğa girer ve gecikmeyi artırır. Bir asyncio.Semaphore, aynı anda çalışan çözüm sayısını plan sınırınıza göre sabitler — böylece hem CaptchaAI tarafını hem de kendi kaynaklarınızı gereksiz yere zorlamazsınız.

Pratik bir başlangıç noktası: max_concurrent değerini planınızdaki thread sayısına eşitleyin. BASIC ($15/ay, 5 thread) planında 5, PREMIUM ($170/ay, 100 thread) planında ise 100 civarı bir değer, kuyruk beklemesi olmadan verimi en üst düzeye çıkarır. Değeri thread sınırınızın çok üzerine çıkarmak ek hız getirmez; yalnızca istekleri sıraya sokar. CaptchaAI thread başına sınırsız çözüm içerdiği için asıl belirleyici, aynı anda kaç görevi açık tuttuğunuzdur.

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

Turnstile örneği

Aynı istemci, method değerini turnstile yaparak Cloudflare Turnstile için de çalışır — kod iskeleti değişmez, yalnızca gönderdiğiniz parametreler farklıdır.

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

Sorun giderme

Sık sorulan sorular

aiohttp mı yoksa httpx mi kullanmalıyım?

Yüksek eşzamanlılıkta aiohttp, olgun bağlantı havuzu ve düşük ek yüküyle güçlü bir seçenektir. httpx de sorunsuz çalışır; senkron ve asenkron API'yi tek pakette istiyorsanız httpx entegrasyon kılavuzuna bakın.

aiohttp ile hangi CAPTCHA türlerini çözebilirim?

Desen tüm CaptchaAI türlerinde aynı kalır: reCAPTCHA v2/v3, Cloudflare Turnstile ve Challenge, GeeTest v3, görüntü/OCR, grid ve BLS. Yalnızca gönderdiğiniz method ve parametreler değişir. hCaptcha ve FunCaptcha desteklenmez.

Yüksek eşzamanlılık için hangi CaptchaAI planı gerekir?

CaptchaAI thread tabanlı faturalandırır: eşzamanlılığınız planınızdaki thread sayısıyla sınırlıdır. Az sayıda paralel görev için BASIC ($15/ay, 5 thread) yeterlidir; yoğun toplu iş yükleri için PREMIUM ($170/ay, 100 thread) gibi bir üst plan daha uygundur. Her plan thread başına sınırsız çözüm içerir.

Bir görev hata verirse asyncio.gather diğerlerini iptal eder mi?

return_exceptions=True ile hayır — hata veren görev sonuç listesine bir istisna olarak döner, kalan görevler tamamlanır. Bu bayrağı kaldırırsanız ilk hata tüm gather çağrısını durdurur.

Çözüm ne kadar sürede tamamlanır?

Süre CAPTCHA türüne ve anlık yüke göre değişir. poll her 5 saniyede bir sorgular ve varsayılan timeout değeri 300 saniyedir. Yavaş yanıt aldığınız türlerde bu süreyi artırın; hızlı türlerde ise sorgulama aralığını düşürerek toplam gecikmeyi azaltabilirsiniz.

İlgili kılavuzlar

• HTTPX + CaptchaAI entegrasyonu
• Paralel CAPTCHA çözümü
• Scrapy + CaptchaAI entegrasyonu