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

كثيرًا ما تستخدم مواقع السفر وشركات الطيران اختبارات CAPTCHA للحد من التحقق الآلي من الأسعار. وأي نظام يراقب الأسعار عبر مسارات متعددة سيصطدم بتحديات reCAPTCHA، وصفحات Cloudflare البينية، وآليات الحد من المعدل. يتولى CaptchaAI خطوة حل CAPTCHA حتى يتمكن خط المراقبة من الاستمرار في جمع بيانات الأسعار بدل التوقف عند كل تحدٍّ.

يشرح هذا الدليل كيفية بناء سير عمل يكتشف CAPTCHA أثناء فحص الأسعار، ثم يعالجه ويكمل استخراج البيانات من دون كسر المسار بالكامل.

سير عمل المراقبة

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

ما تحتاجه

pip install requests

مساعد الحل CaptchaAI

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")

مراقبة الأسعار من خلال التعامل مع اختبار CAPTCHA

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))

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

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

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

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

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

كم مرة يجب أن أتحقق من الأسعار؟

يُعدّ الفحص كل 4 إلى 8 ساعات معدلاً مناسبًا لمعظم حالات الاستخدام. زيادة تكرار عمليات الفحص ترفع احتمالية ظهور CAPTCHA وتكاليف البروكسي.

ما أنواع CAPTCHA التي تستخدمها مواقع شركات الطيران؟;

reCAPTCHA v2 وCloudflare Turnstile هما الأكثر شيوعًا. تستخدم بعض المواقع أيضًا تحدي المتصفحات الكاملة (Challenge Page)، ونادرًا CAPTCHA الصور.

هل أحتاج إلى بروكسيات سكنية؟;

نعم. مواقع السفر تحظر عناوين IP لمراكز البيانات بشكل متكرر، وتحقّق البروكسيات السكنية وبروكسيات الجوال معدلات نجاح أعلى بكثير.

هل يمكنني مراقبة شركات طيران متعددة؟

نعم. قم بتخصيص طريقة _parse_fare لتنسيق الاستجابة لكل شركة طيران وأضف مسارات لكل موقع.

كيف أتعامل مع صفحات Cloudflare Challenge؟

استخدم method=cloudflare_challenge مع بروكسي. ملفف الارتباط cf_clearance المعاد يتيح الوصول إلى الموقع. انظر دليل Cloudflare Challenge.

احصل على مفتاح CaptchaAI API

ابدأ بمراقبة أسعار تذاكر الطيران عبر captchaai.com. عالج اختبارات CAPTCHA تلقائيًا ضمن سير عمل تتبع الأسعار.

أدلة ذات صلة

• كيفية حل reCAPTCHA v2 باستخدام API
• كيفية حل Cloudflare Turnstile باستخدام API
• تجريف البيانات المالية باستخدام معالجة CAPTCHA
• أفضل ممارسات إعداد الوكيل