Pemantauan Tarif Maskapai dengan Penanganan CAPTCHA

Situs perjalanan dan maskapai penerbangan sering kali menggunakan CAPTCHA untuk membatasi pemeriksaan tarif otomatis. Sistem pemantauan harga yang memeriksa tarif di berbagai rute akan menghadapi tantangan reCAPTCHA, interstisial Cloudflare, dan mekanisme pembatasan tarif. CaptchaAI menangani langkah penyelesaian CAPTCHA sehingga saluran pemantauan Anda dapat terus mengumpulkan data tarif.

Panduan ini menunjukkan cara membangun alur kerja pemantauan tarif yang mendeteksi dan menyelesaikan CAPTCHA selama pemeriksaan harga.

Alur kerja pemantauan

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

Apa yang Anda butuhkan

pip install requests

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

Pemantauan tarif dengan penanganan 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))

Penjadwalan pemeriksaan

Jalankan monitor sesuai jadwal menggunakan cron atau penjadwal tugas:

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

Daftar periksa pengambilan tarif

• Catat tarif dasar, pajak, mata uang, dan kelompok tarif dalam satu payload sehingga perbandingan selanjutnya tetap dapat digunakan.
• Simpan setidaknya dua fallback pemilih dan simpan cuplikan HTML mentah setiap kali penguraian gagal.
• Catat rute, tanggal keberangkatan, waktu pengumpulan, dan latensi CAPTCHA pada setiap peristiwa penyelesaian.

Pemecahan Masalah

Pertanyaan Umum

Seberapa sering saya harus memeriksa tarif?

Biasanya setiap 4–8 jam. Pemeriksaan yang lebih sering akan meningkatkan pertemuan CAPTCHA dan biaya proxy.

Jenis CAPTCHA apa yang digunakan situs maskapai penerbangan?

Paling umum reCAPTCHA v2, Cloudflare Turnstile atau halaman Tantangan, dan terkadang gambar CAPTCHA.

Apakah saya memerlukan proxy perumahan?

Ya. Situs perjalanan secara aktif memblokir IP pusat data. egress jaringan yang diotorisasi atau seluler memiliki tingkat keberhasilan yang jauh lebih tinggi.

Bisakah saya memantau beberapa maskapai penerbangan?

Ya. Sesuaikan metode _parse_fare untuk format respons setiap maskapai penerbangan dan tambahkan rute untuk setiap situs.

Bagaimana cara menangani halaman Cloudflare Challenge?

Gunakan method=cloudflare_challenge dengan proxy. Cookie qa_validation_cookie yang dikembalikan memberikan akses ke situs. Lihat Panduan Cloudflare Challenge.

Dapatkan kunci API CaptchaAI Anda

Mulai pantau tarif maskapai dicaptchaai.com. Tangani CAPTCHA secara otomatis dalam alur kerja pelacakan harga Anda.

Panduan terkait

• Cara Mengatasi reCAPTCHA v2 Menggunakan API
• Cara Mengatasi Cloudflare Turnstile Menggunakan API
• Scraping Data Keuangan dengan Penanganan CAPTCHA
• Pengaturan egress jaringan yang diotorisasi Berputar