import json import logging from dataclasses import dataclass, field from pathlib import Path from typing import Any logger = logging.getLogger("iaai_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 RuntimeConfig: sync: RuntimeSyncConfig = field(default_factory=RuntimeSyncConfig) listing: RuntimeListingConfig = field(default_factory=RuntimeListingConfig) filters: RuntimeFiltersConfig = field(default_factory=RuntimeFiltersConfig) @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")), )