AutoScraperX/core/models/spiders_config_models.py

# spiders_config_models.py 修改后
from pydantic import BaseModel, AnyUrl, field_validator, ConfigDict
from typing import Dict, Any, Optional, Union


class BaseConfig(BaseModel):
    model_config = ConfigDict(extra='forbid')  # 禁止额外字段

    base_url: Optional[AnyUrl] = None
    request_timeout: int = 30
    max_retries: int = 3
    headers: Dict[str, Any] = {}

    @field_validator('request_timeout', 'max_retries')
    @classmethod
    def validate_positive_int(cls, v: int) -> int:
        if v <= 0:
            raise ValueError('Value must be positive')
        return v


class PlatformConfig(BaseConfig):
    platform: str
    mode: str
    path: Optional[str] = None
    url: str
    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

    @field_validator('method')
    @classmethod
    def validate_method(cls, v: str) -> str:
        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()

    @field_validator('loop_times')
    @classmethod
    def validate_loop_times(cls, v: int) -> int:
        if v <= 0:
            raise ValueError('loop_times must be positive')
        return v

    @field_validator('loop_interval')
    @classmethod
    def validate_loop_interval(cls, v: Dict[str, int]) -> Dict[str, int]:
        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

    @field_validator('response_parse')
    @classmethod
    def validate_response_parse(cls, v: Dict[str, Any]) -> Dict[str, Any]:
        if 'data_path' not in v:
            raise ValueError('response_parse must contain data_path')
        return v

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

    @field_validator('request_body')
    @classmethod
    def validate_request_body(cls, v: Dict[str, Any]) -> Dict[str, Any]:
        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_val) for k, v_val 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