import os from dataclasses import dataclass, field from pathlib import Path from dotenv import load_dotenv load_dotenv() TRUE_VALUES = {"1", "true", "yes", "on"} # Хелперы для чтения env-переменных с приведением типов def _env_str(name: str, default: str) -> str: value = os.getenv(name) return value if value is not None else default def _env_optional_str(name: str) -> str | None: value = os.getenv(name) if value is None: return None value = value.strip() return value or None def _env_path_str(name: str) -> str | None: value = _env_optional_str(name) if value is None: return None return str(Path(value).expanduser()) def _env_bool(name: str, default: bool) -> bool: fallback = "true" if default else "false" return _env_str(name, fallback).strip().lower() in TRUE_VALUES def _env_int(name: str, default: int) -> int: return int(_env_str(name, str(default)).strip()) def _env_float(name: str, default: float) -> float: return float(_env_str(name, str(default)).strip()) # Конфиг браузерного отпечатка (User-Agent, viewport, timezone) @dataclass(slots=True) class FingerprintConfig: user_agent: str = ( "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " "AppleWebKit/537.36 (KHTML, like Gecko) " "Chrome/135.0.0.0 Safari/537.36" ) viewport_presets: tuple[dict[str, int], ...] = ( {"width": 1920, "height": 1080}, {"width": 1600, "height": 900}, {"width": 1536, "height": 864}, {"width": 1440, "height": 900}, {"width": 1366, "height": 768}, ) timezone_candidates: tuple[str, ...] = ( "America/New_York", "America/Chicago", "America/Los_Angeles", ) locale: str = "en-US" sec_ch_ua: str = '"Google Chrome";v="135", "Chromium";v="135", "Not.A/Brand";v="24"' # --- Конфиг OpenLane (auth, paging, throttle, retry, checkpoint/output) --- @dataclass(slots=True) class OpenLaneConfig: sign_in_url: str = _env_str("OPENLANE_SIGN_IN_URL", "https://app.openlane.com/sign_in") search_url: str = _env_str("OPENLANE_SEARCH_URL", "https://app.openlane.com/search?tab=marketplace") api_search_url: str = _env_str("OPENLANE_API_SEARCH_URL", "https://app.openlane.com/api/v4/search") username: str = _env_str("OPENLANE_USERNAME", "") password: str = _env_str("OPENLANE_PASSWORD", "") source_tab: str = _env_str("OPENLANE_SOURCE_TAB", "marketplace") sale_types: str = _env_str("OPENLANE_SALE_TYPES", "marketplace") page_size: int = _env_int("OPENLANE_PAGE_SIZE", 30) max_pages: int = _env_int("OPENLANE_MAX_PAGES", 512) concurrency: int = _env_int("OPENLANE_CONCURRENCY", 4) throttle_min_seconds: float = _env_float("OPENLANE_THROTTLE_MIN_SECONDS", 0.3) throttle_max_seconds: float = _env_float("OPENLANE_THROTTLE_MAX_SECONDS", 0.8) retry_schedule_seconds: tuple[float, ...] = tuple( float(part.strip()) for part in _env_str("OPENLANE_RETRY_SCHEDULE_SECONDS", "2,5,10").split(",") if part.strip() ) or (2.0, 5.0, 10.0) checkpoint_every_pages: int = _env_int("OPENLANE_CHECKPOINT_EVERY_PAGES", 10) output_dir: str = _env_str("OPENLANE_OUTPUT_DIR", "artifacts/openlane") jsonl_output: str = _env_str("OPENLANE_JSONL_OUTPUT", "artifacts/openlane/openlane_search.jsonl") aggregated_output: str = _env_str("OPENLANE_AGGREGATED_OUTPUT", "artifacts/openlane/openlane_search_aggregated.json") checkpoint_file: str = _env_str("OPENLANE_CHECKPOINT_FILE", "artifacts/openlane/openlane_checkpoint.json") storage_state_file: str = _env_str("OPENLANE_STORAGE_STATE_FILE", "artifacts/openlane/openlane_storage_state.json") persist_storage_state: bool = _env_bool("OPENLANE_PERSIST_STORAGE_STATE", True) progress_log_every_pages: int = _env_int("OPENLANE_PROGRESS_LOG_EVERY_PAGES", 10) request_timeout_ms: int = _env_int("OPENLANE_REQUEST_TIMEOUT_MS", 45000) max_cars_limit: int = _env_int("OPENLANE_MAX_CARS_LIMIT", 20) refresh_token: str = _env_str("OPENLANE_REFRESH_TOKEN", "") okta_client_id: str = _env_str("OPENLANE_OKTA_CLIENT_ID", "") okta_token_endpoint: str = _env_str( "OPENLANE_OKTA_TOKEN_ENDPOINT", "https://okta.iam.karglobal.com/oauth2/default/v1/token", ) okta_redirect_uri: str = _env_str("OPENLANE_OKTA_REDIRECT_URI", "https://app.openlane.com/sign_in") okta_timeout_seconds: int = _env_int("OPENLANE_OKTA_TIMEOUT_SECONDS", 15) # --- Конфиг PostgreSQL (URL, пул соединений, pool_recycle) --- @dataclass(slots=True) class DatabaseConfig: url: str = _env_str("OPENLANE_DATABASE_URL", "postgresql+psycopg2://openlane:openlane@localhost:5432/openlane_scraper") echo: bool = _env_bool("OPENLANE_DATABASE_ECHO", False) pool_size: int = _env_int("OPENLANE_DATABASE_POOL_SIZE", 5) max_overflow: int = _env_int("OPENLANE_DATABASE_MAX_OVERFLOW", 10) pool_recycle_seconds: int = _env_int("OPENLANE_DATABASE_POOL_RECYCLE_SECONDS", 1800) auto_create_tables: bool = _env_bool("OPENLANE_DATABASE_AUTO_CREATE_TABLES", False) # --- Конфиг Redis (URL для Celery broker) --- @dataclass(slots=True) class RedisConfig: url: str = _env_str("OPENLANE_REDIS_URL", "redis://localhost:6379/0") socket_timeout_seconds: float = _env_float("OPENLANE_REDIS_SOCKET_TIMEOUT_SECONDS", 10.0) socket_connect_timeout_seconds: float = _env_float("OPENLANE_REDIS_SOCKET_CONNECT_TIMEOUT_SECONDS", 5.0) health_check_interval_seconds: int = _env_int("OPENLANE_REDIS_HEALTH_CHECK_INTERVAL_SECONDS", 30) # --- Конфиг Celery (лимиты задач, concurrency, beat-расписание) --- @dataclass(slots=True) class CeleryConfig: broker_url: str = _env_str("CELERY_BROKER_URL", "") result_backend: str = _env_str("CELERY_RESULT_BACKEND", "") task_soft_time_limit: int = _env_int("CELERY_TASK_SOFT_TIME_LIMIT", 3300) task_time_limit: int = _env_int("CELERY_TASK_TIME_LIMIT", 3600) worker_concurrency: int = _env_int("CELERY_WORKER_CONCURRENCY", 4) worker_max_tasks_per_child: int = _env_int("CELERY_WORKER_MAX_TASKS_PER_CHILD", 5) broker_visibility_timeout: int = _env_int("CELERY_BROKER_VISIBILITY_TIMEOUT", 7200) beat_sync_interval_minutes: int = _env_int("CELERY_BEAT_SYNC_INTERVAL_MINUTES", 60) beat_sync_limit: int | None = _env_int("CELERY_BEAT_SYNC_LIMIT", 0) or None # --- Конфиг прокси (server, username, password) --- @dataclass(slots=True) class ProxyConfig: server: str | None = _env_optional_str("OPENLANE_PROXY_SERVER") username: str | None = _env_optional_str("OPENLANE_PROXY_USERNAME") password: str | None = _env_optional_str("OPENLANE_PROXY_PASSWORD") @property def enabled(self) -> bool: return bool(self.server) def to_playwright_dict(self) -> dict[str, str] | None: if not self.server: return None result: dict[str, str] = {"server": self.server} if self.username: result["username"] = self.username if self.password: result["password"] = self.password return result # Главный объект настроек: собирает все блоки конфигурации @dataclass(slots=True) class Settings: default_timeout_ms: int = _env_int("OPENLANE_TIMEOUT_MS", 45000) headless: bool = _env_bool("OPENLANE_HEADLESS", True) browser_engine: str = _env_str("OPENLANE_BROWSER_ENGINE", "auto") log_level: str = _env_str("OPENLANE_LOG_LEVEL", "INFO") log_file: str | None = _env_optional_str("OPENLANE_LOG_FILE") enable_trace_id_logs: bool = _env_bool("OPENLANE_ENABLE_TRACE_ID_LOGS", True) runtime_config_file: str | None = _env_path_str("OPENLANE_RUNTIME_CONFIG_FILE") or "runtime_config.json" fingerprint: FingerprintConfig = field(default_factory=FingerprintConfig) openlane: OpenLaneConfig = field(default_factory=OpenLaneConfig) database: DatabaseConfig = field(default_factory=DatabaseConfig) redis: RedisConfig = field(default_factory=RedisConfig) celery: CeleryConfig = field(default_factory=CeleryConfig) proxy: ProxyConfig = field(default_factory=ProxyConfig) # Глобальный синглтон — используется по умолчанию во всех модулях. settings = Settings()