800 lines
33 KiB
Python
800 lines
33 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import re
|
|
from datetime import datetime, timezone
|
|
from string import ascii_letters, digits
|
|
from typing import Any
|
|
from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit
|
|
|
|
from ..storage.schemas import CarRecord, ImageRecord
|
|
from .client import MobileDeClient
|
|
from .models import MobileDeListing
|
|
|
|
PARSER_ID_ALPHABET = ascii_letters + digits
|
|
PARSER_ID_RE = re.compile(r"^car-[A-Za-z0-9]{22}$")
|
|
|
|
_RU_BODY_MAP = {
|
|
"кабриолет": "OPEN",
|
|
"родстер": "OPEN",
|
|
"седан": "SEDAN",
|
|
"внедорожник": "SUV",
|
|
"пикап": "PICKUP",
|
|
"универсал": "STATION_WAGON",
|
|
"минивэн": "MINIVAN",
|
|
"микроавтобус": "MINIVAN",
|
|
"фургон": "MINIVAN",
|
|
"купе": "COUPE",
|
|
"хэтчбек": "HATCHBACK",
|
|
"малолитражный": "HATCHBACK",
|
|
}
|
|
|
|
_RU_GEARBOX_MAP = {
|
|
"автоматическая": "AT",
|
|
"автоматическая коробка": "AT",
|
|
"механика": "MT",
|
|
"механическая": "MT",
|
|
"механическая коробка": "MT",
|
|
"вариатор": "CVT",
|
|
}
|
|
|
|
_RU_COLOR_MAP = {
|
|
"черный": "black",
|
|
"чёрный": "black",
|
|
"белый": "white",
|
|
"серебристый": "silver",
|
|
"серебряный": "silver",
|
|
"серый": "gray",
|
|
"красный": "red",
|
|
"синий": "blue",
|
|
"голубой": "blue",
|
|
"зеленый": "green",
|
|
"зелёный": "green",
|
|
"коричневый": "brown",
|
|
"бежевый": "beige",
|
|
"желтый": "yellow",
|
|
"жёлтый": "yellow",
|
|
"оранжевый": "orange",
|
|
"золотистый": "gold",
|
|
"золотой": "gold",
|
|
"бронзовый": "bronze",
|
|
"фиолетовый": "purple",
|
|
}
|
|
|
|
_RU_DRIVE_MARKERS = {
|
|
"FWD": ("передний привод", "привод на передние колеса", "передние колеса"),
|
|
"RWD": ("задний привод", "привод на задние колеса", "задние колеса"),
|
|
"4WD": ("полный привод", "привод на четыре колеса", "все колеса"),
|
|
"2WD": ("привод на два колеса", "два ведущих колеса"),
|
|
}
|
|
|
|
_SEARCH_SSR_DRIVE_MARKERS = (
|
|
("4WD", re.compile(r"\b(?:xdrive(?:\b|\d)|quattro|4matic|4motion|all4|awd|4wd|4x4|allrad(?:antrieb)?|carrera\s*4s?)\b", re.IGNORECASE)),
|
|
("FWD", re.compile(r"\bfrontantrieb\b", re.IGNORECASE)),
|
|
("RWD", re.compile(r"\bheckantrieb\b", re.IGNORECASE)),
|
|
("2WD", re.compile(r"\bsdrive(?:\b|\d)", re.IGNORECASE)),
|
|
)
|
|
|
|
_IMAGE_FIELD_HINTS = ["image", "images", "media", "gallery", "photo", "pic", "picture", "url", "src", "uri", "ref"]
|
|
_MOBILEDE_IMAGE_HOST = "img.classistatic.de"
|
|
_MOBILEDE_IMAGE_PATH_PREFIX = "/api/v1/mo-prod/images/"
|
|
_MOBILEDE_FULLRES_IMAGE_RULE = "mo-1600"
|
|
_MOBILEDE_PREVIEW_IMAGE_RULE = "mo-360"
|
|
_MOBILEDE_IMAGE_ID_RE = re.compile(r"^[0-9a-f]{2}/[0-9a-f-]{8,}$", re.IGNORECASE)
|
|
_COLOR_VALUE_KEYS = {"ecol", "color", "exteriorcolor", "manufacturercolorname", "vehiclecolor", "paint"}
|
|
_COUNTRY_VALUE_KEYS = {"country", "countrycode"}
|
|
_BODY_VALUE_KEYS = {"category", "bodytype", "vehiclecategory", "body"}
|
|
_ENGINE_VOLUME_VALUE_KEYS = {"cubiccapacity", "enginevolume", "displacement", "enginedisplacement"}
|
|
_DRIVE_VALUE_KEYS = {"wheeldrive", "drivetrain", "drive"}
|
|
_GEARBOX_VALUE_KEYS = {"transmission", "gearbox", "transmissiontype"}
|
|
_STEERING_VALUE_KEYS = {"steeringwheel", "steeringposition", "steering", "righthanddrive"}
|
|
|
|
|
|
class MobileDeMapper:
|
|
"""Map mobile.de payloads into CarRecord."""
|
|
|
|
def listing_to_car_record(self, listing: MobileDeListing) -> CarRecord:
|
|
raw = listing.raw or {}
|
|
attr = raw.get("attr") if isinstance(raw.get("attr"), dict) else {}
|
|
make = raw.get("make") if isinstance(raw.get("make"), dict) else {}
|
|
model_payload = raw.get("model") if isinstance(raw.get("model"), dict) else {}
|
|
|
|
brand = self._text(make.get("localized") or self._brand_from_title(listing.title) or listing.title or "UNKNOWN")
|
|
model = self._text(model_payload.get("localized") or self._model_from_title(listing.title, brand) or listing.subtitle or "UNKNOWN")
|
|
origin_id = self.origin_id(str(listing.id))
|
|
title = " ".join(part for part in [listing.title, listing.subtitle] if part)
|
|
contact = raw.get("contact") if isinstance(raw.get("contact"), dict) else {}
|
|
nested_country = self._find_first_value(raw, _COUNTRY_VALUE_KEYS)
|
|
nested_color = self._find_first_value(raw, _COLOR_VALUE_KEYS)
|
|
nested_body = self._find_first_value(raw, _BODY_VALUE_KEYS)
|
|
nested_engine = self._find_first_value(raw, _ENGINE_VOLUME_VALUE_KEYS)
|
|
nested_drive = self._find_first_value(raw, _DRIVE_VALUE_KEYS)
|
|
nested_gearbox = self._find_first_value(raw, _GEARBOX_VALUE_KEYS)
|
|
nested_steering = self._find_first_value(raw, _STEERING_VALUE_KEYS)
|
|
attr_country = self._mapping_value(attr, "cn", "countryCode", "country")
|
|
attr_color = self._mapping_value(attr, "ecol", "color", "exteriorColor", "manufacturerColorName", "paint")
|
|
attr_body = self._mapping_value(attr, "c", "category", "bodyType", "body")
|
|
attr_engine = self._mapping_value(attr, "cc", "cubicCapacity", "engineVolume", "displacement", "engineDisplacement")
|
|
attr_drive = self._mapping_value(attr, "wd", "wheelDrive", "drivetrain", "drive", "driveType", "antriebsart")
|
|
attr_gearbox = self._mapping_value(attr, "tr", "transmission", "gearbox", "transmissionType")
|
|
attr_steering = self._mapping_value(attr, "steeringWheel", "steeringPosition", "steering", "rightHandDrive")
|
|
raw_features = raw.get("features") if isinstance(raw.get("features"), list) else []
|
|
steering_text = " ".join(
|
|
str(part)
|
|
for part in [attr_steering, nested_steering, *raw_features]
|
|
if part is not None and str(part).strip()
|
|
)
|
|
drive_text = " ".join(
|
|
part
|
|
for part in [
|
|
attr_drive,
|
|
raw.get("wheelDrive"),
|
|
raw.get("drivetrain"),
|
|
raw.get("drive"),
|
|
raw.get("driveType"),
|
|
nested_drive,
|
|
]
|
|
if isinstance(part, str) and part.strip()
|
|
)
|
|
search_ssr_drive = self._normalize_search_ssr_drive(raw.get("searchSsrDriveText"))
|
|
damage_text = self._text(
|
|
raw.get("damageCondition")
|
|
or attr.get("damageCondition")
|
|
or attr.get("dc")
|
|
).lower()
|
|
is_damaged = self._is_damaged_value(raw.get("hasDamage"), damage_text)
|
|
year = self._year_from_first_registration(
|
|
self._first_present(
|
|
listing.first_registration,
|
|
attr.get("fr"),
|
|
attr.get("yc"),
|
|
raw.get("firstRegistration"),
|
|
raw.get("firstRegistrationYear"),
|
|
raw.get("year"),
|
|
)
|
|
)
|
|
|
|
return CarRecord(
|
|
parser_id=self._parser_id(origin_id),
|
|
brand=brand[:50] or "UNKNOWN",
|
|
model=model[:50] or "UNKNOWN",
|
|
year=year,
|
|
price=self._money_to_int(self._first_present(listing.price, raw.get("p"), raw.get("price"))),
|
|
currency="EUR",
|
|
mileage=self._int_from_text(
|
|
self._first_present(listing.mileage, attr.get("ml"), raw.get("mileage"))
|
|
) or 0,
|
|
country=self._normalize_country(
|
|
self._first_present(
|
|
attr_country,
|
|
contact.get("countryCode"),
|
|
contact.get("country"),
|
|
raw.get("countryCode"),
|
|
raw.get("country"),
|
|
nested_country,
|
|
"DE",
|
|
)
|
|
),
|
|
is_sold=False,
|
|
color=self._normalize_color(
|
|
self._first_present(
|
|
attr_color,
|
|
raw.get("color"),
|
|
raw.get("manufacturerColorName"),
|
|
raw.get("exteriorColor"),
|
|
nested_color,
|
|
)
|
|
),
|
|
drive=self._normalize_drive(drive_text) or search_ssr_drive,
|
|
gearbox=self._normalize_gearbox(listing.transmission or attr_gearbox or nested_gearbox),
|
|
steering_wheel=self._normalize_steering(
|
|
steering_text,
|
|
right_hand_drive=self._first_present(raw.get("rightHandDrive"), attr.get("rightHandDrive")),
|
|
),
|
|
body_type=self._normalize_body_from_candidates(
|
|
attr_body,
|
|
raw.get("category"),
|
|
raw.get("bodyType"),
|
|
nested_body,
|
|
listing.subtitle,
|
|
listing.title,
|
|
),
|
|
engine_volume=self._engine_volume_from_candidates(
|
|
attr_engine,
|
|
raw.get("cubicCapacity"),
|
|
raw.get("cc"),
|
|
raw.get("engineVolume"),
|
|
raw.get("displacement"),
|
|
raw.get("engineDisplacement"),
|
|
nested_engine,
|
|
raw.get("modelDescription"),
|
|
raw.get("variant"),
|
|
listing.subtitle,
|
|
listing.title,
|
|
),
|
|
selling_type="STOCK",
|
|
one_owner=self._is_one_owner(attr.get("pvo") or raw.get("numPreviousOwners")),
|
|
new_car=self._bool_value(raw.get("isConditionNew")),
|
|
is_hidden=False,
|
|
origin="MOBILEDE",
|
|
origin_url=listing.url,
|
|
origin_id=origin_id,
|
|
is_damaged=is_damaged,
|
|
evaluation=self._rating_text(raw.get("priceRating") or raw.get("rating")),
|
|
non_smoking=False,
|
|
rental=False,
|
|
repair_history=False,
|
|
slug=self._slugify(" ".join(part for part in [title or f"{brand} {model}", str(year) if year is not None else ""] if part)),
|
|
last_seen_at=datetime.now(timezone.utc),
|
|
details_confirmed=False,
|
|
images_confirmed=False,
|
|
preserve_existing_details=True,
|
|
images=self._images_from_listing(raw),
|
|
)
|
|
|
|
def detail_to_car_record(self, listing_id: str, detail: dict[str, Any]) -> CarRecord:
|
|
attrs = self._detail_attrs_by_tag(detail.get("attributes"))
|
|
make = detail.get("make") if isinstance(detail.get("make"), dict) else {}
|
|
model_payload = detail.get("model") if isinstance(detail.get("model"), dict) else {}
|
|
contact = detail.get("contact") if isinstance(detail.get("contact"), dict) else {}
|
|
|
|
short_title = self._text(detail.get("shortTitle") or make.get("localized") or "UNKNOWN")
|
|
subtitle = self._text(detail.get("subTitle"))
|
|
title = " ".join(part for part in [short_title, subtitle] if part)
|
|
|
|
brand = self._text(make.get("localized") or self._brand_from_title(short_title) or "UNKNOWN")
|
|
model = self._text(model_payload.get("localized") or self._model_from_title(short_title, brand) or subtitle or "UNKNOWN")
|
|
origin_id = self.origin_id(str(listing_id))
|
|
|
|
price_amount, price_currency = self._detail_price(detail.get("price"))
|
|
mileage = self._int_from_text(self._mapping_value(attrs, "mileage")) or 0
|
|
year = self._year_from_first_registration(self._mapping_value(attrs, "firstRegistration", "year", "registrationDate"))
|
|
gearbox = self._normalize_gearbox(self._first_present(self._mapping_value(attrs, "transmission", "gearbox", "transmissionType"), detail.get("transmission"), detail.get("gearbox")))
|
|
body_type = self._normalize_body_from_candidates(
|
|
self._mapping_value(attrs, "category", "bodyType", "body", "vehicleCategory"),
|
|
detail.get("category"),
|
|
detail.get("bodyType"),
|
|
subtitle,
|
|
short_title,
|
|
)
|
|
color = self._normalize_color(
|
|
self._first_present(
|
|
self._mapping_value(attrs, "color", "exteriorColor", "manufacturerColorName", "paint"),
|
|
detail.get("color"),
|
|
detail.get("manufacturerColorName"),
|
|
detail.get("exteriorColor"),
|
|
self._find_first_value(detail, _COLOR_VALUE_KEYS),
|
|
)
|
|
)
|
|
features = detail.get("features") if isinstance(detail.get("features"), list) else []
|
|
feature_text = " ".join(str(feature) for feature in features if isinstance(feature, str))
|
|
explicit_drive_text = " ".join(
|
|
part
|
|
for part in [
|
|
self._mapping_value(attrs, "wheelDrive", "drivetrain", "drive", "driveType", "antriebsart"),
|
|
detail.get("wheelDrive"),
|
|
detail.get("drivetrain"),
|
|
detail.get("drive"),
|
|
detail.get("driveType"),
|
|
self._find_first_value(detail, _DRIVE_VALUE_KEYS),
|
|
]
|
|
if isinstance(part, str) and part.strip()
|
|
)
|
|
drive = self._normalize_drive(explicit_drive_text)
|
|
|
|
damage_text = self._text(attrs.get("damageCondition")).lower()
|
|
is_damaged = self._is_damaged_value(
|
|
self._first_present(detail.get("isDamageCase"), detail.get("hasDamage")),
|
|
damage_text,
|
|
)
|
|
|
|
owners_text = self._text(
|
|
self._mapping_value(attrs, "numPreviousOwners", "numberOfPreviousOwners")
|
|
)
|
|
one_owner = self._is_one_owner(owners_text)
|
|
|
|
return CarRecord(
|
|
parser_id=self._parser_id(origin_id),
|
|
brand=brand[:50] or "UNKNOWN",
|
|
model=model[:50] or "UNKNOWN",
|
|
year=year,
|
|
price=price_amount,
|
|
currency=price_currency or "EUR",
|
|
mileage=mileage,
|
|
country=self._normalize_country(contact.get("countryCode") or contact.get("country") or "DE"),
|
|
is_sold=False,
|
|
color=color,
|
|
drive=drive,
|
|
gearbox=gearbox,
|
|
steering_wheel=self._normalize_steering(
|
|
" ".join(
|
|
str(part)
|
|
for part in [
|
|
self._mapping_value(attrs, "steeringWheel", "steeringPosition", "steering"),
|
|
detail.get("steeringWheel"),
|
|
detail.get("steeringPosition"),
|
|
detail.get("steering"),
|
|
feature_text,
|
|
]
|
|
if part is not None and str(part).strip()
|
|
),
|
|
right_hand_drive=self._first_present(
|
|
detail.get("rightHandDrive"),
|
|
self._mapping_value(attrs, "rightHandDrive"),
|
|
),
|
|
),
|
|
body_type=body_type,
|
|
engine_volume=self._engine_volume_from_candidates(
|
|
self._mapping_value(attrs, "cubicCapacity", "cc", "engineVolume", "displacement", "engineDisplacement"),
|
|
detail.get("cubicCapacity"),
|
|
detail.get("cc"),
|
|
detail.get("engineVolume"),
|
|
detail.get("displacement"),
|
|
detail.get("engineDisplacement"),
|
|
self._find_first_value(detail, _ENGINE_VOLUME_VALUE_KEYS),
|
|
subtitle,
|
|
short_title,
|
|
),
|
|
selling_type="STOCK",
|
|
one_owner=one_owner,
|
|
new_car=self._bool_value(detail.get("isConditionNew")),
|
|
is_hidden=False,
|
|
origin="MOBILEDE",
|
|
origin_url=MobileDeClient.build_transport_detail_url(listing_id),
|
|
origin_id=origin_id,
|
|
is_damaged=is_damaged,
|
|
evaluation=self._rating_text(detail.get("priceRating") or detail.get("rating")),
|
|
non_smoking=self._contains_marker(feature_text, "машина некурящего водителя"),
|
|
rental=self._contains_marker(
|
|
feature_text,
|
|
"прокатный автомобиль",
|
|
"арендный автомобиль",
|
|
"rental vehicle",
|
|
),
|
|
repair_history=self._contains_marker(
|
|
damage_text,
|
|
"устраненные повреждения",
|
|
"устранённые повреждения",
|
|
"repaired damage",
|
|
),
|
|
slug=self._slugify(" ".join(part for part in [title or f"{brand} {model}", str(year) if year is not None else ""] if part)),
|
|
last_seen_at=datetime.now(timezone.utc),
|
|
details_confirmed=True,
|
|
images_confirmed=True,
|
|
preserve_existing_details=False,
|
|
images=self._images_from_detail(detail),
|
|
)
|
|
|
|
@staticmethod
|
|
def origin_id(listing_id: str) -> str:
|
|
return f"mobile.de:{listing_id}"
|
|
|
|
@staticmethod
|
|
def _parser_id(origin_id: str) -> str:
|
|
number = int.from_bytes(hashlib.sha256(origin_id.encode("utf-8")).digest()[:17], "big")
|
|
chars: list[str] = []
|
|
for _ in range(22):
|
|
number, index = divmod(number, len(PARSER_ID_ALPHABET))
|
|
chars.append(PARSER_ID_ALPHABET[index])
|
|
return "car-" + "".join(chars)
|
|
|
|
@staticmethod
|
|
def _text(value: Any) -> str:
|
|
return "" if value is None else str(value).strip()
|
|
|
|
@staticmethod
|
|
def _first_present(*values: Any) -> Any:
|
|
for value in values:
|
|
if value is None:
|
|
continue
|
|
if isinstance(value, str) and not value.strip():
|
|
continue
|
|
return value
|
|
return None
|
|
|
|
@staticmethod
|
|
def _normalized_key(value: Any) -> str:
|
|
return re.sub(r"[^a-z0-9]", "", str(value or "").lower())
|
|
|
|
@classmethod
|
|
def _find_first_value(cls, value: Any, keys: set[str], *, depth: int = 0) -> Any:
|
|
if depth > 6:
|
|
return None
|
|
if isinstance(value, dict):
|
|
for key, item in value.items():
|
|
if cls._normalized_key(key) in keys and cls._first_present(item) is not None:
|
|
if not isinstance(item, (dict, list)):
|
|
return item
|
|
for item in value.values():
|
|
found = cls._find_first_value(item, keys, depth=depth + 1)
|
|
if cls._first_present(found) is not None:
|
|
return found
|
|
elif isinstance(value, list):
|
|
for item in value:
|
|
found = cls._find_first_value(item, keys, depth=depth + 1)
|
|
if cls._first_present(found) is not None:
|
|
return found
|
|
return None
|
|
|
|
@classmethod
|
|
def _mapping_value(cls, mapping: Any, *keys: str) -> Any:
|
|
if not isinstance(mapping, dict):
|
|
return None
|
|
for key in keys:
|
|
if key in mapping and cls._first_present(mapping.get(key)) is not None:
|
|
return mapping.get(key)
|
|
normalized_keys = {cls._normalized_key(key) for key in keys if key}
|
|
for existing_key, value in mapping.items():
|
|
if cls._normalized_key(existing_key) in normalized_keys and cls._first_present(value) is not None:
|
|
return value
|
|
return None
|
|
|
|
@classmethod
|
|
def _money_to_int(cls, value: Any) -> int | None:
|
|
if isinstance(value, dict):
|
|
amount = ((value.get("grs") or {}).get("amount") if isinstance(value.get("grs"), dict) else None) or value.get("amount")
|
|
return cls._int_from_text(amount)
|
|
return cls._int_from_text(value)
|
|
|
|
@staticmethod
|
|
def _int_from_text(value: Any) -> int | None:
|
|
if value is None:
|
|
return None
|
|
if isinstance(value, (int, float)) and not isinstance(value, bool):
|
|
return int(value)
|
|
digits = re.sub(r"[^0-9]", "", str(value))
|
|
return int(digits) if digits else None
|
|
|
|
@staticmethod
|
|
def _year_from_first_registration(value: Any) -> int | None:
|
|
text = "" if value is None else str(value)
|
|
match = re.search(r"(19|20)\d{2}", text)
|
|
return int(match.group(0)) if match else None
|
|
|
|
@staticmethod
|
|
def _brand_from_title(title: str | None) -> str | None:
|
|
if not title:
|
|
return None
|
|
return title.split()[0]
|
|
|
|
@staticmethod
|
|
def _model_from_title(title: str | None, brand: str) -> str | None:
|
|
if not title:
|
|
return None
|
|
rest = title.replace(brand, "", 1).strip()
|
|
return rest or None
|
|
|
|
@staticmethod
|
|
def _normalize_gearbox(value: Any) -> str | None:
|
|
text = "" if value is None else str(value).lower()
|
|
for marker, mapped in _RU_GEARBOX_MAP.items():
|
|
if marker in text:
|
|
return mapped
|
|
return None
|
|
|
|
@staticmethod
|
|
def _normalize_drive(value: Any) -> str | None:
|
|
text = "" if value is None else str(value).lower()
|
|
for mapped, markers in _RU_DRIVE_MARKERS.items():
|
|
if any(marker in text for marker in markers):
|
|
return mapped
|
|
return None
|
|
|
|
@staticmethod
|
|
def _normalize_search_ssr_drive(value: Any) -> str | None:
|
|
"""Map only explicit, high-confidence drive tokens rendered by Search SSR."""
|
|
text = "" if value is None else str(value)
|
|
for mapped, pattern in _SEARCH_SSR_DRIVE_MARKERS:
|
|
if pattern.search(text):
|
|
return mapped
|
|
return None
|
|
|
|
@classmethod
|
|
def _normalize_steering(cls, value: Any, *, right_hand_drive: Any = None) -> str | None:
|
|
if right_hand_drive is not None and cls._bool_value(right_hand_drive):
|
|
return "RIGHT"
|
|
text = "" if value is None else str(value).lower()
|
|
if any(marker in text for marker in ("правый руль", "праворуль")):
|
|
return "RIGHT"
|
|
if any(marker in text for marker in ("левый руль", "леворуль")):
|
|
return "LEFT"
|
|
return None
|
|
|
|
@staticmethod
|
|
def _normalize_country(value: Any) -> str:
|
|
text = "" if value is None else str(value).strip().upper()
|
|
if text in {"DE", "ГЕРМАНИЯ"}:
|
|
return "DE"
|
|
if text in {"US", "USA", "UNITED STATES"}:
|
|
return "US"
|
|
if text in {"CA", "CANADA"}:
|
|
return "CA"
|
|
if text in {"IT", "ITALY", "ITALIA"}:
|
|
return "IT"
|
|
if text in {"JP", "JAPAN"}:
|
|
return "JP"
|
|
if text in {"KR", "KOREA", "SOUTH KOREA"}:
|
|
return "KR"
|
|
if re.fullmatch(r"[A-Z]{2}", text):
|
|
return text
|
|
return "NA"
|
|
|
|
@staticmethod
|
|
def _normalize_body(value: Any) -> str:
|
|
text = "" if value is None else str(value).lower()
|
|
if re.search(r"\bbmw\s+(?:x(?:[1-7]|m)|ix(?:[1-3])?|xm)\b", text):
|
|
return "SUV"
|
|
for marker, mapped in _RU_BODY_MAP.items():
|
|
if marker in text:
|
|
return mapped
|
|
return "OTHER"
|
|
|
|
@staticmethod
|
|
def _contains_marker(value: Any, *markers: str) -> bool:
|
|
text = "" if value is None else str(value).lower()
|
|
return any(marker.lower() in text for marker in markers)
|
|
|
|
@classmethod
|
|
def _normalize_body_from_candidates(cls, *values: Any) -> str:
|
|
for value in values:
|
|
normalized = cls._normalize_body(value)
|
|
if normalized != "OTHER":
|
|
return normalized
|
|
return "OTHER"
|
|
|
|
@classmethod
|
|
def _engine_volume_from_candidates(cls, *values: Any) -> int | None:
|
|
fallback_texts: list[str] = []
|
|
free_text_start = max(0, len(values) - 2)
|
|
for index, value in enumerate(values):
|
|
if value is None:
|
|
continue
|
|
if isinstance(value, str):
|
|
text = value.strip()
|
|
if not text:
|
|
continue
|
|
fallback_texts.append(text)
|
|
if index < free_text_start and re.fullmatch(r"[\d\s.,]+", text):
|
|
parsed = cls._int_from_text(text)
|
|
if parsed and 500 <= parsed <= 9000:
|
|
return parsed
|
|
if re.search(r"(ccm|cm3|cm³|\bcc\b|cubic|displacement)", text, re.IGNORECASE):
|
|
parsed = cls._engine_volume_from_cc_text(text)
|
|
if parsed:
|
|
return parsed
|
|
if re.search(r"\b(?:liter|litre|l)\b", text, re.IGNORECASE):
|
|
parsed = cls._engine_volume_from_liter_text(text)
|
|
if parsed:
|
|
return parsed
|
|
continue
|
|
parsed = cls._int_from_text(value)
|
|
if parsed and 500 <= parsed <= 9000:
|
|
return parsed
|
|
|
|
for text in fallback_texts:
|
|
match = re.search(r"(?<![\w])([1-6])[\.,](\d{1,2})(?!\d)", text.lower())
|
|
if match:
|
|
liters = float(f"{match.group(1)}.{match.group(2)}")
|
|
return int(round(liters * 1000))
|
|
return None
|
|
|
|
@classmethod
|
|
def _engine_volume_from_cc_text(cls, value: str) -> int | None:
|
|
text = str(value or "")
|
|
match = re.search(r"(\d{1,2}(?:[\s.,]\d{3})|\d{3,5})\s*(?:ccm|cm3|cm³|cc)", text, re.IGNORECASE)
|
|
if not match:
|
|
return None
|
|
parsed = cls._int_from_text(match.group(1))
|
|
if parsed and 500 <= parsed <= 9000:
|
|
return parsed
|
|
return None
|
|
|
|
@classmethod
|
|
def _engine_volume_from_liter_text(cls, value: str) -> int | None:
|
|
text = str(value or "")
|
|
match = re.search(r"(?<!\d)([1-8])(?:[\.,](\d{1,2}))?\s*(?:l|liter|litre)(?![a-z])", text, re.IGNORECASE)
|
|
if not match:
|
|
return None
|
|
decimals = (match.group(2) or "0").ljust(1, "0")
|
|
liters = float(f"{match.group(1)}.{decimals}")
|
|
parsed = int(round(liters * 1000))
|
|
if 500 <= parsed <= 9000:
|
|
return parsed
|
|
return None
|
|
|
|
@staticmethod
|
|
def _normalize_color(value: Any) -> str:
|
|
text = "" if value is None else str(value).lower().strip()
|
|
text = re.sub(r"[_\-/]+", " ", text)
|
|
for marker, mapped in _RU_COLOR_MAP.items():
|
|
if marker in text:
|
|
return mapped
|
|
return "other"
|
|
|
|
@staticmethod
|
|
def _is_one_owner(value: Any) -> bool:
|
|
text = "" if value is None else str(value).strip().lower()
|
|
return text in {"1", "01", "1.0", "one", "one owner", "1 owner", "1 previous owner"}
|
|
|
|
@staticmethod
|
|
def _bool_value(value: Any) -> bool:
|
|
if isinstance(value, bool):
|
|
return value
|
|
if isinstance(value, (int, float)):
|
|
return value != 0
|
|
text = "" if value is None else str(value).strip().lower()
|
|
if text in {"1", "true", "yes", "on", "да"}:
|
|
return True
|
|
if text in {"0", "false", "no", "off", "нет", ""}:
|
|
return False
|
|
return False
|
|
|
|
@classmethod
|
|
def _is_damaged_value(cls, flag: Any, description: Any) -> bool:
|
|
if cls._bool_value(flag):
|
|
return True
|
|
text = cls._text(description).lower()
|
|
if not text:
|
|
return False
|
|
if any(marker in text for marker in ("без дтп", "без аварий", "no accident", "accident free")):
|
|
return False
|
|
return any(
|
|
marker in text
|
|
for marker in (
|
|
"дтп",
|
|
"аварийн",
|
|
"с повреждениями",
|
|
"устраненные повреждения",
|
|
"устранённые повреждения",
|
|
"accident",
|
|
"damaged",
|
|
)
|
|
)
|
|
|
|
@staticmethod
|
|
def _rating_text(value: Any) -> str | None:
|
|
if isinstance(value, dict):
|
|
for key in ("rating", "ratingLabel", "label", "value"):
|
|
text = MobileDeMapper._text(value.get(key))
|
|
if text:
|
|
return text
|
|
return None
|
|
text = MobileDeMapper._text(value)
|
|
return text or None
|
|
|
|
@staticmethod
|
|
def _slugify(value: str) -> str:
|
|
slug = re.sub(r"[^a-zA-Z0-9]+", "-", value.lower()).strip("-")
|
|
return slug[:180] or "mobilede-car"
|
|
|
|
@staticmethod
|
|
def _detail_attrs_by_tag(value: Any) -> dict[str, str]:
|
|
result: dict[str, str] = {}
|
|
if not isinstance(value, list):
|
|
return result
|
|
for item in value:
|
|
if not isinstance(item, dict):
|
|
continue
|
|
tag = str(item.get("tag") or "").strip()
|
|
val = str(item.get("value") or "").strip()
|
|
if tag and val and tag not in result:
|
|
result[tag] = val
|
|
return result
|
|
|
|
@staticmethod
|
|
def _detail_price(value: Any) -> tuple[int | None, str | None]:
|
|
if not isinstance(value, dict):
|
|
return None, None
|
|
grs = value.get("grs") if isinstance(value.get("grs"), dict) else {}
|
|
amount = grs.get("amount") if isinstance(grs, dict) else None
|
|
currency = grs.get("currency") if isinstance(grs, dict) else None
|
|
if amount is None:
|
|
amount = value.get("amount")
|
|
if currency is None:
|
|
currency = value.get("currency")
|
|
return MobileDeMapper._int_from_text(amount), (str(currency).strip() if currency else None)
|
|
|
|
@staticmethod
|
|
def _images_from_listing(raw: dict[str, Any]) -> list[ImageRecord]:
|
|
source_urls = dict.fromkeys(MobileDeMapper._extract_image_urls(raw))
|
|
return [
|
|
ImageRecord(
|
|
fullres_image=MobileDeMapper._with_mobilede_image_rule(
|
|
source_url,
|
|
_MOBILEDE_FULLRES_IMAGE_RULE,
|
|
),
|
|
preview_image=MobileDeMapper._with_mobilede_image_rule(
|
|
source_url,
|
|
_MOBILEDE_PREVIEW_IMAGE_RULE,
|
|
),
|
|
order_index=index,
|
|
)
|
|
for index, source_url in enumerate(source_urls)
|
|
]
|
|
|
|
@staticmethod
|
|
def _images_from_detail(raw: dict[str, Any]) -> list[ImageRecord]:
|
|
explicit_images = raw.get("images")
|
|
if isinstance(explicit_images, list) and explicit_images:
|
|
return MobileDeMapper._images_from_listing({"images": explicit_images})
|
|
media_gallery = raw.get("mediaGallery")
|
|
if isinstance(media_gallery, list) and media_gallery:
|
|
return MobileDeMapper._images_from_listing({"mediaGallery": media_gallery})
|
|
return MobileDeMapper._images_from_listing(raw)
|
|
|
|
def detail_to_images(self, raw: dict[str, Any]) -> list[ImageRecord]:
|
|
"""Extract the complete gallery without mapping Detail car fields."""
|
|
return self._images_from_detail(raw)
|
|
|
|
@staticmethod
|
|
def _extract_image_urls(value: Any, *, parent_key: str = "", depth: int = 0) -> list[str]:
|
|
if depth > 8:
|
|
return []
|
|
urls: list[str] = []
|
|
parent_hint = MobileDeMapper._has_image_field_hint(parent_key)
|
|
if isinstance(value, str):
|
|
if parent_hint or MobileDeMapper._looks_like_image_url(value):
|
|
normalized = MobileDeMapper._normalize_image_url(value)
|
|
if normalized:
|
|
urls.append(normalized)
|
|
return urls
|
|
if isinstance(value, list):
|
|
for item in value:
|
|
urls.extend(MobileDeMapper._extract_image_urls(item, parent_key=parent_key, depth=depth + 1))
|
|
return urls
|
|
if isinstance(value, dict):
|
|
for key, item in value.items():
|
|
key_text = str(key or "")
|
|
urls.extend(MobileDeMapper._extract_image_urls(item, parent_key=key_text, depth=depth + 1))
|
|
return urls
|
|
return urls
|
|
|
|
@staticmethod
|
|
def _has_image_field_hint(value: str) -> bool:
|
|
normalized = re.sub(r"[^a-z0-9]", "", str(value or "").lower())
|
|
return normalized in _IMAGE_FIELD_HINTS or any(hint in normalized for hint in _IMAGE_FIELD_HINTS)
|
|
|
|
@staticmethod
|
|
def _looks_like_image_url(value: str) -> bool:
|
|
return bool(MobileDeMapper._normalize_image_url(value))
|
|
|
|
@staticmethod
|
|
def _normalize_image_url(value: str) -> str:
|
|
url = str(value).strip()
|
|
if not url:
|
|
return ""
|
|
if url.startswith("//"):
|
|
url = f"https:{url}"
|
|
elif url.startswith(_MOBILEDE_IMAGE_HOST):
|
|
url = f"https://{url}"
|
|
elif not (url.startswith("http://") or url.startswith("https://")):
|
|
return ""
|
|
|
|
parsed = urlsplit(url)
|
|
if parsed.netloc.lower() != _MOBILEDE_IMAGE_HOST:
|
|
return ""
|
|
if not parsed.path.startswith(_MOBILEDE_IMAGE_PATH_PREFIX):
|
|
return ""
|
|
|
|
image_id = parsed.path[len(_MOBILEDE_IMAGE_PATH_PREFIX):].strip("/")
|
|
if not _MOBILEDE_IMAGE_ID_RE.fullmatch(image_id):
|
|
return ""
|
|
|
|
query_pairs = [
|
|
(key, query_value)
|
|
for key, query_value in parse_qsl(parsed.query, keep_blank_values=True)
|
|
if key.lower() != "rule"
|
|
]
|
|
return urlunsplit(("https", _MOBILEDE_IMAGE_HOST, parsed.path, urlencode(query_pairs), ""))
|
|
|
|
@staticmethod
|
|
def _with_mobilede_image_rule(url: str, rule: str) -> str:
|
|
parsed = urlsplit(url)
|
|
query_pairs = [
|
|
(key, query_value)
|
|
for key, query_value in parse_qsl(parsed.query, keep_blank_values=True)
|
|
if key.lower() != "rule"
|
|
]
|
|
query_pairs.append(("rule", rule))
|
|
return urlunsplit((parsed.scheme, parsed.netloc, parsed.path, urlencode(query_pairs), parsed.fragment))
|