AutoScraperX/core/models/spiders_config_models.py

from pydantic import BaseModel, AnyUrl, validator
from typing import Dict, Any, Optional, Union


class BaseConfig(BaseModel):
    base_url: Optional[AnyUrl]
    request_timeout: int = 30
    max_retries: int = 3
    headers: Dict[str, Any] = {}

    @validator('request_timeout', 'max_retries')
    def validate_positive_int(cls, v, field):
        if v <= 0:
            raise ValueError(f'{field.name} must be positive')
        return v


class PlatformConfig(BaseConfig):
    platform: str
    mode: str
    path: Optional[str]
    url: AnyUrl
    method: str
    request_body: Dict[str, Any] = {}
    loop_times: int = 1
    loop_interval: Dict[str, int] = {}
    response_parse: Dict[str, Any] = {}
    retry_times: int = 0
    feishu_sheetid: Optional[str] = None

    @validator('method')
    def validate_method(cls, v):
        allowed_methods = ['GET', 'POST', 'PUT', 'DELETE', 'PATCH']
        if v.upper() not in allowed_methods:
            raise ValueError(f'Method must be one of {", ".join(allowed_methods)}')
        return v.upper()

    @validator('loop_times')
    def validate_loop_times(cls, v):
        if v <= 0:
            raise ValueError('loop_times must be positive')
        return v

    @validator('loop_interval')
    def validate_loop_interval(cls, v):
        if 'min' not in v or 'max' not in v:
            raise ValueError('loop_interval must contain both min and max keys')
        if v['min'] < 0 or v['max'] < 0:
            raise ValueError('loop_interval values must be non-negative')
        if v['min'] > v['max']:
            raise ValueError('min value cannot be greater than max value')
        return v

    @validator('response_parse')
    def validate_response_parse(cls, v):
        if 'data_path' not in v:
            raise ValueError('response_parse must contain data_path')
        return v

    @validator('retry_times')
    def validate_retry_times(cls, v):
        if v < 0:
            raise ValueError('retry_times must be non-negative')
        return v

    @validator('request_body')
    def validate_request_body(cls, v):
        # 确保request_body中的值是基本类型或字典/列表
        if not isinstance(v, dict):
            raise ValueError('request_body must be a dictionary')
        
        def is_valid_type(value):
            if isinstance(value, (str, int, float, bool, type(None))):
                return True
            elif isinstance(value, (list, tuple)):
                return all(is_valid_type(item) for item in value)
            elif isinstance(value, dict):
                return all(isinstance(k, str) and is_valid_type(v) for k, v in value.items())
            return False
            
        for key, value in v.items():
            if not is_valid_type(value):
                raise ValueError(f'Invalid type for request_body["{key}"]: {type(value)}')
        return v