机票价格监控与验证码处理

想让票价监控长期跑下去，脚本里必须有一步“遇到验证码就去识别”。旅行站点的风控比电商更敏感：同一条航线连查三四次，返回的往往是 reCAPTCHA 或 Cloudflare 挑战，而不是价格页。检测到挑战就把 sitekey 与 pageurl 交给 CaptchaAI，拿回 token 填进对应字段，再继续解析。

安全范围： 本文示例面向你自有或已获授权的采集场景。采集前请确认目标站点的 robots 协议与服务条款，并对齐《网络安全法》与 PIPL 的要求。

旅行站点会给你哪几类验证码

先确认要监控的 URL 落在哪一行，这决定脚本要写几个分支。

国内 OTA 是另一套体系

携程、去哪儿几乎不用 reCAPTCHA，主流是 GeeTest（极验）滑块，参数是 gt 和 challenge，与 googlekey 不是一回事。CaptchaAI 支持 GeeTest v3；v4 的官方口径是即将支持，现在还不能用。另外 reCAPTCHA 依赖 Google 托管脚本，境内加载不稳定，调试时容易误判成超时。

把验证码识别嵌进抓取循环

脚本只做三件事：取页面、遇到挑战调 CaptchaAI 换 token、解析价格写入历史。提交走 in.php，轮询 res.php。token 落点不能混：reCAPTCHA 填回 g-recaptcha-response，Turnstile 填回 cf-turnstile-response。

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

三个容易踩的点

• 先分清 v2 和 v3。 页面里出现 recaptcha/api.js?render= 就是 v3，要多带 version 和 action；action 与页面真实动作对不上，分数会偏低。
• 轮询节奏按类型给。 官方指标：reCAPTCHA v3 <4 秒，Turnstile <10 秒，GeeTest v3 <12 秒，v2 按 <60 秒预留；CAPCHA_NOT_READY 不是报错，继续轮询即可。
• time.sleep(3) 不要删。 请求压得越密，触发挑战的概率越高，省下的几秒会以识别耗时还回来。

价格由 JavaScript 渲染时，把取页面换成 Selenium 或 Playwright，识别流程不变。

票价监控多久跑一次：频率对准调价窗口

航司和 OTA 按批次刷新价格缓存，查得更勤很快进入收益递减区间。

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

个人盯几条航线，4 到 6 小时一轮足够；做代客比价或分销，1 到 2 小时是上限。值得提频的只有几个窗口：出发前 21 天、14 天、7 天这几个调价节点，以及春运、国庆前后。给重点航线单独配一条更密的任务，比全量提频划算。

套餐怎么选：并发线程决定价格

CaptchaAI 按并发线程数计价，不按识别条数：套餐内次数不限，也不因验证码类型加价。所以该问的是“同一时刻最多几个请求卡在验证码上”。

串行执行时并发恒为 1，50 条航线每 4 小时轮一遍也只占 1 个线程；线程池并行铺开才会顶到几十。先定并发模型，再选套餐。

常见问题

跑一段时间后验证码越来越频繁，是被拉黑了吗？

多半不是针对性封禁，而是风控的常规升级：同一会话反复查同一条航线，站点就会插入挑战。拉开间隔、把识别纳入重试流程即可继续。

token 拿到了，页面还是不放行，问题出在哪？

按三步查：pageurl 与实际请求的 URL 是否一致，v3 的 action 是否与页面动作对得上，token 是否填进了正确字段。错一项都会被判为未通过。

遇到 hCaptcha 怎么办？

CaptchaAI 目前不支持 hCaptcha 和 FunCaptcha（Arkose Labs），脚本应把这类页面标记为跳过，别反复重试占用线程。CaptchaFox、Friendly Captcha、Lemin 属于测试版（beta）支持。

监控 500 条航线，是不是必须上 VIP 套餐？

不需要。计费单位是并发线程，不是路线数量。500 条路线串行跑只占个位数线程；只有线程池同时发出几十条时，才需要 ADVANCE（$90/月，50 线程）。

延伸阅读

• 电商价格监控中的验证码处理
• 让抓取任务把验证码纳入重试流程
• 采集流程中代理与验证码的配合方式