fix car mapping
This commit is contained in:
qananasikq
301
dubizzle_scraper/core/runtime_config.py
Normal file
301
dubizzle_scraper/core/runtime_config.py
Normal file
@@ -0,0 +1,301 @@
|
||||
import json
|
||||
import logging
|
||||
from dataclasses import dataclass, field
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
|
||||
logger = logging.getLogger("dubizzle_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")),
|
||||
)
|
||||
Reference in New Issue
Block a user