بناء مسار موحّد لحل CAPTCHA لعدة عملاء عبر CaptchaAI

مسار الحل الموحّد هو مكوّن واحد يستقبل طلبات حل CAPTCHA من كل عملائك، يرتّبها في قائمة انتظار، يرسلها إلى CaptchaAI، ثم يعيد الرمز المحلول لمن طلبه. الفائدة المباشرة: تكتب منطق الحل مرة واحدة بدل نسخه في كل مشروع. هذا الدليل يبني هذا المسار من الصفر في Python وNode.js، مع التحكم في التزامن لكل عميل وطريقة موحّدة للتعامل مع الأخطاء.

إذا كنت تدير وكالة صغيرة في القاهرة أو الرياض تخدم عدة عملاء — مراقبة أسعار متجر إلكتروني، جمع بيانات لوحات الوظائف، اختبار نماذج تسجيل — فإن كل مشروع يصطدم عاجلاً بـ reCAPTCHA v2 أو Cloudflare Turnstile. مسار مشترك يعني أنك تُدير مفتاح API واحداً وقائمة انتظار واحدة تخدم الجميع، بدل خمس نسخ متفرقة يصعب صيانتها.

لماذا مسار موحّد بدل كود لكل مشروع؟

مسار واحد مشترك يمنحك ثلاث فوائد مباشرة عند خدمة عملاء متعددين:

• صيانة أبسط — منطق الإرسال والاستطلاع في مكان واحد بدل نسخة تُكرَّر مع كل مشروع.
• تحكّم مركزي في التزامن — تفرض حداً أقصى لكل عميل من إعداد واحد لا من كود متناثر.
• تتبّع أوضح للتكلفة — الـ Threads تُوزَّع من مفتاح واحد، فتعرف حصة كل عميل بدقة.

المتطلبات المسبقة

قبل كتابة أي سطر، جهّز العناصر التالية:

• حساب على CaptchaAI ومفتاح الـ API جاهز في متغيّر بيئة، لا مكتوباً داخل الكود.
• بيئة Python 3 مع حزمة requests، أو بيئة Node.js مع مكتبة axios.
• قائمة بعملائك وأنواع CAPTCHA التي يواجهها كل منهم (reCAPTCHA v2، Cloudflare Turnstile، وغيرها).
• تقدير مبدئي لعدد الطلبات المتزامنة المتوقّعة، لتختار خطة بعدد Threads مناسب.

بنية المسار

┌──────────────┐    ┌───────────────┐    ┌──────────────┐
│  Client A    │──▶ │               │    │              │
│  Client B    │──▶ │  Task Queue   │──▶ │  CaptchaAI   │
│  Client C    │──▶ │               │    │  API         │
└──────────────┘    └───────────────┘    └──────────────┘
                           │                    │
                           ▼                    ▼
                    ┌───────────────┐    ┌──────────────┐
                    │  Result Store │◀── │  Polling      │
                    │  (Redis/DB)   │    │  Workers      │
                    └───────────────┘    └──────────────┘

مكوّنات المسار الأربعة

يتكوّن المسار من أربع طبقات، لكل منها مسؤولية واحدة واضحة:

1. طبقة الاستقبال — تستقبل طلبات الحل القادمة من سكربتات جمع البيانات لدى كل عميل.
2. قائمة الانتظار — تخزّن المهام مؤقتاً وتفرض حداً أقصى للتزامن لكل عميل على حدة.
3. عمال الحل — يرسلون المهام إلى CaptchaAI ويستطلعون النتيجة دورياً حتى تجهز.
4. مخزن النتائج — يحتفظ بالرموز المحلولة ريثما يسحبها العميل صاحب الطلب.

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

بناء المسار في Python

الفئة الأساسية للحل

الفئة التالية تجمع الطبقات الأربع في مكان واحد: enqueue لإضافة مهمة، وsubmit_task للإرسال إلى نقطة النهاية in.php، وpoll_result للاستطلاع الدوري على res.php، وprocess_queue لإدارة الفتحات النشطة ضمن حد التزامن.

import requests
import time
from dataclasses import dataclass
from typing import Optional
from collections import deque
from threading import Lock

SUBMIT_URL = "https://ocr.captchaai.com/in.php"
RESULT_URL = "https://ocr.captchaai.com/res.php"

@dataclass
class SolveRequest:
    client_id: str
    method: str
    params: dict
    callback: Optional[callable] = None

@dataclass
class SolveResult:
    client_id: str
    task_id: str
    token: Optional[str] = None
    error: Optional[str] = None


class CaptchaPipeline:
    def __init__(self, api_key: str, max_concurrent: int = 10):
        self.api_key = api_key
        self.max_concurrent = max_concurrent
        self.queue = deque()
        self.active = {}
        self.lock = Lock()

    def enqueue(self, request: SolveRequest):
        with self.lock:
            self.queue.append(request)

    def submit_task(self, request: SolveRequest) -> Optional[str]:
        data = {
            "key": self.api_key,
            "method": request.method,
            "json": 1,
            **request.params
        }

        try:
            resp = requests.post(SUBMIT_URL, data=data, timeout=15)
            result = resp.json()

            if result.get("status") == 1:
                return result["request"]
            else:
                print(f"[{request.client_id}] Submit error: {result.get('error_text', result.get('request'))}")
                return None
        except requests.RequestException as e:
            print(f"[{request.client_id}] Network error: {e}")
            return None

    def poll_result(self, task_id: str, max_wait: int = 120) -> Optional[str]:
        elapsed = 0
        interval = 5
        while elapsed < max_wait:
            time.sleep(interval)
            elapsed += interval

            try:
                resp = requests.get(RESULT_URL, params={
                    "key": self.api_key,
                    "action": "get",
                    "id": task_id,
                    "json": 1
                }, timeout=10)
                result = resp.json()

                if result.get("status") == 1:
                    return result["request"]
                elif result.get("request") == "CAPCHA_NOT_READY":
                    continue
                else:
                    print(f"Poll error for {task_id}: {result.get('error_text', result.get('request'))}")
                    return None
            except requests.RequestException:
                continue

        return None

    def process_queue(self):
        while self.queue or self.active:
            # Fill active slots
            with self.lock:
                while self.queue and len(self.active) < self.max_concurrent:
                    request = self.queue.popleft()
                    task_id = self.submit_task(request)
                    if task_id:
                        self.active[task_id] = request

            # Poll active tasks
            completed = []
            for task_id, request in list(self.active.items()):
                token = self.poll_result(task_id, max_wait=10)
                if token:
                    result = SolveResult(
                        client_id=request.client_id,
                        task_id=task_id,
                        token=token
                    )
                    if request.callback:
                        request.callback(result)
                    completed.append(task_id)

            with self.lock:
                for task_id in completed:
                    del self.active[task_id]

لاحظ أن client_id مرافق للطلب من لحظة إضافته حتى عودة الرمز المحلول، فلا تختلط نتائج عميل بآخر حتى لو تعالجت مهامهم في الوقت نفسه.

تشغيل عدة عملاء في وقت واحد

هنا يظهر جوهر الفكرة: طلبان من عميلين مختلفين، أحدهما reCAPTCHA v2 والآخر Cloudflare Turnstile، يدخلان القائمة نفسها ويُعالجان بالتوازي. كل طلب يحمل دالة callback تُستدعى فور جهوز الرمز.

pipeline = CaptchaPipeline(api_key="YOUR_API_KEY", max_concurrent=15)

# Client A — reCAPTCHA v2
pipeline.enqueue(SolveRequest(
    client_id="client_a",
    method="userrecaptcha",
    params={
        "googlekey": "6Le-SITEKEY-A",
        "pageurl": "https://client-a-target.com/form"
    },
    callback=lambda r: print(f"[{r.client_id}] Solved: {r.token[:40]}...")
))

# Client B — Turnstile
pipeline.enqueue(SolveRequest(
    client_id="client_b",
    method="turnstile",
    params={
        "sitekey": "0x4AAAA-SITEKEY-B",
        "pageurl": "https://client-b-target.com/login"
    },
    callback=lambda r: print(f"[{r.client_id}] Solved: {r.token[:40]}...")
))

pipeline.process_queue()

بناء المسار في Node.js

النسخة التالية تبني المنطق نفسه على وعود JavaScript: enqueue تُرجع Promise يُحَل عند جهوز الرمز، و_processNext تحرص على ألا يتجاوز عدد المهام النشطة حدّ maxConcurrent. هذا مناسب إذا كانت خدماتك مبنية أصلاً على Node.js.

الفئة الكاملة مع مثال التشغيل

const axios = require("axios");

const SUBMIT_URL = "https://ocr.captchaai.com/in.php";
const RESULT_URL = "https://ocr.captchaai.com/res.php";

class CaptchaPipeline {
  constructor(apiKey, maxConcurrent = 10) {
    this.apiKey = apiKey;
    this.maxConcurrent = maxConcurrent;
    this.queue = [];
    this.activeCount = 0;
  }

  enqueue(clientId, method, params) {
    return new Promise((resolve, reject) => {
      this.queue.push({ clientId, method, params, resolve, reject });
      this._processNext();
    });
  }

  async _processNext() {
    if (this.activeCount >= this.maxConcurrent || this.queue.length === 0) return;

    this.activeCount++;
    const task = this.queue.shift();

    try {
      const token = await this._solve(task);
      task.resolve({ clientId: task.clientId, token });
    } catch (err) {
      task.reject(err);
    } finally {
      this.activeCount--;
      this._processNext();
    }
  }

  async _solve(task) {
    const submitResp = await axios.post(SUBMIT_URL, null, {
      params: {
        key: this.apiKey,
        method: task.method,
        json: 1,
        ...task.params,
      },
      timeout: 15000,
    });

    if (submitResp.data.status !== 1) {
      throw new Error(submitResp.data.error_text || submitResp.data.request);
    }

    const taskId = submitResp.data.request;
    return this._poll(taskId);
  }

  async _poll(taskId, maxWait = 120000) {
    const interval = 5000;
    let elapsed = 0;

    while (elapsed < maxWait) {
      await new Promise((r) => setTimeout(r, interval));
      elapsed += interval;

      try {
        const resp = await axios.get(RESULT_URL, {
          params: {
            key: this.apiKey,
            action: "get",
            id: taskId,
            json: 1,
          },
          timeout: 10000,
        });

        if (resp.data.status === 1) return resp.data.request;
        if (resp.data.request !== "CAPCHA_NOT_READY") {
          throw new Error(resp.data.error_text || resp.data.request);
        }
      } catch (err) {
        if (err.response) throw err;
      }
    }

    throw new Error(`Timeout waiting for task ${taskId}`);
  }
}

// Usage
(async () => {
  const pipeline = new CaptchaPipeline("YOUR_API_KEY", 15);

  const results = await Promise.allSettled([
    pipeline.enqueue("client_a", "userrecaptcha", {
      googlekey: "6Le-SITEKEY-A",
      pageurl: "https://client-a-target.com/form",
    }),
    pipeline.enqueue("client_b", "turnstile", {
      sitekey: "0x4AAAA-SITEKEY-B",
      pageurl: "https://client-b-target.com/login",
    }),
  ]);

  results.forEach((r) => {
    if (r.status === "fulfilled") {
      console.log(`[${r.value.clientId}] Token: ${r.value.token.slice(0, 40)}...`);
    } else {
      console.error(`Failed: ${r.reason.message}`);
    }
  });
})();

إعدادات مخصّصة لكل عميل

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

CLIENT_CONFIG = {
    "client_a": {
        "proxy": "host:port:user:pass",
        "proxytype": "HTTP",
        "max_concurrent": 5,
        "default_method": "userrecaptcha"
    },
    "client_b": {
        "proxy": None,
        "proxytype": None,
        "max_concurrent": 10,
        "default_method": "turnstile"
    }
}

def build_params(client_id, params):
    config = CLIENT_CONFIG.get(client_id, {})
    if config.get("proxy"):
        params["proxy"] = config["proxy"]
        params["proxytype"] = config["proxytype"]
    return params

توزيع الـ Threads على العملاء والتكلفة

فوترة CaptchaAI قائمة على عدد الـ Threads المتزامنة لا على عدد عمليات الحل، وكل خطة تشمل عدداً غير محدود من الحلول لكل Thread خلال الشهر. هذا يبسّط حساباتك في المسار متعدد العملاء: مجموع حدود max_concurrent لكل عملائك يجب ألا يتجاوز عدد الـ Threads في خطتك.

• خطة ADVANCE بسعر 90 دولاراً شهرياً تمنحك 50 Thread — تكفي مثلاً لخمسة عملاء بحد 10 لكل منهم.
• خطة PREMIUM بسعر 170 دولاراً شهرياً تمنحك 100 Thread إذا نمت الأحمال أو زاد عدد العملاء.
• للبدايات الصغيرة، خطة BASIC بسعر 15 دولاراً شهرياً تشمل 5 Threads، وخطة STANDARD بسعر 30 دولاراً تشمل 15 Thread.

لأن الأسعار بالدولار الأمريكي وثابتة بصرف النظر عن نوع CAPTCHA، تصبح كلفة كل عميل مجرد حصته من الـ Threads — رقم يسهل تضمينه في فاتورتك له. راجع صفحة الأسعار على captchaai.com/pricing للاطلاع على القيم المحدّثة.

نصيحة تشغيلية: اترك هامشاً من الـ Threads غير مخصّص لأي عميل لاستيعاب موجات الطلب المفاجئة، بدل توزيع كامل السعة على العملاء.

استكشاف الأعطال وإصلاحها

قبل الحديث عن رموز الأخطاء، هذه أكثر الأعطال التشغيلية شيوعاً في مسار متعدد العملاء وكيفية إصلاحها:

معالجة رموز الأخطاء من CaptchaAI

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

أفضل الممارسات لتشغيل المسار في الإنتاج

بعد أن يعمل المسار، هذه العادات تبقيه مستقراً تحت الحمل الحقيقي:

• سجّل كل عملية إرسال واستطلاع مع client_id وtask_id لتتبّع أي عطل حتى مصدره.
• خزّن قائمة الانتظار في Redis أو قاعدة بيانات كي تنجو من إعادة التشغيل دون فقدان مهام.
• راقب الرصيد آلياً وأطلق تنبيهاً قبل نفاده، لا بعده.
• افصل حدود التزامن لكل عميل حتى لا يستهلك عميل واحد كامل الـ Threads.

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

هل يتعامل هذا المسار مع أنواع CAPTCHA غير reCAPTCHA v2 وTurnstile؟

نعم. المسار محايد تجاه النوع: كل ما يتغيّر هو قيمة method والمعاملات المرسلة. يدعم CaptchaAI عائلات reCAPTCHA v2/v3 وCloudflare Turnstile وChallenge وGeeTest v3 والصور/OCR والشبكة وBLS، فتضيف عميلاً بنوع جديد بتغيير سطر واحد في إعداده.

كيف أوزّع الـ Threads بين عملاء لهم أحجام مختلفة؟

اجعل مجموع max_concurrent لكل العملاء أقل من أو يساوي عدد الـ Threads في خطتك، ثم امنح العميل ذا الحجم الأكبر حصة أعلى. راقب أوقات الحل ومعدلات الخطأ لكل عميل، وأعد التوزيع دورياً. غالباً ما يكون تجمّع الخوادم الوسيطة لديك هو العنق الحقيقي قبل الوصول إلى حد الـ Threads.

كيف أعزل بيانات كل عميل داخل المسار نفسه؟

استخدم جدول CLIENT_CONFIG لفصل الخادم الوسيط وحد التزامن وطريقة الحل الافتراضية لكل عميل، وافهرس مخزن النتائج بـ client_id. بهذا يبقى كل عميل معزولاً منطقياً رغم مشاركته البنية نفسها.

هل أحتاج قاعدة بيانات أم تكفي الذاكرة لتخزين النتائج؟

للأحمال القصيرة تكفي الذاكرة. أما إذا كانت القائمة تعمل ليلاً أو عبر عمليات إعادة تشغيل، فاحفظها في Redis أو قاعدة بيانات؛ وعند إعادة التشغيل، أعد تحميل المهام المعلقة وتابع المعالجة من حيث توقفت.

ابدأ ببناء مسارك مع CaptchaAI

سجّل الدخول إلى captchaai.com، احصل على مفتاح الـ API، وشغّل أول مسار يخدم كل عملائك من قائمة انتظار واحدة.

أدلة ذات صلة

• بناء قائمة انتظار لحل CAPTCHA في Python
• بناء قائمة انتظار لحل CAPTCHA في Node.js