302 lines
15 KiB
Python
302 lines
15 KiB
Python
import hashlib
|
||
import json
|
||
import re
|
||
from datetime import datetime, timezone
|
||
from typing import Any
|
||
from urllib.parse import urlparse
|
||
|
||
from ..core.utils import first_non_empty
|
||
from ..storage.enums import (
|
||
BODY_TYPE_ENUM_VALUES,
|
||
COUNTRY_ENUM_VALUES,
|
||
CURRENCY_ENUM_VALUES,
|
||
DRIVE_ENUM_VALUES,
|
||
GEARBOX_ENUM_VALUES,
|
||
ORIGIN_ENUM_VALUES,
|
||
SELLING_TYPE_ENUM_VALUES,
|
||
STEERING_WHEEL_ENUM_VALUES,
|
||
)
|
||
from ..storage.schemas import CarRecord, ImageRecord
|
||
|
||
|
||
class CarMapper:
|
||
"""IAAI data → CarRecord."""
|
||
|
||
BODY_MAP = {
|
||
"sedan": "SEDAN", "coupe": "COUPE", "hatchback": "HATCHBACK", "sport utility": "SUV",
|
||
"suv": "SUV", "wagon": "STATION_WAGON", "station wagon": "STATION_WAGON", "pickup": "PICKUP",
|
||
"pickup truck": "PICKUP", "crew cab": "PICKUP", "extended cab": "PICKUP", "regular cab": "PICKUP",
|
||
"quad cab": "PICKUP", "double cab": "PICKUP", "king cab": "PICKUP", "mega cab": "PICKUP",
|
||
"supercab": "PICKUP", "supercrew": "PICKUP", "truck": "TRUCK", "van": "MINIVAN",
|
||
"minivan": "MINIVAN", "convertible": "OPEN", "cabriolet": "OPEN", "rv": "RV", "crossover": "SUV",
|
||
}
|
||
DRIVE_MAP = {
|
||
"front wheel drive": "FWD", "fwd": "FWD", "rear wheel drive": "RWD", "rwd": "RWD",
|
||
"all wheel drive": "4WD", "awd": "4WD", "4x4": "4WD", "four wheel drive": "4WD",
|
||
"4wd": "4WD", "2wd": "2WD", "two wheel drive": "2WD",
|
||
}
|
||
GEARBOX_MAP = {
|
||
"automatic": "AT", "automatic transmission": "AT", "a/t": "AT", "aut": "AT",
|
||
"manual": "MT", "manual transmission": "MT", "m/t": "MT", "cvt": "CVT",
|
||
"continuously variable transmission": "CVT", "electric": "EV", "ev": "EV",
|
||
}
|
||
STEERING_MAP = {"left": "LEFT", "left hand drive": "LEFT", "right": "RIGHT", "right hand drive": "RIGHT"}
|
||
COUNTRY_MAP = {
|
||
"us": "US", "usa": "US", "united states": "US", "ca": "CA", "canada": "CA",
|
||
"jp": "JP", "japan": "JP", "kr": "KR", "korea": "KR", "south korea": "KR",
|
||
}
|
||
NO_DAMAGE_MARKERS = {"normal wear", "normal wear & tear", "normal wear and tear", "n/a", "na", "none", "no damage", "minor dents/scratches"}
|
||
|
||
def map_to_car_record(self, vehicle_url: str, vehicle_summary: dict[str, Any], payload_insights: dict[str, Any]) -> CarRecord:
|
||
# Собираем нормализованную DB-модель из summary и payload insights.
|
||
vehicle_summary = vehicle_summary or {}
|
||
payload_insights = payload_insights or {}
|
||
notes: list[str] = []
|
||
core = payload_insights.get("vehicle_core", {})
|
||
pricing = payload_insights.get("pricing", {})
|
||
damage = payload_insights.get("damage", {})
|
||
auction = payload_insights.get("auction", {})
|
||
images = payload_insights.get("images", {})
|
||
|
||
origin_id = self._build_origin_id(vehicle_url, vehicle_summary, core)
|
||
parser_id = f"iaai:{origin_id}"
|
||
brand = self._as_str(first_non_empty([core.get("make"), vehicle_summary.get("make")])) or "UNKNOWN"
|
||
model = self._as_str(first_non_empty([core.get("model"), vehicle_summary.get("model")])) or "UNKNOWN"
|
||
year = self._to_int(first_non_empty([core.get("year"), vehicle_summary.get("year")]))
|
||
price = self._to_int(first_non_empty([pricing.get("buy_now"), pricing.get("current_bid"), vehicle_summary.get("buy_now"), vehicle_summary.get("current_bid")]))
|
||
mileage = self._to_int(first_non_empty([core.get("odometer"), vehicle_summary.get("odometer"), 0])) or 0
|
||
color = self._normalize_color(first_non_empty([core.get("color"), vehicle_summary.get("color"), "other"]))
|
||
drive = self._normalize_drive(first_non_empty([core.get("drive"), vehicle_summary.get("drive")]))
|
||
gearbox = self._normalize_gearbox(first_non_empty([core.get("gearbox"), vehicle_summary.get("gearbox")]))
|
||
steering = self._normalize_steering(first_non_empty([core.get("steering_wheel"), vehicle_summary.get("steering_wheel")])) or "LEFT"
|
||
body_type = self._normalize_body_type(first_non_empty([core.get("body_type"), vehicle_summary.get("body_type")]))
|
||
engine_volume = self._to_engine_cc(first_non_empty([core.get("engine"), core.get("engine_volume"), vehicle_summary.get("engine"), vehicle_summary.get("engine_volume")]))
|
||
title_text = self._as_str(first_non_empty([core.get("title"), vehicle_summary.get("title"), ""]))
|
||
seller = self._as_str(first_non_empty([core.get("seller"), vehicle_summary.get("seller"), ""]))
|
||
location = self._as_str(first_non_empty([core.get("location"), vehicle_summary.get("location"), auction.get("branch"), ""]))
|
||
country = self._normalize_country(first_non_empty([core.get("country"), location, "US"]))
|
||
is_damaged = self._bool_damage(damage, vehicle_summary)
|
||
is_sold = self._bool_sold(auction)
|
||
one_owner = self._boolish(first_non_empty([core.get("one_owner"), vehicle_summary.get("one_owner"), False]))
|
||
new_car = self._boolish(first_non_empty([core.get("new_car"), vehicle_summary.get("new_car"), False]))
|
||
rental = self._boolish(first_non_empty([core.get("rental"), vehicle_summary.get("rental"), False]))
|
||
repair_history = self._boolish(first_non_empty([core.get("repair_history"), vehicle_summary.get("repair_history"), False]))
|
||
non_smoking = self._boolish(first_non_empty([core.get("non_smoking"), vehicle_summary.get("non_smoking"), True]))
|
||
evaluation = self._as_str(first_non_empty([core.get("grade"), core.get("evaluation"), vehicle_summary.get("evaluation")])) or None
|
||
currency = self._normalize_currency(first_non_empty([pricing.get("currency"), vehicle_summary.get("currency"), "USD"]))
|
||
slug = self._slugify(" ".join(filter(None, [str(year or ""), brand, model, origin_id])))
|
||
images_records = self._build_images(images.get("urls") or vehicle_summary.get("image_urls") or [])
|
||
origin = "IAAI" if "IAAI" in ORIGIN_ENUM_VALUES else "NA"
|
||
if not price:
|
||
notes.append("Price is missing or not parseable from the observed payloads.")
|
||
if not vehicle_summary.get("vin"):
|
||
notes.append("VIN was not observed in the accessible payloads for this account/session.")
|
||
if not images_records:
|
||
notes.append("No image URLs were found in the captured payloads.")
|
||
|
||
raw_attributes = {
|
||
# Здесь сохраняем полезный сырой контекст без жёсткой нормализации.
|
||
"vin": vehicle_summary.get("vin"),
|
||
"lot_number": first_non_empty([core.get("lot_number"), vehicle_summary.get("lot_number")]),
|
||
"trim": first_non_empty([core.get("trim"), vehicle_summary.get("trim")]),
|
||
"fuel_type": first_non_empty([core.get("fuel_type"), vehicle_summary.get("fuel_type")]),
|
||
"cylinders": first_non_empty([core.get("cylinders"), vehicle_summary.get("cylinders")]),
|
||
"engine": first_non_empty([core.get("engine"), vehicle_summary.get("engine")]),
|
||
"manufactured_in": vehicle_summary.get("manufactured_in"),
|
||
"vehicle_class": vehicle_summary.get("vehicle_class"),
|
||
"run_and_drive": first_non_empty([core.get("run_and_drive"), vehicle_summary.get("run_and_drive")]),
|
||
"keys": first_non_empty([core.get("keys"), vehicle_summary.get("keys")]),
|
||
"title": title_text,
|
||
"title_brand": vehicle_summary.get("title_brand"),
|
||
"damage_primary": first_non_empty([damage.get("primary"), vehicle_summary.get("primary_damage")]),
|
||
"damage_secondary": damage.get("secondary"),
|
||
"damage_description": damage.get("description"),
|
||
"buy_now": first_non_empty([pricing.get("buy_now"), vehicle_summary.get("buy_now")]),
|
||
"current_bid": first_non_empty([pricing.get("current_bid"), vehicle_summary.get("current_bid")]),
|
||
"actual_cash_value": pricing.get("actual_cash_value"),
|
||
"estimated_repair_cost": pricing.get("estimated_repair_cost"),
|
||
"seller": seller,
|
||
"location": location,
|
||
"vehicle_location": vehicle_summary.get("vehicle_location"),
|
||
"auction_date": auction.get("auction_date"),
|
||
"lane": auction.get("lane"),
|
||
"branch": auction.get("branch"),
|
||
"sale_status": auction.get("sale_status"),
|
||
"source_endpoints": payload_insights.get("source_endpoints", {}),
|
||
}
|
||
|
||
content_hash = hashlib.sha256(json.dumps({
|
||
# Хеш нужен для пропуска записей без фактических изменений.
|
||
"brand": brand, "model": model, "year": year, "price": price, "mileage": mileage,
|
||
"color": color, "drive": drive, "gearbox": gearbox, "body_type": body_type,
|
||
"engine_volume": engine_volume, "is_damaged": is_damaged, "is_sold": is_sold,
|
||
"image_count": len(images_records),
|
||
}, sort_keys=True, default=str).encode()).hexdigest()
|
||
|
||
return CarRecord(
|
||
parser_id=parser_id, brand=brand, model=model, year=year, price=price, currency=currency,
|
||
mileage=mileage, country=country, is_sold=is_sold, color=color, drive=drive, gearbox=gearbox,
|
||
steering_wheel=steering, body_type=body_type, engine_volume=engine_volume,
|
||
selling_type=self._normalize_selling_type("AUCTION"), one_owner=one_owner, new_car=new_car,
|
||
is_hidden=False, origin=origin, origin_url=vehicle_url, origin_id=origin_id, is_damaged=is_damaged,
|
||
evaluation=evaluation, non_smoking=non_smoking, rental=rental, repair_history=repair_history,
|
||
slug=slug, last_seen_at=datetime.now(timezone.utc), content_hash=content_hash, images=images_records,
|
||
raw_attributes=raw_attributes, mapping_notes=notes,
|
||
)
|
||
|
||
@staticmethod
|
||
def _as_str(value: Any) -> str:
|
||
return "" if value is None else str(value).strip()
|
||
|
||
@staticmethod
|
||
def _to_int(value: Any) -> int | None:
|
||
if value is None:
|
||
return None
|
||
if isinstance(value, bool):
|
||
return int(value)
|
||
if isinstance(value, (int, float)):
|
||
return int(value)
|
||
digits = re.sub(r"[^\d]", "", str(value))
|
||
return int(digits) if digits else None
|
||
|
||
def _to_engine_cc(self, value: Any) -> int | None:
|
||
# Поддерживаем и литры, и уже готовые cc.
|
||
text = str(value).lower().strip() if value is not None else ""
|
||
if not text:
|
||
return None
|
||
if liters_match := re.search(r"(\d+(?:\.\d+)?)\s*l", text):
|
||
return int(float(liters_match.group(1)) * 1000)
|
||
if cubic_match := re.search(r"(\d{3,5})\s*(?:cc|cm3|cubic)", text):
|
||
return int(cubic_match.group(1))
|
||
return int(text) if re.match(r"^\d+$", text) else None
|
||
|
||
def _normalize_currency(self, value: Any) -> str:
|
||
text = self._as_str(value).upper() or "USD"
|
||
if text in CURRENCY_ENUM_VALUES:
|
||
return text
|
||
return "USD" if "$" in str(value) else "USD"
|
||
|
||
def _normalize_drive(self, value: Any) -> str | None:
|
||
return self._map_value(value, self.DRIVE_MAP, DRIVE_ENUM_VALUES, empty_default=None, fallback="NA")
|
||
|
||
def _normalize_gearbox(self, value: Any) -> str | None:
|
||
return self._map_value(value, self.GEARBOX_MAP, GEARBOX_ENUM_VALUES, empty_default=None, fallback="NA")
|
||
|
||
def _normalize_steering(self, value: Any) -> str | None:
|
||
return self._map_value(value, self.STEERING_MAP, STEERING_WHEEL_ENUM_VALUES, empty_default=None, fallback=None)
|
||
|
||
def _normalize_body_type(self, value: Any) -> str:
|
||
return self._map_value(value, self.BODY_MAP, BODY_TYPE_ENUM_VALUES, empty_default="OTHER", fallback="OTHER") or "OTHER"
|
||
|
||
def _normalize_country(self, value: Any) -> str:
|
||
fallback = "US" if "US" in COUNTRY_ENUM_VALUES else "NA"
|
||
return self._map_value(value, self.COUNTRY_MAP, COUNTRY_ENUM_VALUES, empty_default="US", fallback=fallback) or fallback
|
||
|
||
def _normalize_selling_type(self, value: Any) -> str:
|
||
text = self._as_str(value) or "AUCTION"
|
||
return text if text in SELLING_TYPE_ENUM_VALUES else "AUCTION"
|
||
|
||
def _map_value(
|
||
self,
|
||
value: Any,
|
||
mapping: dict[str, str],
|
||
allowed_values: tuple[str, ...],
|
||
*,
|
||
empty_default: str | None,
|
||
fallback: str | None,
|
||
) -> str | None:
|
||
# Общий helper для enum-нормализации по точному или частичному совпадению.
|
||
text = self._as_str(value).lower()
|
||
if not text:
|
||
return empty_default
|
||
if (mapped := mapping.get(text)) and mapped in allowed_values:
|
||
return mapped
|
||
for marker, mapped in mapping.items():
|
||
if marker in text and mapped in allowed_values:
|
||
return mapped
|
||
return fallback
|
||
|
||
@staticmethod
|
||
def _normalize_color(value: Any) -> str:
|
||
text = str(value).strip() if value is not None else "other"
|
||
if not text:
|
||
return "other"
|
||
if "/" in text:
|
||
text = text.split("/")[0].strip()
|
||
return text.lower() or "other"
|
||
|
||
@staticmethod
|
||
def _boolish(value: Any) -> bool:
|
||
return value if isinstance(value, bool) else str(value).strip().lower() in {"1", "true", "yes", "y", "owner", "one owner", "new"}
|
||
|
||
def _bool_damage(self, damage: dict[str, Any], vehicle_summary: dict[str, Any]) -> bool:
|
||
for value in [damage.get("primary"), damage.get("secondary"), damage.get("description"), vehicle_summary.get("primary_damage")]:
|
||
text = self._as_str(value).lower()
|
||
if text and text not in self.NO_DAMAGE_MARKERS:
|
||
return True
|
||
return False
|
||
|
||
def _bool_sold(self, auction: dict[str, Any]) -> bool:
|
||
return any(token in self._as_str(auction.get("sale_status")).lower() for token in ["sold", "closed", "ended"])
|
||
|
||
def _build_images(self, urls: list[Any]) -> list[ImageRecord]:
|
||
# Для imageKeys оставляем ссылку с наибольшим размером.
|
||
best_by_key: dict[str, str] = {}
|
||
key_order: list[str] = []
|
||
non_keyed: list[str] = []
|
||
for item in urls:
|
||
if not isinstance(item, str):
|
||
continue
|
||
url = item.strip()
|
||
if not url:
|
||
continue
|
||
if m := re.search(r'imageKeys=([^&]+)', url):
|
||
img_key = m.group(1)
|
||
w = int(re.search(r'width=(\d+)', url).group(1)) if re.search(r'width=(\d+)', url) else 0
|
||
existing = best_by_key.get(img_key)
|
||
if existing is None:
|
||
best_by_key[img_key] = url
|
||
key_order.append(img_key)
|
||
else:
|
||
existing_w = int(re.search(r'width=(\d+)', existing).group(1)) if re.search(r'width=(\d+)', existing) else 0
|
||
if w > existing_w:
|
||
best_by_key[img_key] = url
|
||
elif url not in non_keyed:
|
||
non_keyed.append(url)
|
||
deduped = [best_by_key[k] for k in key_order] + non_keyed
|
||
return [ImageRecord(fullres_image=self._make_fullres_url(url), preview_image=self._make_preview_url(url), order_index=index) for index, url in enumerate(deduped)]
|
||
|
||
@staticmethod
|
||
def _make_fullres_url(url: str) -> str:
|
||
if "vis.iaai.com" in url:
|
||
return re.sub(r'height=\d+', 'height=633', re.sub(r'width=\d+', 'width=845', url))
|
||
return url
|
||
|
||
@staticmethod
|
||
def _make_preview_url(url: str) -> str:
|
||
if "vis.iaai.com" in url:
|
||
preview = re.sub(r'height=\d+', 'height=300', re.sub(r'width=\d+', 'width=400', url))
|
||
if preview != url:
|
||
return preview
|
||
return url
|
||
|
||
def _build_origin_id(self, vehicle_url: str, vehicle_summary: dict[str, Any], core: dict[str, Any]) -> str:
|
||
# Предпочитаем lot_number, затем vin, затем хвост URL.
|
||
for value in [core.get("lot_number"), vehicle_summary.get("lot_number"), vehicle_summary.get("vin")]:
|
||
text = self._as_str(value)
|
||
if text:
|
||
return text
|
||
tail = urlparse(vehicle_url).path.rstrip("/").split("/")[-1]
|
||
if "~" in tail:
|
||
tail = tail.split("~")[0]
|
||
return tail or self._slugify(vehicle_url)
|
||
|
||
@staticmethod
|
||
def _slugify(value: str) -> str:
|
||
return re.sub(r"[^a-z0-9]+", "-", value.lower()).strip("-") or "car"
|
||
|
||
|
||
class IAAICarMapper(CarMapper):
|
||
"""Совместимое имя маппера."""
|
||
|