import hashlib import re from datetime import datetime, timezone from string import ascii_letters, digits 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-модель. vehicle_summary = vehicle_summary or {} payload_insights = payload_insights or {} 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 = self._generate_parser_id(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_year(first_non_empty([core.get("year"), vehicle_summary.get("year")])) price = self._first_parsed_int( [ pricing.get("buy_now"), pricing.get("current_bid"), vehicle_summary.get("buy_now"), vehicle_summary.get("current_bid"), pricing.get("actual_cash_value"), ], self._to_money_int, ) mileage = self._first_parsed_int( [core.get("odometer"), vehicle_summary.get("odometer"), 0], self._to_int, ) 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"), pricing.get("buy_now"), pricing.get("current_bid"), vehicle_summary.get("buy_now"), vehicle_summary.get("current_bid"), "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" 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), images=images_records, ) @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 @classmethod def _to_money_int(cls, value: Any) -> int | None: # Нормализация стоимости из разных форматов. if value is None: return None if isinstance(value, bool): return None if isinstance(value, (int, float)): return int(value) text = str(value).strip() if not text: return None lowered = text.lower() if any(token in lowered for token in ["n/a", "na", "tbd", "unknown", "call", "contact"]): return None numbers = re.findall(r"\d[\d\s.,]*", text) if not numbers: return None best: int | None = None for number in numbers: clean = number.replace(" ", "") if "," in clean and "." in clean: # Поддержка 1,234.56 и 1.234,56. if clean.rfind(",") > clean.rfind("."): clean = clean.replace(".", "").replace(",", ".") else: clean = clean.replace(",", "") elif "," in clean: parts = clean.split(",") # 123,45 -> 123.45, иначе разделитель тысяч. if len(parts[-1]) in {1, 2} and len(parts) == 2: clean = clean.replace(",", ".") else: clean = clean.replace(",", "") elif "." in clean: parts = clean.split(".") if not (len(parts[-1]) in {1, 2} and len(parts) == 2): clean = clean.replace(".", "") try: parsed = int(float(clean)) except ValueError: continue if parsed > 0 and (best is None or parsed > best): best = parsed return best @staticmethod def _first_parsed_int(values: list[Any], parser) -> int | None: for value in values: parsed = parser(value) if parsed is not None: return parsed return None @staticmethod def _to_year(value: Any) -> int | None: parsed = CarMapper._to_int(value) if parsed is None: return None if 1900 <= parsed <= 2100: return parsed return 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 = text.upper() or "USD" if any(token in text for token in ["€", "EUR"]): return "EUR" if any(token in text for token in ["¥", "JPY"]): return "JPY" if any(token in text for token in ["₩", "KRW"]): return "KRW" if any(token in text for token in ["£", "GBP"]): return "GBP" if any(token in text for token in ["₽", "RUB"]): return "RUB" if any(token in text for token in ["AED", "د.إ"]): return "AED" if any(token in text for token in ["CA$", "CAD"]): return "CAD" text = upper 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 # ---------- parser_id ---------- _PARSER_ID_ALPHABET = ascii_letters + digits # a-zA-Z0-9 @classmethod def _generate_parser_id(cls, origin_id: str) -> str: """Deterministic car-XXXX... (prefix 'car-' + 22 alphanumeric chars).""" digest = hashlib.sha256(origin_id.encode()).digest() alphabet = cls._PARSER_ID_ALPHABET base = len(alphabet) num = int.from_bytes(digest[:17], "big") # 17 bytes = 136 bits, enough for 22 chars chars: list[str] = [] for _ in range(22): num, idx = divmod(num, base) chars.append(alphabet[idx]) return "car-" + "".join(chars) def _build_origin_id(self, vehicle_url: str, vehicle_summary: dict[str, Any], core: dict[str, Any]) -> str: # Формат: iaai:{lot_number} (аналог copart:94323985). for value in [core.get("lot_number"), vehicle_summary.get("lot_number"), vehicle_summary.get("vin")]: text = self._as_str(value) if text: return f"iaai:{text}" tail = urlparse(vehicle_url).path.rstrip("/").split("/")[-1] if "~" in tail: tail = tail.split("~")[0] raw = tail or self._slugify(vehicle_url) return f"iaai:{raw}" @staticmethod def _slugify(value: str) -> str: return re.sub(r"[^a-z0-9]+", "-", value.lower()).strip("-") or "car"