إنشاء قائمة انتظار حل اختبار CAPTCHA في Python باستخدام CaptchaAI

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

نبني هنا أربعة أنماط جاهزة للإنتاج فوق CaptchaAI API — قائمة على الخيوط، وأخرى غير متزامنة عبر asyncio، ونمط المنتِج/المستهلِك، وقائمة ذات أولوية — مع طبقة لمراقبة الأداء وضبطٍ للتزامن حسب خطتك. كل مقطع كود جاهز للنسخ بعد وضع مفتاح الـ API الخاص بك.

متى تحتاج قائمة انتظار بدل الحل الفردي؟

حين تحلّ عدداً قليلاً من اختبارات CAPTCHA، يكفي طلبٌ واحد يُرسل ثم ينتظر النتيجة. لكن مع مئات الصفحات في مهمة استخراج بيانات واحدة، يتحوّل هذا الانتظار المتسلسل إلى العنق الأضيق: كل اختبار قد يستغرق ثوانيَ عدة، وجمعها بالتتابع يعني دقائق ضائعة على الانتظار وحده.

قائمة الانتظار تكسر هذا القيد لأنها:

• ترسل جميع اختبارات CAPTCHA فور توفّرها دون انتظار السابق
• تستطلع معرّفات المهام المتعددة بالتوازي
• تعيد محاولة المهام الفاشلة تلقائياً دون تكرار الناجحة
• تضبط مستوى التزامن لاحترام حدود معدل الطلبات في الـ API
• توفّر تتبعاً للتقدّم ونقاط ربط callbacks تُستدعى عند كل نتيجة

نستعرض فيما يلي أربعة أنماط، لكل منها موضع يتفوّق فيه؛ ابدأ بالأبسط ثم انتقل إلى الأنسب لحجم عملك.

النمط 1: قائمة انتظار قائمة على الخيوط (threading)

أبسط نقطة انطلاق قائمة انتظار مبنية على وحدة threading وصنف Queue القياسي. كل خيط عامل يسحب مهمة من الطابور، يرسلها إلى نقطة النهاية in.php، ثم يستطلع res.php حتى تجهز النتيجة. هذا النمط مثالي إذا كان مشروعك متزامناً بالفعل وتريد إضافة التوازي دون إعادة كتابته. لاحظ كيف يفصل الصنف CaptchaQueue طابور المهام عن طابور النتائج، ويمرّر عدد الخيوط عبر max_workers:

import time
import threading
import requests
from queue import Queue, Empty

API_KEY = "YOUR_API_KEY"


class CaptchaQueue:
    """Thread-based CAPTCHA solving queue."""

    def __init__(self, api_key, max_workers=10):
        self.api_key = api_key
        self.task_queue = Queue()
        self.result_queue = Queue()
        self.max_workers = max_workers
        self.workers = []

    def submit(self, method, callback=None, **params):
        """Add a CAPTCHA task to the queue."""
        task = {
            "method": method,
            "params": params,
            "callback": callback,
        }
        self.task_queue.put(task)

    def start(self):
        """Start worker threads."""
        for _ in range(self.max_workers):
            t = threading.Thread(target=self._worker, daemon=True)
            t.start()
            self.workers.append(t)

    def wait(self):
        """Wait for all tasks to complete."""
        self.task_queue.join()

    def get_results(self):
        """Get all available results."""
        results = []
        while not self.result_queue.empty():
            try:
                results.append(self.result_queue.get_nowait())
            except Empty:
                break
        return results

    def _worker(self):
        while True:
            try:
                task = self.task_queue.get(timeout=1)
            except Empty:
                continue

            try:
                result = self._solve(task["method"], **task["params"])
                entry = {"status": "solved", "result": result, "task": task}
                self.result_queue.put(entry)
                if task["callback"]:
                    task["callback"](result)
            except Exception as e:
                entry = {"status": "error", "error": str(e), "task": task}
                self.result_queue.put(entry)
            finally:
                self.task_queue.task_done()

    def _solve(self, method, **params):
        submit = requests.post("https://ocr.captchaai.com/in.php", data={
            "key": self.api_key, "method": method, "json": 1, **params,
        }, timeout=30).json()

        if submit.get("status") != 1:
            raise Exception(f"Submit error: {submit.get('request')}")

        task_id = submit["request"]
        for _ in range(30):
            time.sleep(5)
            result = requests.get("https://ocr.captchaai.com/res.php", params={
                "key": self.api_key, "action": "get", "id": task_id, "json": 1,
            }, timeout=30).json()
            if result.get("status") == 1:
                return result["request"]
            if result.get("request") == "ERROR_CAPTCHA_UNSOLVABLE":
                raise Exception("CAPTCHA unsolvable")
        raise TimeoutError("Solve timed out")


# Usage
queue = CaptchaQueue(API_KEY, max_workers=5)
queue.start()

# Submit multiple CAPTCHAs
urls_and_sitekeys = [
    ("https://example.com/page1", "SITEKEY_1"),
    ("https://example.com/page2", "SITEKEY_2"),
    ("https://example.com/page3", "SITEKEY_3"),
]

for url, sitekey in urls_and_sitekeys:
    queue.submit("userrecaptcha", googlekey=sitekey, pageurl=url)

queue.wait()
results = queue.get_results()
print(f"Solved {len(results)} CAPTCHAs")
for r in results:
    print(f"  {r['status']}: {r.get('result', r.get('error', ''))[:50]}")

النمط 2: قائمة انتظار غير متزامنة عبر asyncio

للمشاريع الحديثة التي يغلب عليها الانتظار على الشبكة، يقدّم asyncio كفاءةً أعلى من الخيوط: فبدل حجز خيط لكل مهمة، تدير حلقةُ الأحداث آلافَ الطلبات المعلّقة على خيط واحد. نستخدم asyncio.Semaphore للتحكم في عدد الطلبات المتزامنة، وasyncio.gather لتشغيل الدفعة كاملة، فيما يمثّل max_concurrent سقف الحلول الجارية في آنٍ واحد.

import asyncio
import aiohttp

API_KEY = "YOUR_API_KEY"


class AsyncCaptchaQueue:
    """Async CAPTCHA solving queue with concurrency control."""

    def __init__(self, api_key, max_concurrent=10):
        self.api_key = api_key
        self.semaphore = asyncio.Semaphore(max_concurrent)
        self.results = []

    async def solve_batch(self, tasks):
        """Solve a batch of CAPTCHA tasks concurrently."""
        coros = [self._solve_task(task) for task in tasks]
        self.results = await asyncio.gather(*coros, return_exceptions=True)
        return self.results

    async def _solve_task(self, task):
        async with self.semaphore:
            return await self._solve(task["method"], **task["params"])

    async def _solve(self, method, **params):
        async with aiohttp.ClientSession() as session:
            # Submit
            async with session.post("https://ocr.captchaai.com/in.php", data={
                "key": self.api_key, "method": method, "json": 1, **params,
            }) as resp:
                data = await resp.json(content_type=None)
                if data.get("status") != 1:
                    raise Exception(f"Submit error: {data.get('request')}")
                task_id = data["request"]

            # Poll
            for _ in range(30):
                await asyncio.sleep(5)
                async with session.get("https://ocr.captchaai.com/res.php", params={
                    "key": self.api_key, "action": "get", "id": task_id, "json": 1,
                }) as resp:
                    result = await resp.json(content_type=None)
                    if result.get("status") == 1:
                        return result["request"]
                    if result.get("request") == "ERROR_CAPTCHA_UNSOLVABLE":
                        raise Exception("CAPTCHA unsolvable")

            raise TimeoutError("Solve timed out")


# Usage
async def main():
    queue = AsyncCaptchaQueue(API_KEY, max_concurrent=5)

    tasks = [
        {"method": "userrecaptcha", "params": {"googlekey": f"SITEKEY_{i}", "pageurl": f"https://example.com/page{i}"}}
        for i in range(10)
    ]

    results = await queue.solve_batch(tasks)
    for i, result in enumerate(results):
        if isinstance(result, Exception):
            print(f"Task {i}: ERROR — {result}")
        else:
            print(f"Task {i}: {result[:50]}...")


asyncio.run(main())

خيار return_exceptions=True يضمن ألا يُسقط فشلُ مهمة واحدة بقيةَ الدفعة، بل يعيد الاستثناء ضمن النتائج لتعالجه لاحقاً.

النمط 3: المنتِج والمستهلِك للتدفّق المستمر

تفترض الأنماط السابقة أنك تعرف قائمة المهام مسبقاً. أما في أحمال العمل المستمرة — حيث تُكتشف الصفحات أثناء التشغيل — فتحتاج إلى نمط المنتِج والمستهلِك: منتِجٌ يغذّي الطابور كلما ظهرت مهمة، ومستهلِكون يسحبون منه ويحلّون بالتوازي. ويحدّ حجمُ الطابور (maxsize) من تراكم الذاكرة عندما يفوق الإنتاجُ قدرةَ الاستهلاك.

import asyncio
import aiohttp

API_KEY = "YOUR_API_KEY"


class ProducerConsumerQueue:
    """Continuous CAPTCHA solving with producer-consumer pattern."""

    def __init__(self, api_key, queue_size=100, num_consumers=5):
        self.api_key = api_key
        self.queue = asyncio.Queue(maxsize=queue_size)
        self.num_consumers = num_consumers
        self.solved_count = 0
        self.error_count = 0
        self.running = True

    async def produce(self, tasks):
        """Producer: feed CAPTCHA tasks into the queue."""
        for task in tasks:
            await self.queue.put(task)
        # Signal consumers to stop
        for _ in range(self.num_consumers):
            await self.queue.put(None)

    async def consume(self, result_handler):
        """Consumer: solve CAPTCHAs and call result handler."""
        async with aiohttp.ClientSession() as session:
            while True:
                task = await self.queue.get()
                if task is None:
                    self.queue.task_done()
                    break

                try:
                    result = await self._solve(session, task["method"], **task["params"])
                    self.solved_count += 1
                    if result_handler:
                        await result_handler(task, result)
                except Exception as e:
                    self.error_count += 1
                    print(f"Error: {e}")
                finally:
                    self.queue.task_done()

    async def run(self, tasks, result_handler=None):
        """Run the producer-consumer pipeline."""
        # Start producer
        producer = asyncio.create_task(self.produce(tasks))

        # Start consumers
        consumers = [
            asyncio.create_task(self.consume(result_handler))
            for _ in range(self.num_consumers)
        ]

        # Wait for everything to finish
        await producer
        await asyncio.gather(*consumers)

        print(f"Complete: {self.solved_count} solved, {self.error_count} errors")

    async def _solve(self, session, method, **params):
        async with session.post("https://ocr.captchaai.com/in.php", data={
            "key": self.api_key, "method": method, "json": 1, **params,
        }) as resp:
            data = await resp.json(content_type=None)
            if data.get("status") != 1:
                raise Exception(f"Submit: {data.get('request')}")
            task_id = data["request"]

        for _ in range(30):
            await asyncio.sleep(5)
            async with session.get("https://ocr.captchaai.com/res.php", params={
                "key": self.api_key, "action": "get", "id": task_id, "json": 1,
            }) as resp:
                result = await resp.json(content_type=None)
                if result.get("status") == 1:
                    return result["request"]
        raise TimeoutError("Timed out")


# Usage
async def handle_result(task, token):
    url = task["params"]["pageurl"]
    print(f"Solved for {url}: {token[:30]}...")


async def main():
    queue = ProducerConsumerQueue(API_KEY, num_consumers=5)

    tasks = [
        {"method": "userrecaptcha", "params": {"googlekey": f"SITEKEY_{i}", "pageurl": f"https://example.com/page{i}"}}
        for i in range(20)
    ]

    await queue.run(tasks, result_handler=handle_result)


asyncio.run(main())

إشارة الإيقاف هي القيمة None التي يدفعها المنتِج مرةً لكل مستهلِك، فينهي كلٌّ منهم حلقته بأمان عند استقبالها.

النمط 4: قائمة انتظار ذات أولوية

ليست كل اختبارات CAPTCHA متساوية في الأهمية. تخيّل أداة مراقبة أسعار تخدم متجراً إلكترونياً في منطقة الخليج: صفحات إتمام الشراء يجب أن تُحلّ فوراً لأنها الأقرب إلى التحويل، بينما تحتمل صفحات المنتجات تأخيراً بسيطاً، وصفحات المعلومات تنتظر. يتيح لك asyncio.PriorityQueue ترتيب المهام برقم أولوية — الأصغر يُخدَم أولاً — فلا تعلق المهام الحرجة خلف طابور من المهام الأقل أهمية.

import asyncio
from dataclasses import dataclass, field

API_KEY = "YOUR_API_KEY"


@dataclass(order=True)
class PriorityTask:
    priority: int
    task: dict = field(compare=False)


class PriorityCaptchaQueue:
    """CAPTCHA queue with priority levels."""

    def __init__(self, api_key, num_workers=5):
        self.api_key = api_key
        self.queue = asyncio.PriorityQueue()
        self.num_workers = num_workers
        self.results = {}

    async def submit(self, task_id, method, priority=5, **params):
        """Submit with priority (lower number = higher priority)."""
        await self.queue.put(PriorityTask(
            priority=priority,
            task={"id": task_id, "method": method, "params": params},
        ))

    async def process(self):
        """Process all queued tasks by priority."""
        workers = [asyncio.create_task(self._worker()) for _ in range(self.num_workers)]

        # Wait for queue to drain
        await self.queue.join()

        # Cancel workers
        for w in workers:
            w.cancel()

        return self.results

    async def _worker(self):
        import aiohttp
        async with aiohttp.ClientSession() as session:
            while True:
                item = await self.queue.get()
                task = item.task
                try:
                    result = await self._solve(session, task["method"], **task["params"])
                    self.results[task["id"]] = {"status": "solved", "token": result}
                except Exception as e:
                    self.results[task["id"]] = {"status": "error", "error": str(e)}
                finally:
                    self.queue.task_done()

    async def _solve(self, session, method, **params):
        import aiohttp
        async with session.post("https://ocr.captchaai.com/in.php", data={
            "key": self.api_key, "method": method, "json": 1, **params,
        }) as resp:
            data = await resp.json(content_type=None)
            if data.get("status") != 1:
                raise Exception(data.get("request"))
            task_id = data["request"]

        for _ in range(30):
            await asyncio.sleep(5)
            async with session.get("https://ocr.captchaai.com/res.php", params={
                "key": self.api_key, "action": "get", "id": task_id, "json": 1,
            }) as resp:
                result = await resp.json(content_type=None)
                if result.get("status") == 1:
                    return result["request"]
        raise TimeoutError()


# Usage
async def main():
    pq = PriorityCaptchaQueue(API_KEY, num_workers=3)

    # High priority — checkout pages
    await pq.submit("checkout_1", "turnstile", priority=1, sitekey="KEY", pageurl="https://shop.com/checkout")

    # Normal priority — product pages
    for i in range(5):
        await pq.submit(f"product_{i}", "userrecaptcha", priority=5, googlekey="KEY", pageurl=f"https://shop.com/p/{i}")

    # Low priority — info pages
    for i in range(3):
        await pq.submit(f"info_{i}", "userrecaptcha", priority=10, googlekey="KEY", pageurl=f"https://shop.com/info/{i}")

    results = await pq.process()
    for task_id, result in results.items():
        print(f"{task_id}: {result['status']}")


asyncio.run(main())

في المثال، تأخذ صفحة إتمام الشراء الأولوية 1، والمنتجات 5، وصفحات المعلومات 10؛ عدّل الأرقام لتعكس قيمة كل مسار في عملك.

مراقبة الأداء والإبلاغ

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

import time
from dataclasses import dataclass, field


@dataclass
class QueueMetrics:
    submitted: int = 0
    solved: int = 0
    failed: int = 0
    total_solve_time: float = 0.0
    start_time: float = field(default_factory=time.time)

    @property
    def avg_solve_time(self):
        return self.total_solve_time / self.solved if self.solved else 0

    @property
    def success_rate(self):
        total = self.solved + self.failed
        return (self.solved / total * 100) if total else 0

    @property
    def throughput(self):
        elapsed = time.time() - self.start_time
        return self.solved / elapsed * 60 if elapsed > 0 else 0

    def report(self):
        return (
            f"Submitted: {self.submitted} | "
            f"Solved: {self.solved} | "
            f"Failed: {self.failed} | "
            f"Avg time: {self.avg_solve_time:.1f}s | "
            f"Success: {self.success_rate:.1f}% | "
            f"Throughput: {self.throughput:.0f}/min"
        )

ضبط التزامن حسب خطتك في CaptchaAI

القاعدة الذهبية عند ضبط max_workers أو max_concurrent: لا تتجاوز عدد الخيوط في خطتك. يعتمد CaptchaAI تسعيراً قائماً على الخيوط المتزامنة لا على عدد الحلول؛ فكل خطة تمنحك عدداً ثابتاً من الخيوط مع حلول غير محدودة شهرياً، والخيط الواحد اختبار CAPTCHA واحد قيد التنفيذ يتحرّر فور انتهائه ليلتقط التالي. فإذا شغّلت عمّالاً أكثر من خيوط خطتك، قابلك الخطأ ERROR_NO_SLOT_AVAILABLE بدل مكاسبَ في السرعة:

ابدأ قرب سقف خطتك ثم خفّض قليلاً إذا ظهر ERROR_NO_SLOT_AVAILABLE. أما الخطط الأعلى — CORPORATE وENTERPRISE وباقات VIP — فتوفّر خيوطاً أكثر لأحمال العمل الضخمة.

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

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

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

هل أحتاج إلى Redis أو وسيط رسائل لبناء قائمة الانتظار؟

لا في الغالب. تعتمد الأنماط الأربعة على أدوات Python القياسية queue وasyncio داخل عملية واحدة، وهو ما يكفي لأغلب مهام الاستخراج ومراقبة الأسعار. لا تلجأ إلى Redis أو RabbitMQ إلا حين توزّع العمل على عدة خوادم أو تحتاج إلى بقاء الطابور بعد إعادة التشغيل.

كيف أعيد محاولة المهام الفاشلة دون تكرار الناجحة؟

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

asyncio أم threading: أيّهما أختار لمشروعي؟

اختر asyncio للمشاريع الجديدة؛ فهو أكفأ في التعامل مع الحمل المقيَّد بعمليات الإدخال/الإخراج مثل حل اختبار CAPTCHA، ويتوسّع إلى آلاف المهام دون تكلفة خيوط النظام. أبقِ على threading حين تدمج في قاعدة شيفرة متزامنة قائمة يصعب تحويلها بالكامل.

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

اجعله مساوياً لعدد الخيوط في خطتك أو أقل قليلاً — خمسة مع BASIC، وخمسة عشر مع STANDARD. زيادة العمّال فوق سقف الخطة لا ترفع السرعة، بل تُنتج ERROR_NO_SLOT_AVAILABLE لأن التسعير قائم على الخيوط لا الحلول.

كيف أنشر قائمة الانتظار في بيئة إنتاجية؟

غلّف السكربت في حاوية Docker كخدمة طويلة الأمد، واقرأ مفتاح الـ API من متغيّر بيئة لا من داخل الكود. أضف طبقة QueueMetrics إلى سجلاتك، واضبط إعادة التشغيل التلقائي ليستأنف العمل بعد أي انقطاع.

خلاصة

تفصل قائمة انتظار حل اختبار CAPTCHA الإرسالَ عن الاستطلاع، فتتيح حلاً متوازياً لمئات المهام عبر CaptchaAI. اختر الخيوط لدمجٍ سريع في كود متزامن، وasyncio لمشاريع Python الحديثة، ونمط المنتِج/المستهلِك للتدفّق المستمر، والأولوية حين تتفاوت أهمية المهام — واضبط التزامن دائماً على حدود خطتك.

مقالات ذات صلة

• دليل Playwright مع CaptchaAI في Python
• بناء خطوط معالجة الكابتشا من جهة العميل
• أتمتة مسؤولة وآمنة مع CaptchaAI

أدلة ذات صلة

• بناء قائمة انتظار لحل الكابتشا في Node.js
• بناء مجمّع لإعلانات الوظائف باستخدام CaptchaAI