نتائج الدفعة المتدفقة: معالجة حلول اختبار CAPTCHA عند وصولها

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

ستجد هنا نمطين جاهزين للتطبيق: مولّد غير متزامن في Python يُخرج كل حل فور وصوله، وفئة مبنية على EventEmitter في Node.js تُصدر حدثًا لكل نتيجة. وقبل الكود، معياران عمليان: متى يستحق البث العناء، وكيف تربط حجم التزامن بعدد threads المتاحة في خطتك.

لماذا تعالج النتائج فور وصولها بدل انتظار الدفعة

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

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

ثلاثة نماذج لاستهلاك دفعة واحدة

متى يفيد البث ومتى يكون جمع النتائج أفضل

البث ليس الخيار الأمثل دائمًا؛ القرار يعتمد على ما يحدث بعد وصول الرمز:

Python: مولّد غير متزامن يبثّ كل نتيجة فور جهوزها

يعتمد المثال على asyncio وaiohttp. الفكرة المحورية في stream_results: بدل انتظار اكتمال كل المهام، ينتظر asyncio.wait أول مهمة تنتهي عبر FIRST_COMPLETED، ثم يُخرج نتيجتها مباشرة ويعود لانتظار البقية. يضبط Semaphore سقف المهام المتزامنة، بينما تفصل poll_task بين انتهاء المهلة والأخطاء الفعلية، ويحمل كل عنصر مُخرَج حقل index يربطه بمهمته الأصلية لأن الترتيب هنا ترتيب اكتمال لا ترتيب إرسال.

import asyncio
import aiohttp
import time

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


async def submit_task(session, task_data):
    """Submit a single CAPTCHA task."""
    params = {
        "key": API_KEY,
        "method": task_data.get("method", "userrecaptcha"),
        "json": 1,
    }
    if params["method"] == "userrecaptcha":
        params["googlekey"] = task_data["sitekey"]
        params["pageurl"] = task_data["pageurl"]
    elif params["method"] == "turnstile":
        params["sitekey"] = task_data["sitekey"]
        params["pageurl"] = task_data["pageurl"]

    async with session.post(SUBMIT_URL, data=params) as resp:
        result = await resp.json(content_type=None)
        if result.get("status") != 1:
            return None, result.get("request", "unknown")
        return result["request"], None


async def poll_task(session, task_id, timeout=300):
    """Poll until solved or timeout."""
    start = time.monotonic()
    while time.monotonic() - start < timeout:
        await asyncio.sleep(5)
        params = {"key": API_KEY, "action": "get", "id": task_id, "json": 1}
        async with session.get(RESULT_URL, params=params) as resp:
            result = await resp.json(content_type=None)

        if result.get("request") == "CAPCHA_NOT_READY":
            continue
        if result.get("status") == 1:
            return result["request"], None
        return None, result.get("request", "unknown")

    return None, "TIMEOUT"


async def solve_one(session, index, task_data, semaphore):
    """Solve a single task within concurrency limits."""
    async with semaphore:
        start = time.monotonic()
        task_id, error = await submit_task(session, task_data)
        if error:
            return {"index": index, "status": "failed", "error": error, "time": 0}

        token, error = await poll_task(session, task_id)
        elapsed = time.monotonic() - start

        if token:
            return {"index": index, "status": "solved", "token": token, "time": round(elapsed, 1)}
        return {"index": index, "status": "failed", "error": error, "time": round(elapsed, 1)}


async def stream_results(tasks, max_concurrent=20):
    """
    Async generator that yields each result as it completes.
    Results arrive in completion order, not submission order.
    """
    semaphore = asyncio.Semaphore(max_concurrent)

    async with aiohttp.ClientSession() as session:
        pending = set()
        for i, task in enumerate(tasks):
            coro = solve_one(session, i, task, semaphore)
            pending.add(asyncio.ensure_future(coro))

        while pending:
            done, pending = await asyncio.wait(pending, return_when=asyncio.FIRST_COMPLETED)
            for future in done:
                yield future.result()


async def main():
    tasks = [
        {"sitekey": "SITE_KEY", "pageurl": f"https://example.com/page{i}"}
        for i in range(50)
    ]

    solved = 0
    failed = 0

    async for result in stream_results(tasks, max_concurrent=15):
        # Process each result immediately
        if result["status"] == "solved":
            solved += 1
            print(f"  [{solved + failed}/{len(tasks)}] Task {result['index']} SOLVED in {result['time']}s")

            # Use token immediately — don't wait for batch
            # await submit_form(result["token"])
            # await save_to_database(result)
        else:
            failed += 1
            print(f"  [{solved + failed}/{len(tasks)}] Task {result['index']} FAILED: {result['error']}")

    print(f"\nDone: {solved} solved, {failed} failed")


asyncio.run(main())

تثبيت التبعيات:

pip install aiohttp

Node.js: نمط EventEmitter لمعالجة النتائج فور وصولها

في Node.js يأتي البث من بنية اللغة نفسها: فئة ترث EventEmitter وتُصدر حدث result لكل مهمة تنتهي، وحدث done عند اكتمال العدّاد. المستمع الذي تسجله على result هو مكان المعالجة الفورية — إرسال النموذج، أو الكتابة في قائمة الانتظار، أو تحديث الواجهة. لاحظ أن processNext تُستدعى مجددًا بعد كل مهمة، فيبقى عدد الطلبات النشطة عند السقف المحدد بدل إطلاقها دفعة واحدة.

const { EventEmitter } = require("events");

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

class CaptchaStream extends EventEmitter {
  constructor(maxConcurrent = 15) {
    super();
    this.maxConcurrent = maxConcurrent;
    this.active = 0;
    this.queue = [];
    this.total = 0;
    this.completed = 0;
  }

  async submitAndPoll(index, taskData) {
    const params = new URLSearchParams({
      key: API_KEY,
      method: taskData.method || "userrecaptcha",
      googlekey: taskData.sitekey,
      pageurl: taskData.pageurl,
      json: "1",
    });

    const start = Date.now();
    const submitResp = await (await fetch(SUBMIT_URL, { method: "POST", body: params })).json();

    if (submitResp.status !== 1) {
      return { index, status: "failed", error: submitResp.request, time: 0 };
    }

    const taskId = submitResp.request;
    for (let i = 0; i < 60; i++) {
      await new Promise((r) => setTimeout(r, 5000));
      const url = `${RESULT_URL}?key=${API_KEY}&action=get&id=${taskId}&json=1`;
      const poll = await (await fetch(url)).json();

      if (poll.request === "CAPCHA_NOT_READY") continue;
      const elapsed = ((Date.now() - start) / 1000).toFixed(1);
      if (poll.status === 1) return { index, status: "solved", token: poll.request, time: elapsed };
      return { index, status: "failed", error: poll.request, time: elapsed };
    }
    return { index, status: "failed", error: "TIMEOUT", time: ((Date.now() - start) / 1000).toFixed(1) };
  }

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

    const { index, taskData } = this.queue.shift();
    this.active++;

    try {
      const result = await this.submitAndPoll(index, taskData);
      this.emit("result", result);
    } catch (err) {
      this.emit("result", { index, status: "failed", error: err.message });
    } finally {
      this.active--;
      this.completed++;

      if (this.completed === this.total) {
        this.emit("done");
      } else {
        this.processNext();
      }
    }
  }

  start(tasks) {
    this.total = tasks.length;
    this.queue = tasks.map((taskData, index) => ({ index, taskData }));

    // Launch initial batch
    const initial = Math.min(this.maxConcurrent, tasks.length);
    for (let i = 0; i < initial; i++) {
      this.processNext();
    }
    return this;
  }
}

// Usage
const tasks = Array.from({ length: 50 }, (_, i) => ({
  sitekey: "SITE_KEY",
  pageurl: `https://example.com/page${i}`,
}));

const stream = new CaptchaStream(15);
let solved = 0, failed = 0;

stream.on("result", (result) => {
  if (result.status === "solved") {
    solved++;
    console.log(`[${solved + failed}/${tasks.length}] Task ${result.index} SOLVED (${result.time}s)`);
    // Use token immediately
    // submitForm(result.token);
  } else {
    failed++;
    console.log(`[${solved + failed}/${tasks.length}] Task ${result.index} FAILED: ${result.error}`);
  }
});

stream.on("done", () => {
  console.log(`\nComplete: ${solved} solved, ${failed} failed`);
});

stream.start(tasks);

اضبط التزامن على عدد threads في خطتك

القيمة التي تمرّرها إلى max_concurrent في Python أو إلى مُنشئ CaptchaStream في Node.js ليست رقمًا اعتباطيًا: فوترة CaptchaAI قائمة على عدد الـ threads المتزامنة لا على عدد الحلول، وكل خطة تتضمن حلولًا غير محدودة لكل thread خلال الشهر. إذا ضبطت التزامن أعلى من حصتك، فالطلبات الزائدة تنتظر دورها بلا فائدة، وقد تظهر أخطاء رفض من الخدمة.

• خطة BASIC — $15 شهريًا مع 5 threads: مناسبة لتجربة النمط على دفعات صغيرة.
• خطة STANDARD — $30 شهريًا مع 15 thread: تطابق قيمة maxConcurrent في مثال Node.js أعلاه.
• خطة ADVANCE — $90 شهريًا مع 50 thread: نقطة الانطلاق العملية للدفعات التي تتجاوز بضع مئات من المهام.

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

مثال تطبيقي: اختبار نماذج الدفع قبل موسم التخفيضات

تخيّل فريق ضمان جودة في متجر إلكتروني بالرياض يستعد لموسم «الجمعة البيضاء». قبل انطلاق الحملة يشغّل الفريق فحصًا موثوقًا على مئات من نماذج التسجيل والدفع في بيئة الاختبار، وكل نموذج محمي بـ reCAPTCHA v2.

مع انتظار الدفعة كاملة يبقى الفريق معطلًا حتى أبطأ مهمة، وقد تنتهي صلاحية الرموز الأولى قبل استخدامها. ومع البث يُرسل كل نموذج فور وصول رمزه، فتظهر أعطال البيئة — حقل ناقص أو مسار إعادة توجيه خاطئ — خلال الدقائق الأولى، ويزداد عدد دورات الإصلاح الممكنة في نافذة الليل الواحدة قبل انطلاق الحملة.

أخطاء شائعة في خط البث وكيفية معالجتها

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

هل أحتاج إلى asyncio أم تكفي الخيوط في Python؟

الاثنان يعملان، فالمهمة هنا انتظار شبكة لا حساب معالج. لكن asyncio أخف عند عشرات المهام المتزامنة، ويمنحك صيغة async for التي تستهلك النتائج تباعًا. وإن كان مشروعك متزامنًا بالكامل، فـ ThreadPoolExecutor مع as_completed يعطي النمط نفسه.

كم مهمة متزامنة أضبط في خط البث؟

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

ماذا يحدث لو فشلت مهمة واحدة أثناء البث؟

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

هل يصلح النمط نفسه مع أنواع CAPTCHA الأخرى؟

نعم؛ بنية الإرسال والاستطلاع واحدة لدى CaptchaAI، والمتغيّر هو قيمة method والمعلمات المرافقة لها. النمط الذي تراه هنا مع reCAPTCHA v2 ينطبق كما هو على reCAPTCHA v3 وCloudflare Turnstile وGeeTest v3 وكابتشا الصور، دون أي تغيير في منطق البث.

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

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

أدلة ذات صلة

• دليل البدء السريع مع CaptchaAI
• حلّ reCAPTCHA v2 عبر الـ API: الدليل العملي