مراقبة أسعار السفر مع حلّ اختبارات CAPTCHA

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

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

في ما يلي خريطة أنواع الحماية على مواقع السفر، ومراقب أسعار كامل بلغة Python، ثم الجدولة وتقدير عدد الـ threads المناسب.

لماذا تتعثّر مراقبة الأسعار عند أول اختبار تحقق

سكربت مراقبة الأسعار لا يفشل لأن الكود خاطئ، بل لأن الصفحة التي وصلت إليه ليست الصفحة التي توقّعها. بدل جدول الأسعار يستقبل صفحة تحقق، فيخرج الاستخراج بقائمة فارغة ويُسجَّل اليوم كأنه بلا بيانات. الأسباب المتكرّرة معروفة:

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

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

خريطة اختبارات CAPTCHA على منصات الحجز

كل الأنواع الواردة أعلاه مدعومة رسمياً لدى CaptchaAI: reCAPTCHA v2 وv3 بنسخهما المختلفة، وCloudflare Turnstile وCloudflare Challenge، والاختبارات الصورية وشبكات الصور، إضافة إلى GeeTest v3. أي أن مسار المراقبة يكفيه مزوّد حل واحد بدل توزيع الاختبارات على أكثر من خدمة.

سيناريو من السوق: وكالة سفر تراقب مسارات الخليج

وكالة سفر متوسطة الحجم في الرياض تتابع 60 مساراً — الرياض–القاهرة، وجدة–إسطنبول، والدمام–دبي وغيرها — عبر أربع شركات طيران وموقعَي حجز إلكترونيين. الفحص يجري أربع مرات يومياً، أي 240 طلباً، يقابل نحو ثلثها اختبار تحقق في أوقات الذروة المسائية. النتيجة العملية: ما يقارب 80 اختباراً يومياً يجب حلّها دون أن يتوقف الجدول.

الرقم الذي يحدّد الخطة هنا ليس مجموع الاختبارات، بل عدد الطلبات المتزامنة في لحظة الذروة. مع ستة إلى ثمانية فحوصات متوازية، تكفي خطة STANDARD — ‏$30 شهرياً مع 15 thread — وتبقى مساحة للتوسّع في موسم العمرة دون تغيير الاشتراك. الأسعار كلها بالدولار الأمريكي كما هي على صفحة الأسعار الرسمية.

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

بناء مراقب الأسعار في Python

المراقب يتكوّن من ثلاث طبقات فقط: دالة حل موحّدة تتحدث إلى نقطتي in.php وres.php، وجلسة requests تحتفظ بالكوكيز بين الطلبات، وطبقة استخراج تحوّل HTML إلى قائمة أسعار قابلة للمقارنة. السكربت التالي يفحص محتوى الصفحة، يميّز reCAPTCHA v3 عن v2 من وجود render= في وسم السكربت، ويعالج Turnstile عبر مسار منفصل، ثم يسجّل كل قراءة في سجل تاريخي يُستخدم لاحقاً لرصد الانخفاضات.

import requests
import time
import re
import json
import os
from datetime import datetime, timedelta

API_KEY = os.environ["CAPTCHAAI_API_KEY"]


def solve_captcha(params):
    params["key"] = API_KEY
    resp = requests.get("https://ocr.captchaai.com/in.php", params=params)
    if not resp.text.startswith("OK|"):
        raise Exception(f"Submit: {resp.text}")

    task_id = resp.text.split("|")[1]
    for _ in range(60):
        time.sleep(5)
        result = requests.get("https://ocr.captchaai.com/res.php", params={
            "key": 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: {result.text}")
    raise TimeoutError()


class FareMonitor:
    def __init__(self):
        self.session = requests.Session()
        self.session.headers["User-Agent"] = (
            "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
            "AppleWebKit/537.36 Chrome/120.0.0.0"
        )
        self.history = []

    def fetch_with_captcha(self, url):
        """Fetch a travel page, solving CAPTCHAs if encountered."""
        resp = self.session.get(url)

        # reCAPTCHA v2/v3
        match = re.search(
            r'data-sitekey=["\']([A-Za-z0-9_-]+)["\']', resp.text
        )
        if match:
            site_key = match.group(1)

            # Detect v3 vs v2
            if "recaptcha/api.js?render=" in resp.text:
                token = solve_captcha({
                    "method": "userrecaptcha",
                    "googlekey": site_key,
                    "pageurl": url,
                    "version": "v3",
                    "action": "search",
                })
            else:
                token = solve_captcha({
                    "method": "userrecaptcha",
                    "googlekey": site_key,
                    "pageurl": url,
                })

            resp = self.session.post(url, data={
                "g-recaptcha-response": token,
            })

        # Cloudflare Turnstile
        if "cf-turnstile" in resp.text:
            match = re.search(
                r'data-sitekey=["\']([^"\']+)', resp.text
            )
            if match:
                token = solve_captcha({
                    "method": "turnstile",
                    "sitekey": match.group(1),
                    "pageurl": url,
                })
                resp = self.session.post(url, data={
                    "cf-turnstile-response": token,
                })

        return resp.text

    def check_fares(self, routes):
        """Check fares for a list of routes."""
        results = []
        for route in routes:
            try:
                html = self.fetch_with_captcha(route["url"])
                prices = self._extract_prices(html)

                result = {
                    "route": f"{route['origin']}-{route['destination']}",
                    "date": route["date"],
                    "prices": prices,
                    "min_price": min(prices) if prices else None,
                    "timestamp": datetime.utcnow().isoformat(),
                }
                results.append(result)
                self.history.append(result)

                if prices:
                    print(f"  {result['route']} ({route['date']}): "
                          f"${min(prices)}-${max(prices)}")
                else:
                    print(f"  {result['route']}: No prices found")

                time.sleep(3)  # Respectful delay

            except Exception as e:
                print(f"  {route.get('origin', '?')}-"
                      f"{route.get('destination', '?')}: ERROR - {e}")

        return results

    def _extract_prices(self, html):
        """Extract prices from travel page HTML."""
        prices = []
        # Common price patterns
        for match in re.finditer(
            r'\$\s*([\d,]+(?:\.\d{2})?)', html
        ):
            price = float(match.group(1).replace(",", ""))
            if 20 < price < 10000:  # Filter noise
                prices.append(price)
        return sorted(set(prices))

    def detect_price_drops(self, threshold_pct=5):
        """Detect significant price drops in history."""
        route_prices = {}
        for entry in self.history:
            key = f"{entry['route']}_{entry['date']}"
            if key not in route_prices:
                route_prices[key] = []
            if entry["min_price"]:
                route_prices[key].append(entry["min_price"])

        alerts = []
        for key, prices in route_prices.items():
            if len(prices) >= 2:
                prev = prices[-2]
                current = prices[-1]
                change_pct = ((current - prev) / prev) * 100
                if change_pct < -threshold_pct:
                    alerts.append({
                        "route": key,
                        "previous": prev,
                        "current": current,
                        "change": f"{change_pct:.1f}%",
                    })

        return alerts

    def export_report(self, filename="fare_report.json"):
        """Export fare history to JSON."""
        with open(filename, "w") as f:
            json.dump(self.history, f, indent=2)
        print(f"Exported {len(self.history)} fare checks to {filename}")


# Define routes to monitor
routes = [
    {
        "origin": "JFK",
        "destination": "LAX",
        "date": "2025-03-15",
        "url": "https://example-airline.com/flights?from=JFK&to=LAX&date=2025-03-15",
    },
    {
        "origin": "SFO",
        "destination": "ORD",
        "date": "2025-03-20",
        "url": "https://example-airline.com/flights?from=SFO&to=ORD&date=2025-03-20",
    },
]

monitor = FareMonitor()
results = monitor.check_fares(routes)
monitor.export_report()

ماذا يفعل السكربت في كل مرحلة

1. يطلب صفحة النتائج بجلسة ثابتة تحمل ترويسة User-Agent واقعية.
2. يبحث في HTML عن data-sitekey، ويحدّد ما إذا كان الاختبار reCAPTCHA v3 أم v2 أم Turnstile.
3. يرسل الاختبار إلى CaptchaAI مع رابط الصفحة ومفتاح الموقع، ثم يستطلع النتيجة كل خمس ثوانٍ حتى وصول الرمز.
4. يعيد إرسال الطلب ومعه الرمز في الحقل الصحيح — g-recaptcha-response لعائلة reCAPTCHA وcf-turnstile-response لـ Turnstile.
5. يستخرج الأسعار، يحفظ أدنى سعر بطابع زمني UTC، ثم يقارن القراءة بسابقتها ويطلق تنبيهاً عند انخفاض يتجاوز النسبة المحدّدة.

جدولة الفحص الدوري

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

# Check fares every 4 hours
0 */4 * * * cd /opt/fare-monitor && python fare_monitor.py

تقدير الحجم واختيار الخطة

الفوترة في CaptchaAI قائمة على عدد الـ threads المتزامنة، لا على عدد عمليات الحل: كل خطة تتيح عدداً غير محدود من العمليات خلال الشهر، والحدّ الحقيقي هو كم اختباراً يمكن معالجته في اللحظة نفسها. لذلك احسب ذروة التزامن لديك — عدد المسارات التي تُفحص في اللحظة ذاتها — واختر الخطة على أساسها.

أخطاء شائعة تُفسد بيانات الأسعار

• ترك نطاق التصفية الافتراضي بين 20 و10000 كما هو مع أسعار بعملة محلية، فتُستبعد قراءات صحيحة أو تدخل أرقام ليست أسعاراً.
• خلط أسعار الذهاب فقط مع أسعار الذهاب والعودة في السجل نفسه، ما يجعل التنبيهات تُظهر انخفاضاً وهمياً.
• تجاهل رسوم الأمتعة واختيار المقعد عند شركات الطيران الاقتصادية، وهي أحياناً أكبر من فرق السعر المرصود.
• تخزين الطابع الزمني بالتوقيت المحلي بدل UTC، فتختلط قراءات الخوادم عند تغيّر التوقيت.

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

هل مراقبة أسعار الرحلات مسموحة؟

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

كم عدد الـ threads الذي أحتاجه لمراقبة 100 مسار؟

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

لماذا يختلف السعر الذي يجمعه السكربت عن السعر الذي أراه في المتصفح؟

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

ما الفرق بين حلّ CAPTCHA وتدوير البروكسي هنا؟

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

أدلة ذات صلة

• مراقبة أسعار المتاجر الإلكترونية عند ظهور اختبارات CAPTCHA
• تدوير البروكسي في مسارات جمع البيانات