344 lines
13 KiB
Python
344 lines
13 KiB
Python
from __future__ import annotations
|
||
|
||
import hashlib
|
||
import re
|
||
from datetime import datetime, timezone
|
||
from typing import Any
|
||
|
||
from ..storage.schemas import CarRecord, ImageRecord
|
||
from .client import MobileDeClient
|
||
from .models import MobileDeListing
|
||
|
||
_BODY_MAP = {
|
||
"cabrio": "OPEN",
|
||
"roadster": "OPEN",
|
||
"кабрио": "OPEN",
|
||
"limousine": "SEDAN",
|
||
"sedan": "SEDAN",
|
||
"сeдан": "SEDAN",
|
||
"suv": "SUV",
|
||
"внедорож": "SUV",
|
||
"kombi": "STATION_WAGON",
|
||
"estate": "STATION_WAGON",
|
||
"универсал": "STATION_WAGON",
|
||
"van": "MINIVAN",
|
||
"минивэн": "MINIVAN",
|
||
"coupe": "COUPE",
|
||
"купе": "COUPE",
|
||
"hatchback": "HATCHBACK",
|
||
"kleinwagen": "HATCHBACK",
|
||
}
|
||
|
||
_GEARBOX_MAP = {
|
||
"автомат": "AT",
|
||
"automatik": "AT",
|
||
"automatic": "AT",
|
||
"механ": "MT",
|
||
"manual": "MT",
|
||
"cvt": "CVT",
|
||
}
|
||
|
||
_COLOR_MAP = {
|
||
"schwarz": "black",
|
||
"черн": "black",
|
||
"black": "black",
|
||
"weiss": "white",
|
||
"weiß": "white",
|
||
"бел": "white",
|
||
"white": "white",
|
||
"grau": "gray",
|
||
"сер": "gray",
|
||
"gray": "gray",
|
||
"silber": "silver",
|
||
"сереб": "silver",
|
||
"silver": "silver",
|
||
"rot": "red",
|
||
"красн": "red",
|
||
"red": "red",
|
||
"blau": "blue",
|
||
"син": "blue",
|
||
"blue": "blue",
|
||
"grün": "green",
|
||
"gruen": "green",
|
||
"зелен": "green",
|
||
"green": "green",
|
||
}
|
||
|
||
|
||
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)
|
||
|
||
return CarRecord(
|
||
parser_id=self._parser_id(origin_id),
|
||
brand=brand[:50] or "UNKNOWN",
|
||
model=model[:50] or "UNKNOWN",
|
||
year=self._year_from_first_registration(listing.first_registration or attr.get("fr")),
|
||
price=self._money_to_int(listing.price or raw.get("p")),
|
||
currency="EUR",
|
||
mileage=self._int_from_text(listing.mileage or attr.get("ml")) or 0,
|
||
country=self._normalize_country("DE"),
|
||
is_sold=False,
|
||
color=self._normalize_color(attr.get("ecol")),
|
||
drive=None,
|
||
gearbox=self._normalize_gearbox(listing.transmission or attr.get("tr")),
|
||
steering_wheel="LEFT",
|
||
body_type=self._normalize_body(attr.get("c")),
|
||
engine_volume=self._int_from_text(attr.get("cc")),
|
||
selling_type="CLASSIFIED",
|
||
one_owner=(str(attr.get("pvo") or "").strip() == "1"),
|
||
new_car=False,
|
||
is_hidden=False,
|
||
origin="MOBILE_DE",
|
||
origin_url=listing.url,
|
||
origin_id=origin_id,
|
||
is_damaged=bool(raw.get("hasDamage")),
|
||
evaluation=self._text(raw.get("priceRating") or raw.get("rating")) or None,
|
||
non_smoking=True,
|
||
rental=False,
|
||
repair_history=bool(raw.get("hasDamage")),
|
||
slug=self._slugify(title or f"{brand} {model}"),
|
||
last_seen_at=datetime.now(timezone.utc),
|
||
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(attrs.get("mileage")) or 0
|
||
year = self._year_from_first_registration(attrs.get("firstRegistration"))
|
||
gearbox = self._normalize_gearbox(attrs.get("transmission"))
|
||
body_type = self._normalize_body(attrs.get("category") or detail.get("category"))
|
||
color = self._normalize_color(attrs.get("color") or attrs.get("manufacturerColorName"))
|
||
|
||
damage_text = self._text(attrs.get("damageCondition")).lower()
|
||
is_damaged = ("дтп" in damage_text and "без дтп" not in damage_text) or bool(detail.get("hasDamage"))
|
||
|
||
owners_text = self._text(attrs.get("numPreviousOwners"))
|
||
one_owner = owners_text in {"1", "01", "1.0"}
|
||
|
||
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=self._normalize_drive(attrs.get("wheelDrive") or attrs.get("drivetrain")),
|
||
gearbox=gearbox,
|
||
steering_wheel="LEFT",
|
||
body_type=body_type,
|
||
engine_volume=self._int_from_text(attrs.get("cubicCapacity") or attrs.get("cc")),
|
||
selling_type="CLASSIFIED",
|
||
one_owner=one_owner,
|
||
new_car=bool(detail.get("isNew") or detail.get("isConditionNew")),
|
||
is_hidden=False,
|
||
origin="MOBILE_DE",
|
||
origin_url=MobileDeClient.build_detail_url(listing_id),
|
||
origin_id=origin_id,
|
||
is_damaged=is_damaged,
|
||
evaluation=self._text(detail.get("priceRating") or detail.get("rating")) or None,
|
||
non_smoking=True,
|
||
rental=False,
|
||
repair_history=is_damaged,
|
||
slug=self._slugify(title or f"{brand} {model}"),
|
||
last_seen_at=datetime.now(timezone.utc),
|
||
images=self._images_from_listing(detail),
|
||
)
|
||
|
||
@staticmethod
|
||
def origin_id(listing_id: str) -> str:
|
||
return f"mobile.de:{listing_id}"
|
||
|
||
@staticmethod
|
||
def _parser_id(origin_id: str) -> str:
|
||
digest = hashlib.sha1(origin_id.encode("utf-8")).hexdigest()[:16]
|
||
return f"mobilede-{digest}"
|
||
|
||
@staticmethod
|
||
def _text(value: Any) -> str:
|
||
return "" if value is None else str(value).strip()
|
||
|
||
@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 _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()
|
||
if any(marker in text for marker in ("front", "fwd", "перед")):
|
||
return "FWD"
|
||
if any(marker in text for marker in ("rear", "rwd", "зад")):
|
||
return "RWD"
|
||
if any(marker in text for marker in ("all", "awd", "4x4", "quattro", "полный")):
|
||
return "4WD"
|
||
return None
|
||
|
||
@staticmethod
|
||
def _normalize_country(value: Any) -> str:
|
||
text = "" if value is None else str(value).strip().upper()
|
||
if text in {"DE", "GERMANY", "DEUTSCHLAND"}:
|
||
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"
|
||
return "NA"
|
||
|
||
@staticmethod
|
||
def _normalize_body(value: Any) -> str:
|
||
text = "" if value is None else str(value).lower()
|
||
for marker, mapped in _BODY_MAP.items():
|
||
if marker in text:
|
||
return mapped
|
||
return "OTHER"
|
||
|
||
@staticmethod
|
||
def _normalize_color(value: Any) -> str:
|
||
text = "" if value is None else str(value).lower().strip()
|
||
for marker, mapped in _COLOR_MAP.items():
|
||
if marker in text:
|
||
return mapped
|
||
return text[:50] if text else "other"
|
||
|
||
@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]:
|
||
urls: list[str] = []
|
||
image = raw.get("image")
|
||
if isinstance(image, str):
|
||
urls.append(MobileDeMapper._normalize_image_url(image))
|
||
images = raw.get("images")
|
||
if isinstance(images, list):
|
||
for item in images:
|
||
if isinstance(item, str):
|
||
urls.append(MobileDeMapper._normalize_image_url(item))
|
||
elif isinstance(item, dict):
|
||
src = item.get("src") or item.get("url") or item.get("uri")
|
||
if src:
|
||
urls.append(MobileDeMapper._normalize_image_url(str(src)))
|
||
media_gallery = raw.get("mediaGallery")
|
||
if isinstance(media_gallery, list):
|
||
for item in media_gallery:
|
||
if isinstance(item, dict):
|
||
src = item.get("uri") or item.get("url")
|
||
if src:
|
||
urls.append(MobileDeMapper._normalize_image_url(str(src)))
|
||
return [
|
||
ImageRecord(fullres_image=url, preview_image=url, order_index=index)
|
||
for index, url in enumerate(dict.fromkeys(url for url in urls if url))
|
||
]
|
||
|
||
@staticmethod
|
||
def _normalize_image_url(value: str) -> str:
|
||
url = str(value).strip()
|
||
if not url:
|
||
return ""
|
||
if url.startswith("//"):
|
||
return f"https:{url}"
|
||
if url.startswith("http://") or url.startswith("https://"):
|
||
return url
|
||
return f"https://{url.lstrip('/')}"
|