Selenium Grid مع CaptchaAI: حل CAPTCHA الموزّع على عدة عقد

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

يغطي ما يلي: توزيع العمل بين Hub والعقد، تشغيل الشبكة عبر Docker Compose، العميل المشترك، الإطلاق المتوازي، قراءة السعة، ثم التوسّع على Kubernetes ونسخة Java.

متى تحتاج فعلاً إلى Selenium Grid؟

جهاز واحد يشغّل Chrome يكفي لعشرات النماذج يومياً. الحاجة إلى شبكة تبدأ مع واحدة من هذه العلامات:

• اختبارات الانحدار الليلية لا تنتهي قبل بداية الدوام.
• الذاكرة تنفد قبل أن تنفد المهام؛ كل مثيل Chrome يطلب 1-2 GB واقعياً.
• تحتاج إلى الجولة نفسها على متصفحات أو إصدارات مختلفة في وقت واحد.
• الفريق موزّع وتريد بنية اختبار مركزية بدل أجهزة متفرقة.

إن لم تنطبق أي علامة، فالمسار المفرد في دليل Selenium وPython مع CaptchaAI أبسط وأرخص.

كيف يوزّع Selenium Grid العمل بين Hub والعقد

الـ Hub موجّه لا أكثر: يستقبل طلب الجلسة ويسنده إلى أول عقدة فيها فتحة حرة، والعقد لا تتنافس إلا على موارد جهازها المحلي. أما مفتاح الـ API فواحد للجميع؛ لا داعي لمفتاح لكل عقدة، لأن حساب الـ threads يُدار على مستوى الحساب لا الجهاز.

┌─────────────┐     ┌──────────────┐     ┌──────────────┐
│  Test Script │────▶│  Grid Hub    │────▶│  Node 1      │
│  (Client)    │     │  (Router)    │     │  Chrome x 5  │
└─────────────┘     └──────────────┘     └──────────────┘
                           │              ┌──────────────┐
                           ├─────────────▶│  Node 2      │
                           │              │  Chrome x 5  │
                           │              └──────────────┘
                           │              ┌──────────────┐
                           └─────────────▶│  Node 3      │
                                          │  Chrome x 5  │
                                          └──────────────┘

All nodes share ──▶ CaptchaAI API (single API key)

الخطوة 1: شغّل الشبكة عبر Docker Compose

الملف التالي يرفع Hub وثلاث عقد Chrome بخمس جلسات لكل عقدة — 15 جلسة متزامنة. ثبّت رقم الإصدار نفسه على الـ Hub والعقد معاً؛ اختلافه بينهما أشهر سبب لجلسات ترفض الانطلاق دون رسالة واضحة.

version: "3"
services:
  selenium-hub:
    image: selenium/hub:4.21.0
    container_name: selenium-hub
    ports:

      - "4442:4442"
      - "4443:4443"
      - "4444:4444"

  chrome-node-1:
    image: selenium/node-chrome:4.21.0
    depends_on:

      - selenium-hub
    environment:

      - SE_EVENT_BUS_HOST=selenium-hub
      - SE_EVENT_BUS_PUBLISH_PORT=4442
      - SE_EVENT_BUS_SUBSCRIBE_PORT=4443
      - SE_NODE_MAX_SESSIONS=5
      - SE_NODE_OVERRIDE_MAX_SESSIONS=true

  chrome-node-2:
    image: selenium/node-chrome:4.21.0
    depends_on:

      - selenium-hub
    environment:

      - SE_EVENT_BUS_HOST=selenium-hub
      - SE_EVENT_BUS_PUBLISH_PORT=4442
      - SE_EVENT_BUS_SUBSCRIBE_PORT=4443
      - SE_NODE_MAX_SESSIONS=5
      - SE_NODE_OVERRIDE_MAX_SESSIONS=true

  chrome-node-3:
    image: selenium/node-chrome:4.21.0
    depends_on:

      - selenium-hub
    environment:

      - SE_EVENT_BUS_HOST=selenium-hub
      - SE_EVENT_BUS_PUBLISH_PORT=4442
      - SE_EVENT_BUS_SUBSCRIBE_PORT=4443
      - SE_NODE_MAX_SESSIONS=5
      - SE_NODE_OVERRIDE_MAX_SESSIONS=true

ثم ارفع المجموعة في الخلفية:

docker-compose up -d

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

الخطوة 2: عميل CaptchaAI الذي تتشاركه العقد

العميل التالي يفصل بين مسؤوليتين: فتح جلسة بعيدة على الشبكة، وحل الاختبار عبر CaptchaAI ثم حقن الرمز في الصفحة. الإرسال إلى in.php، والاستطلاع الدوري للنتيجة عبر res.php حتى يعود الرمز. استبدل YOUR_API_KEY بمفتاحك من لوحة التحكم، ومرّره كمتغيّر بيئة إلى الحاوية بدل كتابته داخل الملف.

import requests
import time
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
from concurrent.futures import ThreadPoolExecutor, as_completed


class GridCaptchaSolver:
    CAPTCHAAI_URL = "https://ocr.captchaai.com"

    def __init__(self, api_key, grid_url="http://localhost:4444"):
        self.api_key = api_key
        self.grid_url = grid_url

    def create_session(self):
        """Create a new browser session on the Grid."""
        options = webdriver.ChromeOptions()
        options.add_argument("--no-sandbox")
        options.add_argument("--disable-blink-features=AutomationControlled")
        options.add_argument("--window-size=1920,1080")

        driver = webdriver.Remote(
            command_executor=self.grid_url,
            options=options,
        )
        return driver

    def solve_recaptcha_v2(self, site_url, sitekey):
        """Solve reCAPTCHA v2 via CaptchaAI API."""
        # Submit
        resp = requests.post(f"{self.CAPTCHAAI_URL}/in.php", data={
            "key": self.api_key,
            "method": "userrecaptcha",
            "googlekey": sitekey,
            "pageurl": site_url,
            "json": 1,
        })
        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.CAPTCHAAI_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 solve_turnstile(self, site_url, sitekey):
        resp = requests.post(f"{self.CAPTCHAAI_URL}/in.php", data={
            "key": self.api_key, "method": "turnstile",
            "sitekey": sitekey, "pageurl": site_url, "json": 1,
        })
        data = resp.json()
        if data["status"] != 1:
            raise Exception(f"Submit: {data['request']}")

        task_id = data["request"]
        for _ in range(60):
            time.sleep(5)
            resp = requests.get(f"{self.CAPTCHAAI_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 process_task(self, task):
        """Process a single CAPTCHA-protected task on a Grid node."""
        driver = self.create_session()

        try:
            driver.get(task["url"])
            time.sleep(2)

            # Detect sitekey
            sitekey = task.get("sitekey")
            if not sitekey:
                sitekey = driver.execute_script(
                    "return document.querySelector('[data-sitekey]')?.getAttribute('data-sitekey')"
                )

            if not sitekey:
                return {"url": task["url"], "status": "no_captcha", "data": driver.page_source[:500]}

            # Solve
            token = self.solve_recaptcha_v2(task["url"], sitekey)

            # Inject
            driver.execute_script(f"""
                document.querySelector('#g-recaptcha-response').value = '{token}';
                document.querySelectorAll('[name="g-recaptcha-response"]').forEach(
                    el => el.value = '{token}'
                );
            """)

            # Fill form and submit
            if task.get("form_data"):
                for field, value in task["form_data"].items():
                    driver.find_element(By.NAME, field).send_keys(value)

            if task.get("submit_selector"):
                driver.find_element(By.CSS_SELECTOR, task["submit_selector"]).click()
                time.sleep(3)

            return {
                "url": task["url"],
                "status": "success",
                "result_url": driver.current_url,
                "data": driver.page_source[:1000],
            }

        except Exception as e:
            return {"url": task["url"], "status": "error", "error": str(e)}

        finally:
            driver.quit()

تفصيلتان تفرقان بين سكربت يعمل مرة وآخر يعمل كل ليلة: إغلاق الجلسة داخل finally مهما كانت النتيجة حتى لا تتراكم جلسات قديمة تلتهم الفتحات، واستخراج مفتاح الموقع من الصفحة عند غيابه بدل افتراض قيمة ثابتة.

الخطوة 3: أطلق المهام بالتوازي

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

def run_parallel_tasks(api_key, tasks, max_workers=10):
    """Run CAPTCHA tasks in parallel across Grid nodes."""
    solver = GridCaptchaSolver(api_key)
    results = []

    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = {
            executor.submit(solver.process_task, task): task
            for task in tasks
        }

        for future in as_completed(futures):
            task = futures[future]
            try:
                result = future.result(timeout=600)
                results.append(result)
                print(f"[{result['status']}] {result['url']}")
            except Exception as e:
                results.append({
                    "url": task["url"],
                    "status": "exception",
                    "error": str(e),
                })

    return results


# Usage
tasks = [
    {
        "url": "https://site-a.com/form",
        "sitekey": "6Le-wvkSAAAAAPBMRTvw0Q4Muexq9bi0DJwx_mJ-",
        "form_data": {"name": "Test User", "email": "[email protected]"},
        "submit_selector": "#submit",
    },
    {
        "url": "https://site-b.com/register",
        "sitekey": "6LdKlZEpAAAAAAOQjzC2v_mJ-",
        "form_data": {"username": "testuser"},
        "submit_selector": "button[type='submit']",
    },
    # Add more tasks...
]

results = run_parallel_tasks("YOUR_API_KEY", tasks, max_workers=15)

# Summary
success = sum(1 for r in results if r["status"] == "success")
print(f"\nCompleted: {success}/{len(results)} successful")

الخطوة 4: اقرأ سعة الشبكة قبل رفع التوازي

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

import requests

def check_grid_status(grid_url="http://localhost:4444"):
    """Check Selenium Grid status and available nodes."""
    try:
        resp = requests.get(f"{grid_url}/status")
        data = resp.json()

        nodes = data.get("value", {}).get("nodes", [])
        total_slots = 0
        available_slots = 0

        print(f"Grid Status: {data['value']['ready']}")
        print(f"Nodes: {len(nodes)}")

        for i, node in enumerate(nodes):
            slots = node.get("slots", [])
            free = sum(1 for s in slots if not s.get("session"))
            total_slots += len(slots)
            available_slots += free
            print(f"  Node {i+1}: {free}/{len(slots)} slots available")

        print(f"Total capacity: {available_slots}/{total_slots} available")
        return available_slots

    except Exception as e:
        print(f"Grid check failed: {e}")
        return 0


# Adjust workers based on grid capacity
available = check_grid_status()
optimal_workers = min(available, 20)
print(f"Optimal workers: {optimal_workers}")

مواءمة عدد العمال مع خطة الـ threads

القاعدة بسيطة: عدد الـ threads في حسابك لا يقل عن عدد الجلسات التي قد تنتظر رمزاً في اللحظة نفسها. الفوترة عند CaptchaAI على الـ threads المتزامنة لا على عمليات الحل، وكل خطة تتيح عمليات حل غير محدودة لكل thread شهرياً، فالسقف هو التزامن لا الاستهلاك.

الأسعار بالدولار الأمريكي شهرياً؛ راجع صفحة الأسعار الرسمية قبل أي ترقية. وإذا زادت الفتحات عن الـ threads المتاحة فستدفع ثمن أجهزة تنتظر دورها.

الخطوة 5: التوسّع التلقائي على Kubernetes

حين يتغيّر حجم الجولات بين ليلة وأخرى، انقل العقد إلى Kubernetes واترك ضبط العدد لـ HPA حسب استهلاك المعالج:

# selenium-grid-k8s.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: selenium-chrome-node
spec:
  replicas: 5
  selector:
    matchLabels:
      app: selenium-chrome
  template:
    metadata:
      labels:
        app: selenium-chrome
    spec:
      containers:

        - name: chrome
          image: selenium/node-chrome:4.21.0
          env:

            - name: SE_EVENT_BUS_HOST
              value: selenium-hub

            - name: SE_EVENT_BUS_PUBLISH_PORT
              value: "4442"

            - name: SE_EVENT_BUS_SUBSCRIBE_PORT
              value: "4443"

            - name: SE_NODE_MAX_SESSIONS
              value: "3"
          resources:
            limits:
              memory: "2Gi"
              cpu: "1"
            requests:
              memory: "1Gi"
              cpu: "500m"
---
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: chrome-node-hpa
spec:
  scaleRef:
    apiVersion: apps/v1
    kind: Deployment
    name: selenium-chrome-node
  minReplicas: 2
  maxReplicas: 20
  metrics:

    - type: Resource
      resource:
        name: cpu
        target:
          type: Utilization
          averageUtilization: 70

اضبط maxReplicas حتى لا يزيد التوازي الناتج عن عدد الـ threads في خطتك، وأبقِ minReplicas عند عقدتين حتى لا تبدأ كل جولة بانتظار سحب الصور.

نفس المسار في Java

فرق JVM التي تشغّل TestNG أو JUnit لا تحتاج إلى تغيير لغة الأتمتة؛ البنية نفسها بـ RemoteWebDriver وExecutorService، ويؤدي processTask الإرسال إلى in.php والاستطلاع الدوري لـ res.php بمنطق نسخة Python نفسه:

import org.openqa.selenium.WebDriver;
import org.openqa.selenium.remote.RemoteWebDriver;
import org.openqa.selenium.chrome.ChromeOptions;
import java.net.URL;
import java.net.http.*;
import java.net.URI;
import java.util.concurrent.*;

public class GridCaptchaSolver {
    private final String apiKey;
    private final String gridUrl;
    private final HttpClient httpClient;

    public GridCaptchaSolver(String apiKey, String gridUrl) {
        this.apiKey = apiKey;
        this.gridUrl = gridUrl;
        this.httpClient = HttpClient.newHttpClient();
    }

    public WebDriver createSession() throws Exception {
        ChromeOptions options = new ChromeOptions();
        options.addArguments("--no-sandbox", "--window-size=1920,1080");
        return new RemoteWebDriver(new URL(gridUrl), options);
    }

    public List<Map<String, String>> runParallel(
        List<Map<String, String>> tasks, int workers
    ) throws Exception {
        ExecutorService executor = Executors.newFixedThreadPool(workers);
        List<Future<Map<String, String>>> futures = new ArrayList<>();

        for (Map<String, String> task : tasks) {
            futures.add(executor.submit(() -> processTask(task)));
        }

        List<Map<String, String>> results = new ArrayList<>();
        for (Future<Map<String, String>> future : futures) {
            results.add(future.get(600, TimeUnit.SECONDS));
        }

        executor.shutdown();
        return results;
    }
}

سيناريو من السوق المحلي: ليلة تحقق قبل موسم الجمعة البيضاء

تخيّل فريق ضمان جودة في القاهرة يخدم متجراً خليجياً، أمامه 120 مساراً محمياً بـ reCAPTCHA v2 بين التسجيل واسترجاع كلمة المرور وإتمام الشراء، ونافذة تشغيل من 1:00 إلى 4:00 صباحاً بتوقيت مكة قبل ذروة الطلبات. على جهاز واحد تخرج الجولة عن النافذة وتُقطع في منتصفها.

الترتيب الذي يعالج هذا: ست عقد Chrome على خادمين قريبين زمنياً من الخليج ومصر، 30 فتحة، عدد عمال أقل قليلاً من الفتحات لترك هامش لإعادة المحاولة، وخطة threads تغطي التوازي. وتفصيلة تنقذ ليلة كاملة: ثبّت المنطقة الزمنية داخل الحاويات، لأن اختبارات العروض والقسائم المعتمدة على تاريخ محلي تفشل بصمت حين تعمل العقد على UTC ويقرأ المتجر توقيت الرياض.

أعطال شائعة في Selenium Grid وكيف تعالجها

قائمة تحقق قبل أول جولة ليلية

• مفتاح الـ API يصل عبر متغيّر بيئة، ولا نسخة منه في المستودع.
• عدد العمال محسوب من الفتحات الحرة لحظة الإطلاق، لا رقم ثابت.
• كل مهمة تكتب نتيجتها ورابطها في سجل يُقرأ صباحاً.
• تنبيه عند سقوط عقدة، فالجولة تستمر بصمت بفتحات أقل.

أسئلة شائعة

كم thread أحتاج مقابل حجم الشبكة؟

اربط الرقم بالتوازي لا بعدد المهام: إذا كانت 24 جلسة قد تنتظر رمزاً في اللحظة نفسها، فاحسب 24 thread على الأقل. عدد عمليات الحل الشهري لا يغيّر الفاتورة، لأن الفوترة على الـ threads المتزامنة.

أشغّل الشبكة على خادم واحد كبير أم عدة خوادم أصغر؟

عدة خوادم أصغر أفضل غالباً: سقوط عقدة لا يوقف الجولة، وتوزيع الذاكرة أسهل. احسب نحو 2 GB لكل جلسة Chrome مع هامش للنظام.

ماذا أفعل حين تنتهي مهلة الحل وكل الفتحات مشغولة؟

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

هل يمكن تشغيل هذا الإعداد داخل CI؟

نعم، لكن اجعل الشبكة خدمة دائمة لا حاويات تُرفع وتُهدم مع كل تشغيل، واجعل خطوة الـ CI عميلاً يتصل بـ webdriver.Remote فقط؛ هذا يختصر زمن التشغيل ويوحّد سلوك الاختبارات.

أي أنواع CAPTCHA أعتمد عليها في هذا المسار؟

يحل CaptchaAI اختبارات reCAPTCHA v2 وv3 بصيغها المختلفة، وCloudflare Turnstile وCloudflare Challenge، وGeeTest v3، واختبارات الصور وOCR وGrid Image وBLS، إضافة إلى CaptchaFox وFriendly Captcha وLemin في مرحلة beta. أما hCaptcha وFunCaptcha فلا يدعمهما CaptchaAI حالياً، وGeeTest v4 معلن على أنه قريباً ولم يُتَح بعد.

أدلة ذات صلة

• تكامل Selenium مع CaptchaAI في Python
• البدء السريع: أول اختبار CAPTCHA محلول

جهّز الشبكة، ثم افتح حسابك على CaptchaAI واربط العقد بمفتاح واحد قبل أول جولة ليلية.