import json import logging from dataclasses import dataclass, field from pathlib import Path from typing import Any logger = logging.getLogger("MOBILEDE_scraper.runtime_config") def _normalize_text(value: str) -> str: return value.strip().casefold() def _text_tuple(values: Any) -> tuple[str, ...]: return tuple( str(item).strip() for item in (values or []) if str(item).strip() ) def _int_tuple(values: Any) -> tuple[int, ...]: result: list[int] = [] for value in values or []: try: result.append(int(value)) except (TypeError, ValueError): continue return tuple(result) def _optional_int(value: Any) -> int | None: if value in (None, ""): return None try: return int(value) except (TypeError, ValueError): return None def _optional_bool(value: Any) -> bool | None: if value is None: return None if isinstance(value, bool): return value if isinstance(value, str): normalized = value.strip().casefold() if normalized in {"1", "true", "yes", "on"}: return True if normalized in {"0", "false", "no", "off"}: return False return None @dataclass(slots=True) class RuntimeSyncConfig: name: str | None = None ids_initial_size: int | None = None ids_next_size: int | None = None ids_max_pages: int | None = None condition_check_enabled: bool | None = None lane: str | None = None only_new: bool | None = None limit: int | None = None @classmethod def from_dict(cls, data: dict[str, Any] | None) -> "RuntimeSyncConfig": data = data or {} name = str(data.get("name")).strip() if data.get("name") else None lane = str(data.get("lane")).strip() if data.get("lane") else None return cls( name=name or None, ids_initial_size=_optional_int(data.get("ids_initial_size")), ids_next_size=_optional_int(data.get("ids_next_size")), ids_max_pages=_optional_int(data.get("ids_max_pages")), condition_check_enabled=_optional_bool(data.get("condition_check_enabled")), lane=lane or None, only_new=_optional_bool(data.get("only_new")), limit=_optional_int(data.get("limit")), ) @dataclass(slots=True) class RuntimeListingConfig: make: str | None = None model: str | None = None @classmethod def from_dict(cls, data: dict[str, Any] | None) -> "RuntimeListingConfig": data = data or {} make = str(data.get("make")).strip() if data.get("make") else None model = str(data.get("model")).strip() if data.get("model") else None return cls(make=make or None, model=model or None) @dataclass(slots=True) class RuntimeFieldFilters: brands: tuple[str, ...] = () models: tuple[str, ...] = () years: tuple[int, ...] = () body_types: tuple[str, ...] = () colors: tuple[str, ...] = () drives: tuple[str, ...] = () gearboxes: tuple[str, ...] = () locations: tuple[str, ...] = () @classmethod def from_dict(cls, data: dict[str, Any] | None) -> "RuntimeFieldFilters": data = data or {} return cls( brands=_text_tuple(data.get("brands")), models=_text_tuple(data.get("models")), years=_int_tuple(data.get("years")), body_types=_text_tuple(data.get("body_types")), colors=_text_tuple(data.get("colors")), drives=_text_tuple(data.get("drives")), gearboxes=_text_tuple(data.get("gearboxes")), locations=_text_tuple(data.get("locations")), ) def is_empty(self) -> bool: return not any([ self.brands, self.models, self.years, self.body_types, self.colors, self.drives, self.gearboxes, self.locations, ]) def matches(self, values: dict[str, Any]) -> bool: return all([ self._match_text(self.brands, values.get("brand")), self._match_text(self.models, values.get("model")), self._match_int(self.years, values.get("year")), self._match_text(self.body_types, values.get("body_type")), self._match_text(self.colors, values.get("color")), self._match_text(self.drives, values.get("drive")), self._match_text(self.gearboxes, values.get("gearbox")), self._match_text(self.locations, values.get("location")), ]) @staticmethod def _match_text(allowed: tuple[str, ...], value: Any) -> bool: if not allowed: return True normalized = _normalize_text(str(value or "")) return normalized in {_normalize_text(item) for item in allowed} @staticmethod def _match_int(allowed: tuple[int, ...], value: Any) -> bool: if not allowed: return True parsed = _optional_int(value) return parsed in set(allowed) @dataclass(slots=True) class RuntimeRangeFilter: min: int | None = None max: int | None = None @classmethod def from_dict(cls, data: dict[str, Any] | None) -> "RuntimeRangeFilter": data = data or {} return cls(min=_optional_int(data.get("min")), max=_optional_int(data.get("max"))) def is_empty(self) -> bool: return self.min is None and self.max is None def matches(self, value: Any) -> bool: parsed = _optional_int(value) if parsed is None: return self.is_empty() if self.min is not None and parsed < self.min: return False if self.max is not None and parsed > self.max: return False return True @dataclass(slots=True) class RuntimeFlagFilters: damaged_only: bool | None = None run_and_drive: bool | None = None @classmethod def from_dict(cls, data: dict[str, Any] | None) -> "RuntimeFlagFilters": data = data or {} return cls( damaged_only=_optional_bool(data.get("damaged_only")), run_and_drive=_optional_bool(data.get("run_and_drive")), ) def is_empty(self) -> bool: return self.damaged_only is None and self.run_and_drive is None def matches(self, values: dict[str, Any]) -> bool: if self.damaged_only is not None and _optional_bool(values.get("is_damaged")) is not self.damaged_only: return False if self.run_and_drive is not None and _optional_bool(values.get("run_and_drive")) is not self.run_and_drive: return False return True @dataclass(slots=True) class RuntimeFiltersConfig: include: RuntimeFieldFilters = field(default_factory=RuntimeFieldFilters) exclude: RuntimeFieldFilters = field(default_factory=RuntimeFieldFilters) price: RuntimeRangeFilter = field(default_factory=RuntimeRangeFilter) mileage: RuntimeRangeFilter = field(default_factory=RuntimeRangeFilter) flags: RuntimeFlagFilters = field(default_factory=RuntimeFlagFilters) @classmethod def from_dict(cls, data: dict[str, Any] | None) -> "RuntimeFiltersConfig": data = data or {} legacy_fields = RuntimeFieldFilters.from_dict(data) include_payload = data.get("include") exclude_payload = data.get("exclude") flat_exclude_payload = { "brands": data.get("exclude_brands"), "models": data.get("exclude_models"), "years": data.get("exclude_years"), "body_types": data.get("exclude_body_types"), "colors": data.get("exclude_colors"), "drives": data.get("exclude_drives"), "gearboxes": data.get("exclude_gearboxes"), "locations": data.get("exclude_locations"), } include = RuntimeFieldFilters.from_dict(include_payload) if include_payload is not None else legacy_fields exclude = ( RuntimeFieldFilters.from_dict(exclude_payload) if exclude_payload is not None else RuntimeFieldFilters.from_dict(flat_exclude_payload) ) return cls( include=include, exclude=exclude, price=RuntimeRangeFilter.from_dict(data.get("price")), mileage=RuntimeRangeFilter.from_dict(data.get("mileage")), flags=RuntimeFlagFilters.from_dict(data.get("flags")), ) def is_empty(self) -> bool: return all([ self.include.is_empty(), self.exclude.is_empty(), self.price.is_empty(), self.mileage.is_empty(), self.flags.is_empty(), ]) def matches(self, values: dict[str, Any]) -> bool: if not self.include.matches(values): return False if not self._matches_exclude(values): return False if not self.price.matches(values.get("price")): return False if not self.mileage.matches(values.get("mileage")): return False if not self.flags.matches(values): return False return True def _matches_exclude(self, values: dict[str, Any]) -> bool: if self.exclude.is_empty(): return True return not self.exclude.matches(values) @dataclass(slots=True) class RuntimeMobileDeSegment: make: str | None = None make_id: str | None = None model: str | None = None model_id: str | None = None search_url: str | None = None only_new: bool | None = None price_min: str | None = None price_max: str | None = None year_min: str | None = None year_max: str | None = None start_page: int = 1 max_pages: int | None = None label: str | None = None @classmethod def from_dict(cls, data: dict[str, Any] | None) -> "RuntimeMobileDeSegment | None": data = data or {} make = str(data.get("make") or "").strip() or None make_id = str(data.get("make_id") or data.get("makeId") or "").strip() or None model = str(data.get("model") or "").strip() or None model_id = str(data.get("model_id") or data.get("modelId") or "").strip() or None search_url = str(data.get("search_url") or data.get("searchUrl") or data.get("listing_url") or "").strip() or None if not make_id and not make and not search_url: return None label = str(data.get("label") or "").strip() or None return cls( make=make, make_id=make_id, model=model, model_id=model_id, search_url=search_url, only_new=_optional_bool(data.get("only_new")), price_min=str(data.get("price_min") or "").strip() or None, price_max=str(data.get("price_max") or "").strip() or None, year_min=str(data.get("year_min") or "").strip() or None, year_max=str(data.get("year_max") or "").strip() or None, start_page=max(1, _optional_int(data.get("start_page")) or 1), max_pages=_optional_int(data.get("max_pages")), label=label, ) def to_task_kwargs(self) -> dict[str, Any]: return { "make": self.make, "make_id": self.make_id, "model": self.model, "model_id": self.model_id, "search_url": self.search_url, "only_new": self.only_new, "price_min": self.price_min, "price_max": self.price_max, "year_min": self.year_min, "year_max": self.year_max, "start_page": self.start_page, "max_pages": self.max_pages, "label": self.label or self.display_name, } @property def display_name(self) -> str: if self.label: return self.label if self.search_url: return "filtered-url" parts = [part for part in [self.make, self.model] if part] return " / ".join(parts) or self.make_id or "mobilede-segment" @dataclass(slots=True) class RuntimeMobileDeConfig: segments: tuple[RuntimeMobileDeSegment, ...] = () @classmethod def from_dict(cls, data: dict[str, Any] | None) -> "RuntimeMobileDeConfig": data = data or {} raw_segments = data.get("segments") segments: list[RuntimeMobileDeSegment] = [] if isinstance(raw_segments, list): for item in raw_segments: if not isinstance(item, dict): continue segment = RuntimeMobileDeSegment.from_dict(item) if segment is not None: segments.append(segment) return cls(segments=tuple(segments)) def is_empty(self) -> bool: return not self.segments @dataclass(slots=True) class RuntimeConfig: sync: RuntimeSyncConfig = field(default_factory=RuntimeSyncConfig) listing: RuntimeListingConfig = field(default_factory=RuntimeListingConfig) filters: RuntimeFiltersConfig = field(default_factory=RuntimeFiltersConfig) mobilede: RuntimeMobileDeConfig = field(default_factory=RuntimeMobileDeConfig) @classmethod def from_file(cls, config_path: str | None) -> "RuntimeConfig": if not config_path: return cls() path = Path(config_path) if not path.exists(): logger.info("Runtime config file not found: %s", path) return cls() try: payload = json.loads(path.read_text(encoding="utf-8")) except Exception as exc: logger.warning("Failed to read runtime config %s: %s", path, exc) return cls() return cls( sync=RuntimeSyncConfig.from_dict(payload.get("sync")), listing=RuntimeListingConfig.from_dict(payload.get("listing")), filters=RuntimeFiltersConfig.from_dict(payload.get("filters")), mobilede=RuntimeMobileDeConfig.from_dict(payload.get("mobilede")), )