AIエージェント向けの旅行空き状況データの収集方法

# CapSolverコアエンジン
pip install git+https://github.com/capsolver-ai/capsolver-core.git

# データ収集の依存関係
pip install aiohttp beautifulsoup4 pandas schedule

export CAPSOLVER_API_KEY="your-capsolver-api-key"

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