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"

REAL_ESTATE_PLATFORMS = {
    "zillow": {
        "name": "Zillow",
        "captcha_type": "ReCaptchaV2TaskProxyLess",
        "trigger": "after_50_listing_views",
        "data_fields": ["price", "beds", "baths", "sqft", "lot_size", "year_built", "zestimate"]
    },
    "realtor_com": {
        "name": "Realtor.com",
        "captcha_type": "AntiTurnstileTaskProxyLess",
        "trigger": "rate_limit_80_per_hour",
        "data_fields": ["price", "beds", "baths", "sqft", "listing_date", "agent"]
    },
    "redfin": {
        "name": "Redfin",
        "captcha_type": "ReCaptchaV3TaskProxyLess",
        "trigger": "behavioral_score",
        "data_fields": ["price", "beds", "baths", "sqft", "redfin_estimate", "hot_home"]
    }
}

import asyncio
from capsolver_core import create_capsolver, CaptchaType, CaptchaInfo

class RealEstateCollector:
    """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_listings(self, market: str, platform: str, max_pages: int = 10) -> list:
        """CAPTCHA処理を備えた市場の不動産リストを収集。"""
        listings = []
        
        for page in range(max_pages):
            proxy = self.proxies[self.proxy_idx % len(self.proxies)]
            self.proxy_idx += 1
            
            html = await self._fetch_page(platform, market, page, proxy)
            
            # CAPTCHAがトリガーされた場合の処理
            if self._is_captcha(html):
                token = await self._solve(platform)
                html = await self._retry_with_token(platform, market, page, proxy, token)
            
            page_listings = self._parse_listings(html)
            listings.extend(page_listings)
            
            if not page_listings:
                break
            
            await asyncio.sleep(5)  # レートリミット
        
        return listings
    
    async def _solve(self, platform_key: str) -> str:
        """不動産プラットフォームのCAPTCHAを解決。"""
        platform = REAL_ESTATE_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 RealEstateMonitor:
    """リスト変更を検出する不動産市場のモニタリング。"""
    
    def __init__(self, collector: RealEstateCollector):
        self.collector = collector
        self.listing_history = {}
    
    async def monitor_market(self, market: str, platforms: list) -> dict:
        """複数のプラットフォームで市場をモニタリングし、変更を検出。"""
        changes = {"new_listings": [], "price_changes": [], "removed": []}
        
        for platform in platforms:
            current = await self.collector.collect_listings(market, platform)
            
            for listing in current:
                listing_id = listing.get("id")
                if listing_id in self.listing_history:
                    prev = self.listing_history[listing_id]
                    if listing["price"] != prev["price"]:
                        changes["price_changes"].append({
                            "id": listing_id,
                            "address": listing["address"],
                            "old_price": prev["price"],
                            "new_price": listing["price"]
                        })
                else:
                    changes["new_listings"].append(listing)
                
                self.listing_history[listing_id] = listing
        
        return changes