425 lines
12 KiB
Python
425 lines
12 KiB
Python
"""Маппер: JSON-запись OpenLane API → CarRecord для сохранения в БД."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import logging
|
|
import re
|
|
from datetime import datetime, timezone
|
|
from typing import Any
|
|
|
|
from ..storage.schemas import CarRecord, ImageRecord
|
|
|
|
logger = logging.getLogger("openlane_scraper.openlane.mapper")
|
|
|
|
# Маппинг body_type из OpenLane в наш enum.
|
|
BODY_MAP: dict[str, str] = {
|
|
"sedan": "SEDAN",
|
|
"coupe": "COUPE",
|
|
"suv": "SUV",
|
|
"sport utility": "SUV",
|
|
"crossover": "SUV",
|
|
"hatchback": "HATCHBACK",
|
|
"minivan": "MINIVAN",
|
|
"van": "MINIVAN",
|
|
"wagon": "STATION_WAGON",
|
|
"station wagon": "STATION_WAGON",
|
|
"pickup": "PICKUP",
|
|
"truck": "TRUCK",
|
|
"convertible": "OPEN",
|
|
"cabriolet": "OPEN",
|
|
"roadster": "OPEN",
|
|
"rv": "RV",
|
|
"motorhome": "RV",
|
|
}
|
|
|
|
DRIVE_MAP: dict[str, str] = {
|
|
"fwd": "FWD",
|
|
"front wheel drive": "FWD",
|
|
"front-wheel drive": "FWD",
|
|
"rwd": "RWD",
|
|
"rear wheel drive": "RWD",
|
|
"rear-wheel drive": "RWD",
|
|
"awd": "4WD",
|
|
"4wd": "4WD",
|
|
"all wheel drive": "4WD",
|
|
"all-wheel drive": "4WD",
|
|
"4x4": "4WD",
|
|
"2wd": "2WD",
|
|
"two wheel drive": "2WD",
|
|
}
|
|
|
|
GEARBOX_MAP: dict[str, str] = {
|
|
"automatic": "AT",
|
|
"auto": "AT",
|
|
"at": "AT",
|
|
"manual": "MT",
|
|
"mt": "MT",
|
|
"cvt": "CVT",
|
|
"continuously variable": "CVT",
|
|
"electric": "EV",
|
|
"ev": "EV",
|
|
}
|
|
|
|
_MILEAGE_RE = re.compile(r"[\d,]+")
|
|
_ENGINE_RE = re.compile(r"(\d+\.?\d*)\s*[lL]")
|
|
|
|
|
|
def _safe_str(value: Any, default: str = "") -> str:
|
|
if value is None:
|
|
return default
|
|
return str(value).strip() or default
|
|
|
|
|
|
def _safe_int(value: Any) -> int | None:
|
|
if value is None:
|
|
return None
|
|
try:
|
|
cleaned = str(value).replace(",", "").strip()
|
|
if not cleaned:
|
|
return None
|
|
return int(float(cleaned))
|
|
except (ValueError, TypeError):
|
|
return None
|
|
|
|
|
|
def _normalize_enum(raw: Any, mapping: dict[str, str], default: str = "NA") -> str:
|
|
if not raw:
|
|
return default
|
|
key = str(raw).strip().lower()
|
|
return mapping.get(key, default)
|
|
|
|
|
|
def _generate_parser_id(origin_id: str) -> str:
|
|
return hashlib.sha256(origin_id.encode()).hexdigest()[:40]
|
|
|
|
|
|
def _generate_slug(year: int | None, brand: str, model: str, origin_id: str) -> str:
|
|
parts = []
|
|
if year:
|
|
parts.append(str(year))
|
|
parts.append(brand.lower())
|
|
parts.append(model.lower())
|
|
parts.append(origin_id.replace(":", "-"))
|
|
slug = "-".join(parts)
|
|
slug = re.sub(r"[^a-z0-9\-]", "-", slug)
|
|
slug = re.sub(r"-+", "-", slug).strip("-")
|
|
return slug[:200]
|
|
|
|
|
|
def _build_openlane_origin_id(raw_id: Any) -> str | None:
|
|
"""Строит origin_id в формате openlane:id."""
|
|
normalized = _safe_str(raw_id)
|
|
if not normalized:
|
|
return None
|
|
if normalized.lower().startswith("openlane:"):
|
|
normalized = _safe_str(normalized.split(":", 1)[1])
|
|
if not normalized:
|
|
return None
|
|
return f"openlane:{normalized}"
|
|
|
|
|
|
def _extract_nested(record: dict[str, Any], *keys: str) -> Any:
|
|
"""Извлекает значение из вложенного словаря по цепочке ключей."""
|
|
obj: Any = record
|
|
for key in keys:
|
|
if isinstance(obj, dict):
|
|
obj = obj.get(key)
|
|
else:
|
|
return None
|
|
return obj
|
|
|
|
|
|
def _extract_images(record: dict[str, Any]) -> list[ImageRecord]:
|
|
"""Извлекает изображения из записи OpenLane."""
|
|
images: list[ImageRecord] = []
|
|
seen_urls: set[str] = set()
|
|
|
|
raw_images = (
|
|
record.get("images")
|
|
or record.get("photos")
|
|
or record.get("media", {}).get("images")
|
|
or record.get("image_urls")
|
|
or []
|
|
)
|
|
|
|
if isinstance(raw_images, list):
|
|
for idx, img in enumerate(raw_images):
|
|
if isinstance(img, str):
|
|
url = img.strip()
|
|
if url and url not in seen_urls:
|
|
seen_urls.add(url)
|
|
images.append(ImageRecord(
|
|
fullres_image=url,
|
|
preview_image=url,
|
|
order_index=idx,
|
|
))
|
|
elif isinstance(img, dict):
|
|
fullres = _safe_str(
|
|
img.get("full") or img.get("fullres") or img.get("url")
|
|
or img.get("original") or img.get("large") or img.get("href")
|
|
or img.get("large_resolution_url")
|
|
)
|
|
preview = _safe_str(
|
|
img.get("thumbnail") or img.get("thumb") or img.get("preview")
|
|
or img.get("small") or img.get("low_resolution_url")
|
|
or fullres
|
|
)
|
|
if fullres and fullres not in seen_urls:
|
|
seen_urls.add(fullres)
|
|
images.append(ImageRecord(
|
|
fullres_image=fullres,
|
|
preview_image=preview,
|
|
order_index=idx,
|
|
))
|
|
|
|
# Fallback: одиночное изображение.
|
|
if not images:
|
|
single = (
|
|
record.get("image_url")
|
|
or record.get("image")
|
|
or record.get("large_resolution_url")
|
|
or record.get("low_resolution_url")
|
|
or record.get("thumbnail")
|
|
or record.get("photo_url")
|
|
or _extract_nested(record, "media", "primary")
|
|
)
|
|
if single and isinstance(single, str) and single.strip():
|
|
images.append(ImageRecord(
|
|
fullres_image=single.strip(),
|
|
preview_image=single.strip(),
|
|
order_index=0,
|
|
))
|
|
|
|
return images
|
|
|
|
|
|
def _parse_mileage(raw: Any) -> int:
|
|
if raw is None:
|
|
return 0
|
|
if isinstance(raw, (int, float)):
|
|
return max(0, int(raw))
|
|
text = str(raw)
|
|
match = _MILEAGE_RE.search(text)
|
|
if match:
|
|
try:
|
|
return max(0, int(match.group().replace(",", "")))
|
|
except ValueError:
|
|
pass
|
|
return 0
|
|
|
|
|
|
def _parse_engine_volume_cc(raw: Any) -> int | None:
|
|
"""Парсит объём двигателя и возвращает значение в кубических сантиметрах."""
|
|
if raw is None:
|
|
return None
|
|
if isinstance(raw, (int, float)):
|
|
value = float(raw)
|
|
# Если значение < 20 — скорее всего это литры, конвертируем в cc.
|
|
if 0 < value < 20:
|
|
return int(value * 1000)
|
|
if value >= 100:
|
|
return int(value)
|
|
return None
|
|
text = str(raw)
|
|
match = _ENGINE_RE.search(text)
|
|
if match:
|
|
liters = float(match.group(1))
|
|
return int(liters * 1000)
|
|
return None
|
|
|
|
|
|
def map_openlane_record(record: dict[str, Any]) -> CarRecord | None:
|
|
"""Маппит одну запись из OpenLane API в CarRecord.
|
|
|
|
Возвращает None, если запись не содержит минимально необходимых данных.
|
|
"""
|
|
# Извлекаем идентификатор.
|
|
raw_id = (
|
|
record.get("id")
|
|
or record.get("vehicle_id")
|
|
or record.get("listing_id")
|
|
or record.get("vin")
|
|
)
|
|
if not raw_id:
|
|
logger.debug("Skipping record without id: %s", record.get("vin", "unknown"))
|
|
return None
|
|
|
|
origin_id = _build_openlane_origin_id(raw_id)
|
|
if not origin_id:
|
|
logger.debug("Skipping record with malformed id: %s", raw_id)
|
|
return None
|
|
|
|
# Бренд и модель — обязательные поля.
|
|
brand = _safe_str(
|
|
record.get("make")
|
|
or record.get("brand")
|
|
or record.get("manufacturer")
|
|
or _extract_nested(record, "vehicle", "make")
|
|
).upper()
|
|
|
|
model = _safe_str(
|
|
record.get("model")
|
|
or record.get("model_name")
|
|
or _extract_nested(record, "vehicle", "model")
|
|
).upper()
|
|
|
|
if not brand or not model:
|
|
logger.debug("Skipping record without brand/model: %s", origin_id)
|
|
return None
|
|
|
|
year = _safe_int(
|
|
record.get("year")
|
|
or record.get("model_year")
|
|
or _extract_nested(record, "vehicle", "year")
|
|
)
|
|
|
|
price = _safe_int(
|
|
record.get("price")
|
|
or record.get("current_bid")
|
|
or record.get("buy_now_price")
|
|
or record.get("asking_price")
|
|
or record.get("sale_price")
|
|
or _extract_nested(record, "pricing", "current")
|
|
or _extract_nested(record, "pricing", "buy_now")
|
|
)
|
|
|
|
currency = _safe_str(
|
|
record.get("currency")
|
|
or record.get("currency_code")
|
|
or _extract_nested(record, "pricing", "currency"),
|
|
"USD",
|
|
).upper()
|
|
if currency not in {"JPY", "USD", "EUR", "RUB", "KRW", "AED", "GBP", "CAD"}:
|
|
currency = "USD"
|
|
|
|
mileage = _parse_mileage(
|
|
record.get("mileage")
|
|
or record.get("odometer")
|
|
or record.get("odometer_reading")
|
|
or _extract_nested(record, "vehicle", "mileage")
|
|
)
|
|
|
|
color = _safe_str(
|
|
record.get("color")
|
|
or record.get("exterior_color")
|
|
or _extract_nested(record, "vehicle", "color"),
|
|
"other",
|
|
).lower()
|
|
|
|
body_type = _normalize_enum(
|
|
record.get("body_type")
|
|
or record.get("body_style")
|
|
or record.get("vehicle_type")
|
|
or _extract_nested(record, "vehicle", "body_type"),
|
|
BODY_MAP,
|
|
"OTHER",
|
|
)
|
|
|
|
drive = _normalize_enum(
|
|
record.get("drive_type")
|
|
or record.get("drivetrain")
|
|
or record.get("drive")
|
|
or _extract_nested(record, "vehicle", "drivetrain"),
|
|
DRIVE_MAP,
|
|
)
|
|
|
|
gearbox = _normalize_enum(
|
|
record.get("transmission")
|
|
or record.get("gearbox")
|
|
or _extract_nested(record, "vehicle", "transmission"),
|
|
GEARBOX_MAP,
|
|
)
|
|
|
|
engine_volume = _parse_engine_volume_cc(
|
|
record.get("engine")
|
|
or record.get("engine_size")
|
|
or record.get("displacement")
|
|
or _extract_nested(record, "vehicle", "engine")
|
|
)
|
|
|
|
# URL записи на OpenLane.
|
|
origin_url = _safe_str(
|
|
record.get("url")
|
|
or record.get("listing_url")
|
|
or record.get("detail_url")
|
|
or record.get("permalink")
|
|
)
|
|
if not origin_url:
|
|
origin_url = f"https://app.openlane.com/vehicles/{raw_id}"
|
|
|
|
country = _safe_str(
|
|
record.get("country")
|
|
or record.get("location_country")
|
|
or _extract_nested(record, "location", "country"),
|
|
"US",
|
|
).upper()
|
|
if country not in {"JP", "KR", "US", "CA", "NA"}:
|
|
country = "US"
|
|
|
|
is_damaged = bool(
|
|
record.get("is_damaged")
|
|
or record.get("has_damage")
|
|
or record.get("damage_type")
|
|
)
|
|
|
|
vin = _safe_str(record.get("vin") or _extract_nested(record, "vehicle", "vin"))
|
|
evaluation = vin if vin else None
|
|
|
|
selling_type = "AUCTION"
|
|
sale_type = _safe_str(record.get("sale_type") or record.get("listing_type")).lower()
|
|
if sale_type in {"buy_now", "fixed_price", "stock"}:
|
|
selling_type = "STOCK"
|
|
elif sale_type in {"tender"}:
|
|
selling_type = "TENDER"
|
|
|
|
images = _extract_images(record)
|
|
|
|
parser_id = _generate_parser_id(origin_id)
|
|
slug = _generate_slug(year, brand, model, origin_id)
|
|
|
|
return CarRecord(
|
|
parser_id=parser_id,
|
|
brand=brand,
|
|
model=model,
|
|
year=year,
|
|
price=price,
|
|
currency=currency,
|
|
mileage=mileage,
|
|
country=country,
|
|
is_sold=False,
|
|
color=color,
|
|
drive=drive if drive != "NA" else None,
|
|
gearbox=gearbox if gearbox != "NA" else None,
|
|
body_type=body_type,
|
|
engine_volume=engine_volume,
|
|
selling_type=selling_type,
|
|
origin="OPENLANE",
|
|
origin_url=origin_url,
|
|
origin_id=origin_id,
|
|
is_damaged=is_damaged,
|
|
evaluation=evaluation,
|
|
slug=slug,
|
|
images=images,
|
|
)
|
|
|
|
|
|
def map_openlane_records(records: list[dict[str, Any]]) -> list[CarRecord]:
|
|
"""Маппит список записей OpenLane API в список CarRecord.
|
|
|
|
Пропускает записи без минимально необходимых данных.
|
|
"""
|
|
result: list[CarRecord] = []
|
|
for record in records:
|
|
try:
|
|
car = map_openlane_record(record)
|
|
if car is not None:
|
|
result.append(car)
|
|
except Exception:
|
|
logger.warning(
|
|
"Failed to map OpenLane record id=%s",
|
|
record.get("id", "unknown"),
|
|
exc_info=True,
|
|
)
|
|
return result
|