98 lines
3.8 KiB
Python
98 lines
3.8 KiB
Python
"""Загрузчик runtime_config.json для управления фильтрами синхронизации."""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
from dataclasses import dataclass, field
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
logger = logging.getLogger("encar_scraper.runtime_config")
|
|
|
|
DEFAULT_CONFIG_PATH = Path(__file__).resolve().parent.parent.parent / "runtime_config.json"
|
|
|
|
|
|
@dataclass
|
|
class SyncConfig:
|
|
name: str | None = None
|
|
lane: str = "encar_cars"
|
|
only_new: bool = False
|
|
limit: int | None = None
|
|
probe_all_photos: bool = False
|
|
|
|
|
|
@dataclass
|
|
class FiltersConfig:
|
|
price_min: int | None = None
|
|
price_max: int | None = None
|
|
mileage_min: int | None = None
|
|
mileage_max: int | None = None
|
|
brands: list[str] = field(default_factory=list)
|
|
models: list[str] = field(default_factory=list)
|
|
years: list[int] = field(default_factory=list)
|
|
body_types: list[str] = field(default_factory=list)
|
|
colors: list[str] = field(default_factory=list)
|
|
drives: list[str] = field(default_factory=list)
|
|
gearboxes: list[str] = field(default_factory=list)
|
|
exclude_brands: list[str] = field(default_factory=list)
|
|
exclude_models: list[str] = field(default_factory=list)
|
|
exclude_years: list[int] = field(default_factory=list)
|
|
exclude_body_types: list[str] = field(default_factory=list)
|
|
|
|
|
|
@dataclass
|
|
class RuntimeConfig:
|
|
sync: SyncConfig = field(default_factory=SyncConfig)
|
|
filters: FiltersConfig = field(default_factory=FiltersConfig)
|
|
|
|
|
|
def load_runtime_config(path: str | Path | None = None) -> RuntimeConfig:
|
|
"""Загружает runtime_config.json. Если файл не найден — возвращает дефолт."""
|
|
config_path = Path(path) if path else DEFAULT_CONFIG_PATH
|
|
if not config_path.exists():
|
|
logger.info("runtime_config.json not found at %s, using defaults", config_path)
|
|
return RuntimeConfig()
|
|
|
|
try:
|
|
raw: dict[str, Any] = json.loads(config_path.read_text(encoding="utf-8"))
|
|
except Exception as exc:
|
|
logger.warning("Failed to parse runtime_config.json: %s, using defaults", exc)
|
|
return RuntimeConfig()
|
|
|
|
sync_raw = raw.get("sync") or {}
|
|
filters_raw = raw.get("filters") or {}
|
|
price_raw = filters_raw.get("price") or {}
|
|
mileage_raw = filters_raw.get("mileage") or {}
|
|
|
|
sync = SyncConfig(
|
|
name=sync_raw.get("name"),
|
|
lane=sync_raw.get("lane", "encar_cars"),
|
|
only_new=bool(sync_raw.get("only_new", False)),
|
|
limit=sync_raw.get("limit"),
|
|
probe_all_photos=bool(sync_raw.get("probe_all_photos", False)),
|
|
)
|
|
|
|
filters = FiltersConfig(
|
|
price_min=price_raw.get("min"),
|
|
price_max=price_raw.get("max"),
|
|
mileage_min=mileage_raw.get("min"),
|
|
mileage_max=mileage_raw.get("max"),
|
|
brands=[b.lower() for b in (filters_raw.get("brands") or [])],
|
|
models=[m.lower() for m in (filters_raw.get("models") or [])],
|
|
years=filters_raw.get("years") or [],
|
|
body_types=[b.lower() for b in (filters_raw.get("body_types") or [])],
|
|
colors=[c.lower() for c in (filters_raw.get("colors") or [])],
|
|
drives=[d.lower() for d in (filters_raw.get("drives") or [])],
|
|
gearboxes=[g.lower() for g in (filters_raw.get("gearboxes") or [])],
|
|
exclude_brands=[b.lower() for b in (filters_raw.get("exclude_brands") or [])],
|
|
exclude_models=[m.lower() for m in (filters_raw.get("exclude_models") or [])],
|
|
exclude_years=filters_raw.get("exclude_years") or [],
|
|
exclude_body_types=[b.lower() for b in (filters_raw.get("exclude_body_types") or [])],
|
|
)
|
|
|
|
logger.info(
|
|
"runtime_config loaded: lane=%s, limit=%s, brands=%d, exclude_brands=%d",
|
|
sync.lane, sync.limit, len(filters.brands), len(filters.exclude_brands),
|
|
)
|
|
return RuntimeConfig(sync=sync, filters=filters)
|