Python ThreadPoolExecutor ile Paralel CAPTCHA Çözme

Bir CAPTCHA'nın çözülmesi ortalama 15 saniye sürüyorsa, 100 CAPTCHA'yı sırayla çözmek 25 dakikanızı alır. Aynı 100 isteği paralel çalıştırdığınızda bu süre saniyelere iner — üstelik bunun için tüm çağrı zincirinizi yeniden yazmanız gerekmez.

Python'ın standart kütüphanesindeki ThreadPoolExecutor, mevcut senkron kodunuza olduğu gibi girer ve CaptchaAI isteklerinizi eşzamanlı hale getirir. Bu rehber, tek bir havuzla toplu CAPTCHA çözümünü baştan sona kurmayı; bağlantı yeniden kullanımı, zaman aşımı ve doğru max_workers değerini seçmeyi çalışan örneklerle gösterir.

Neden ThreadPoolExecutor ile paralel çözüm?

CAPTCHA çözümünde geçen sürenin neredeyse tamamı beklemekle geçer: isteği CaptchaAI'ye gönderir, çözülmesini bekler, sonucu sorgularsınız. Bu sırada CPU boştur; darboğaz ağdır. Yani iş I/O-bound'dir ve tam olarak thread'lerin parladığı senaryodur.

Python'da "GIL yüzünden thread işe yaramaz" klişesi yalnızca CPU-bound iş için doğrudur. Bir thread ağ yanıtı beklerken veya time.sleep içindeyken GIL'i serbest bırakır; böylece diğer thread'ler aynı anda kendi isteklerini yürütür. Onlarca CAPTCHA'yı bekleme sürelerini üst üste bindirerek eşzamanlı çözebilmenizin nedeni budur.

Aşağıdaki tablo dört yaklaşımı, senkron bir projeye entegrasyon maliyetiyle birlikte karşılaştırır:

ThreadPoolExecutor'ın cazip yanı, bu paralelliği tek bir with bloğuyla, hiçbir fonksiyonu async yapmadan sağlamasıdır.

Senaryo: yoğun QA hattında toplu doğrulama

Diyelim ki bir e-ticaret müşterisi için ödeme ve kayıt akışlarının QA testini yürüten serbest bir otomasyon geliştiricisisiniz. Her gece 20 staging sayfasındaki reCAPTCHA'yı doğrulamanız ve teslim tarihine yetişmeniz gerekiyor. Tek tek çözerseniz her sayfa ~15 saniye, toplamda beş dakikadan fazla sürer; havuzla hepsini aynı anda gönderdiğinizde iş saniyelere iner.

Türkiye'deki geliştiriciler için ek bir avantaj var: CaptchaAI thread bazlı ve USD üzerinden ücretlendirir. TL'nin oynaklığı düşünüldüğünde, çözüm başına değil de sabit aylık thread ücreti ödemek maliyeti öngörülebilir kılar. Kazıdığınız veri kişisel veri içeriyorsa bunun KVKK kapsamına girdiğini ve yalnızca yetkili QA/veri toplama akışlarında çalışmanız gerektiğini unutmayın.

Temel kurulum: havuzla toplu çözüm

solve_captcha fonksiyonu tek bir CAPTCHA için tüm döngüyü yürütür: görevi in.php uç noktasına gönderir, sonra sonuç hazır olana kadar res.php'yi sorgular. Bu fonksiyonu bir ThreadPoolExecutor içine koyduğunuzda havuz onu onlarca görev için paralel çağırır. as_completed, sonuçları tamamlanma sırasına göre teslim eder; böylece en yavaş görevi beklemeden biteni işlersiniz.

import os
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
import requests

API_KEY = os.environ["CAPTCHAAI_API_KEY"]


def solve_captcha(sitekey, pageurl):
    """Synchronous CAPTCHA solve — submit and poll."""
    # Submit
    resp = requests.post("https://ocr.captchaai.com/in.php", data={
        "key": API_KEY,
        "method": "userrecaptcha",
        "googlekey": sitekey,
        "pageurl": pageurl,
        "json": 1
    })
    data = resp.json()

    if data.get("status") != 1:
        raise RuntimeError(data.get("request", "Submit failed"))

    captcha_id = data["request"]

    # Poll for result
    for _ in range(60):
        time.sleep(5)
        result = requests.get("https://ocr.captchaai.com/res.php", params={
            "key": API_KEY,
            "action": "get",
            "id": captcha_id,
            "json": 1
        }).json()

        if result.get("status") == 1:
            return result["request"]
        if result.get("request") != "CAPCHA_NOT_READY":
            raise RuntimeError(result.get("request", "Unknown error"))

    raise TimeoutError("Solve timeout after 300s")


# Batch solve with ThreadPoolExecutor
tasks = [
    {"sitekey": "6Le-wvkSAAAAAPBMRTvw0Q4Muexq9bi0DJwx_mJ-", "pageurl": f"https://example.com/page/{i}"}
    for i in range(20)
]

start = time.time()

with ThreadPoolExecutor(max_workers=10) as executor:
    futures = {
        executor.submit(solve_captcha, t["sitekey"], t["pageurl"]): t
        for t in tasks
    }

    solved = 0
    failed = 0

    for future in as_completed(futures):
        task = futures[future]
        try:
            solution = future.result()
            solved += 1
            print(f"[OK] {task['pageurl']}: {solution[:30]}...")
        except Exception as e:
            failed += 1
            print(f"[ERR] {task['pageurl']}: {e}")

elapsed = time.time() - start
print(f"\nDone: {solved} solved, {failed} failed in {elapsed:.1f}s")

Bağlantıları yeniden kullanmak için thread-local Session

import threading

# Thread-local storage for sessions
thread_local = threading.local()


def get_session():
    """Get or create a thread-local session."""
    if not hasattr(thread_local, "session"):
        thread_local.session = requests.Session()
        # Configure connection pooling
        adapter = requests.adapters.HTTPAdapter(
            pool_connections=10,
            pool_maxsize=10,
            max_retries=2
        )
        thread_local.session.mount("https://", adapter)
    return thread_local.session


def solve_captcha_pooled(sitekey, pageurl):
    """Solve using thread-local connection pooling."""
    session = get_session()

    resp = session.post("https://ocr.captchaai.com/in.php", data={
        "key": API_KEY,
        "method": "userrecaptcha",
        "googlekey": sitekey,
        "pageurl": pageurl,
        "json": 1
    })
    data = resp.json()

    if data.get("status") != 1:
        raise RuntimeError(data.get("request"))

    captcha_id = data["request"]

    for _ in range(60):
        time.sleep(5)
        result = session.get("https://ocr.captchaai.com/res.php", params={
            "key": API_KEY,
            "action": "get",
            "id": captcha_id,
            "json": 1
        }).json()

        if result.get("status") == 1:
            return result["request"]
        if result.get("request") != "CAPCHA_NOT_READY":
            raise RuntimeError(result.get("request"))

    raise TimeoutError("Solve timeout")

Yukarıdaki get_session, her thread'e threading.local() ile kendi requests.Session'ını verir — Session nesneleri thread-safe olmadığından bu şarttır. Böylece bağlantı havuzu korunur ve her istekte yeniden TLS el sıkışması kurulmaz; yeni bir TCP bağlantısı açmanın maliyetinden kurtulursunuz.

Basit toplu işler için map()

def solve_task(task):
    """Wrapper that returns result dict."""
    try:
        solution = solve_captcha_pooled(task["sitekey"], task["pageurl"])
        return {"url": task["pageurl"], "solution": solution, "error": None}
    except Exception as e:
        return {"url": task["pageurl"], "solution": None, "error": str(e)}


with ThreadPoolExecutor(max_workers=10) as executor:
    results = list(executor.map(solve_task, tasks))

solved = [r for r in results if r["solution"]]
failed = [r for r in results if r["error"]]
print(f"Solved: {len(solved)}, Failed: {len(failed)}")

executor.map(), görev başına ayrı hata yönetimine ihtiyacınız olmadığında daha kısa bir yoldur. Her çözümü; sonucu ve olası hatayı bir dict içinde döndüren küçük bir sarmalayıcıyla birleştirirsiniz. map girdinin sırasını koruduğundan hangi URL'nin hangi sonucu verdiğini takip etmek kolaydır.

Kaçak thread'lere karşı zaman aşımı

from concurrent.futures import TimeoutError as FuturesTimeout

with ThreadPoolExecutor(max_workers=10) as executor:
    futures = {
        executor.submit(solve_captcha_pooled, t["sitekey"], t["pageurl"]): t
        for t in tasks
    }

    for future in as_completed(futures, timeout=600):  # 10 min global timeout
        task = futures[future]
        try:
            solution = future.result(timeout=120)  # 2 min per task
            print(f"[OK] {task['pageurl']}")
        except FuturesTimeout:
            print(f"[TIMEOUT] {task['pageurl']}")
        except Exception as e:
            print(f"[ERR] {task['pageurl']}: {e}")

Hedef site yanıt vermeyi keserse ya da bir çözüm beklenenden uzun sürerse, tek bir kaçak thread tüm havuzu tıkayabilir. Yukarıdaki iki katmanlı zaman aşımı bunu önler: as_completed(..., timeout=600) tüm parti için genel bir tavan, future.result(timeout=120) ise görev başına bir sınır koyar. Süresi dolan görevi FuturesTimeout ile yakalar, diğerlerini akıtmaya devam edersiniz.

İlerlemeyi canlı izleme

import threading

progress_lock = threading.Lock()
progress = {"done": 0, "total": 0}


def solve_with_progress(task):
    result = solve_task(task)
    with progress_lock:
        progress["done"] += 1
        pct = progress["done"] / progress["total"] * 100
        print(f'\r  Progress: {progress["done"]}/{progress["total"]} ({pct:.0f}%)', end="")
    return result


progress["total"] = len(tasks)

with ThreadPoolExecutor(max_workers=10) as executor:
    results = list(executor.map(solve_with_progress, tasks))

print()  # Newline after progress

Uzun süren partilerde bir kilit (Lock) ile korunan sayaç, tamamlanan görev sayısını thread-safe biçimde günceller. Aynı satırı yeniden yazarak yüzde ilerlemeyi terminalde canlı gösterir; hangi aşamada olduğunuzu görmek uzun toplu işlerde işe yarar.

max_workers'ı seçme ve plan thread'lerinizle eşleştirme

max_workers, havuzun aynı anda kaç CAPTCHA'yı işleyeceğini belirler. Daha fazla worker, daha fazla eşzamanlı API bağlantısı demektir; 10 ile başlayıp hata oranlarını izleyerek artırın.

Burada gözden kaçan kritik bağlantı şu: CaptchaAI eşzamanlı thread başına ücretlendirir ve planınızdaki thread sayısı, aynı anda çözebileceğiniz CAPTCHA sayısını belirler. max_workers değeriniz plan thread'lerinizi aşarsa, fazladan worker'lar sıraya girip boşta bekler — daha hızlı olmazsınız, yalnızca bellek harcarsınız.

Pratik eşleştirme:

• BASIC ($15/ay, 5 thread): max_workers=5 — küçük partiler ve deneme için yeterli.
• ADVANCE ($90/ay, 50 thread): max_workers değerini 50'ye kadar çıkarabilirsiniz.
• ENTERPRISE ($300/ay, 200 thread): yüksek hacimli, sürekli çalışan hatlar için.

Havuz boyutunuzu planınızın thread sayısıyla hizalayın; ikisini birbirine denk tutmak hem verimi hem de maliyeti öngörülebilir kılar.

ThreadPoolExecutor mı, asyncio mu?

İki yaklaşım da eşzamanlılık sağlar; fark, projenize giriş maliyetindedir.

# ThreadPoolExecutor — drop into existing sync code
with ThreadPoolExecutor(max_workers=10) as executor:
    results = list(executor.map(solve_task, tasks))

# asyncio — requires async function chain
async def main():
    async with aiohttp.ClientSession() as session:
        tasks = [solve_async(session, t) for t in task_list]
        results = await asyncio.gather(*tasks)

Şu durumlarda ThreadPoolExecutor'ı tercih edin:

• Mevcut kod tabanınız senkron.
• async desteklemeyen kütüphanelerle çalışıyorsunuz (Selenium, bazı ORM'ler).
• Yeniden yapılandırmadan hızlı paralellik istiyorsunuz.

Şu durumlarda asyncio'ya geçin:

• Projeyi sıfırdan kuruyorsunuz.
• İşletim sistemi thread'lerini en aza indirmek (maksimum verim) önemli.
• Zaten async bir çerçevedesiniz (FastAPI, aiohttp).

Bu arada ProcessPoolExecutor'a gerek yoktur: CAPTCHA çözümü I/O-bound olduğundan süreçler arası iletişim yükü getirir ama hız kazandırmaz. Thread'lerde kalın.

Sorun giderme

Bu belirtilerin çoğu havuz boyutuyla ilgilidir; max_workers değerini planınıza göre ayarlamak çoğunu baştan önler.

Sık sorulan sorular

max_workers'ı planımın thread sayısıyla nasıl eşleştiririm?

En basit kural: max_workers değerini planınızın thread sayısına eşit ya da altında tutun. BASIC 5 thread verdiği için max_workers=5, ADVANCE 50 thread için 50'ye kadar uygundur. Plan sınırının ötesinde worker açmak ek hız getirmez, worker'lar yalnızca sıraya girer.

ThreadPoolExecutor'ı Selenium ile birlikte kullanabilir miyim?

Evet — hatta thread'lerin en mantıklı olduğu yerlerden biri budur. Selenium senkron bir kütüphanedir ve async'e uygun değildir; her thread kendi işini yürütürken CAPTCHA çözümünü havuza devrederek tarayıcı oturumlarını paralel ilerletebilirsiniz.

Bu kurulumla saatte kaç CAPTCHA çözebilirim?

10 worker ve ortalama 15 saniyelik çözüm süresiyle saatte yaklaşık 2.400 CAPTCHA; 25 worker ile ~6.000. Darboğaz Python thread'i değil, çözüm süresi ve planınızın thread sayısıdır — daha yüksek hacim için worker sayısını ve planı birlikte artırın.

asyncio'ya ne zaman geçmeliyim?

Onlarca değil binlerce eşzamanlı isteğe çıkıyorsanız ve her thread'in işletim sistemi maliyeti sizi zorlamaya başladıysa asyncio aynı işi daha az kaynakla yapar. Yeni bir projeye sıfırdan başlıyorsanız ya da zaten FastAPI/aiohttp gibi async bir çerçevedeyseniz doğrudan asyncio ile kurun. Aksi halde ThreadPoolExecutor fazlasıyla yeterlidir.

Sonraki adımlar

CAPTCHA çözümünü paralelleştirmek için tek ihtiyacınız bir API anahtarı ve yukarıdaki havuz kodu. CaptchaAI API anahtarınızı alın, max_workers değerini planınıza göre ayarlayın ve ThreadPoolExecutor'ı işlem hattınıza ekleyin.

İlgili rehberler:

• Paralel CAPTCHA çözümü
• Paralel ve sıralı performans karşılaştırması
• Saatte 10.000 görev çözme