""" 系统配置设置 包含系统级别的基础配置。 """ import os from dataclasses import dataclass from pathlib import Path from typing import Optional, Tuple @dataclass class BrowserConfig: """浏览器配置""" headless: bool = False timeout: int = 30000 # 30秒 viewport_width: int = 1920 viewport_height: int = 1080 user_agent: Optional[str] = None locale: str = "zh-CN" timezone: str = "Asia/Shanghai" proxy: Optional[str] = None executable_path: Optional[str] = None @dataclass class HumanBehaviorConfig: """人类行为配置""" typing_speed: Tuple[int, int] = (80, 150) # 打字速度范围（毫秒/字符） pause_probability: float = 0.15 # 随机暂停概率 random_delay_range: Tuple[float, float] = (0.3, 0.8) # 随机延迟范围（秒） click_delay_range: Tuple[float, float] = (0.1, 0.3) # 点击延迟范围 scroll_delay_range: Tuple[float, float] = (0.2, 0.5) # 滚动延迟范围 page_load_wait: Tuple[float, float] = (1.0, 3.0) # 页面加载等待时间 @dataclass class ConcurrencyConfig: """并发配置""" max_concurrent_tasks: int = 3 task_retry_count: int = 3 retry_backoff_factor: float = 2.0 task_timeout: int = 300 # 5分钟 @dataclass class FileConfig: """文件配置""" max_image_size: int = 10 * 1024 * 1024 # 10MB max_video_size: int = 500 * 1024 * 1024 # 500MB supported_image_formats: list = None supported_video_formats: list = None temp_dir: str = "temp" upload_timeout: int = 600 # 10分钟 def __post_init__(self): if self.supported_image_formats is None: self.supported_image_formats = ["jpg", "jpeg", "png", "webp"] if self.supported_video_formats is None: self.supported_video_formats = ["mp4", "mov", "avi", "mkv"] @dataclass class LogConfig: """日志配置""" level: str = "INFO" format: str = ( "{time:YYYY-MM-DD HH:mm:ss} | " "{level: <8} | " "{name}:{function}:{line} - " "{message}" ) log_dir: str = "logs" rotation: str = "1 day" retention: str = "30 days" compression: str = "zip" enable_colors: bool = True @dataclass class SecurityConfig: """安全配置""" encrypt_cookies: bool = True cookie_expiry_days: int = 30 max_login_attempts: int = 3 session_timeout: int = 7200 # 2小时 enable_rate_limit: bool = True rate_limit_per_minute: int = 10 class Settings: """系统主配置类""" def __init__(self): # 项目基础路径 self.BASE_DIR = Path(__file__).resolve().parent.parent # 环境变量 self.ENV = os.getenv("SMP_ENV", "development") self.DEBUG = os.getenv("SMP_DEBUG", "false").lower() == "true" # 初始化各个配置模块 self.browser = BrowserConfig() self.human_behavior = HumanBehaviorConfig() self.concurrency = ConcurrencyConfig() self.file = FileConfig() self.log = LogConfig() self.security = SecurityConfig() # 加载环境变量配置 self._load_env_config() # 根据环境调整配置 self._adjust_for_environment() def _load_env_config(self): """加载环境变量配置""" # 浏览器配置 if os.getenv("SMP_BROWSER_HEADLESS"): self.browser.headless = os.getenv("SMP_BROWSER_HEADLESS").lower() == "true" if os.getenv("SMP_BROWSER_TIMEOUT"): self.browser.timeout = int(os.getenv("SMP_BROWSER_TIMEOUT")) if os.getenv("SMP_BROWSER_USER_AGENT"): self.browser.user_agent = os.getenv("SMP_BROWSER_USER_AGENT") if os.getenv("SMP_BROWSER_PROXY"): self.browser.proxy = os.getenv("SMP_BROWSER_PROXY") # 并发配置 if os.getenv("SMP_MAX_CONCURRENT_TASKS"): self.concurrency.max_concurrent_tasks = int(os.getenv("SMP_MAX_CONCURRENT_TASKS")) if os.getenv("SMP_TASK_RETRY_COUNT"): self.concurrency.task_retry_count = int(os.getenv("SMP_TASK_RETRY_COUNT")) # 文件配置 if os.getenv("SMP_MAX_IMAGE_SIZE"): self.file.max_image_size = int(os.getenv("SMP_MAX_IMAGE_SIZE")) if os.getenv("SMP_MAX_VIDEO_SIZE"): self.file.max_video_size = int(os.getenv("SMP_MAX_VIDEO_SIZE")) # 日志配置 if os.getenv("SMP_LOG_LEVEL"): self.log.level = os.getenv("SMP_LOG_LEVEL") if os.getenv("SMP_LOG_DIR"): self.log.log_dir = os.getenv("SMP_LOG_DIR") def _adjust_for_environment(self): """根据环境调整配置""" if self.ENV == "production": # 生产环境配置 self.browser.headless = True self.log.level = "WARNING" self.concurrency.max_concurrent_tasks = 2 # 降低并发 self.human_behavior.pause_probability = 0.25 # 增加暂停频率 elif self.ENV == "testing": # 测试环境配置 self.browser.headless = True self.concurrency.max_concurrent_tasks = 1 self.concurrency.task_retry_count = 1 self.log.level = "DEBUG" @property def cookies_dir(self) -> Path: """Cookie目录路径""" return self.BASE_DIR / "auth" / "cookies" @property def logs_dir(self) -> Path: """日志目录路径""" return self.BASE_DIR / self.log.log_dir @property def temp_dir(self) -> Path: """临时目录路径""" return self.BASE_DIR / self.file.temp_dir @property def media_dir(self) -> Path: """媒体文件目录路径""" return self.BASE_DIR / "media" def ensure_directories(self): """确保所有必要的目录存在""" directories = [ self.cookies_dir, self.logs_dir, self.temp_dir, self.media_dir / "images", self.media_dir / "videos" ] for directory in directories: directory.mkdir(parents=True, exist_ok=True) def to_dict(self) -> dict: """转换为字典格式""" return { "ENV": self.ENV, "DEBUG": self.DEBUG, "BASE_DIR": str(self.BASE_DIR), "browser": { "headless": self.browser.headless, "timeout": self.browser.timeout, "viewport_width": self.browser.viewport_width, "viewport_height": self.browser.viewport_height, "user_agent": self.browser.user_agent, "proxy": self.browser.proxy }, "human_behavior": { "typing_speed": self.human_behavior.typing_speed, "pause_probability": self.human_behavior.pause_probability, "random_delay_range": self.human_behavior.random_delay_range }, "concurrency": { "max_concurrent_tasks": self.concurrency.max_concurrent_tasks, "task_retry_count": self.concurrency.task_retry_count, "retry_backoff_factor": self.concurrency.retry_backoff_factor }, "file": { "max_image_size": self.file.max_image_size, "max_video_size": self.file.max_video_size, "supported_image_formats": self.file.supported_image_formats, "supported_video_formats": self.file.supported_video_formats }, "log": { "level": self.log.level, "log_dir": self.log.log_dir, "rotation": self.log.rotation, "retention": self.log.retention } } # 全局配置实例 settings = Settings() # 确保目录存在 settings.ensure_directories()