369 lines
16 KiB
Python
369 lines
16 KiB
Python
from __future__ import annotations
|
|
|
|
import re
|
|
from datetime import datetime, timezone
|
|
from typing import Any
|
|
from urllib.parse import urlparse
|
|
|
|
from ..browser.fast_client import FastListingVehicle, build_resizer_images_from_keys, parse_int, parse_text
|
|
from ..storage.enums import BODY_TYPE_ENUM_VALUES, DRIVE_ENUM_VALUES, GEARBOX_ENUM_VALUES
|
|
from ..storage.schemas import CarRecord, ImageRecord
|
|
|
|
ORIGIN_PREFIX = "iaai:"
|
|
ORIGIN_URL_BASE = "https://www.iaai.com/VehicleDetail"
|
|
DAMAGE_NEUTRAL_VALUES = {
|
|
"NORMAL WEAR & TEAR",
|
|
"NORMAL WEAR",
|
|
"NORMALWEAR&TEAR",
|
|
"NONE",
|
|
"NO DAMAGE",
|
|
"NO VISIBLE DAMAGE",
|
|
"MINOR DENT/SCRATCHES",
|
|
}
|
|
INACTIVE_STATUS_VALUES = {"SOLD", "SO", "CLOSED", "CN", "DELIVERED", "WITHDRAWN", "WDR", "COMPLETE", "COMPLETED"}
|
|
|
|
|
|
class FastCarMapper:
|
|
"""Maps IAAI ProductDetailsVM payloads directly to project CarRecord."""
|
|
|
|
def map_payload_to_record(
|
|
self,
|
|
*,
|
|
detail_payload: dict[str, Any],
|
|
vehicle_url: str | None = None,
|
|
listing_vehicle: FastListingVehicle | None = None,
|
|
) -> CarRecord:
|
|
inventory_view = detail_payload.get("inventoryView")
|
|
if not isinstance(inventory_view, dict):
|
|
raise RuntimeError("detail payload missing inventoryView")
|
|
attributes = inventory_view.get("attributes")
|
|
if not isinstance(attributes, dict):
|
|
raise RuntimeError("detail payload missing inventoryView.attributes")
|
|
|
|
inventory_id = parse_text(attributes.get("Id"))
|
|
if not inventory_id and listing_vehicle is not None:
|
|
inventory_id = listing_vehicle.inventory_id
|
|
if not inventory_id and vehicle_url:
|
|
inventory_id = self._inventory_id_from_url(vehicle_url)
|
|
if not inventory_id:
|
|
raise RuntimeError("missing inventory id")
|
|
|
|
brand = self._limit_text(parse_text(attributes.get("Make")) or "UNKNOWN", 50)
|
|
model = self._build_model_name(parse_text(attributes.get("Model")), parse_text(attributes.get("Series"))) or "UNKNOWN"
|
|
model = self._limit_text(model, 50)
|
|
year = self._parse_year(attributes.get("Year"))
|
|
|
|
auction_info = detail_payload.get("auctionInformation")
|
|
auction_info = auction_info if isinstance(auction_info, dict) else {}
|
|
bidding_info = auction_info.get("biddingInformation")
|
|
bidding_info = bidding_info if isinstance(bidding_info, dict) else {}
|
|
prebid_info = auction_info.get("prebidInformation")
|
|
prebid_info = prebid_info if isinstance(prebid_info, dict) else {}
|
|
|
|
high_bid = self._first_positive_int(
|
|
prebid_info.get("decimalHighBidAmount"),
|
|
prebid_info.get("highBidAmount"),
|
|
bidding_info.get("highBidAmount"),
|
|
)
|
|
buy_now = self._first_positive_int(
|
|
bidding_info.get("buyNowAmount"),
|
|
prebid_info.get("buyNowPrice"),
|
|
bidding_info.get("buyNowPrice"),
|
|
)
|
|
price = high_bid if high_bid is not None else buy_now
|
|
|
|
image_dimensions = inventory_view.get("imageDimensions")
|
|
image_dimensions = image_dimensions if isinstance(image_dimensions, dict) else {}
|
|
keys_container = image_dimensions.get("keys")
|
|
keys_container = keys_container if isinstance(keys_container, dict) else {}
|
|
image_keys = keys_container.get("$values")
|
|
image_keys = image_keys if isinstance(image_keys, list) else []
|
|
images = [ImageRecord.model_validate(row) for row in build_resizer_images_from_keys(image_keys)]
|
|
|
|
primary_damage = parse_text(attributes.get("PrimaryDamageDesc"))
|
|
secondary_damage = parse_text(attributes.get("SecondaryDamageDesc"))
|
|
origin_url = vehicle_url or f"{ORIGIN_URL_BASE}/{inventory_id}"
|
|
normalized_origin_id = self._normalize_origin_inventory_id(inventory_id)
|
|
origin_id = f"{ORIGIN_PREFIX}{normalized_origin_id}"
|
|
|
|
return CarRecord(
|
|
parser_id=self._generate_parser_id(origin_id),
|
|
brand=brand,
|
|
model=model,
|
|
year=year,
|
|
price=price,
|
|
currency=self._map_currency(parse_text(attributes.get("Currency")) or (listing_vehicle.currency if listing_vehicle else None)),
|
|
mileage=self._parse_non_negative_int(attributes.get("ODOValue")) or 0,
|
|
country=self._map_country(inventory_id),
|
|
is_sold=self._is_sold(listing_vehicle),
|
|
color=self._normalize_color(parse_text(attributes.get("ExteriorColor")) or parse_text(attributes.get("ColorDesc"))),
|
|
drive=self._map_drive(parse_text(attributes.get("DriveLineTypeDesc"))),
|
|
gearbox=self._map_gearbox(parse_text(attributes.get("Transmission"))),
|
|
steering_wheel="LEFT",
|
|
body_type=self._map_body_type(parse_text(attributes.get("BodyStyleName")) or parse_text(attributes.get("VehicleClass"))),
|
|
engine_volume=self._parse_engine_volume(parse_text(attributes.get("EngineSize")) or parse_text(attributes.get("EngineInformation"))),
|
|
selling_type="AUCTION",
|
|
one_owner=False,
|
|
new_car=False,
|
|
is_hidden=not bool(images),
|
|
origin="IAAI",
|
|
origin_url=origin_url,
|
|
origin_id=origin_id,
|
|
is_damaged=self._derive_is_damaged(primary_damage=primary_damage, secondary_damage=secondary_damage),
|
|
evaluation=parse_text(attributes.get("VehicleGrade")),
|
|
non_smoking=True,
|
|
rental=False,
|
|
repair_history=False,
|
|
slug=self._slugify(" ".join(filter(None, [brand, model, str(year or "")]))),
|
|
last_seen_at=datetime.now(timezone.utc),
|
|
images=images,
|
|
)
|
|
|
|
def payload_to_summary(self, detail_payload: dict[str, Any], vehicle_url: str) -> dict[str, Any]:
|
|
inventory_view = detail_payload.get("inventoryView") if isinstance(detail_payload, dict) else {}
|
|
inventory_view = inventory_view if isinstance(inventory_view, dict) else {}
|
|
attr = inventory_view.get("attributes")
|
|
attr = attr if isinstance(attr, dict) else {}
|
|
auction_info = detail_payload.get("auctionInformation") if isinstance(detail_payload, dict) else {}
|
|
auction_info = auction_info if isinstance(auction_info, dict) else {}
|
|
bidding_info = auction_info.get("biddingInformation")
|
|
bidding_info = bidding_info if isinstance(bidding_info, dict) else {}
|
|
prebid_info = auction_info.get("prebidInformation")
|
|
prebid_info = prebid_info if isinstance(prebid_info, dict) else {}
|
|
image_dimensions = inventory_view.get("imageDimensions")
|
|
image_dimensions = image_dimensions if isinstance(image_dimensions, dict) else {}
|
|
keys_container = image_dimensions.get("keys")
|
|
keys_container = keys_container if isinstance(keys_container, dict) else {}
|
|
image_keys = keys_container.get("$values")
|
|
image_keys = image_keys if isinstance(image_keys, list) else []
|
|
image_urls = [str(row["fullres_image"]) for row in build_resizer_images_from_keys(image_keys)]
|
|
return {
|
|
"source_url": vehicle_url,
|
|
"lot_number": attr.get("Id") or attr.get("StockNumber") or attr.get("SalvageId"),
|
|
"year": attr.get("Year"),
|
|
"make": attr.get("Make"),
|
|
"model": attr.get("Model"),
|
|
"trim": attr.get("Series"),
|
|
"body_type": attr.get("BodyStyleName"),
|
|
"drive": attr.get("DriveLineTypeDesc"),
|
|
"engine": attr.get("EngineInformation") or attr.get("EngineSize"),
|
|
"fuel_type": attr.get("FuelTypeCode"),
|
|
"gearbox": attr.get("Transmission"),
|
|
"color": attr.get("ExteriorColor"),
|
|
"primary_damage": attr.get("PrimaryDamageDesc"),
|
|
"secondary_damage": attr.get("SecondaryDamageDesc"),
|
|
"odometer": attr.get("ODOValue"),
|
|
"location": attr.get("BranchName"),
|
|
"auction_date": attr.get("AuctionDateTime"),
|
|
"title": attr.get("Title"),
|
|
"current_bid": prebid_info.get("highBidAmount") or bidding_info.get("highBidAmount"),
|
|
"buy_now": prebid_info.get("buyNowPrice") or bidding_info.get("buyNowPrice"),
|
|
"actual_cash_value": attr.get("ProviderACV"),
|
|
"estimated_repair_cost": attr.get("EstRepairCost"),
|
|
"image_urls": image_urls,
|
|
}
|
|
|
|
@staticmethod
|
|
def _inventory_id_from_url(vehicle_url: str) -> str | None:
|
|
tail = urlparse(vehicle_url).path.rstrip("/").split("/")[-1]
|
|
return tail or None
|
|
|
|
@staticmethod
|
|
def _normalize_origin_inventory_id(inventory_id: str) -> str:
|
|
return inventory_id.strip().split("~", 1)[0]
|
|
|
|
@staticmethod
|
|
def _build_model_name(model: str | None, series: str | None) -> str:
|
|
unique_parts: list[str] = []
|
|
seen: set[str] = set()
|
|
for value in (model, series):
|
|
if not value:
|
|
continue
|
|
normalized = " ".join(value.split())
|
|
key = normalized.casefold()
|
|
if key in seen:
|
|
continue
|
|
seen.add(key)
|
|
unique_parts.append(normalized)
|
|
return " ".join(unique_parts).strip()
|
|
|
|
@staticmethod
|
|
def _parse_year(value: Any) -> int | None:
|
|
parsed = parse_int(value)
|
|
if parsed is None or parsed < 1900 or parsed > 2100:
|
|
return None
|
|
return parsed
|
|
|
|
@staticmethod
|
|
def _parse_non_negative_int(value: Any) -> int | None:
|
|
parsed = parse_int(value)
|
|
if parsed is None or parsed < 0:
|
|
return None
|
|
return parsed
|
|
|
|
@classmethod
|
|
def _first_positive_int(cls, *values: Any) -> int | None:
|
|
for value in values:
|
|
parsed = cls._parse_non_negative_int(value)
|
|
if parsed is not None and parsed > 0:
|
|
return parsed
|
|
return None
|
|
|
|
@staticmethod
|
|
def _normalize_color(value: str | None) -> str:
|
|
if not value:
|
|
return "other"
|
|
normalized = value.strip().lower()
|
|
if "/" in normalized:
|
|
normalized = normalized.split("/", 1)[0].strip()
|
|
return normalized or "other"
|
|
|
|
@staticmethod
|
|
def _map_currency(value: str | None) -> str:
|
|
normalized = (value or "USD").strip().upper()
|
|
return normalized if normalized in {"USD", "CAD", "EUR", "JPY", "RUB", "KRW", "AED", "GBP"} else "USD"
|
|
|
|
@staticmethod
|
|
def _map_country(inventory_id: str) -> str:
|
|
upper_id = inventory_id.strip().upper()
|
|
if upper_id.endswith("~CA"):
|
|
return "CA"
|
|
if upper_id.endswith("~US"):
|
|
return "US"
|
|
return "NA"
|
|
|
|
@staticmethod
|
|
def _map_drive(value: str | None) -> str | None:
|
|
if not value:
|
|
return None
|
|
normalized = value.strip().lower()
|
|
candidates: tuple[str, ...] | None = None
|
|
if "front" in normalized or normalized == "fwd":
|
|
candidates = ("FWD",)
|
|
elif "all wheel" in normalized or "4x4" in normalized or normalized == "awd" or "four wheel" in normalized:
|
|
candidates = ("4WD", "FOUR_WD")
|
|
elif "rear" in normalized or normalized == "rwd":
|
|
candidates = ("RWD",)
|
|
elif "2wd" in normalized or "two wheel" in normalized:
|
|
candidates = ("2WD", "TWO_WD")
|
|
elif "unknown" in normalized or normalized in {"na", "n/a"}:
|
|
candidates = ("NA",)
|
|
return FastCarMapper._select_allowed(candidates, DRIVE_ENUM_VALUES) if candidates else None
|
|
|
|
@staticmethod
|
|
def _map_gearbox(value: str | None) -> str | None:
|
|
if not value:
|
|
return None
|
|
normalized = value.strip().lower()
|
|
candidates: tuple[str, ...] | None = None
|
|
if "cvt" in normalized:
|
|
candidates = ("CVT",)
|
|
elif "manual" in normalized or normalized == "mt":
|
|
candidates = ("MT",)
|
|
elif "electric" in normalized or normalized == "ev":
|
|
candidates = ("EV",)
|
|
elif "auto" in normalized or normalized == "at":
|
|
candidates = ("AT",)
|
|
elif "unknown" in normalized or normalized in {"na", "n/a"}:
|
|
candidates = ("NA",)
|
|
return FastCarMapper._select_allowed(candidates, GEARBOX_ENUM_VALUES) if candidates else None
|
|
|
|
@staticmethod
|
|
def _map_body_type(value: str | None) -> str:
|
|
if not value:
|
|
return "OTHER"
|
|
normalized = value.strip().lower()
|
|
candidates: tuple[str, ...] | None = None
|
|
if "sedan" in normalized:
|
|
candidates = ("SEDAN",)
|
|
elif "sport utility" in normalized or normalized == "suv" or "crossover" in normalized:
|
|
candidates = ("SUV",)
|
|
elif "hatch" in normalized:
|
|
candidates = ("HATCHBACK",)
|
|
elif "wagon" in normalized:
|
|
candidates = ("STATION_WAGON", "Station Wagon")
|
|
elif "coupe" in normalized:
|
|
candidates = ("COUPE",)
|
|
elif "pickup" in normalized or ("crew" in normalized and "cab" in normalized):
|
|
candidates = ("PICKUP", "Pickup")
|
|
elif "convertible" in normalized or "roadster" in normalized or "cabrio" in normalized:
|
|
candidates = ("OPEN", "Open")
|
|
elif "van" in normalized:
|
|
candidates = ("MINIVAN",)
|
|
elif "truck" in normalized or "chassis" in normalized:
|
|
candidates = ("TRUCK", "Truck")
|
|
elif "rv" in normalized or "motorized" in normalized:
|
|
candidates = ("RV",)
|
|
elif normalized in {"other", "unknown"}:
|
|
candidates = ("OTHER", "Other")
|
|
return FastCarMapper._select_allowed(candidates, BODY_TYPE_ENUM_VALUES, fallback="OTHER") or "OTHER"
|
|
|
|
@staticmethod
|
|
def _parse_engine_volume(value: str | None) -> int | None:
|
|
if not value:
|
|
return None
|
|
match = re.search(r"(\d+(?:\.\d+)?)\s*[lL]\b", value)
|
|
if not match:
|
|
return None
|
|
try:
|
|
liters = float(match.group(1))
|
|
except ValueError:
|
|
return None
|
|
cc = int(round(liters * 1000))
|
|
if cc <= 0 or cc > 10000:
|
|
return None
|
|
return cc
|
|
|
|
@staticmethod
|
|
def _derive_is_damaged(*, primary_damage: str | None, secondary_damage: str | None) -> bool:
|
|
neutral = {item.replace(" ", "").strip().upper() for item in DAMAGE_NEUTRAL_VALUES}
|
|
for value in (primary_damage, secondary_damage):
|
|
if not value:
|
|
continue
|
|
normalized = value.replace(" ", "").strip().upper()
|
|
if normalized and normalized not in neutral:
|
|
return True
|
|
return False
|
|
|
|
@staticmethod
|
|
def _is_sold(listing_vehicle: FastListingVehicle | None) -> bool:
|
|
if listing_vehicle is None:
|
|
return False
|
|
if listing_vehicle.timed_auction_closed:
|
|
return True
|
|
status = (listing_vehicle.inventory_status or "").strip().upper()
|
|
return status in INACTIVE_STATUS_VALUES
|
|
|
|
@staticmethod
|
|
def _select_allowed(candidates: tuple[str, ...] | None, allowed: tuple[str, ...], fallback: str | None = None) -> str | None:
|
|
if not candidates:
|
|
return fallback if fallback in allowed else None
|
|
allowed_set = set(allowed)
|
|
for candidate in candidates:
|
|
if candidate in allowed_set:
|
|
return candidate
|
|
return fallback if fallback in allowed_set else None
|
|
|
|
@staticmethod
|
|
def _limit_text(value: str, max_length: int) -> str:
|
|
return value if len(value) <= max_length else value[:max_length].rstrip()
|
|
|
|
@staticmethod
|
|
def _slugify(value: str) -> str:
|
|
return re.sub(r"[^a-z0-9]+", "-", value.lower()).strip("-") or "car"
|
|
|
|
@staticmethod
|
|
def _generate_parser_id(origin_id: str) -> str:
|
|
import hashlib
|
|
from string import ascii_letters, digits
|
|
|
|
digest = hashlib.sha256(origin_id.encode()).digest()
|
|
alphabet = ascii_letters + digits
|
|
base = len(alphabet)
|
|
num = int.from_bytes(digest[:17], "big")
|
|
chars: list[str] = []
|
|
for _ in range(22):
|
|
num, idx = divmod(num, base)
|
|
chars.append(alphabet[idx])
|
|
return "car-" + "".join(chars)
|