mobile.de/mobilede_scraper/core/config.py

import json
import os
from dataclasses import dataclass, field
from pathlib import Path
from urllib.parse import quote, urlsplit, urlunsplit

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())


# Конфиг ссылок.

@dataclass(slots=True)
class ListingConfig:
    filtered_search_url: str | None = _env_optional_str("MOBILEDE_FILTERED_SEARCH_URL")
    filtered_search_urls_raw: str | None = _env_optional_str("MOBILEDE_FILTERED_SEARCH_URLS")

    @property
    def filtered_search_urls(self) -> list[str]:
        urls: list[str] = []
        if self.filtered_search_url and self.filtered_search_url.strip():
            urls.append(self.filtered_search_url.strip())
        if self.filtered_search_urls_raw and self.filtered_search_urls_raw.strip():
            raw = self.filtered_search_urls_raw.strip()
            try:
                parsed = json.loads(raw)
            except (json.JSONDecodeError, ValueError):
                parsed = None
            if isinstance(parsed, list):
                candidates = [str(item or "").strip() for item in parsed]
            else:
                candidates = [part.strip() for part in raw.replace("\n", ",").split(",")]
            for candidate in candidates:
                if candidate and candidate not in urls:
                    urls.append(candidate)
        return urls


# Конфиг PostgreSQL.

@dataclass(slots=True)
class DatabaseConfig:
    url: str = _env_str("MOBILEDE_DATABASE_URL", "postgresql+psycopg2://mobilede:MOBILEDE@localhost:5432/MOBILEDE_scraper")
    echo: bool = _env_bool("MOBILEDE_DATABASE_ECHO", False)
    pool_size: int = _env_int("MOBILEDE_DATABASE_POOL_SIZE", 5)
    max_overflow: int = _env_int("MOBILEDE_DATABASE_MAX_OVERFLOW", 10)
    pool_recycle_seconds: int = _env_int("MOBILEDE_DATABASE_POOL_RECYCLE_SECONDS", 1800)
    auto_create_tables: bool = _env_bool("MOBILEDE_DATABASE_AUTO_CREATE_TABLES", False)


# Конфиг Redis.

@dataclass(slots=True)
class RedisConfig:
    url: str = _env_str("MOBILEDE_REDIS_URL", "redis://localhost:6379/0")
    socket_timeout_seconds: float = _env_float("MOBILEDE_REDIS_SOCKET_TIMEOUT_SECONDS", 10.0)
    socket_connect_timeout_seconds: float = _env_float("MOBILEDE_REDIS_SOCKET_CONNECT_TIMEOUT_SECONDS", 5.0)
    health_check_interval_seconds: int = _env_int("MOBILEDE_REDIS_HEALTH_CHECK_INTERVAL_SECONDS", 30)


# Конфиг Celery.

@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)
    task_stall_timeout_seconds: int = _env_int("CELERY_TASK_STALL_TIMEOUT_SECONDS", 600)
    worker_concurrency: int = _env_int("CELERY_WORKER_CONCURRENCY", 4)
    worker_pool: str = _env_str("CELERY_WORKER_POOL", "prefork")
    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)

# Конфиг прокси.

@dataclass(slots=True)
class ProxyConfig:
    server: str | None = _env_optional_str("MOBILEDE_PROXY_SERVER")
    username: str | None = _env_optional_str("MOBILEDE_PROXY_USERNAME")
    password: str | None = _env_optional_str("MOBILEDE_PROXY_PASSWORD")

    @property
    def enabled(self) -> bool:
        return bool(self.server)

    def to_requests_proxy_url(self) -> str | None:
        if not self.server:
            return None
        if not self.username:
            return self.server

        parts = urlsplit(self.server)
        if not parts.scheme or not parts.hostname:
            return self.server

        username = quote(self.username, safe="")
        password = quote(self.password or "", safe="")
        host = parts.hostname
        if ":" in host and not host.startswith("["):
            host = f"[{host}]"
        if parts.port is not None:
            host = f"{host}:{parts.port}"
        netloc = f"{username}:{password}@{host}"
        return urlunsplit((parts.scheme, netloc, parts.path, parts.query, parts.fragment))

    def to_requests_proxies(self) -> dict[str, str] | None:
        proxy_url = self.to_requests_proxy_url()
        if not proxy_url:
            return None
        return {"http": proxy_url, "https": proxy_url}


# Общие настройки.

@dataclass(slots=True)
class Settings:
    headless: bool = _env_bool("MOBILEDE_HEADLESS", True)
    log_level: str = _env_str("MOBILEDE_LOG_LEVEL", "INFO")
    runtime_config_file: str | None = _env_path_str("MOBILEDE_RUNTIME_CONFIG_FILE")
    listing: ListingConfig = field(default_factory=ListingConfig)
    database: DatabaseConfig = field(default_factory=DatabaseConfig)
    redis: RedisConfig = field(default_factory=RedisConfig)
    celery: CeleryConfig = field(default_factory=CeleryConfig)

# Глобальные настройки.
settings = Settings()