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"

JOB_BOARDS = {
    "indeed": {
        "name": "Indeed",
        "captcha_type": "ReCaptchaV2TaskProxyLess",
        "trigger": "after_30_searches",
        "data_fields": ["title", "company", "location", "salary", "description", "posted_date"]
    },
    "linkedin_jobs": {
        "name": "LinkedIn Jobs",
        "captcha_type": "AntiTurnstileTaskProxyLess",
        "trigger": "rate_limit_and_behavioral",
        "data_fields": ["title", "company", "location", "level", "employment_type", "applicants"]
    },
    "glassdoor": {
        "name": "Glassdoor",
        "captcha_type": "ReCaptchaV2TaskProxyLess",
        "trigger": "after_40_page_views",
        "data_fields": ["title", "company", "salary_range", "rating", "reviews", "benefits"]
    },
    "ziprecruiter": {
        "name": "ZipRecruiter",
        "captcha_type": "AntiTurnstileTaskProxyLess",
        "trigger": "rate_limit_60_per_hour",
        "data_fields": ["title", "company", "salary", "location", "skills", "posted_date"]
    }
}

import asyncio
from capsolver_core import create_capsolver, CaptchaType, CaptchaInfo

class JobBoardCollector:
    """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 = {"listings_collected": 0, "captchas_solved": 0}
    
    async def search_jobs(self, query: str, location: str, platform: str = "indeed") -> list:
        """CAPTCHA処理付きで求人リストを検索します。"""
        proxy = self.proxies[self.proxy_idx % len(self.proxies)]
        self.proxy_idx += 1
        
        html = await self._fetch_search(platform, query, location, proxy)
        
        if self._is_captcha(html):
            token = await self._solve(platform)
            html = await self._retry(platform, query, location, proxy, token)
            self.stats["captchas_solved"] += 1
        
        listings = self._parse_listings(html)
        self.stats["listings_collected"] += len(listings)
        return listings
    
    async def _solve(self, platform_key: str) -> str:
        """求人ボードのCAPTCHAを解決します。"""
        platform = JOB_BOARDS[platform_key]
        type_map = {
            "ReCaptchaV2TaskProxyLess": CaptchaType.RECAPTCHA_V2,
            "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 collect_salary_data(self, role: str, location: str) -> dict:
        """特定の役割と場所の給与データを収集します。"""
        results = {}
        for platform in ["indeed", "glassdoor", "ziprecruiter"]:
            listings = await self.search_jobs(f"{role} salary", location, platform)
            salaries = [l.get("salary") for l in listings if l.get("salary")]
            if salaries:
                results[platform] = {
                    "min": min(salaries),
                    "max": max(salaries),
                    "median": sorted(salaries)[len(salaries)//2],
                    "sample_size": len(salaries)
                }
            await asyncio.sleep(5)
        return results
    
    async def close(self):
        await self.cap.aclose()

class RecruitingIntelligence:
    """AI採用エージェントのデータレイヤー。"""
    
    def __init__(self, collector: JobBoardCollector):
        self.collector = collector
    
    async def market_analysis(self, role: str, locations: list) -> dict:
        """特定の役割の求人市場を場所ごとに分析します。"""
        analysis = {}
        for location in locations:
            listings = await self.collector.search_jobs(role, location)
            analysis[location] = {
                "total_openings": len(listings),
                "companies_hiring": list(set(l.get("company") for l in listings if l.get("company"))),
                "common_skills": self._extract_skills(listings),
                "salary_range": self._salary_range(listings)
            }
            await asyncio.sleep(5)
        return analysis
    
    async def track_hiring_trends(self, companies: list, period_days: int = 30) -> dict:
        """特定の会社の採用活動を追跡します。"""
        trends = {}
        for company in companies:
            listings = await self.collector.search_jobs(company, "remote")
            trends[company] = {
                "active_listings": len(listings),
                "roles": [l.get("title") for l in listings[:10]],
                "locations": list(set(l.get("location") for l in listings if l.get("location")))
            }
            await asyncio.sleep(5)
        return trends
    
    def _extract_skills(self, listings: list) -> list:
        """求人説明文から一般的なスキルを抽出します。"""
        # 簡易的なスキル抽出
        all_skills = []
        for listing in listings:
            desc = listing.get("description", "").lower()
            common_skills = ["python", "javascript", "sql", "aws", "react", "machine learning", "docker", "kubernetes"]
            for skill in common_skills:
                if skill in desc:
                    all_skills.append(skill)
        from collections import Counter
        return [s for s, _ in Counter(all_skills).most_common(10)]