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