Python中如何实现基于多个打码平台的验证码识别服务？

import abc
import requests
from typing import Optional, Dict, List
from dataclasses import dataclass
from concurrent.futures import ThreadPoolExecutor, as_completed
import time

@dataclass
class CaptchaResult:
    """验证码识别结果数据类"""
    success: bool
    code: Optional[str] = None
    platform: Optional[str] = None
    error_msg: Optional[str] = None
    cost_time: Optional[float] = None

class BaseCaptchaPlatform(abc.ABC):
    """验证码平台基类"""
    
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.base_url = self.get_base_url()
    
    @abc.abstractmethod
    def get_base_url(self) -> str:
        """获取平台API基础URL"""
        pass
    
    @abc.abstractmethod
    def recognize(self, image_data: bytes, **kwargs) -> CaptchaResult:
        """识别验证码核心方法"""
        pass
    
    def _make_request(self, url: str, data: Dict, files: Dict = None) -> Dict:
        """统一请求处理"""
        try:
            response = requests.post(url, data=data, files=files, timeout=30)
            response.raise_for_status()
            return response.json()
        except Exception as e:
            raise Exception(f"请求失败: {str(e)}")

class RuoKuaiPlatform(BaseCaptchaPlatform):
    """若快平台实现"""
    
    def get_base_url(self) -> str:
        return "http://api.ruokuai.com"
    
    def recognize(self, image_data: bytes, **kwargs) -> CaptchaResult:
        start_time = time.time()
        try:
            url = f"{self.base_url}/create.json"
            data = {
                'username': kwargs.get('username', ''),
                'password': kwargs.get('password', ''),
                'typeid': kwargs.get('typeid', 3040),  # 4位字母数字
                'softid': kwargs.get('softid', ''),
                'softkey': kwargs.get('softkey', '')
            }
            
            files = {'image': ('captcha.jpg', image_data)}
            result = self._make_request(url, data, files)
            
            return CaptchaResult(
                success=True,
                code=result.get('Result'),
                platform='ruokuai',
                cost_time=time.time() - start_time
            )
        except Exception as e:
            return CaptchaResult(
                success=False,
                platform='ruokuai',
                error_msg=str(e),
                cost_time=time.time() - start_time
            )

class ChaoJiYingPlatform(BaseCaptchaPlatform):
    """超级鹰平台实现"""
    
    def get_base_url(self) -> str:
        return "http://upload.chaojiying.net"
    
    def recognize(self, image_data: bytes, **kwargs) -> CaptchaResult:
        start_time = time.time()
        try:
            url = f"{self.base_url}/Upload/Processing.php"
            data = {
                'user': kwargs.get('username', ''),
                'pass': kwargs.get('password', ''),
                'softid': kwargs.get('softid', ''),
                'codetype': kwargs.get('codetype', 1004)  # 4位英文数字
            }
            
            files = {'userfile': ('captcha.jpg', image_data)}
            result = self._make_request(url, data, files)
            
            return CaptchaResult(
                success=True,
                code=result.get('pic_str'),
                platform='chaojiying',
                cost_time=time.time() - start_time
            )
        except Exception as e:
            return CaptchaResult(
                success=False,
                platform='chaojiying',
                error_msg=str(e),
                cost_time=time.time() - start_time
            )

class CaptchaService:
    """多平台验证码识别服务"""
    
    def __init__(self, platforms: List[BaseCaptchaPlatform]):
        self.platforms = platforms
        self.executor = ThreadPoolExecutor(max_workers=len(platforms))
    
    def recognize_parallel(self, image_data: bytes, **kwargs) -> CaptchaResult:
        """并行调用所有平台，返回最快成功结果"""
        futures = {
            self.executor.submit(platform.recognize, image_data, **kwargs): platform 
            for platform in self.platforms
        }
        
        successful_results = []
        for future in as_completed(futures):
            result = future.result()
            if result.success:
                successful_results.append(result)
                # 取消其他未完成的任务
                for f in futures:
                    if not f.done():
                        f.cancel()
                return successful_results[0]
        
        # 所有平台都失败时返回第一个错误
        for future in futures:
            if future.done():
                return future.result()
        
        return CaptchaResult(success=False, error_msg="所有平台识别失败")
    
    def recognize_fallback(self, image_data: bytes, **kwargs) -> CaptchaResult:
        """顺序调用平台，失败时自动切换"""
        for platform in self.platforms:
            result = platform.recognize(image_data, **kwargs)
            if result.success:
                return result
        return CaptchaResult(success=False, error_msg="所有平台识别失败")

# 使用示例
def main():
    # 初始化平台（实际使用时需要配置正确的API密钥）
    ruokuai = RuoKuaiPlatform(api_key="your_ruokuai_key")
    chaojiying = ChaoJiYingPlatform(api_key="your_chaojiying_key")
    
    # 创建多平台服务
    service = CaptchaService(platforms=[ruokuai, chaojiying])
    
    # 读取验证码图片
    with open("captcha.jpg", "rb") as f:
        image_data = f.read()
    
    # 并行识别（获取最快结果）
    result = service.recognize_parallel(
        image_data,
        username="your_username",
        password="your_password",
        typeid=3040
    )
    
    if result.success:
        print(f"识别成功: {result.code} (来自: {result.platform}, 耗时: {result.cost_time:.2f}s)")
    else:
        print(f"识别失败: {result.error_msg}")

if __name__ == "__main__":
    main()

这个实现的关键点：

1. 抽象基类定义统一接口，方便扩展新平台
2. 每个平台独立实现，互不干扰
3. 提供两种调用策略：并行竞速和顺序降级
4. 统一的结果格式，包含平台信息和耗时统计
5. 线程池管理并发请求

扩展新平台只需继承BaseCaptchaPlatform并实现recognize方法。

总结：用策略模式封装不同平台，结合线程池实现高效调度。

打码都是人工识别的，机器识别率达不到要求的，别想了。既然是人工的，那就简单了，平台无非就是一个下单一个接单。