"""Crawler configuration Handles proxy settings, rate limiting, and retry configuration. """ import os from urllib.parse import urlparse from typing import Optional, Dict, Any from pydantic import BaseModel, Field class ProxyConfig(BaseModel): """Proxy configuration for crawler requests. Attributes: server: Proxy server address (e.g., http://127.0.0.1:7890) username: Optional proxy username for authentication password: Optional proxy password for authentication """ server: str = Field(..., description="Proxy server address") username: Optional[str] = Field(None, description="Proxy username") password: Optional[str] = Field(None, description="Proxy password") def to_playwright_format(self) -> Dict[str, Any]: """Convert to Playwright proxy format. Returns: Dictionary compatible with browser.new_context(proxy=...) """ config = {"server": self.server} if self.username: config["username"] = self.username if self.password: config["password"] = self.password return config class CrawlerConfig(BaseModel): """Crawler configuration. Handles proxy settings, rate limiting, and retry behavior. Attributes: proxy: Optional proxy configuration proxy_required: Whether proxy is mandatory (raises error if not configured) min_delay: Minimum delay between requests (seconds) max_delay: Maximum delay between requests (seconds) max_retries: Maximum number of retry attempts retry_min_wait: Minimum wait time between retries (seconds) retry_max_wait: Maximum wait time between retries (seconds) """ # Proxy settings proxy: Optional[ProxyConfig] = Field(None, description="Proxy configuration") proxy_required: bool = Field(default=False, description="Whether proxy is mandatory (False for 旁路由 auto-proxy)") # Rate limiting min_delay: float = Field(default=1.0, description="Minimum delay between requests (seconds)") max_delay: float = Field(default=3.0, description="Maximum delay between requests (seconds)") # Retry settings max_retries: int = Field(default=3, description="Maximum retry attempts") retry_min_wait: float = Field(default=2.0, description="Minimum retry wait (seconds)") retry_max_wait: float = Field(default=10.0, description="Maximum retry wait (seconds)") @classmethod def from_env(cls) -> "CrawlerConfig": """Load configuration from environment variables. Reads HTTP_PROXY or HTTPS_PROXY for proxy settings. Returns: CrawlerConfig instance with proxy settings from environment """ proxy = cls._parse_proxy_env() return cls(proxy=proxy) @staticmethod def _parse_proxy_env() -> Optional[ProxyConfig]: """Parse proxy configuration from environment variables. Checks HTTPS_PROXY first, then HTTP_PROXY. Returns: ProxyConfig if proxy URL found, None otherwise """ proxy_url = os.environ.get("HTTPS_PROXY") or os.environ.get("HTTP_PROXY") if not proxy_url: return None parsed = urlparse(proxy_url) server = f"{parsed.scheme}://{parsed.hostname}" if parsed.port: server += f":{parsed.port}" return ProxyConfig( server=server, username=parsed.username, password=parsed.password ) def validate_proxy(self) -> None: """Validate proxy configuration. Raises: ValueError: If proxy_required=True but no proxy is configured """ if self.proxy_required and not self.proxy: raise ValueError( "Proxy configuration missing! Please set HTTP_PROXY or HTTPS_PROXY environment variable.\n" "Example: export HTTPS_PROXY=http://127.0.0.1:7890" ) # Module-level cached config _crawler_config: Optional[CrawlerConfig] = None def get_crawler_config() -> CrawlerConfig: """Get crawler configuration singleton. Returns: CrawlerConfig instance loaded from environment variables """ global _crawler_config if _crawler_config is None: _crawler_config = CrawlerConfig.from_env() return _crawler_config def reset_crawler_config() -> None: """Reset the cached crawler configuration. Useful when environment variables change during runtime. """ global _crawler_config _crawler_config = None