Pemantauan Tarif Penerbangan dan Perjalanan dengan Penanganan CAPTCHA

Script pemantau tarif jarang gagal karena pola ekstraksi harganya salah. Ia berhenti karena halaman yang dikembalikan server bukan hasil pencarian, melainkan tantangan CAPTCHA. Solusinya bukan menambah retry, melainkan memperlakukan CAPTCHA sebagai bagian normal dari alur pengambilan halaman: deteksi, selesaikan, lanjutkan. Artikel ini menunjukkan cara mengenali reCAPTCHA v2/v3 dan Cloudflare Turnstile di halaman pencarian tiket, menyelesaikannya lewat API CaptchaAI, lalu melanjutkan parsing harga seperti biasa.

Konteksnya nyata untuk pasar Indonesia. Banyak pekerjaan monitoring harga di sini datang dari agensi kecil dan freelancer kontrak yang memantau puluhan rute domestik — CGK–DPS, CGK–SUB, SUB–UPG — untuk klien travel. Biaya per pengecekan menentukan margin, dan itulah alasan model harga berbasis thread lebih masuk akal daripada tarif per solve.

Contoh di artikel ini memakai domain contoh. Pantau hanya endpoint yang Anda miliki atau yang izin aksesnya sudah Anda kantongi, dan jauhi data pribadi — UU 27/2022 tentang Pelindungan Data Pribadi berlaku atas data yang Anda kumpulkan dan simpan.

Jenis CAPTCHA yang muncul di situs perjalanan

Petakan dulu apa yang Anda hadapi sebelum menulis kode. Kategori situs cukup andal memprediksi jenis tantangan dan frekuensinya.

Semua tipe di tabel ini termasuk yang ditangani CaptchaAI. Dua di antaranya menyumbang sebagian besar kasus di jalur perjalanan: reCAPTCHA v2 pada form pencarian OTA dan Cloudflare Turnstile pada halaman yang dilindungi Cloudflare. Untuk situs yang memakai GeeTest v3, polanya identik — yang berubah hanya nilai parameter method saat Anda mengirim task.

Satu fungsi fetch yang sadar CAPTCHA

Kesalahan arsitektur paling umum adalah menyebar logika CAPTCHA ke seluruh script. Bungkus semuanya dalam satu fungsi pengambil halaman. Alurnya tetap empat langkah yang sama seperti integrasi CaptchaAI lainnya: kirim task ke in.php → simpan task ID → polling res.php → pakai token pada request berikutnya.

Deteksi tipe dibaca dari HTML respons. Kehadiran recaptcha/api.js?render= menandakan v3, yang butuh parameter version dan action; data-sitekey tanpa penanda itu berarti v2. Penanda cf-turnstile menandakan Turnstile, dan tokennya dikirim balik lewat field cf-turnstile-response, bukan g-recaptcha-response.

Implementasi pemantau tarif

Kelas FareMonitor berikut menyatukan tiga tanggung jawab yang tetap terpisah rapi: pengambilan halaman yang sadar CAPTCHA, ekstraksi harga, dan pendeteksian penurunan harga terhadap riwayat.

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

Perhatikan time.sleep(3) di antara rute. Jeda itu bukan formalitas: pengecekan yang terlalu rapat memicu lebih banyak tantangan, sehingga menaikkan biaya dan waktu total. Untuk halaman yang harganya baru muncul setelah JavaScript berjalan, ganti bagian session.get dengan Selenium atau Playwright dan biarkan sisa alurnya tetap sama.

Menjadwalkan pengecekan

Cron sudah lebih dari cukup untuk sebagian besar kebutuhan. Jalankan dari server yang dekat dengan target — misalnya region AWS ap-southeast-1 (Singapura) atau ap-southeast-3 (Jakarta) — agar latensi pengambilan halaman tetap rendah.

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

Sebarkan jadwal antar-rute; jangan menembakkan 50 rute sekaligus pada menit yang sama. Antrean yang merata membuat jumlah thread yang Anda butuhkan jauh lebih kecil.

Perkiraan biaya dan pemilihan paket

Angka berikut adalah perkiraan beban kerja, bukan tagihan pasti — hasilnya bergantung pada situs target, frekuensi, dan seberapa sering tantangan muncul.

Ini bagian yang paling sering disalahpahami: CaptchaAI menagih per thread yang berjalan bersamaan, bukan per solve, dan setiap paket memberi solve tanpa batas selama bulan berjalan. Jadi pertanyaannya bukan "berapa banyak CAPTCHA per hari", melainkan "berapa banyak pengecekan yang berjalan bersamaan pada puncak jadwal". Agensi dengan 200 CAPTCHA per hari yang tersebar rata sepanjang hari hampir tidak pernah membutuhkan lebih dari belasan thread. Harga selalu dalam USD; daftar paket lengkap ada di halaman pricing CaptchaAI.

Pertanyaan umum

Berapa thread yang saya butuhkan untuk memantau 50 rute?

Hitung dari puncak konkurensi, bukan total harian. Kalau 50 rute dicek empat kali sehari dan Anda menyebarnya dalam jendela sepuluh menit, belasan thread sudah memadai — STANDARD ($30/bulan, 15 thread) jadi titik awal yang wajar. Naikkan paket hanya kalau antrean mulai menumpuk.

Situs target saya memakai hCaptcha — apa yang harus saya lakukan?

Belum. hCaptcha dan FunCaptcha tidak didukung saat ini, jadi untuk target yang memakai keduanya Anda memerlukan layanan lain. GeeTest v4 juga belum tersedia — statusnya segera hadir — sementara GeeTest v3 sudah didukung penuh.

Bagaimana kalau harga baru muncul setelah JavaScript dijalankan?

Ganti tahap pengambilan halaman dengan browser headless (Selenium atau Playwright), lalu tetap kirim sitekey ke CaptchaAI dan suntikkan token hasilnya ke halaman. Logika CAPTCHA tidak berubah, hanya cara halaman diambil.

Bisakah script yang sama dipakai untuk harga hotel?

Bisa. Alur deteksi dan penyelesaian CAPTCHA identik; yang perlu disesuaikan hanya pola ekstraksi harga dan parameter pencarian, karena halaman hotel memuat rentang tanggal dan jumlah tamu.

Kenapa jumlah CAPTCHA melonjak padahal jadwal tidak saya ubah?

Biasanya karena pengecekan menumpuk di menit yang sama. Cron 0 */4 * * * untuk semua rute berarti puluhan request serempak dari satu IP, dan situs merespons dengan lebih banyak tantangan. Sebarkan jadwal per rute ke menit yang berbeda sebelum menambah thread.

Panduan terkait

• Pemantauan harga di e-commerce
• Scraping yang andal tanpa terhenti blokir
• Rotasi proxy untuk scraping dengan CAPTCHA