HTTPX + CaptchaAI Entegrasyonu

Tek bir HTTP istemcisiyle hem tek tek hem de yüzlerce CAPTCHA'yı aynı anda çözmek istiyorsanız, Python tarafında en pratik seçenek httpx'tir. httpx senkron ve asenkron API'yi tek pakette verir, HTTP/2 destekler ve requests'e alışkın bir geliştirici için neredeyse birebir aynı arayüzü sunar. Bu rehberde httpx'i CaptchaAI ile hem senkron hem asenkron modda çalıştıran, üretime hazır bir istemci kuracaksınız — teoriye girmeden, doğrudan çalışan kodla.

CaptchaAI görev akışı her CAPTCHA türü için aynıdır: görevi in.php uç noktasına gönderirsiniz, dönen görev kimliğini res.php üzerinden periyodik olarak sorgularsınız, hazır olduğunda token'ı alırsınız. httpx'in tek katkısı bu iki isteği taşımak; asıl kazanç, asenkron mod ve HTTP/2 ile bu istekleri çok daha verimli paralelleştirebilmenizdir.

Gereksinimler

Başlamak için tek paket yeterli:

pip install httpx

API anahtarınızı koda gömmek yerine bir ortam değişkeninde tutun (CAPTCHAAI_API_KEY); aşağıdaki örneklerin tamamı anahtarı buradan okur.

Senkron istemci: en hızlı başlangıç

Tek seferlik betikler, cron işleri veya bir CAPTCHA'yı çözüp devam eden basit iş akışları için senkron istemci en sade seçenektir. Aşağıdaki sınıf görevi gönderir, res.php'yi 5 saniyede bir sorgular ve 300 saniyelik bir zaman aşımı uygular. CAPCHA_NOT_READY yanıtı geldikçe döngü sürer; OK| önekiyle gelen ilk yanıtta token'ı döndürür.

import httpx
import time
import os


class CaptchaAISync:
    def __init__(self, api_key):
        self.api_key = api_key
        self.base_url = "https://ocr.captchaai.com"
        self.client = httpx.Client(timeout=30)

    def solve(self, params, timeout=300):
        params["key"] = self.api_key

        # Submit
        resp = self.client.get(f"{self.base_url}/in.php", params=params)
        text = resp.text

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

        task_id = text.split("|")[1]

        # Poll
        deadline = time.time() + timeout
        poll_params = {"key": self.api_key, "action": "get", "id": task_id}

        while time.time() < deadline:
            time.sleep(5)
            result = self.client.get(
                f"{self.base_url}/res.php", params=poll_params
            )

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

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

    def get_balance(self):
        resp = self.client.get(f"{self.base_url}/res.php", params={
            "key": self.api_key, "action": "getbalance"
        })
        return float(resp.text)

    def close(self):
        self.client.close()


# Usage
solver = CaptchaAISync(os.environ["CAPTCHAAI_API_KEY"])

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

Dönen token'ı reCAPTCHA v2 için hedef formdaki g-recaptcha-response alanına yerleştirirsiniz. get_balance() yöntemi de aynı res.php uç noktasını kullanarak bakiyenizi USD cinsinden döndürür — kotanızı izlemek için pratik bir kontrol.

Asenkron istemci: eşzamanlı çözüm

İşin ölçek tarafı burada başlar. Bir kazıma işinde onlarca sayfayı işliyorsanız, her CAPTCHA'yı sırayla beklemek dakikalar kaybettirir. Asenkron istemci asyncio.gather ile birden fazla görevi aynı anda gönderip sorgular; toplam süre en yavaş tek çözüme yaklaşır, çözümlerin toplamına değil.

import httpx
import asyncio
import os


class CaptchaAIAsync:
    def __init__(self, api_key):
        self.api_key = api_key
        self.base_url = "https://ocr.captchaai.com"
        self.client = httpx.AsyncClient(timeout=30)

    async def solve(self, params, timeout=300):
        params["key"] = self.api_key

        # Submit
        resp = await self.client.get(
            f"{self.base_url}/in.php", params=params
        )
        text = resp.text

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

        task_id = text.split("|")[1]

        # Poll
        deadline = asyncio.get_event_loop().time() + timeout
        poll_params = {"key": self.api_key, "action": "get", "id": task_id}

        while asyncio.get_event_loop().time() < deadline:
            await asyncio.sleep(5)
            result = await self.client.get(
                f"{self.base_url}/res.php", params=poll_params
            )

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

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

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

    async def close(self):
        await self.client.aclose()


# Usage
async def main():
    solver = CaptchaAIAsync(os.environ["CAPTCHAAI_API_KEY"])

    # Solve multiple concurrently
    tasks = [
        solver.solve({
            "method": "userrecaptcha",
            "googlekey": "6Le-wvkS...",
            "pageurl": f"https://example.com/page{i}",
        })
        for i in range(5)
    ]

    results = await asyncio.gather(*tasks, return_exceptions=True)
    for i, r in enumerate(results):
        if isinstance(r, Exception):
            print(f"Page {i}: FAILED - {r}")
        else:
            print(f"Page {i}: solved ({len(r)} chars)")

    await solver.close()

asyncio.run(main())

return_exceptions=True sayesinde tek bir görevin başarısız olması diğerlerini düşürmez — beş görevden biri zaman aşımına uğrarsa kalan dördü yine token döndürür. Üretimde bu yalıtım kritiktir.

Eşzamanlılık, thread ve maliyet

Kaç görevi aynı anda çalıştırabileceğiniz kodunuzla değil, CaptchaAI planınızın thread sayısıyla belirlenir. Bir thread, o anda çözülmekte olan tek bir CAPTCHA demektir; çözüm bitince o thread bir sonrakine geçer. CaptchaAI çözüm başına değil, eşzamanlı thread başına ücretlendirir ve her planda thread başına çözüm limitsizdir.

Pratik karşılığı şu: asyncio.gather ile 50 görevi aynı anda başlatsanız da, BASIC ($15/ay, 5 thread) planında bunların yalnızca beşi aynı anda işlenir, geri kalanı kuyruğa girer. Düzenli olarak yüksek eşzamanlılıkla çalışıyorsanız ADVANCE ($90/ay, 50 thread) gibi bir plan asenkron kodunuzun kapasitesine denk düşer. Doğru planı seçerken şuna dikkat edin:

• Eşzamanlı görev sayısı ≈ thread sayısı. Kodunuzdaki range(5) değil, planınız gerçek paralelliği belirler.
• Çözüm başına değil, thread başına ödersiniz. Her planda thread başına çözüm limitsizdir; günlük tavan veya CAPTCHA türüne göre ek ücret yoktur.
• Fiyatlar USD ve aylık sabittir. Türkiye'deki geliştiriciler için avantaj burada: TL kurundaki oynaklıktan bağımsız, öngörülebilir bir maliyet. Fiyatları TL'ye çevirmeyin; güncel değerleri captchaai.com/pricing sayfasından doğrulayın.

HTTP/2 ile bağlantı yükünü azaltma

httpx, requests ve aiohttp'ten farklı olarak HTTP/2'yi destekler. Ek bir paketle etkinleşir:

pip install httpx[http2]

client = httpx.AsyncClient(http2=True, timeout=30)

HTTP/2, istekleri tek bir bağlantı üzerinden çoğullar (multiplexing). Çok sayıda CAPTCHA'yı gönderip sorgularken her istek için yeni bir TCP/TLS el sıkışması kurmak yerine mevcut bağlantıyı yeniden kullanır; bu da yoğun sorgulama trafiğinde gecikmeyi ve bağlantı yükünü ölçülebilir biçimde düşürür. getbalance gibi sık tekrarlanan küçük isteklerde fark özellikle belirginleşir.

CAPTCHA işleme ile kazıma örneği

Gerçek bir senaryoda genellikle önce sayfayı çeker, içinde bir reCAPTCHA olup olmadığını kontrol eder, varsa sitekey'i çıkarıp çözer ve token'la formu gönderirsiniz. Aşağıdaki örnek bu akışın tamamını tek bir asenkron fonksiyonda birleştirir.

import httpx
import re
import os

async def scrape_with_captcha(url, solver):
    async with httpx.AsyncClient() as client:
        # Fetch page
        resp = await client.get(url)
        html = resp.text

        # Check for reCAPTCHA
        match = re.search(
            r'data-sitekey=["\']([A-Za-z0-9_-]+)["\']', html
        )
        if not match:
            return html

        site_key = match.group(1)
        token = await solver.solve({
            "method": "userrecaptcha",
            "googlekey": site_key,
            "pageurl": url,
        })

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


async def main():
    solver = CaptchaAIAsync(os.environ["CAPTCHAAI_API_KEY"])
    content = await scrape_with_captcha("https://example.com", solver)
    print(f"Got {len(content)} chars")
    await solver.close()

asyncio.run(main())

Bu tür akışları yalnızca yetkiniz olan veya kendi QA ortamınızdaki hedeflerde çalıştırın; testlerinizi staging.example.com/qa-form gibi bir aşama (staging) adresine yönlendirmek en temiz yaklaşımdır. Topladığınız veri kişisel veri içeriyorsa KVKK kapsamına girdiğini unutmayın — otomasyonu yetkili veri toplama iş akışları içinde tutun.

Üretim için pratik notlar

Örnek istemciler doğrudan çalışır, ancak üretime almadan önce birkaç noktayı sağlamlaştırın:

• Zaman aşımı ve yeniden deneme. solve çağrısındaki 300 saniyelik sınırı iş yükünüze göre ayarlayın; ağ kaynaklı httpx hatalarında görevi birkaç kez yeniden deneyin.
• Bakiye kontrolü. Toplu iş öncesi get_balance() ile kotanızı doğrulayın; boşta kalan bir işin ortasında bakiye tükenmesi en pahalı hatadır.
• İstemciyi yeniden kullanın. Her istek için yeni bir AsyncClient açmayın; tek bir istemciyi paylaşmak bağlantı havuzundan ve HTTP/2 çoğullamasından tam verim almanızı sağlar.

Karşılaştırma: httpx, requests ve aiohttp

Kısacası: mevcut requests tabanlı bir kodu bozmadan asenkron ve HTTP/2 kazanmak istiyorsanız httpx, aradığınız tek pakettir.

Sık sorulan sorular

httpx ile aynı anda kaç CAPTCHA çözebilirim?

Kod tarafında sınır yok; asyncio.gather ile yüzlerce görev başlatabilirsiniz. Gerçek eşzamanlılık planınızın thread sayısıyla belirlenir: BASIC 5, ADVANCE 50 thread verir. Fazlası kuyruğa girer ve sırayla işlenir.

Asenkron çözümde zaman aşımı hatası alıyorum, ne yapmalıyım?

Önce solve çağrısındaki timeout değerinin (varsayılan 300 saniye) çözüm süresinden uzun olduğundan emin olun. gather ile çok sayıda görev başlatıp thread limitinizi aştıysanız, görevler kuyrukta beklerken zaman aşımına uğrayabilir; eşzamanlı görev sayısını thread'inize göre sınırlayın.

httpx yerine requests kullanmaya devam etmeli miyim?

Yeni projeler için httpx daha iyi bir varsayılan: requests uyumlu bir arayüzün yanında asenkron ve HTTP/2 desteği sunar. Çalışan requests kodunuz varsa geçiş genellikle birkaç satırlık bir değişikliktir.

HTTP/2 gerçekten çözüm hızını artırır mı?

Tek bir CAPTCHA'da farkı hissetmezsiniz. Fark, çok sayıda görevi tek bağlantı üzerinden çoğullarken ve res.php'yi sık sorgularken ortaya çıkar; bağlantı kurma yükü ortadan kalktığı için genel gecikme düşer.

İlgili kılavuzlar

• aiohttp ile asenkron CAPTCHA çözümü
• Axios ile tarayıcısız çözüm (Node.js)
• Python ile CAPTCHA'lı kazıma rehberi