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)