用于AI代理的车辆列表数据收集：构建更智能的汽车智能

pip install git+https://github.com/capsolver-ai/capsolver-core.git
pip install aiohttp beautifulsoup4 pandas

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

AUTOMOTIVE_PLATFORMS = {
    "autotrader": {
        "name": "AutoTrader",
        "captcha_type": "ReCaptchaV2TaskProxyLess",
        "trigger": "after_50_searches",
        "data": ["year", "make", "model", "price", "mileage", "dealer", "vin", "features"]
    },
    "cargurus": {
        "name": "CarGurus",
        "captcha_type": "AntiTurnstileTaskProxyLess",
        "trigger": "rate_limit_80_per_hour",
        "data": ["price", "deal_rating", "mileage", "dealer_rating", "days_on_market"]
    },
    "cars_com": {
        "name": "Cars.com",
        "captcha_type": "ReCaptchaV2TaskProxyLess",
        "trigger": "after_40_listing_views",
        "data": ["price", "mileage", "features", "dealer", "history_report", "photos"]
    },
    "facebook_marketplace": {
        "name": "Facebook Marketplace",
        "captcha_type": "ReCaptchaV2TaskProxyLess",
        "trigger": "behavioral_and_rate",
        "data": ["price", "mileage", "location", "seller_type", "description"]
    }
}

import asyncio
from capsolver_core import create_capsolver, CaptchaType, CaptchaInfo
from dataclasses import dataclass
from typing import Optional

@dataclass
class VehicleListing:
    vin: Optional[str]
    year: int
    make: str
    model: str
    price: float
    mileage: int
    dealer: str
    location: str
    days_on_market: int
    platform: str
    url: str

class VehicleDataCollector:
    """使用CAPTCHA处理收集车辆列表数据。"""
    
    def __init__(self, api_key: str, proxies: list):
        self.cap = create_capsolver(api_key=api_key)
        self.proxies = proxies
        self.proxy_idx = 0
    
    async def search_vehicles(self, make: str, model: str, year_range: tuple, 
                              location: str, platform: str = "autotrader") -> list:
        """使用CAPTCHA处理搜索车辆列表。"""
        proxy = self.proxies[self.proxy_idx % len(self.proxies)]
        self.proxy_idx += 1
        
        html = await self._fetch_search(platform, make, model, year_range, location, proxy)
        
        if self._is_captcha(html):
            token = await self._solve(platform)
            html = await self._retry(platform, make, model, year_range, location, proxy, token)
        
        return self._parse_listings(html)
    
    async def _solve(self, platform_key: str) -> str:
        config = AUTOMOTIVE_PLATFORMS[platform_key]
        type_map = {
            "ReCaptchaV2TaskProxyLess": CaptchaType.RECAPTCHA_V2,
            "AntiTurnstileTaskProxyLess": CaptchaType.CLOUDFLARE
        }
        info = CaptchaInfo(
            type=type_map[config["captcha_type"]],
            website_url=f"https://www.{platform_key.replace('_', '')}.com",
            website_key=config.get("site_key", "")
        )
        solution = await self.cap.solve(info)
        return solution.token
    
    async def monitor_market(self, make: str, model: str, region: str) -> dict:
        """监控特定车辆的市场价格。"""
        all_listings = []
        for platform in ["autotrader", "cargurus", "cars_com"]:
            listings = await self.search_vehicles(make, model, (2020, 2024), region, platform)
            all_listings.extend(listings)
            await asyncio.sleep(5)
        
        prices = [l.price for l in all_listings if l.price > 0]
        return {
            "make_model": f"{make} {model}",
            "total_listings": len(all_listings),
            "avg_price": sum(prices) / len(prices) if prices else 0,
            "min_price": min(prices) if prices else 0,
            "max_price": max(prices) if prices else 0,
            "platforms_checked": 3
        }
    
    async def close():
        await self.cap.aclose()