通过验证码处理监控航空公司票价

旅游和航空公司网站经常使用验证码来限制自动票价检查。检查多条路线票价的价格监控系统将遇到 reCAPTCHA 挑战、Cloudflare 插页式广告和速率限制机制。 CaptchaAI 处理验证码解决步骤，以便您的监控管道可以继续收集票价数据。

本指南展示了如何构建票价监控工作流程，以在价格检查期间检测和解决验证码问题。

监控工作流程

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

通过验证码处理进行票价监控

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 或任务调度程序按计划运行监视器：

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

票价捕获清单

• Capture base fare, taxes, currency, and fare family in one payload so later comparisons stay usable.
• 至少保留两个选择器后备，并在解析失败时存储原始 HTML 快照。
• 记录每个解决事件的路线、出发日期、集合时间和验证码延迟。

故障排除

常问问题

我应该多久检查一次票价？

通常每 4 至 8 小时检查一次。更频繁的检查会增加验证码遇到次数和代理成本。

航空公司网站使用哪些验证码类型？

最常见的是 reCAPTCHA v2、Cloudflare Turnstile 或挑战页面，偶尔也会使用图像验证码。

我需要自有服务器基础设施吗？

是的。旅游网站主动阻止数据中心 IP。住宅或自有服务器基础设施的成功率明显更高。

我可以监控多家航空公司吗？

是的。为每个航空公司的响应格式自定义 _parse_fare 方法，并为每个站点添加路线。

如何处理 Cloudflare 验证流程 页面？

将 method=cloudflare_challenge 与代理一起使用。返回的 qa_session_cookie cookie 授予对站点的访问权限。请参阅Cloudflare 验证流程指南。

获取您的 CaptchaAI API 密钥

开始监控机票价格验证码网站。在价格跟踪工作流程中自动处理验证码。

相关指南

• 如何使用API​​解决reCAPTCHA v2
• 如何使用API​​解决Cloudflare Turnstile
• 使用验证码处理进行财务数据抓取
• 代理设置最佳实践