حل CAPTCHA بالتوازي في Python باستخدام ThreadPoolExecutor

حين ترسل عشرات طلبات CAPTCHA واحدة تلو الأخرى، يقضي سكربتك معظم وقته في الانتظار. الحل ليس إعادة كتابة مشروعك بالكامل ليصبح غير متزامن، بل تشغيل الطلبات جنباً إلى جنب عبر ThreadPoolExecutor: أنت تحصل على التوازي داخل كود متزامن عادي، وتضيفه إلى مشروعك القائم دون لمس بنيته. هذا الدليل يبني مسار حل CAPTCHA بالتوازي مع CaptchaAI خطوة بخطوة، من التطبيق الأساسي حتى ضبط عدد العمال.

لماذا يناسب ThreadPoolExecutor حل CAPTCHA بالتوازي؟

حل CAPTCHA عملية مقيّدة بالإدخال/الإخراج (I/O-bound): معظم الزمن يمر في انتظار استجابة HTTP من الخادم، لا في حسابات المعالج. وبما أن خيوط Python تُحرّر الـ GIL أثناء عمليات I/O والانتظار، فإن ThreadPoolExecutor يمنحك توازياً حقيقياً لهذا النوع من الأحمال دون تعقيد asyncio. يبرز هذا الخيار لثلاثة أسباب:

• يعمل داخل كودك المتزامن الحالي دون إعادة كتابته
• يستغل زمن الانتظار الشبكي لتشغيل عدة طلبات معاً
• يبقى أبسط في التصحيح والصيانة مقارنةً بالكود غير المتزامن

الجدول التالي يوازن بين المقاربات المتاحة:

التطبيق الأساسي لحل CAPTCHA بالتوازي

الفكرة بسيطة: دالة متزامنة واحدة تُرسل المهمة ثم تستطلع النتيجة، وThreadPoolExecutor يشغّل نسخاً منها بالتوازي عبر مجموعة من العمال. في المثال التالي، تُرسَل عشرون مهمة ويعالجها عشرة عمال في آنٍ واحد، مع تجميع نتائج النجاح والفشل عبر as_completed:

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

تتبع الدالة أربع خطوات ثابتة لكل مهمة:

1. إرسال المهمة إلى نقطة النهاية in.php والحصول على معرّف
2. استطلاع res.php دورياً حتى تجهز النتيجة
3. إرجاع الرمز عند النجاح أو رفع استثناء عند الخطأ
4. تجميع نتائج العمّال عبر as_completed فور اكتمال كل منها

إعادة استخدام الاتصال عبر Session لكل Thread

فتح اتصال TCP جديد مع كل طلب يهدر وقتاً ثميناً عند العمل بالحجم الكبير. الأفضل أن يحتفظ كل Thread بجلسة requests.Session خاصة به ويعيد استخدامها، بحيث تبقى الاتصالات مفتوحة ضمن تجمّع واحد لكل عامل. نستخدم هنا تخزيناً محلياً للخيط (thread-local) كي لا تتداخل الجلسات بين العمال:

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

تمنحك هذه البنية ثلاث فوائد مباشرة:

• إعادة استخدام اتصالات HTTP بدل فتحها من جديد في كل طلب
• عزل الجلسات بين العمّال دون أي تعارض
• إعادة محاولة تلقائية محدودة عبر إعداد المحوّل (adapter)

استخدام map() للمعالجة الدُفعية البسيطة

حين لا تحتاج إلى منطق منفصل لمعالجة الخطأ في كل مهمة، تختصر executor.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)}")

حماية المهلة الزمنية للمهام

مهمة واحدة عالقة قد تحجز عاملاً وتُبطئ الدفعة كلها. لذلك نضع مهلتين: مهلة كلية على مستوى as_completed تمنع الانتظار اللانهائي، ومهلة لكل مهمة عبر future.result(timeout=...) تُنهي أي مسار تنفيذ متعثّر قبل أن يعطّل التجمّع:

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

نصيحة: اضبط مهلة كل مهمة أعلى قليلاً من متوسط وقت الحل المتوقع، لا مساوية له، حتى لا تُلغى مهام كانت ستكتمل بعد ثوانٍ قليلة.

متابعة التقدّم لحظة بلحظة

عند تشغيل دفعات كبيرة يفيد أن ترى نسبة الإنجاز أثناء العمل بدل انتظار النهاية. نحمي العدّاد المشترك بقفل Lock كي لا يتصادم العمّال عند تحديثه، ونطبع نسبة مئوية متجدّدة على السطر نفسه:

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

كيف تختار عدد العمال (max_workers)؟

عدد العمال هو أهم رقم تضبطه، وهو مرتبط مباشرة بعدد الخيوط (threads) المتاحة في خطتك على CaptchaAI، لأن التسعير قائم على الخيوط المتزامنة لا على عدد عمليات الحل. القاعدة العملية: لا تجعل max_workers أكبر من عدد خيوط خطتك، وإلا ستقف الطلبات الزائدة في قائمة انتظار بلا فائدة.

تخيّل فريقاً في متجر إلكتروني بمنطقة الخليج يراقب أسعار آلاف صفحات المنافسين يومياً، وكل صفحة محميّة بـ reCAPTCHA. مع خطة ADVANCE ($90/شهر، 50 خيطاً) يمكن ضبط max_workers=50 لمعالجة خمسين صفحة في آنٍ واحد؛ أما إذا بدأ الفريق بخطة BASIC ($15/شهر، 5 خيوط)، فإبقاء max_workers=5 يمنع ازدحام الطلبات. ابدأ عند 10 عمّال وارفع الرقم تدريجياً مع مراقبة معدلات الخطأ وسعة خطتك.

ThreadPoolExecutor مقابل asyncio: متى تستخدم كلاً منهما

كلا المقاربتين توفّران التوازي، لكن الفرق في التكلفة الهندسية. ThreadPoolExecutor يندمج مع كود متزامن قائم، بينما يفرض asyncio سلسلة دوال غير متزامنة من البداية للنهاية:

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

اختر ThreadPoolExecutor في هذه الحالات

• تكون قاعدة الكود الحالية لديك متزامنة
• تعتمد على مكتبات لا تدعم البرمجة غير المتزامنة مثل Selenium وبعض أطر الـ ORM
• تريد توازياً سريعاً دون إعادة هيكلة المشروع

اختر asyncio في هذه الحالات

• تبني المشروع من الصفر
• تكون أقصى كفاءة مطلوبة (عدد أقل من خيوط نظام التشغيل)
• تعمل أصلاً داخل إطار غير متزامن مثل FastAPI أو aiohttp

معالجة الأخطاء الشائعة

أسئلة شائعة

كم عدد العمال المناسب لخطتي في CaptchaAI؟

اجعل max_workers مساوياً لعدد الخيوط في خطتك أو أقل منه. خطة BASIC ($15/شهر) تتيح 5 خيوط، فيناسبها max_workers=5، بينما تتيح ADVANCE ($90/شهر) خمسين خيطاً تسمح بخمسين عاملاً متزامناً. تجاوز هذا الحد لا يزيد الإنتاجية بل يضع الطلبات في قائمة انتظار فقط.

هل يعمل ThreadPoolExecutor مع Selenium؟

نعم، وهذه إحدى أقوى حالاته. Selenium مكتبة متزامنة لا تدعم asyncio، لذا يمنحك ThreadPoolExecutor توازياً في حل CAPTCHA دون تغيير طريقة قيادتك للمتصفح. احرص فقط على أن يمتلك كل Thread نسخة المتصفح أو الجلسة الخاصة به.

ماذا أفعل عند ارتفاع أخطاء ConnectionError فجأة؟

غالباً يعني ذلك أن عدد الاتصالات المتزامنة تجاوز ما تحتمله بيئتك أو تجمّع الاتصالات. قلّل max_workers، وفعّل تجميع الاتصالات عبر requests.Session كما في المثال أعلاه، وارفع العدد تدريجياً مع مراقبة معدل الأخطاء.

هل يصلح هذا النمط لكل أنواع CAPTCHA أم لـ reCAPTCHA فقط؟

النمط نفسه محايد للنوع: تغيّر قيمة method والمعاملات فقط. يدعم CaptchaAI عبر النمط ذاته reCAPTCHA v2/v3 وCloudflare Turnstile وGeeTest v3 والصور وشبكات الصور، بينما لا يدعم حالياً hCaptcha أو FunCaptcha. تحقّق من نوعك في وثائق CaptchaAI قبل البناء.

الخطوات التالية

• البدء السريع مع CaptchaAI: حلّ أول كابتشا في 5 دقائق
• كيفية حلّ reCAPTCHA v2 عبر الـ API: دليل خطوة بخطوة
• كيفية حل Cloudflare Turnstile باستخدام واجهة API
• كيفية حل GeeTest v3 باستخدام API

أدلة ذات صلة

• البدء السريع مع CaptchaAI: حلّ أول كابتشا في 5 دقائق
• كيفية حلّ reCAPTCHA v2 عبر الـ API: دليل خطوة بخطوة