مراقبة أسعار تذاكر الطيران على المواقع المحمية بـ CAPTCHA

يمكنك تتبّع أسعار تذاكر الطيران آليًا حتى على المواقع التي تفرض اختبارات CAPTCHA، دون أن تتوقف عملية المراقبة عند أول تحدٍّ. الفكرة مباشرة: يكتشف السكربت وجود اختبار reCAPTCHA v2 أو Cloudflare Turnstile في الصفحة، فيرسله إلى CaptchaAI للحصول على الرمز، ثم يعيد الطلب ويكمل قراءة السعر — كل ذلك ضمن الحلقة نفسها.

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

لماذا تعترضك اختبارات CAPTCHA أثناء تتبّع الأسعار؟

تعامل مواقع شركات الطيران عمليات الفحص المتكررة على أنها حركة آلية، فتضيف طبقات تحقق تتصاعد كلما زاد الضغط على عنوان IP واحد. أكثر الأنواع التي ستقابلها على هذه المواقع:

• reCAPTCHA v2 — يظهر غالبًا في نموذج البحث عن الرحلات أو عند تكرار الطلبات بسرعة.
• Cloudflare Turnstile — بديل خفيف بدأت مواقع كثيرة تعتمده على صفحات النتائج.
• Cloudflare Challenge — صفحة تحقق كامل للمتصفح تسبق الوصول إلى المحتوى.
• تحديد معدل الطلبات — حجب مؤقت أو دائم لعناوين IP التي تتجاوز حدًّا معينًا.

يتولى CaptchaAI خطوة الحل لهذه الأنواع المدعومة ويعيد الرمز الذي تحقنه في الطلب لتكمل قراءة السعر. أما الأنواع التي لا يحلّها بعد — مثل hCaptcha وFunCaptcha — فيُستحسن أن تتأكد من نوع الاختبار على موقعك المستهدف قبل بناء المسار، لأن حقن رمز لنوع غير مدعوم لن يجدي.

كيف يسير مسار المراقبة من الجدولة إلى التنبيه

يتكوّن المسار من حلقة واحدة تتكرر وفق جدول زمني، وتفصل بوضوح بين اكتشاف الاختبار وحلّه وقراءة السعر:

Schedule check → Request fare page → CAPTCHA detected?
                                         ↓ Yes
                                    Solve via CaptchaAI → Inject token → Retry request
                                         ↓ No
                                    Parse fare data → Store → Alert on price change

تبدأ الحلقة بجدولة زمنية ثم تطلب صفحة الأسعار. فإن ظهر اختبار CAPTCHA، تُرسله إلى CaptchaAI وتحقن الرمز الناتج قبل إعادة الطلب؛ وإن لم يظهر، تنتقل مباشرةً إلى قراءة السعر وتخزينه وإطلاق تنبيه عند أي تغيّر. هذا الفصل بين المراحل يجعل المسار مرنًا: أي تعثّر في مسار واحد لا يوقف بقية المسارات.

المتطلبات الأساسية

تحتاج إلى ثلاثة عناصر فقط لتشغيل النموذج التالي:

ثبّت المكتبة الوحيدة المطلوبة:

pip install requests

بناء مساعد الحل عبر CaptchaAI

يتبع الحل في CaptchaAI نمطًا واحدًا لكل الأنواع: أرسل بيانات الاختبار إلى in.php، احتفظ بمعرّف المهمة، استطلع النتيجة من res.php حتى تجهز، ثم استخدم الرمز. الدالتان التاليتان تغلّفان هذا النمط لنوعي reCAPTCHA v2 (باستخدام method=userrecaptcha) وCloudflare Turnstile (باستخدام method=turnstile)، وتعيد كل منهما الرمز جاهزًا للحقن في الطلب:

import requests
import time

API_KEY = "YOUR_API_KEY"


def solve_recaptcha_v2(sitekey, pageurl):
    """Solve reCAPTCHA v2 and return the token."""
    submit = requests.post("https://ocr.captchaai.com/in.php", data={
        "key": API_KEY, "method": "userrecaptcha",
        "googlekey": sitekey, "pageurl": pageurl, "json": 1
    }).json()

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

    task_id = submit["request"]
    time.sleep(20)

    for _ in range(30):
        result = requests.get("https://ocr.captchaai.com/res.php", params={
            "key": API_KEY, "action": "get", "id": task_id, "json": 1
        }).json()
        if result.get("status") == 1:
            return result["request"]
        if result.get("request") != "CAPCHA_NOT_READY":
            raise RuntimeError(f"Solve error: {result['request']}")
        time.sleep(5)
    raise TimeoutError("Solve timed out")


def solve_turnstile(sitekey, pageurl):
    """Solve Cloudflare Turnstile and return the token."""
    submit = requests.post("https://ocr.captchaai.com/in.php", data={
        "key": API_KEY, "method": "turnstile",
        "sitekey": sitekey, "pageurl": pageurl, "json": 1
    }).json()

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

    task_id = submit["request"]
    time.sleep(10)

    for _ in range(30):
        result = requests.get("https://ocr.captchaai.com/res.php", params={
            "key": API_KEY, "action": "get", "id": task_id, "json": 1
        }).json()
        if result.get("status") == 1:
            return result["request"]
        if result.get("request") != "CAPCHA_NOT_READY":
            raise RuntimeError(f"Solve error: {result['request']}")
        time.sleep(5)
    raise TimeoutError("Solve timed out")

دمج الحل داخل حلقة فحص الأسعار

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

import json
from datetime import datetime


class FareMonitor:
    def __init__(self, proxy=None):
        self.session = requests.Session()
        if proxy:
            self.session.proxies = {
                "http": f"http://{proxy}",
                "https": f"http://{proxy}"
            }
        self.fare_history = {}

    def check_fare(self, route):
        """Check fare for a route, solving CAPTCHAs if needed."""
        url = route["url"]

        response = self.session.get(url)

        # Detect CAPTCHA in response
        if self._has_recaptcha(response.text):
            sitekey = self._extract_sitekey(response.text)
            token = solve_recaptcha_v2(sitekey, url)
            response = self.session.post(url, data={
                "g-recaptcha-response": token,
                **route.get("params", {})
            })

        elif self._has_turnstile(response.text):
            sitekey = self._extract_turnstile_key(response.text)
            token = solve_turnstile(sitekey, url)
            response = self.session.post(url, data={
                "cf-turnstile-response": token,
                **route.get("params", {})
            })

        return self._parse_fare(response.text, route)

    def _has_recaptcha(self, html):
        return "g-recaptcha" in html or "recaptcha/api" in html

    def _has_turnstile(self, html):
        return "cf-turnstile" in html or "turnstile" in html

    def _extract_sitekey(self, html):
        # Extract data-sitekey from reCAPTCHA div
        if 'data-sitekey="' in html:
            start = html.index('data-sitekey="') + 14
            end = html.index('"', start)
            return html[start:end]
        return None

    def _extract_turnstile_key(self, html):
        if 'data-sitekey="' in html:
            idx = html.index("cf-turnstile")
            start = html.index('data-sitekey="', idx) + 14
            end = html.index('"', start)
            return html[start:end]
        return None

    def _parse_fare(self, html, route):
        """Parse fare data from the response. Customize per target site."""
        # Placeholder — implement per site
        return {
            "route": route["name"],
            "timestamp": datetime.now().isoformat(),
            "raw_length": len(html)
        }

    def monitor_routes(self, routes):
        """Check all routes and report price changes."""
        results = []
        for route in routes:
            try:
                fare = self.check_fare(route)
                results.append(fare)
                print(f"[OK] {route['name']}: checked")
            except Exception as e:
                print(f"[ERROR] {route['name']}: {e}")
        return results


# Usage
routes = [
    {
        "name": "NYC-LAX",
        "url": "https://example-airline.com/search?from=JFK&to=LAX&date=2025-08-15",
        "params": {"adults": 1}
    },
    {
        "name": "SFO-ORD",
        "url": "https://example-airline.com/search?from=SFO&to=ORD&date=2025-08-20",
        "params": {"adults": 1}
    }
]

monitor = FareMonitor(proxy="user:pass@proxy.example.com:8080")
results = monitor.monitor_routes(routes)

for r in results:
    print(json.dumps(r, indent=2))

لاحظ الحقل fare_history: خصّصه لتخزين آخر سعر لكل مسار حتى تتمكن من المقارنة وإطلاق تنبيه عند الانخفاض. أما دالة _parse_fare فهي نقطة التخصيص الأساسية — عدّلها لتناسب بنية صفحة كل شركة طيران، لأن تنسيق السعر يختلف من موقع إلى آخر.

جدولة عمليات الفحص التلقائية

شغّل أداة المراقبة وفق جدول زمني منتظم باستخدام cron على Linux أو "مجدول المهام" على Windows:

# فحص الأسعار كل 6 ساعات
0 */6 * * * cd /path/to/project && python fare_monitor.py >> /var/log/fares.log 2>&1

اختر تكرارًا يوازن بين حداثة البيانات وكلفة التشغيل. الفحص كل 4 إلى 8 ساعات يكفي لمعظم الحالات، بينما تحتاج مسارات مواسم الذروة فحصًا أكثف. انتبه إلى أن زيادة التكرار ترفع احتمال ظهور CAPTCHA وتكلفة البروكسي، ويُفضَّل توزيع عمليات الفحص على مدار اليوم بدل تشغيلها دفعة واحدة.

التعامل مع صفحات Cloudflare Challenge

تعرض بعض مواقع الطيران صفحة Cloudflare Challenge — وهي تحقق كامل للمتصفح — قبل السماح بالوصول. للتعامل معها، استخدم method=cloudflare_challenge مع بروكسي ثابت خلال الجلسة؛ يعيد CaptchaAI ملف الارتباط cf_clearance الذي يمنح جلستك حق الوصول إلى الموقع لبقية الطلبات. للاطلاع على التفاصيل الكاملة، راجع دليل حلّ Cloudflare Challenge عبر الـ API.

معالجة المشكلات الشائعة أثناء المراقبة

أغلب مشكلات المراقبة تعود إلى إدارة عناوين IP والتخزين المؤقت. يلخّص الجدول أكثرها شيوعًا مع الإجراء العملي لكل حالة:

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

التكلفة وتخطيط عدد الـ threads

يفوتر CaptchaAI حسب عدد الـ threads المتزامنة لا حسب عدد عمليات الحل، مع عمليات حل غير محدودة داخل كل thread طوال الشهر. عمليًا، يحدد عدد الـ threads كم مسارًا يمكنك فحصه في آنٍ واحد: خطة BASIC ($15 شهريًا، 5 threads) تكفي لمتابعة بضعة مسارات على فترات متباعدة، بينما تناسب خطة ADVANCE ($90 شهريًا، 50 thread) وكالة تراقب عشرات المسارات بالتوازي في مواسم الذروة. ولأن الحل غير محدود داخل الـ thread، تبقى الكلفة الشهرية ثابتة ومتوقعة مهما ارتفع عدد الاختبارات التي تظهر أثناء الفحص.

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

كم يكلّف تشغيل نظام مراقبة أسعار عبر CaptchaAI؟

تعتمد التكلفة على خطة الـ threads لا على عدد الاختبارات. تبدأ الخطط من BASIC ($15 شهريًا، 5 threads) وترتفع مع عدد المسارات المتزامنة. وبما أن الحل غير محدود داخل كل thread، لا توجد رسوم إضافية لكل عملية حل، ما يجعل الفاتورة الشهرية متوقعة سلفًا.

كم مسارًا يمكنني مراقبته في الوقت نفسه؟

يحدد عدد الـ threads في خطتك سقف المسارات المتزامنة تقريبًا: كل thread يعالج اختبارًا واحدًا في كل لحظة، وبمجرد انتهائه يلتقط التالي. لمتابعة عشرات المسارات في مواسم الذروة، تناسبك خطة أكبر مثل ADVANCE ($90 شهريًا، 50 thread) تمنحك تزامنًا أوسع.

ماذا لو استخدم موقع الطيران نوع CAPTCHA لا يدعمه CaptchaAI؟

يحل CaptchaAI أنواع reCAPTCHA v2/v3 وCloudflare Turnstile وCloudflare Challenge وGeeTest v3 والصور، لكنه لا يدعم حاليًا hCaptcha ولا FunCaptcha. لذلك تحقّق من نوع الاختبار على صفحتك المستهدفة أولًا؛ فإن كان غير مدعوم، لن ينفع حقن الرمز وستحتاج إلى إعادة النظر في المصدر أو المسار.

كيف أخزّن سجل الأسعار وأُطلق تنبيهًا عند الانخفاض؟

استخدم حقل fare_history لتخزين آخر سعر لكل مسار، ثم قارن السعر الجديد بالقديم بعد استدعاء _parse_fare. عند هبوط السعر تحت عتبة تحددها، أرسل تنبيهًا عبر البريد أو رسالة فورية أو Webhook. ويمكنك حفظ السجل في قاعدة بيانات خفيفة لمتابعة اتجاه الأسعار عبر الزمن.

هل مراقبة الأسعار الآلية متوافقة مع شروط المواقع؟

راقب فقط الأسعار المتاحة للعموم، والتزم بشروط استخدام كل موقع وبملف robots.txt، واضبط معدل الطلبات ضمن حدود معقولة. يُستخدم هذا المسار لمراقبة الأسعار المصرّح بها والاختبار المعتمد ضمن بيئة مملوكة أو متفق عليها، لا لأي نشاط يخالف سياسات الموقع.

ابدأ مع CaptchaAI الآن

أنشئ حسابك على captchaai.com واحصل على مفتاح الـ API، ثم اربط دالتي الحل بمسار المراقبة لديك لتبدأ بتتبّع أسعار الطيران دون توقف عند اختبارات CAPTCHA.

أدلة ذات صلة

• التعامل مع التحقق أثناء مراجعة الإعلانات
• جمع بيانات الأبحاث الأكاديمية من مواقع محمية بـ CAPTCHA