工商注册数据提取面向人工智能代理：自动化公司验证

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": "特拉华州州务卿",
        "url": "https://icis.corp.delaware.gov",
        "captcha_type": "ReCaptchaV2TaskProxyLess",
        "data": ["entity_name", "file_number", "status", "formation_date", "agent"]
    },
    "california_sos": {
        "name": "加利福尼亚州州务卿",
        "url": "https://bizfileonline.sos.ca.gov",
        "captcha_type": "ReCaptchaV2TaskProxyLess",
        "data": ["entity_name", "number", "status", "type", "jurisdiction", "agent"]
    },
    "uk_companies_house": {
        "name": "英国公司局",
        "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")
        }