add mobilede scraper
This commit is contained in:
qananasikq
395
mobilede_scraper/core/runtime_config.py
Normal file
395
mobilede_scraper/core/runtime_config.py
Normal file
@@ -0,0 +1,395 @@
|
||||
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")),
|
||||
)
|
||||
Reference in New Issue
Block a user