302 lines
10 KiB
Python
302 lines
10 KiB
Python
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")),
|
|
)
|