ソーシャルメディアの公開データ 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"

SOCIAL_PLATFORMS = {
    "reddit_public": {
        "name": "Reddit (Public)",
        "captcha_type": "ReCaptchaV2TaskProxyLess",
        "trigger": "rate_limit_after_30_pages",
        "public_data": ["posts", "comments", "upvotes", "subreddit_stats"]
    },
    "twitter_public": {
        "name": "X/Twitter (Public)",
        "captcha_type": "AntiTurnstileTaskProxyLess",
        "trigger": "behavioral_and_rate",
        "public_data": ["tweets", "likes", "retweets", "follower_count"]
    },
    "linkedin_public": {
        "name": "LinkedIn (Public Profiles)",
        "captcha_type": "AntiTurnstileTaskProxyLess",
        "trigger": "after_20_profile_views",
        "public_data": ["name", "headline", "company", "public_posts"]
    },
    "review_platforms": {
        "name": "G2/Trustpilot/Capterra",
        "captcha_type": "ReCaptchaV2TaskProxyLess",
        "trigger": "after_50_reviews_viewed",
        "public_data": ["reviews", "ratings", "company_scores", "feature_mentions"]
    }
}

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

@dataclass
class SocialDataPoint:
    platform: str
    content_type: str
    text: str
    engagement: dict
    author: str
    timestamp: float
    url: str

class SocialDataCollector:
    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_public_posts(self, platform: str, query: str, max_pages: int = 5) -> List[SocialDataPoint]:
        results = []
        for page in range(max_pages):
            proxy = self.proxies[self.proxy_idx % len(self.proxies)]
            self.proxy_idx += 1
            html = await self._fetch(platform, query, page, proxy)
            if self._is_captcha(html):
                token = await self._solve(platform)
                html = await self._retry(platform, query, page, proxy, token)
            posts = self._parse_posts(html, platform)
            results.extend(posts)
            if not posts:
                break
            await asyncio.sleep(5)
        return results
    
    async def _solve(self, platform_key: str) -> str:
        config = SOCIAL_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.split('_')[0]}.com",
            website_key=config.get("site_key", "")
        )
        solution = await self.cap.solve(info)
        return solution.token
    
    async def monitor_brand_mentions(self, brand: str, platforms: list) -> dict:
        all_mentions = []
        for platform in platforms:
            posts = await self.collect_public_posts(platform, brand)
            mentions = [p for p in posts if brand.lower() in p.text.lower()]
            all_mentions.extend(mentions)
        return {
            "brand": brand,
            "total_mentions": len(all_mentions),
            "by_platform": {p: len([m for m in all_mentions if m.platform == p]) for p in platforms}
        }
    
    async def close(self):
        await self.cap.aclose()