Giám sát giá vé du lịch bằng cách xử lý CAPTCHA

Crawler theo dõi giá vé chạy tốt vài ngày rồi đột nhiên chỉ nhận về trang trắng hoặc màn hình "Đang xác minh trình duyệt"? Vấn đề gần như luôn nằm ở một lớp: trang đặt vé đã bật CAPTCHA. Cách xử lý gọn nhất là để crawler tự phát hiện thử thách, gửi sang API giải CAPTCHA, nhận token rồi gửi lại request. Bài này đi đúng luồng đó bằng Python.

Bối cảnh quen với dev Việt Nam: một agency ở TP.HCM theo dõi giá chặng SGN–HAN và SGN–ICN cho khách doanh nghiệp, hoặc startup du lịch cần biết giá phòng Đà Nẵng biến động ra sao trước kỳ nghỉ lễ. Đều là dữ liệu công khai — chỉ là bạn không thể mở 200 tab mỗi sáng để chép tay.

Trang du lịch dùng loại CAPTCHA nào

Mỗi nhóm trang có thói quen riêng, quyết định bạn gọi method nào:

Cả năm nhóm trên đều rơi vào loại CaptchaAI hỗ trợ chính thức — reCAPTCHA v2/v3 (kể cả Enterprise), Cloudflare Turnstile, Cloudflare Challenge, GeeTest v3, CAPTCHA hình ảnh/OCR và grid image. Ngược lại, CaptchaAI hiện không giải hCaptcha và FunCaptcha (Arkose Labs); GeeTest v4 cũng chưa có, mới ở trạng thái sắp ra mắt.

Cách kiểm tra nhanh: xem source trang mục tiêu và tìm data-sitekey, cf-turnstile hoặc recaptcha/api.js?render=. Chuỗi cuối là dấu hiệu reCAPTCHA v3 — khác biệt quan trọng, vì v3 cần thêm tham số action khi gửi task.

Luồng xử lý bốn bước

Dù là trang nào, luồng vẫn giống nhau:

1. Request trang tìm kiếm bằng session có User-Agent thật.
2. Dò trong HTML xem có sitekey của reCAPTCHA hay Turnstile không.
3. Nếu có, gửi task tới in.php rồi polling res.php cho tới khi nhận được token.
4. Gửi lại request kèm token vào đúng field (g-recaptcha-response cho reCAPTCHA, cf-turnstile-response cho Turnstile) và parse giá từ HTML trả về.

Chỉ tham số gửi lên in.php và tên field nhận token là khác nhau giữa các loại.

Script theo dõi giá vé hoàn chỉnh

Class FareMonitor dưới đây gói cả bốn bước: fetch trang, nhận diện v2 hay v3, giải CAPTCHA, trích giá, lưu lịch sử và so sánh hai lần đo gần nhất để báo giảm giá.

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

Hai chỗ nên chỉnh trước khi đưa vào production. _extract_prices đang bắt chuỗi bắt đầu bằng dấu đô la và lọc trong khoảng 20–10000; với trang hiển thị giá VND bạn phải đổi regex sang định dạng 1.250.000 và bỏ bộ lọc theo biên độ USD. datetime.utcnow() lưu mốc UTC, nên khi dựng báo cáo cho team ở Việt Nam nhớ cộng bù 7 giờ, nếu không biểu đồ "giá theo giờ" sẽ lệch hẳn một buổi.

Vòng polling mặc định 60 lần cách nhau 5 giây, tức chờ tối đa 5 phút mỗi CAPTCHA — rộng rãi, chủ yếu để script không chết giữa chừng khi mạng chập chờn.

Lên lịch chạy định kỳ

Theo dõi giá là bài toán chạy nền. Một dòng cron là đủ cho phần lớn trường hợp:

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

Với nhiều tuyến, đừng để tất cả cùng khởi động ở phút 0 — chia ra vài offset (0, 15, 30) để số thread dùng đồng thời không bị dồn cục.

Chi phí thực tế tốn bao nhiêu

CaptchaAI tính tiền theo thread, không theo từng lần giải: một thread là một CAPTCHA đang xử lý, giải xong thì nhận việc tiếp theo, và số lượt giải trong tháng không giới hạn. Câu hỏi đúng vì thế là "cần bao nhiêu thread chạy song song".

Cột cuối là gợi ý theo mức chạy song song: dồn 3.000 lượt kiểm tra vào một khung giờ hẹp thì cần nhiều thread hơn là rải đều 24 giờ. Giá niêm yết bằng USD; bảng giá đầy đủ tới VIP-3 ($7,500/tháng, 5.000 thread) nằm trên trang pricing của CaptchaAI.

Vài lưu ý trước khi bật chạy thật

• Chỉ lấy dữ liệu công khai. Giá trên trang tìm kiếm là công khai; dữ liệu sau màn hình đăng nhập của người khác thì không.
• Ghi log tối thiểu. Nếu pipeline chạm tới dữ liệu cá nhân, Nghị định 13/2023/NĐ-CP về bảo vệ dữ liệu cá nhân là lý do đủ tốt để chỉ lưu đúng trường cần dùng và giữ audit log.
• Theo dõi mã lỗi. ERROR_WRONG_GOOGLEKEY xuất hiện dồn dập thường nghĩa là trang đã đổi sitekey, không phải API hỏng — regex của bạn đang bắt nhầm một key cũ trong script inline.

Câu hỏi thường gặp

Làm sao biết trang đang dùng reCAPTCHA v2 hay v3?

Tìm recaptcha/api.js?render= trong HTML. Có chuỗi này là v3 và bạn phải gửi kèm version cùng action; không có thì xử lý như v2. Script ở trên đã tự kiểm tra trước khi gửi task.

Script bị Cloudflare Challenge chặn ngay từ request đầu tiên thì làm gì?

Cloudflare Challenge được hỗ trợ chính thức nhưng khác Turnstile: thử thách nằm ở tầng trang chứ không phải widget trong form. Gửi task theo đúng method của nó và giữ nguyên session cookie sau khi giải, thay vì tạo session mới mỗi request.

Có bắt buộc dùng Selenium hay Playwright không?

Chỉ khi giá được render bằng JavaScript sau khi trang tải xong. Nhiều trang meta-search trả giá qua một API JSON nội bộ — mở tab Network trong DevTools, tìm được endpoint đó thì gọi thẳng, nhanh và nhẹ hơn hẳn.

Nên kiểm tra giá bao nhiêu lần một ngày?

Mỗi 4–6 giờ cho nhu cầu cá nhân, 1–2 giờ cho công việc. Các hãng cập nhật giá theo đợt, nên chạy mỗi 15 phút hầu như chỉ đốt thread.

Theo dõi giá khách sạn có dùng lại được script này không?

Được, phần fetch và giải CAPTCHA giữ nguyên. Chỉ cần viết lại _extract_prices cho layout trang khách sạn và thêm hai trường ngày nhận phòng, ngày trả phòng vào key định danh, vì giá phòng phụ thuộc độ dài kỳ lưu trú.

Bài liên quan

• Giám sát giá thương mại điện tử
• Scraping ổn định khi gặp CAPTCHA
• Luân phiên proxy khi scraping có CAPTCHA