Python ThreadPoolExecutor para CAPTCHA resolvendo paralelismo

asyncio é poderoso, mas requer reescrever toda a sua cadeia de chamadas como assíncrona. ThreadPoolExecutor oferece paralelismo com código síncrono padrão – coloque-o em projetos existentes sem reestruturação.

Por que ThreadPoolExecutor para CAPTCHAs

A resolução de CAPTCHA é I/O-bound (aguardando respostas HTTP). Threads Python liberam o GIL durante operações I/O, tornando ThreadPoolExecutor eficiente para esta carga de trabalho:

Implementação Básica

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

Usando sessão para reutilização de conexão

Criar uma nova conexão TCP por solicitação é uma perda de tempo. Compartilhe um requests.Session por tópico:

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

map() para operações em lote simples

Quando você não precisa de tratamento de erros por tarefa:

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

Proteção de tempo limite

Evite que threads descontrolados bloqueiem seu pool:

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

Retorno de chamada de progresso

Acompanhe a conclusão em tempo real:

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

Escolhendo max_workers

Mais trabalhadores significam mais conexões API simultâneas. Comece em 10, aumente enquanto monitora as taxas de erro.

ThreadPoolExecutor vs assíncrono

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

Use ThreadPoolExecutor quando:

• Sua base de código existente é síncrona
• Você usa bibliotecas que não suportam assíncrono (Selenium, alguns ORMs)
• Você quer paralelismo rápido sem reestruturação

Use assíncio quando:

• Construindo do zero
• A eficiência máxima é importante (menos threads do sistema operacional)
• Já em um framework assíncrono (FastAPI, aiohttp)

Solução de problemas

Perguntas frequentes

O GIL evita o paralelismo real?

Não - para I/O-bound funcionar como solicitações HTTP e time.sleep, Python libera o GIL. Seus threads são executados de forma verdadeiramente simultânea durante chamadas de rede. O GIL limita apenas o paralelismo vinculado à CPU.

Quantos CAPTCHAs o ThreadPoolExecutor pode manipular por hora?

Com 10 trabalhadores e tempo médio de resolução de 15 segundos: aproximadamente 2.400 por hora. Com 25 trabalhadores: ~6.000 por hora. O gargalo é o tempo de resolução do CaptchaAI, não o threading do Python.

Devo usar ProcessPoolExecutor?

A solução CAPTCHA é I/O-bound. ProcessPoolExecutor adiciona sobrecarga de comunicação entre processos sem nenhum benefício. Fique com fios.

Próximas etapas

Paralelizar a resolução de CAPTCHA -obtenha sua chave API CaptchaAIe coloque ThreadPoolExecutor em seu pipeline.

Guias relacionados:

• Resolução paralela de CAPTCHA
• Desempenho paralelo versus sequencial
• Resolvendo 10.000 tarefas por hora