イーコマース商品データの収集 アイ・エージェント向け 完全ガイド

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

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

ECOMMERCE_PLATFORMS = {
    "amazon": {
        "captcha_type": "ImageToTextTask",
        "trigger": "after_80_requests_or_rapid_navigation",
        "data": ["title", "price", "rating", "reviews_count", "availability", "seller"]
    },
    "shopify_stores": {
        "captcha_type": "AntiTurnstileTaskProxyLess",
        "trigger": "cloudflare_managed_challenge",
        "data": ["title", "price", "variants", "inventory", "description", "images"]
    },
    "ebay": {
        "captcha_type": "ReCaptchaV2TaskProxyLess",
        "trigger": "after_60_searches",
        "data": ["title", "price", "bids", "seller_rating", "shipping", "condition"]
    },
    "walmart": {
        "captcha_type": "AntiTurnstileTaskProxyLess",
        "trigger": "rate_limit_and_behavioral",
        "data": ["title", "price", "availability", "pickup_options", "reviews"]
    }
}

import asyncio
from capsolver_core import create_capsolver, CaptchaType, CaptchaInfo

class EcommerceDataCollector:
    """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 collect_product(self, url: str, platform: str) -> dict:
        """特定のURLから製品データを収集し、CAPTCHA処理を行います。"""
        proxy = self.proxies[self.proxy_idx % len(self.proxies)]
        self.proxy_idx += 1
        
        html = await self._fetch(url, proxy)
        
        if self._is_captcha(html):
            token = await self._solve(platform, url)
            html = await self._retry(url, proxy, token)
        
        return self._parse_product(html, platform)
    
    async def _solve(self, platform: str, url: str) -> str:
        config = ECOMMERCE_PLATFORMS[platform]
        type_map = {
            "ImageToTextTask": CaptchaType.RECAPTCHA_V2,  # 別の方法で処理
            "ReCaptchaV2TaskProxyLess": CaptchaType.RECAPTCHA_V2,
            "AntiTurnstileTaskProxyLess": CaptchaType.CLOUDFLARE
        }
        
        info = CaptchaInfo(
            type=type_map.get(config["captcha_type"], CaptchaType.RECAPTCHA_V2),
            website_url=url,
            website_key=config.get("site_key", "")
        )
        solution = await self.cap.solve(info)
        return solution.token
    
    async def monitor_prices(self, product_urls: list, platform: str) -> list:
        """製品リストの価格をモニタリングします。"""
        results = []
        for url in product_urls:
            data = await self.collect_product(url, platform)
            results.append(data)
            await asyncio.sleep(5)
        return results
    
    async def close(self):
        await self.cap.aclose()