如何为AI代理收集旅行可用性数据

# CapSolver核心引擎
pip install git+https://github.com/capsolver-ai/capsolver-core.git

# 数据收集依赖
pip install aiohttp beautifulsoup4 pandas schedule

export CAPSOLVER_API_KEY="你的CapSolver API密钥"

TRAVEL_PLATFORMS = {
    "booking_com": {
        "name": "Booking.com",
        "captcha_type": "AntiTurnstileTaskProxyLess",
        "trigger": "after_20_searches_per_session",
        "data_available": ["hotels", "apartments", "pricing", "availability"]
    },
    "skyscanner": {
        "name": "Skyscanner",
        "captcha_type": "ReCaptchaV2TaskProxyLess",
        "trigger": "rate_limit_50_per_hour",
        "data_available": ["flights", "prices", "routes", "airlines"]
    },
    "expedia": {
        "name": "Expedia",
        "captcha_type": "ReCaptchaV3TaskProxyLess",
        "trigger": "behavioral_score_based",
        "data_available": ["flights", "hotels", "packages", "car_rental"]
    },
    "airbnb": {
        "name": "Airbnb",
        "captcha_type": "AntiTurnstileTaskProxyLess",
        "trigger": "after_search_and_listing_views",
        "data_available": ["listings", "pricing", "availability", "reviews"]
    }
}

import asyncio
import aiohttp
import time
from capsolver_core import create_capsolver, CaptchaType, CaptchaInfo

class TravelDataCollector:
    """收集旅行可用性数据并处理CAPTCHA。"""
    
    def __init__(self, api_key: str, proxies: list):
        self.cap = create_capsolver(api_key=api_key)
        self.proxies = proxies
        self.proxy_idx = 0
        self.stats = {"queries": 0, "captchas_solved": 0}
    
    async def search_flights(self, origin: str, destination: str, date: str) -> dict:
        """搜索航班可用性并处理CAPTCHA。"""
        proxy = self.proxies[self.proxy_idx % len(self.proxies)]
        self.proxy_idx += 1
        
        # 尝试搜索
        result = await self._make_search_request(
            platform="skyscanner", 
            params={"origin": origin, "dest": destination, "date": date},
            proxy=proxy
        )
        
        # 处理触发的CAPTCHA
        if self._is_captcha_response(result):
            token = await self._solve_captcha("skyscanner")
            result = await self._retry_with_token(token, proxy)
            self.stats["captchas_solved"] += 1
        
        self.stats["queries"] += 1
        return self._parse_flight_results(result)
    
    async def search_hotels(self, location: str, checkin: str, checkout: str) -> dict:
        """搜索酒店可用性并处理CAPTCHA。"""
        proxy = self.proxies[self.proxy_idx % len(self.proxies)]
        self.proxy_idx += 1
        
        result = await self._make_search_request(
            platform="booking_com",
            params={"location": location, "checkin": checkin, "checkout": checkout},
            proxy=proxy
        )
        
        if self._is_captcha_response(result):
            token = await self._solve_captcha("booking_com")
            result = await self._retry_with_token(token, proxy)
            self.stats["captchas_solved"] += 1
        
        self.stats["queries"] += 1
        return self._parse_hotel_results(result)
    
    async def _solve_captcha(self, platform_key: str) -> str:
        """为特定旅行平台解决CAPTCHA。"""
        platform = TRAVEL_PLATFORMS[platform_key]
        
        type_map = {
            "ReCaptchaV2TaskProxyLess": CaptchaType.RECAPTCHA_V2,
            "ReCaptchaV3TaskProxyLess": CaptchaType.RECAPTCHA_V3,
            "AntiTurnstileTaskProxyLess": CaptchaType.CLOUDFLARE
        }
        
        info = CaptchaInfo(
            type=type_map[platform["captcha_type"]],
            website_url=f"https://www.{platform_key.replace('_', '')}.com",
            website_key=platform.get("site_key", "")
        )
        
        solution = await self.cap.solve(info)
        return solution.token
    
    async def close(self):
        await self.cap.aclose()

class TravelPriceMonitor:
    """监控旅行价格并在价格变化时发出警报。"""
    
    def __init__(self, collector: TravelDataCollector):
        self.collector = collector
        self.price_history = {}
    
    async def monitor_route(self, origin: str, dest: str, dates: list) -> list:
        """监控多天的航班路线。"""
        alerts = []
        for date in dates:
            current = await self.collector.search_flights(origin, dest, date)
            
            route_key = f"{origin}-{dest}-{date}"
            if route_key in self.price_history:
                prev_price = self.price_history[route_key]
                curr_price = current.get("lowest_price", 0)
                
                if curr_price < prev_price * 0.9:  # 10%以上的价格下降
                    alerts.append({
                        "type": "PRICE_DROP",
                        "route": f"{origin} → {dest}",
                        "date": date,
                        "old_price": prev_price,
                        "new_price": curr_price,
                        "savings": f"{(1 - curr_price/prev_price)*100:.0f}%"
                    })
            
            self.price_history[route_key] = current.get("lowest_price", 0)
            await asyncio.sleep(5)  # 限速
        
        return alerts