update scraper package
This commit is contained in:
qananasikq
622
dubizzle_scraper/parsing/mapper.py
Normal file
622
dubizzle_scraper/parsing/mapper.py
Normal file
@@ -0,0 +1,622 @@
|
||||
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:
|
||||
# Маппер DUBIZZLE в 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",
|
||||
"ae": "AE", "uae": "AE", "united arab emirates": "AE", "dubai": "AE", "abu dhabi": "AE",
|
||||
}
|
||||
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._resolve_make(core, vehicle_summary) or "UNKNOWN"
|
||||
model = self._resolve_model(core, vehicle_summary) or "UNKNOWN"
|
||||
year = self._to_year(first_non_empty([
|
||||
core.get("year"),
|
||||
vehicle_summary.get("year"),
|
||||
self._extract_detail_v2_value(vehicle_summary, "year"),
|
||||
self._extract_detail_value(vehicle_summary, "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"),
|
||||
vehicle_summary.get("price"),
|
||||
self._extract_detail_v2_value(vehicle_summary, "price"),
|
||||
self._extract_detail_value(vehicle_summary, "Price"),
|
||||
],
|
||||
self._to_money_int,
|
||||
)
|
||||
mileage = self._parse_odometer(
|
||||
first_non_empty([
|
||||
core.get("odometer"),
|
||||
core.get("kilometers"),
|
||||
vehicle_summary.get("odometer"),
|
||||
vehicle_summary.get("kilometers"),
|
||||
self._extract_detail_v2_value(vehicle_summary, "kilometers"),
|
||||
self._extract_detail_value(vehicle_summary, "Kilometers"),
|
||||
])
|
||||
)
|
||||
color = self._normalize_color(first_non_empty([
|
||||
core.get("color"),
|
||||
vehicle_summary.get("color"),
|
||||
vehicle_summary.get("exterior_color"),
|
||||
self._extract_detail_v2_value(vehicle_summary, "exterior_color"),
|
||||
self._extract_detail_value(vehicle_summary, "Exterior Color"),
|
||||
"other",
|
||||
]))
|
||||
drive = self._normalize_drive(first_non_empty([
|
||||
core.get("drive"),
|
||||
vehicle_summary.get("drive"),
|
||||
vehicle_summary.get("drive_type"),
|
||||
self._extract_detail_v2_value(vehicle_summary, "drive_system"),
|
||||
self._extract_detail_value(vehicle_summary, "Drive Type"),
|
||||
self._extract_detail_value(vehicle_summary, "Drive System"),
|
||||
]))
|
||||
# Пробуем взять привод из двигателя.
|
||||
if not drive or drive == "NA":
|
||||
engine_text = self._as_str(first_non_empty([
|
||||
core.get("engine"),
|
||||
vehicle_summary.get("engine"),
|
||||
self._extract_detail_v2_value(vehicle_summary, "engine_capacity_cc"),
|
||||
self._extract_detail_value(vehicle_summary, "Engine Capacity (cc)"),
|
||||
]))
|
||||
if engine_text:
|
||||
inferred_drive = self._normalize_drive(engine_text)
|
||||
if inferred_drive and inferred_drive != "NA":
|
||||
drive = inferred_drive
|
||||
gearbox = self._normalize_gearbox(first_non_empty([
|
||||
core.get("gearbox"),
|
||||
vehicle_summary.get("gearbox"),
|
||||
vehicle_summary.get("transmission_type"),
|
||||
vehicle_summary.get("transmission"),
|
||||
self._extract_detail_v2_value(vehicle_summary, "transmission_type"),
|
||||
self._extract_detail_value(vehicle_summary, "Transmission Type"),
|
||||
]))
|
||||
steering = self._normalize_steering(first_non_empty([
|
||||
core.get("steering_wheel"),
|
||||
vehicle_summary.get("steering_wheel"),
|
||||
self._extract_detail_v2_value(vehicle_summary, "steering_side"),
|
||||
self._extract_detail_value(vehicle_summary, "Steering Side"),
|
||||
])) or "LEFT"
|
||||
body_type = self._normalize_body_type(first_non_empty([
|
||||
core.get("body_type"),
|
||||
vehicle_summary.get("body_type"),
|
||||
self._extract_detail_v2_value(vehicle_summary, "body_type"),
|
||||
self._extract_detail_value(vehicle_summary, "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"),
|
||||
self._extract_detail_v2_value(vehicle_summary, "engine_capacity_cc"),
|
||||
self._extract_detail_value(vehicle_summary, "Engine Capacity (cc)"),
|
||||
]))
|
||||
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"),
|
||||
vehicle_summary.get("location_name"),
|
||||
self._extract_nested_text(vehicle_summary.get("neighbourhood"), "en"),
|
||||
self._extract_location_country(vehicle_summary),
|
||||
auction.get("branch"),
|
||||
"",
|
||||
]))
|
||||
country = self._normalize_country(first_non_empty([core.get("country"), location, "AE"]))
|
||||
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, [brand, model, str(year or "")])))
|
||||
images_records = self._build_images(images.get("urls") or vehicle_summary.get("image_urls") or [])
|
||||
origin = "DUBIZZLE"
|
||||
|
||||
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(first_non_empty([core.get("selling_type"), "STOCK"])), 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()
|
||||
|
||||
def _resolve_make(self, core: dict[str, Any], vehicle_summary: dict[str, Any]) -> str | None:
|
||||
category_make, _ = self._extract_category_make_model(vehicle_summary)
|
||||
slug_make, _ = self._extract_slug_make_model(vehicle_summary)
|
||||
return self._first_text(
|
||||
[
|
||||
core.get("make"),
|
||||
vehicle_summary.get("make"),
|
||||
self._extract_detail_v2_value(vehicle_summary, "make"),
|
||||
self._extract_detail_value(vehicle_summary, "Make"),
|
||||
category_make,
|
||||
slug_make,
|
||||
]
|
||||
)
|
||||
|
||||
def _resolve_model(self, core: dict[str, Any], vehicle_summary: dict[str, Any]) -> str | None:
|
||||
_, category_model = self._extract_category_make_model(vehicle_summary)
|
||||
_, slug_model = self._extract_slug_make_model(vehicle_summary)
|
||||
return self._first_text(
|
||||
[
|
||||
core.get("model"),
|
||||
vehicle_summary.get("model"),
|
||||
self._extract_detail_v2_value(vehicle_summary, "model"),
|
||||
self._extract_detail_value(vehicle_summary, "Model"),
|
||||
category_model,
|
||||
slug_model,
|
||||
]
|
||||
)
|
||||
|
||||
def _first_text(self, values: list[Any]) -> str | None:
|
||||
for value in values:
|
||||
text = self._coerce_text(value)
|
||||
if text:
|
||||
return text
|
||||
return None
|
||||
|
||||
def _coerce_text(self, value: Any) -> str | None:
|
||||
if value is None:
|
||||
return None
|
||||
if isinstance(value, list):
|
||||
for item in value:
|
||||
text = self._coerce_text(item)
|
||||
if text:
|
||||
return text
|
||||
return None
|
||||
if isinstance(value, dict):
|
||||
for key in ("en", "value", "name", "text", "label"):
|
||||
if key in value:
|
||||
text = self._coerce_text(value.get(key))
|
||||
if text:
|
||||
return text
|
||||
for nested in value.values():
|
||||
text = self._coerce_text(nested)
|
||||
if text:
|
||||
return text
|
||||
return None
|
||||
text = self._as_str(value)
|
||||
return text or None
|
||||
|
||||
def _extract_nested_text(self, value: Any, preferred_key: str = "en") -> str | None:
|
||||
if not isinstance(value, dict):
|
||||
return self._coerce_text(value)
|
||||
return self._coerce_text(value.get(preferred_key)) or self._coerce_text(value)
|
||||
|
||||
def _extract_detail_v2_value(self, vehicle_summary: dict[str, Any], slug: str) -> str | None:
|
||||
details_v2 = vehicle_summary.get("details_v2")
|
||||
if not isinstance(details_v2, dict):
|
||||
return None
|
||||
target = slug.strip().lower()
|
||||
for section in details_v2.values():
|
||||
if not isinstance(section, list):
|
||||
continue
|
||||
for item in section:
|
||||
if not isinstance(item, dict):
|
||||
continue
|
||||
if self._as_str(item.get("slug")).lower() != target:
|
||||
continue
|
||||
text = self._coerce_text(item.get("value"))
|
||||
if text:
|
||||
return text
|
||||
return None
|
||||
|
||||
def _extract_detail_value(self, vehicle_summary: dict[str, Any], detail_key: str) -> str | None:
|
||||
details = vehicle_summary.get("details")
|
||||
if not isinstance(details, dict):
|
||||
return None
|
||||
target = detail_key.strip().lower()
|
||||
for key, value in details.items():
|
||||
if self._as_str(key).lower() != target:
|
||||
continue
|
||||
text = self._coerce_text(value)
|
||||
if text:
|
||||
return text
|
||||
return None
|
||||
|
||||
def _extract_category_make_model(self, vehicle_summary: dict[str, Any]) -> tuple[str | None, str | None]:
|
||||
category_v2 = vehicle_summary.get("category_v2")
|
||||
if isinstance(category_v2, dict):
|
||||
names_en = category_v2.get("names_en")
|
||||
if isinstance(names_en, list) and len(names_en) >= 4:
|
||||
return self._coerce_text(names_en[2]), self._coerce_text(names_en[3])
|
||||
|
||||
category = vehicle_summary.get("category")
|
||||
if isinstance(category, dict):
|
||||
names_en = category.get("en")
|
||||
if isinstance(names_en, list) and len(names_en) >= 3:
|
||||
return self._coerce_text(names_en[1]), self._coerce_text(names_en[2])
|
||||
|
||||
return None, None
|
||||
|
||||
def _extract_slug_make_model(self, vehicle_summary: dict[str, Any]) -> tuple[str | None, str | None]:
|
||||
category_v2 = vehicle_summary.get("category_v2")
|
||||
if not isinstance(category_v2, dict):
|
||||
return None, None
|
||||
slug_paths = category_v2.get("slug_paths")
|
||||
if not isinstance(slug_paths, list) or len(slug_paths) < 3:
|
||||
return None, None
|
||||
make = self._humanize_slug_path_part(slug_paths[2])
|
||||
model = self._humanize_slug_path_part(slug_paths[3]) if len(slug_paths) >= 4 else None
|
||||
return make, model
|
||||
|
||||
def _humanize_slug_path_part(self, value: Any) -> str | None:
|
||||
text = self._as_str(value)
|
||||
if not text:
|
||||
return None
|
||||
slug = text.split("/")[-1].strip().strip("-")
|
||||
if not slug:
|
||||
return None
|
||||
return slug.replace("-", " ").title()
|
||||
|
||||
def _extract_location_country(self, vehicle_summary: dict[str, Any]) -> str | None:
|
||||
site = vehicle_summary.get("site")
|
||||
if isinstance(site, dict):
|
||||
return self._coerce_text(site.get("en"))
|
||||
location_list = vehicle_summary.get("location_list")
|
||||
if isinstance(location_list, dict):
|
||||
en_values = location_list.get("en")
|
||||
if isinstance(en_values, list) and en_values:
|
||||
return self._coerce_text(en_values[0])
|
||||
return None
|
||||
|
||||
@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:
|
||||
# Поддержка двух форматов.
|
||||
if clean.rfind(",") > clean.rfind("."):
|
||||
clean = clean.replace(".", "").replace(",", ".")
|
||||
else:
|
||||
clean = clean.replace(",", "")
|
||||
elif "," in clean:
|
||||
parts = clean.split(",")
|
||||
# Десятичный или тысячный разделитель.
|
||||
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
|
||||
|
||||
@staticmethod
|
||||
def _parse_odometer(value: Any) -> int:
|
||||
# Разбор пробега.
|
||||
if value is None:
|
||||
return 0
|
||||
text = str(value).strip()
|
||||
if not text:
|
||||
return 0
|
||||
lowered = text.lower()
|
||||
if any(skip in lowered for skip in ["not required", "exempt", "n/a", "na", "unknown", "tbd"]):
|
||||
return 0
|
||||
# Ищем число.
|
||||
numbers = re.findall(r"[\d,]+", text)
|
||||
for num_str in numbers:
|
||||
clean = num_str.replace(",", "")
|
||||
if clean.isdigit() and int(clean) > 0:
|
||||
return int(clean)
|
||||
return 0
|
||||
|
||||
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 = "AE" if "AE" in COUNTRY_ENUM_VALUES else "NA"
|
||||
return self._map_value(value, self.COUNTRY_MAP, COUNTRY_ENUM_VALUES, empty_default="AE", fallback=fallback) or fallback
|
||||
|
||||
def _normalize_selling_type(self, value: Any) -> str:
|
||||
text = (self._as_str(value) or "STOCK").upper()
|
||||
if text in SELLING_TYPE_ENUM_VALUES:
|
||||
return text
|
||||
if text in {"DEALER", "PRIVATE", "CLASSIFIED"}:
|
||||
return "STOCK"
|
||||
return "STOCK"
|
||||
|
||||
def _map_value(
|
||||
self,
|
||||
value: Any,
|
||||
mapping: dict[str, str],
|
||||
allowed_values: tuple[str, ...],
|
||||
*,
|
||||
empty_default: str | None,
|
||||
fallback: str | None,
|
||||
) -> str | None:
|
||||
# Общая нормализация 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.dubizzle.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.dubizzle.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
|
||||
|
||||
@classmethod
|
||||
def _generate_parser_id(cls, origin_id: str) -> str:
|
||||
# Стабильный parser_id.
|
||||
digest = hashlib.sha256(origin_id.encode()).digest()
|
||||
alphabet = cls._PARSER_ID_ALPHABET
|
||||
base = len(alphabet)
|
||||
num = int.from_bytes(digest[:17], "big") # Хватает на 22 символа.
|
||||
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:
|
||||
# Формат: dubizzle:{lot_number}.
|
||||
for value in [
|
||||
core.get("lot_number"),
|
||||
vehicle_summary.get("lot_number"),
|
||||
vehicle_summary.get("id"),
|
||||
vehicle_summary.get("objectID"),
|
||||
vehicle_summary.get("uuid"),
|
||||
]:
|
||||
text = self._as_str(value)
|
||||
if text:
|
||||
return f"dubizzle:{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"dubizzle:{raw}"
|
||||
|
||||
@staticmethod
|
||||
def _slugify(value: str) -> str:
|
||||
return re.sub(r"[^a-z0-9]+", "-", value.lower()).strip("-") or "car"
|
||||
|
||||
|
||||
Reference in New Issue
Block a user