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, "country": country, "selling_type": "AUCTION", "one_owner": one_owner, "new_car": new_car, "evaluation": evaluation, "non_smoking": non_smoking, "rental": rental, "repair_history": repair_history, "images": [image.fullres_image for image in 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): """Совместимое имя маппера."""