Selenium Wire و CaptchaAI: اعتراض الطلبات لحل CAPTCHA

أدقّ طريقة لمعرفة نوع CAPTCHA في صفحة ما ليست قراءة الـ DOM، بل مراقبة ما يطلبه المتصفح فعلياً من الشبكة. Selenium Wire طبقة رقيقة فوق Selenium WebDriver تمنحك قائمة driver.requests قابلة للقراءة والتعديل: منها تستخرج مفتاح الموقع وعنوان الصفحة، ثم ترسلهما إلى CaptchaAI عبر in.php، وتستطلع النتيجة من res.php، وتحقن الرمز في النموذج قبل الضغط على زر الإرسال.

الدليل موجّه لمهندس Python يشغّل اختبارات جودة أو مهام جمع بيانات مصرّحاً بها، ويريد مساراً واحداً يعمل على reCAPTCHA v2 وv3 وCloudflare Turnstile بدل منطق منفصل لكل صفحة.

لماذا يخذلك اكتشاف CAPTCHA من الـ DOM وحده

حين تعتمد على find_element وحده فأنت تقرأ ما وصل إلى شجرة العناصر لحظة نظرك. طبقة الشبكة أصدق وأبكر:

• يُحمَّل ودجت reCAPTCHA داخل iframe، فيغيب data-sitekey عن الصفحة الأم.
• تسحب بعض الصفحات إعدادات التحقق عبر AJAX بعد ثوانٍ من التحميل.
• في reCAPTCHA v3 لا ودجت مرئي أصلاً؛ الدليل الوحيد استدعاء يحمل render=.
• مع خادم وسيط بطيء يتغيّر ترتيب ظهور العناصر، فينكسر أي انتظار مبني على الـ DOM.

مسار اعتراض الطلبات في أربع خطوات

1. شغّل المتصفح عبر Selenium Wire مع إعدادات البروكسي إن وُجدت.
2. افتح الصفحة، انتظر اكتمال طلبات التحقق، ثم اقرأ driver.requests لتحديد النوع ومفتاح الموقع.
3. أرسل المهمة إلى in.php، واستطلع res.php دورياً حتى يعود الرمز.
4. احقن الرمز في الحقل المخفي المناسب، ثم أرسل النموذج وتحقق من العنوان الناتج.

ما الذي يضيفه Selenium Wire فوق Selenium القياسي

الخطوة 1: ثبّت الحزم

pip install seleniumwire selenium webdriver-manager requests

الحزمة webdriver-manager تعفيك من تثبيت ChromeDriver يدوياً ومطابقة إصداره مع المتصفح.

الخطوة 2: فئة واحدة للاكتشاف والحل والحقن

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

from seleniumwire import webdriver
from selenium.webdriver.chrome.service import Service
from webdriver_manager.chrome import ChromeDriverManager
import requests
import time
import json

class SeleniumWireCaptchaSolver:
    BASE_URL = "https://ocr.captchaai.com"

    def __init__(self, api_key, proxy=None):
        self.api_key = api_key
        self.proxy = proxy
        self.driver = None

    def create_driver(self, headless=True):
        options = webdriver.ChromeOptions()
        if headless:
            options.add_argument("--headless=new")
        options.add_argument("--no-sandbox")
        options.add_argument("--disable-blink-features=AutomationControlled")
        options.add_argument("--window-size=1920,1080")

        wire_options = {}
        if self.proxy:
            wire_options["proxy"] = {
                "http": self.proxy,
                "https": self.proxy,
            }

        self.driver = webdriver.Chrome(
            service=Service(ChromeDriverManager().install()),
            options=options,
            seleniumwire_options=wire_options,
        )
        return self.driver

    def detect_captcha_from_requests(self):
        """Detect CAPTCHA type from intercepted network requests."""
        for request in self.driver.requests:
            url = request.url

            if "recaptcha/api2/anchor" in url or "recaptcha/enterprise/anchor" in url:
                # Extract sitekey from reCAPTCHA iframe URL
                import re
                match = re.search(r"k=([A-Za-z0-9_-]+)", url)
                if match:
                    return {
                        "type": "recaptcha_v2",
                        "sitekey": match.group(1),
                        "page_url": self.driver.current_url,
                    }

            if "recaptcha/api2/reload" in url or "recaptcha/enterprise/reload" in url:
                return {
                    "type": "recaptcha_v3",
                    "sitekey": self._extract_v3_sitekey(),
                    "page_url": self.driver.current_url,
                }

            if "challenges.cloudflare.com" in url:
                sitekey = self._extract_turnstile_sitekey()
                if sitekey:
                    return {
                        "type": "turnstile",
                        "sitekey": sitekey,
                        "page_url": self.driver.current_url,
                    }

        return None

    def _extract_v3_sitekey(self):
        for request in self.driver.requests:
            if "recaptcha" in request.url and "render=" in request.url:
                import re
                match = re.search(r"render=([A-Za-z0-9_-]+)", request.url)
                if match:
                    return match.group(1)
        return self.driver.execute_script(
            "return document.querySelector('[data-sitekey]')?.getAttribute('data-sitekey')"
        )

    def _extract_turnstile_sitekey(self):
        return self.driver.execute_script(
            "return document.querySelector('.cf-turnstile[data-sitekey]')?.getAttribute('data-sitekey')"
        )

    def solve_captcha(self, captcha_info):
        """Solve detected CAPTCHA via CaptchaAI API."""
        params = {"key": self.api_key, "json": 1}

        if captcha_info["type"] == "recaptcha_v2":
            params.update({
                "method": "userrecaptcha",
                "googlekey": captcha_info["sitekey"],
                "pageurl": captcha_info["page_url"],
            })
        elif captcha_info["type"] == "recaptcha_v3":
            params.update({
                "method": "userrecaptcha",
                "googlekey": captcha_info["sitekey"],
                "pageurl": captcha_info["page_url"],
                "version": "v3",
                "action": "verify",
            })
        elif captcha_info["type"] == "turnstile":
            params.update({
                "method": "turnstile",
                "key": captcha_info["sitekey"],
                "pageurl": captcha_info["page_url"],
            })

        # Submit
        resp = requests.post(f"{self.BASE_URL}/in.php", data=params)
        data = resp.json()
        if data["status"] != 1:
            raise Exception(f"Submit: {data['request']}")

        task_id = data["request"]

        # Poll
        for _ in range(60):
            time.sleep(5)
            resp = requests.get(f"{self.BASE_URL}/res.php", params={
                "key": self.api_key,
                "action": "get",
                "id": task_id,
                "json": 1,
            })
            data = resp.json()
            if data["request"] == "CAPCHA_NOT_READY":
                continue
            if data["status"] != 1:
                raise Exception(f"Solve: {data['request']}")
            return data["request"]

        raise Exception("Timeout")

    def inject_token(self, captcha_type, token):
        """Inject solved token into the page."""
        if captcha_type in ("recaptcha_v2", "recaptcha_v3"):
            self.driver.execute_script(f"""
                document.querySelector('#g-recaptcha-response').value = '{token}';
                // Try hidden textareas in iframes
                document.querySelectorAll('[name="g-recaptcha-response"]').forEach(
                    el => el.value = '{token}'
                );
            """)
        elif captcha_type == "turnstile":
            self.driver.execute_script(f"""
                const input = document.querySelector('[name="cf-turnstile-response"]');
                if (input) input.value = '{token}';
            """)

    def close(self):
        if self.driver:
            self.driver.quit()

الخطوة 3: سيناريو كامل على صفحة تسجيل دخول

يربط المثال التالي الخطوات معاً: تنقّل، تعبئة، اكتشاف، حل، حقن، ثم إرسال.

def automate_login():
    solver = SeleniumWireCaptchaSolver(
        api_key="YOUR_API_KEY",
        proxy="http://user:[email protected]:8080"
    )

    try:
        driver = solver.create_driver(headless=True)

        # Navigate and wait for CAPTCHA resources to load
        driver.get("https://example.com/login")
        time.sleep(3)

        # Fill form
        driver.find_element("id", "email").send_keys("[email protected]")
        driver.find_element("id", "password").send_keys("password123")

        # Detect CAPTCHA from network requests
        captcha = solver.detect_captcha_from_requests()

        if captcha:
            print(f"Found {captcha['type']} (sitekey: {captcha['sitekey'][:20]}...)")
            token = solver.solve_captcha(captcha)
            solver.inject_token(captcha["type"], token)
            print(f"Token injected: {token[:50]}...")

        # Submit
        driver.find_element("id", "submit").click()
        time.sleep(3)

        print(f"Current URL: {driver.current_url}")

    finally:
        solver.close()

automate_login()

سيناريو من السوق العربي: اختبار ليلي لبوابة حجز بواجهة RTL

افترض فريق جودة في القاهرة يشغّل اختبار انحدار ليلياً على بوابة حجز عربية بواجهة RTL تضع Cloudflare Turnstile أمام خطوة تأكيد الحجز. المشكلة هنا ليست الحل بل التوقيت: تُحمَّل الخطوط العربية وسكربتات التتبّع قبل ودجت التحقق، فيصل الاختبار إلى زر الإرسال قبل ظهور الحقل المخفي. مع اعتراض الطلبات يصبح الشرط صريحاً — لا ترسل النموذج قبل ظهور طلب challenges.cloudflare.com في driver.requests.

المنطق نفسه يفيد فرق التجارة الإلكترونية في الخليج التي تختبر إتمام الشراء في ذروة المساء. أما التكلفة فمرتبطة بعدد الجلسات المتوازية لا بعدد الحلول: تبدأ BASIC من $15 شهرياً مع 5 threads، وترفع STANDARD العدد إلى 15 thread عند $30، وتصل ADVANCE إلى 50 thread عند $90 لفريق يشغّل عشرات المتصفحات ليلاً. والحلول داخل الـ thread الواحد غير محدودة.

اعتراض الطلبات وتعديلها أثناء التشغيل

أضف رؤوساً مخصصة لكل طلب

def add_headers_interceptor(request):
    request.headers["X-Custom-Header"] = "value"
    request.headers["Accept-Language"] = "en-US,en;q=0.9"

driver = solver.create_driver()
driver.request_interceptor = add_headers_interceptor

مفيد لتثبيت لغة الواجهة أو تمرير رأس تعريفي تميّز به حركة الاختبار.

أوقف سكربتات التتبّع لتسريع الجلسة

def block_trackers(request):
    blocked = ["analytics", "tracking", "ads", "facebook.net", "doubleclick"]
    if any(b in request.url for b in blocked):
        request.abort()

driver.request_interceptor = block_trackers

إسقاط طلبات التحليلات والإعلانات يقصّر زمن التحميل ويقلّل الاختبارات المتذبذبة، خصوصاً حين تمرّ كل الحركة عبر خادم وسيط.

عدّل جسم الاستجابة قبل وصولها إلى الصفحة

def modify_response(request, response):
    if "captcha-config" in request.url:
        # Modify CAPTCHA configuration response
        import json
        body = json.loads(response.body.decode("utf-8"))
        body["difficulty"] = "easy"
        response.body = json.dumps(body).encode("utf-8")

driver.response_interceptor = modify_response

استعمل هذا النمط داخل بيئتك أو بيئة اختبار تملك صلاحية عليها فقط؛ الهدف محاكاة حالات محددة أثناء اختبار الجودة، لا تغيير سلوك مواقع الآخرين.

اقرأ معلمات CAPTCHA من استدعاءات AJAX

بعض الصفحات لا تضع مفتاح الموقع في الـ HTML، بل تجلبه ضمن استجابة JSON بعد التحميل. الدالة التالية تراقب الشبكة مدة محدودة وتلتقط أول استجابة تحمل مفتاح موقع:

def extract_captcha_from_api(driver, timeout=10):
    """Watch network for CAPTCHA configuration API calls."""
    start = time.time()

    while time.time() - start < timeout:
        for request in driver.requests:
            if request.response and "captcha" in request.url.lower():
                try:
                    data = json.loads(request.response.body.decode())
                    if "sitekey" in str(data) or "siteKey" in str(data):
                        return data
                except (json.JSONDecodeError, UnicodeDecodeError):
                    pass
        time.sleep(0.5)

    return None

دوّر البروكسي بين الجلسات

يقبل Selenium Wire إعداد بروكسي بمصادقة مباشرة في seleniumwire_options، وهو ما يعجز عنه Selenium القياسي دون إضافات:

from seleniumwire import webdriver

proxies = [
    "http://user:[email protected]:8080",
    "http://user:[email protected]:8080",
    "http://user:[email protected]:8080",
]

current_proxy = [0]

def rotate_proxy(request):
    proxy = proxies[current_proxy[0] % len(proxies)]
    request.headers["Proxy-Authorization"] = None  # Reset
    current_proxy[0] += 1

wire_options = {
    "proxy": {
        "http": proxies[0],
        "https": proxies[0],
    }
}

driver = webdriver.Chrome(seleniumwire_options=wire_options)

اجعل الخادم الوسيط ثابتاً طوال الجلسة الواحدة: تغيّر العنوان بين لحظة الحل ولحظة إرسال النموذج سبب شائع لرفض الرمز رغم صحته.

حلّل حركة CAPTCHA قبل أن تتّهم الخدمة

قبل فتح تذكرة دعم، اعرف أين ذهب الوقت: في تحميل الودجت، أم في الاستطلاع الدوري، أم في الخادم الوسيط. تلخّص الدالة التالية طلبات التحقق مع رموز الحالة والأحجام:

def analyze_captcha_traffic(driver):
    """Analyze all CAPTCHA-related network traffic."""
    captcha_requests = []

    for request in driver.requests:
        if any(k in request.url for k in ["recaptcha", "turnstile", "hcaptcha", "geetest"]):
            captcha_requests.append({
                "url": request.url,
                "method": request.method,
                "status": request.response.status_code if request.response else None,
                "content_type": request.response.headers.get("Content-Type") if request.response else None,
                "size": len(request.response.body) if request.response and request.response.body else 0,
            })

    print(f"\nCAPTCHA Network Traffic ({len(captcha_requests)} requests):")
    for req in captcha_requests:
        print(f"  [{req['status']}] {req['method']} {req['url'][:80]}...")

    return captcha_requests

صدّر ملف HAR لتشخيص الأعطال المتقطعة

from seleniumwire.utils import decode

def export_har(driver, filename="captcha_traffic.har"):
    """Export HAR file for debugging CAPTCHA issues."""
    import json

    har = {
        "log": {
            "version": "1.2",
            "entries": []
        }
    }

    for request in driver.requests:
        if not request.response:
            continue

        entry = {
            "request": {
                "method": request.method,
                "url": request.url,
                "headers": [{"name": k, "value": v} for k, v in request.headers.items()],
            },
            "response": {
                "status": request.response.status_code,
                "headers": [{"name": k, "value": v} for k, v in request.response.headers.items()],
            },
        }
        har["log"]["entries"].append(entry)

    with open(filename, "w") as f:
        json.dump(har, f, indent=2)

    print(f"HAR exported: {filename} ({len(har['log']['entries'])} entries)")

افتح الملف في لوحة الشبكة داخل أي متصفح وقارن جلسة ناجحة بأخرى فاشلة؛ الفارق الزمني يكشف السبب أسرع من أي سجل نصي.

أعطال شائعة وطريقة إصلاحها

قائمة تحقق قبل النقل إلى الإنتاج

• خزّن مفتاح الـ API في متغيّر بيئة، وأبقِ YOUR_API_KEY قيمة نائبة في المستودع.
• ضع حداً أقصى لدورات الاستطلاع بدل الانتظار المفتوح، وسجّل وقت كل حل.
• أبقِ عدد المتصفحات المتوازية ضمن عدد الـ threads في خطتك؛ الزائد ينتظر في الطابور.
• أفرغ driver.requests بين الحالات حتى لا يلتقط الاكتشاف مفتاح موقع من صفحة سابقة.

الأسئلة الشائعة

هل يحل CaptchaAI كل نوع تحدٍّ قد يظهر في الشبكة؟

لا. المسار أعلاه يغطي عائلة reCAPTCHA بنسخها v2 وInvisible وEnterprise وv3، وCloudflare Turnstile وChallenge، وGeeTest v3، وتحديات الصور وشبكات الصور وBLS CAPTCHA. أما hCaptcha وFunCaptcha فغير مدعومين، وGeeTest v4 ما زال «قريباً». وثلاثة أنواع في مرحلة beta: CaptchaFox (beta) وFriendly Captcha (beta) وLemin (beta).

كم thread أحتاج لتشغيل متصفحات متوازية؟

عدد الـ threads يساوي عدد الحلول الجارية في اللحظة نفسها، لا عدد الحلول اليومية. 5 جلسات متزامنة تعمل ضمن BASIC عند $15 شهرياً، و15 جلسة تحتاج STANDARD عند $30، ومزارع الاختبار الليلية تبدأ من ADVANCE عند $90 مع 50 thread.

كيف أتأكد أن الرمز حُقن فعلاً قبل إرسال النموذج؟

اقرأ قيمة الحقل المخفي بعد الحقن عبر execute_script وتأكد أن طولها أكبر من بضع عشرات من الأحرف. إن تكرر الخطأ رغم وجود قيمة، فالسبب غالباً في pageurl المُرسَل إلى CaptchaAI: يجب أن يطابق العنوان الذي ظهر فيه الودجت، بروتوكولاً ومساراً.

هل أحتاج وكيلاً سكنياً لكل جلسة اختبار؟

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

متى ألجأ إلى ملف HAR بدل قراءة السجلات؟

عندما يكون العطل متقطعاً. السجل النصي يقول إن الحل تأخر فقط، بينما يُظهر HAR ترتيب الطلبات وأزمنتها: هل تأخّر ودجت التحقق، أم طال الرد، أم أضاف الخادم الوسيط الثواني؟

أدلة ذات صلة

• الدليل الكامل لتكامل Selenium مع CaptchaAI في Python
• أتمتة CAPTCHA عبر Chrome DevTools Protocol

جهّز مساراً واحداً للاكتشاف والحل بدل منطق منفصل لكل صفحة — احصل على مفتاح CaptchaAI وابدأ من أول اختبار الليلة.