تكامل Scrapy مع CaptchaAI لحل reCAPTCHA v2

عندما تعترض صفحة تحقق مسار زحف قائم على Scrapy، الحل ليس إعادة كتابة العناكب واحداً واحداً: يكفي downloader middleware واحد يقرأ الـ sitekey من HTML، يرسله إلى CaptchaAI، ثم يضع التوكن العائد في request.meta ليكمل العنكبوت عمله. هذا الدليل يبني المسار كاملاً: وحدة حل مستقلة، ثم middleware للكشف، ثم ربطه في settings.py، ثم عنكبوت يستهلك التوكن، وأخيراً طبقة إعادة محاولة تحمي الزحف من التوقف. القاعدة التي تختصر ما يلي: اجعل التحقق تفصيلة في طبقة الشبكة لا استثناءً مكرراً في كل دالة parse.

أين يقع الـ middleware في دورة حياة طلب Scrapy

كل استجابة في Scrapy تمر عبر سلسلة downloader middlewares قبل أن تصل إلى parse، وهذه هي النقطة المناسبة للتعامل مع التحقق: الاستجابة كاملة بين يديك والطلب الأصلي ما زال متاحاً لإعادة الإرسال. أما توزيع منطق الحل على العناكب فيعني تكرار الكود وصعوبة تتبّع تكلفة الحل.

نقسم المسؤوليات في هذا الدليل على ثلاث طبقات واضحة:

• الكشف داخل process_response، لأن الـ sitekey لا يظهر إلا في HTML النهائي.
• الحل داخل وحدة منفصلة يمكن اختبارها وحدها دون تشغيل زحف كامل.
• الاستهلاك داخل العنكبوت عبر request.meta، فيبقى الـ middleware جاهلاً بمنطق العمل.

المتطلبات

ثبّت الحزمتين داخل بيئة افتراضية خاصة بالمشروع حتى لا تتعارض إصدارات Scrapy مع مشاريع أخرى على الجهاز نفسه:

pip install scrapy requests

الخطوة 1: وحدة الحل التي تتحدث مع CaptchaAI

أنشئ ملف captcha_solver.py في جذر مشروع Scrapy. الوحدة تتعامل مع نقطتَي النهاية in.php وres.php: الأولى تستقبل المهمة وتعيد معرّفها، والثانية تُستطلع دورياً حتى تعود بالتوكن الجاهز للاستخدام.

انتبه إلى فارق المهلة بين الدالتين: solve_recaptcha تنتظر حتى 300 ثانية بينما تكتفي solve_image بـ 120 ثانية، لأن reCAPTCHA v2 أثقل من صورة OCR بسيطة. المهلة هنا هامش أمان لا متوسط متوقع؛ الأرقام المعلنة تضع سقف reCAPTCHA v2 عند أقل من 60 ثانية.

import requests
import time


class CaptchaAISolver:
    def __init__(self, api_key):
        self.api_key = api_key
        self.base_url = "https://ocr.captchaai.com"

    def solve_recaptcha(self, site_key, page_url, timeout=300):
        resp = requests.get(f"{self.base_url}/in.php", params={
            "key": self.api_key,
            "method": "userrecaptcha",
            "googlekey": site_key,
            "pageurl": page_url,
        })

        if not resp.text.startswith("OK|"):
            raise Exception(f"Submit failed: {resp.text}")

        task_id = resp.text.split("|")[1]
        deadline = time.time() + timeout

        while time.time() < deadline:
            time.sleep(5)
            result = requests.get(f"{self.base_url}/res.php", params={
                "key": self.api_key,
                "action": "get",
                "id": task_id,
            })

            if result.text == "CAPCHA_NOT_READY":
                continue
            if result.text.startswith("OK|"):
                return result.text.split("|", 1)[1]
            raise Exception(f"Solve failed: {result.text}")

        raise TimeoutError(f"Task {task_id} timed out")

    def solve_image(self, image_base64, timeout=120):
        resp = requests.get(f"{self.base_url}/in.php", params={
            "key": self.api_key,
            "method": "base64",
            "body": image_base64,
        })

        if not resp.text.startswith("OK|"):
            raise Exception(f"Submit failed: {resp.text}")

        task_id = resp.text.split("|")[1]
        deadline = time.time() + timeout

        while time.time() < deadline:
            time.sleep(5)
            result = requests.get(f"{self.base_url}/res.php", params={
                "key": self.api_key,
                "action": "get",
                "id": task_id,
            })

            if result.text == "CAPCHA_NOT_READY":
                continue
            if result.text.startswith("OK|"):
                return result.text.split("|", 1)[1]
            raise Exception(f"Solve failed: {result.text}")

        raise TimeoutError(f"Task {task_id} timed out")

اختبر هذا الملف وحده من الطرفية على صفحة واحدة قبل ربطه بأي عنكبوت. اكتشاف خطأ في مفتاح الـ API أثناء زحف يمتد لآلاف الصفحات أغلى بكثير من دقيقتين في اختبار مباشر.

الخطوة 2: middleware يكتشف التحقق ويحلّه

أنشئ middlewares.py. الفئة التالية تقرأ المفتاح من إعدادات المشروع عبر from_crawler، ثم تفحص كل استجابة بحثاً عن data-sitekey أو صورة تحقق مضمّنة بصيغة base64، وتخزّن الناتج في request.meta: التوكن تحت captcha_token والنص المستخرج تحت captcha_text. لاحظ أنها ترفع خطأً صريحاً عند غياب CAPTCHAAI_API_KEY بدل أن تفشل في منتصف الزحف.

import base64
import re
from scrapy import signals
from scrapy.http import HtmlResponse
from captcha_solver import CaptchaAISolver


class CaptchaAIMiddleware:
    """Scrapy downloader middleware that detects and solves CAPTCHAs."""

    def __init__(self, api_key):
        self.solver = CaptchaAISolver(api_key)

    @classmethod
    def from_crawler(cls, crawler):
        api_key = crawler.settings.get("CAPTCHAAI_API_KEY")
        if not api_key:
            raise ValueError("CAPTCHAAI_API_KEY setting is required")
        return cls(api_key)

    def process_response(self, request, response, spider):
        # Check for reCAPTCHA on the page
        site_key = self._find_recaptcha_key(response.text)
        if site_key:
            spider.logger.info(f"reCAPTCHA detected on {response.url}")
            token = self.solver.solve_recaptcha(site_key, response.url)
            request.meta["captcha_token"] = token
            spider.logger.info("CAPTCHA solved successfully")

        # Check for image CAPTCHA
        captcha_img = self._find_image_captcha(response)
        if captcha_img:
            spider.logger.info(f"Image CAPTCHA detected on {response.url}")
            text = self.solver.solve_image(captcha_img)
            request.meta["captcha_text"] = text
            spider.logger.info(f"Image CAPTCHA solved: {text}")

        return response

    def _find_recaptcha_key(self, html):
        match = re.search(
            r'data-sitekey=["\']([A-Za-z0-9_-]+)["\']', html
        )
        return match.group(1) if match else None

    def _find_image_captcha(self, response):
        img = response.css("img#captcha-image::attr(src)").get()
        if img and img.startswith("data:image"):
            return img.split(",", 1)[1]
        return None

الخطوة 3: تفعيل الـ middleware داخل settings.py

يبقى الـ middleware معطّلاً حتى تسجّله في DOWNLOADER_MIDDLEWARES. الرقم 560 يحدد ترتيبه في السلسلة: بعد الطبقات التي تفكّ ضغط الاستجابة، فيصل إليه response.text كـ HTML صالح للفحص. اقرأ المفتاح من متغير بيئة ولا تكتبه داخل الملف:

import os

CAPTCHAAI_API_KEY = os.environ.get("CAPTCHAAI_API_KEY")

DOWNLOADER_MIDDLEWARES = {
    "myproject.middlewares.CaptchaAIMiddleware": 560,
}

الخطوة 4: العنكبوت الذي يستهلك التوكن

العنكبوت نفسه لا يعرف شيئاً عن CaptchaAI؛ كل ما يفعله هو قراءة captcha_token من meta. إن وجده أعاد إرسال الصفحة عبر FormRequest مع الحقل g-recaptcha-response، وإن لم يجده مضى مباشرة إلى استخراج البيانات:

import scrapy


class ProductSpider(scrapy.Spider):
    name = "products"
    start_urls = ["https://example.com/products"]

    def parse(self, response):
        # If CAPTCHA was solved, the token is in meta
        token = response.meta.get("captcha_token")
        if token:
            # Resubmit the page with the token
            yield scrapy.FormRequest(
                url=response.url,
                formdata={"g-recaptcha-response": token},
                callback=self.parse_products,
            )
        else:
            yield from self.parse_products(response)

    def parse_products(self, response):
        for product in response.css(".product-item"):
            yield {
                "name": product.css("h2::text").get(),
                "price": product.css(".price::text").get(),
                "url": response.urljoin(
                    product.css("a::attr(href)").get()
                ),
            }

        next_page = response.css("a.next-page::attr(href)").get()
        if next_page:
            yield scrapy.Request(response.urljoin(next_page))

الخطوة 5: إعادة المحاولة عند ظهور صفحة تحقق

أحياناً يستبدل الموقع الصفحة كلها بصفحة تحقق، فلا يوجد sitekey لتلتقطه وتبدو الاستجابة ناجحة برمز 200 بينما محتواها بلا قيمة. الحل طبقة ثانية تتعرف على بصمات تلك الصفحة وتعيد الطلب بحد أقصى ثلاث محاولات، مع عدّاد في meta يمنع الدوران اللانهائي:

class CaptchaRetryMiddleware:
    """Retry requests that return CAPTCHA challenge pages."""

    max_retries = 3

    def process_response(self, request, response, spider):
        if self._is_captcha_page(response):
            retries = request.meta.get("captcha_retries", 0)
            if retries < self.max_retries:
                request.meta["captcha_retries"] = retries + 1
                spider.logger.info(
                    f"CAPTCHA page detected, retry {retries + 1}"
                )
                return request.copy()

        return response

    def _is_captcha_page(self, response):
        indicators = [
            "g-recaptcha",
            "cf-turnstile",
            "captcha-image",
            "Please verify you are human",
        ]
        return any(ind in response.text for ind in indicators)

الخطوة 6: تشغيل عنكبوت Scrapy

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

export CAPTCHAAI_API_KEY="YOUR_API_KEY"
scrapy crawl products -o products.json

كم thread تحتاج فعلياً؟

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

اربط هذا الرقم بإعداد CONCURRENT_REQUESTS: القاعدة أن يقارب عدد الـ threads عدد الصفحات التي تتوقع أن تعرض تحققاً في اللحظة نفسها، لا إجمالي طلباتك.

سيناريو تطبيقي من السوق المحلي

تخيّل فريق تسعير في متجر إلكتروني بالسعودية يراقب أسعار المنافسين في فئتَي الإلكترونيات والعناية الشخصية: نحو 12,000 صفحة منتج كل ليلة، تُشغَّل بعد منتصف الليل بتوقيت الرياض لتكون التقارير جاهزة قبل اجتماع الصباح. من هذه الصفحات تعرض 4% تقريباً — أي 480 صفحة — اختبار تحقق عند تسارع معدل الطلبات.

لو افترضنا 30 ثانية وسطياً لكل عملية حل، فذلك 14,400 ثانية معالجة إجمالية. مع 15 thread متزامناً تنتهي هذه الدفعة في نحو 16 دقيقة، وهي مدة تذوب داخل نافذة الزحف الليلية. أما مع 5 threads فتقارب 48 دقيقة، وقد تصطدم بموعد التقرير.

نقطة يعرفها من يعمل على مواقع عربية: كثير من المتاجر تعرض واجهتين، عربية وإنجليزية، وقد يظهر التحقق في إحداهما دون الأخرى بحسب توجيه الـ CDN. سجّل response.url كلما عمل الـ middleware؛ هذا السجل وحده يكشف أي فرع يستهلك ميزانية الحل.

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

أسئلة شائعة

أين أحفظ مفتاح الـ API في بيئة إنتاج؟

في متغير بيئة يُحقن وقت التشغيل، أو في مدير أسرار لدى مزوّد الاستضافة؛ ولهذا يقرأ الكود أعلاه os.environ. لا تضع المفتاح في settings.py ولا في أي ملف يصل إلى مستودع Git، وراقب الرصيد من لوحة التحكم لاكتشاف أي استهلاك غير متوقع مبكراً.

هل ما زلت بحاجة إلى بروكسي إذا كنت أستخدم CaptchaAI؟

غالباً نعم، لأنهما يعالجان مشكلتين مختلفتين. CaptchaAI يعيد إليك توكناً صالحاً لاختبار التحقق، لكن سمعة عنوان IP وتكرار الطلبات من المصدر نفسه يظلان من مسؤوليتك. إن لاحظت عودة صفحات التحقق مباشرة بعد كل حل ناجح، فالمشكلة في معدل الطلبات أو في الـ IP، لا في وحدة الحل.

ماذا لو ظهر hCaptcha أو FunCaptcha في أحد المواقع المستهدفة؟

هذان النوعان غير مدعومين حالياً في CaptchaAI، وGeeTest v4 مدرج ضمن الدعم القادم ولم يتوفر بعد. الأنواع المتاحة تشمل عائلة reCAPTCHA v2 وv3، وCloudflare Turnstile وChallenge، وGeeTest v3، والصور وGrid وBLS، إضافة إلى CaptchaFox وFriendly Captcha وLemin في مرحلة beta. صمّم الـ middleware ليمرّ على مثل هذه الصفحات ويسجّل تحذيراً بدل أن ينهار.

كيف أتعامل مع أكثر من نوع تحقق داخل الزحف نفسه؟

وسّع process_response بحيث يفحص بصمات كل نوع بالترتيب — data-sitekey لـ reCAPTCHA، وعنصر Turnstile، وصورة التحقق — ثم يستدعي دالة الحل المناسبة. أبقِ الكشف في دوال صغيرة منفصلة حتى تضيف نوعاً جديداً دون المساس بالمنطق القائم.

اقرأ أيضاً

• دليل استخراج بيانات الويب بـ Python مع التعامل مع CAPTCHA
• تكامل HTTPX مع CaptchaAI للطلبات المباشرة