Flugpreisüberwachung mit CAPTCHA-Handhabung

Reise- und Flugreise-Websites verwenden häufig CAPTCHAs, um die automatische Fahrpreisüberprüfung einzuschränken. Ein Preisüberwachungssystem, das die Tarife auf mehreren Strecken überprüft, wird auf reCAPTCHA-Herausforderungen, Cloudflare-Interstitials und Tarifbegrenzungsmechanismen stoßen. CaptchaAI übernimmt den CAPTCHA-Lösungsschritt, damit Ihre Überwachungspipeline weiterhin Tarifdaten sammeln kann.

Dieser Leitfaden zeigt, wie Sie einen Workflow zur Tarifüberwachung aufbauen, der CAPTCHAs bei Preisprüfungen erkennt und löst.

Welche Prüf-Taktung ist realistisch?

Der Überwachungsworkflow

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

Was Sie brauchen

pip install requests

CaptchaAI Solver-Helfer

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

Fahrpreisüberwachung mit CAPTCHA-Verwaltung

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 with resilient selector fallbacks."""
        from bs4 import BeautifulSoup

        soup = BeautifulSoup(html, "html.parser")
        price_text = ""
        for selector in [
            ".fare-price",
            "[data-testid='fare-price']",
            ".price-amount",
            ".total-price",
        ]:
            element = soup.select_one(selector)
            if element:
                price_text = element.get_text(strip=True)
                break

        currency_match = re.search(r"(USD|EUR|GBP|\$|€|£)", price_text)
        return {
            "route": route["name"],
            "timestamp": datetime.now().isoformat(),
            "price_text": price_text,
            "currency": currency_match.group(1) if currency_match else "",
            "available": bool(price_text),
            "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))

Terminkontrollen

Führen Sie den Monitor nach einem Zeitplan mit cron oder einem Taskplaner aus:

# Check fares every 6 hours
0 */6 * * * cd /path/to/project && python fare_monitor.py >> /var/log/fares.log 2>&1

Fehlerbehebung

FAQ

Wie oft sollte ich die Tarife überprüfen?

Üblicherweise erfolgt dies alle 4 bis 8 Stunden. Häufigere Überprüfungen erhöhen die Zahl der CAPTCHA-Begegnungen und die Proxy-Kosten.

Welche CAPTCHA-Typen verwenden Airline-Websites?

Am häufigsten reCAPTCHA v2, Cloudflare Turnstile oder Challenge-Seiten und gelegentlich Bild-CAPTCHAs.

Benötige ich Residential-Proxys?

Ja. Reiseseiten blockieren aktiv Rechenzentrum-IPs. Residential- oder Mobile-Proxys weisen deutlich höhere Erfolgsraten auf.

Kann ich mehrere Fluggesellschaften überwachen?

Ja. Passen Sie die _parse_fare-Methode an das Antwortformat jeder Fluggesellschaft an und fügen Sie Routen für jeden Standort hinzu.

Wie gehe ich mit Cloudflare Challenge-Seiten um?

Verwenden Sie method=cloudflare_challenge mit einem Proxy. Das zurückgegebene cf_clearance-Cookie gewährt Zugriff auf die Website. Siehe dieCloudflare Challenge-Anleitung.

Nutzen Sie CaptchaAI in einem realistischen Monitoring-Rhythmus, der Preisrelevanz, Proxy-Kosten und CAPTCHA-Druck gegeneinander abwägt.

Verwandte Leitfäden

• So lösen Sie reCAPTCHA v2 mithilfe der API
• So lösen Sie Cloudflare Turnstile mithilfe der API
• Finanzdaten-Scraping mit CAPTCHA-Verwaltung
• Rotierende Residential-Proxys für höhere Erfolgsraten