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"

BUSINESS_REGISTRIES = {
    "delaware_sos": {
        "name": "デラウェア州 Secretary of State",
        "url": "https://icis.corp.delaware.gov",
        "captcha_type": "ReCaptchaV2TaskProxyLess",
        "data": ["entity_name", "file_number", "status", "formation_date", "agent"]
    },
    "california_sos": {
        "name": "カリフォルニア州 Secretary of State",
        "url": "https://bizfileonline.sos.ca.gov",
        "captcha_type": "ReCaptchaV2TaskProxyLess",
        "data": ["entity_name", "number", "status", "type", "jurisdiction", "agent"]
    },
    "uk_companies_house": {
        "name": "英国 Companies House",
        "url": "https://find-and-update.company-information.service.gov.uk",
        "captcha_type": "ReCaptchaV2TaskProxyLess",
        "data": ["company_name", "number", "status", "type", "incorporated", "officers"]
    },
    "sec_edgar": {
        "name": "SEC EDGAR",
        "url": "https://www.sec.gov/cgi-bin/browse-edgar",
        "captcha_type": "ImageToTextTask",
        "data": ["company_name", "cik", "filings", "sic_code", "state"]
    }
}

import asyncio
from capsolver_core import create_capsolver, CaptchaType, CaptchaInfo

class BusinessRegistryExtractor:
    """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 verify_company(self, company_name: str, jurisdiction: str) -> dict:
        """特定の管轄区域の登録データベースで会社を確認します。"""
        registry = BUSINESS_REGISTRIES.get(jurisdiction)
        if not registry:
            return {"error": f"サポートされていない管轄区域: {jurisdiction}"}
        
        proxy = self.proxies[self.proxy_idx % len(self.proxies)]
        self.proxy_idx += 1
        
        # 登録データベースを検索
        html = await self._search_registry(registry, company_name, proxy)
        
        # CAPTCHAを処理
        if self._is_captcha(html):
            token = await self._solve(registry)
            html = await self._retry_search(registry, company_name, proxy, token)
        
        # 結果を解析
        return self._parse_registry_result(html, registry)
    
    async def _solve(self, registry: dict) -> str:
        type_map = {
            "ReCaptchaV2TaskProxyLess": CaptchaType.RECAPTCHA_V2,
            "ImageToTextTask": CaptchaType.RECAPTCHA_V2  # 異なる処理
        }
        info = CaptchaInfo(
            type=type_map[registry["captcha_type"]],
            website_url=registry["url"],
            website_key=registry.get("site_key", "")
        )
        solution = await self.cap.solve(info)
        return solution.token
    
    async def bulk_verify(self, companies: list, jurisdiction: str) -> list:
        """複数の会社を一括で確認します。"""
        results = []
        for company in companies:
            result = await self.verify_company(company, jurisdiction)
            results.append(result)
            await asyncio.sleep(3)  # レートリミットを尊重
        return results
    
    async def close(self):
        await self.cap.aclose()

class CompanyVerificationAgent:
    """ビジネス確認ワークフロー用のAIエージェントレイヤー。"""
    
    def __init__(self, extractor: BusinessRegistryExtractor):
        self.extractor = extractor
    
    async def full_verification(self, company_name: str) -> dict:
        """複数の登録機関で完全な確認を実行します。"""
        results = {}
        
        # 主な米国登録機関をチェック
        for jurisdiction in ["delaware_sos", "california_sos"]:
            result = await self.extractor.verify_company(company_name, jurisdiction)
            if result.get("status") == "Active":
                results["us_registration"] = result
                break
        
        # 公開企業用にSECをチェック
        sec_result = await self.extractor.verify_company(company_name, "sec_edgar")
        if sec_result.get("cik"):
            results["sec_filing"] = sec_result
        
        # 確認レポートをまとめます
        return {
            "company": company_name,
            "verified": bool(results),
            "registrations": results,
            "verification_date": time.strftime("%Y-%m-%d")
        }