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", "кабриолет": "OPEN", "limousine": "SEDAN", "седан": "SEDAN", "suv": "SUV", "внедорожник": "SUV", "kombi": "STATION_WAGON", "универсал": "STATION_WAGON", "van": "MINIVAN", "фургон": "MINIVAN", "coupe": "COUPE", "купе": "COUPE", "kleinwagen": "HATCHBACK", "маленький": "HATCHBACK", } _GEARBOX_MAP = { "автомат": "AT", "automatic": "AT", "механ": "MT", "manual": "MT", "cvt": "CVT", } _COLOR_MAP = { "schwarz": "black", "черный": "black", "weiß": "white", "weiss": "white", "белый": "white", "серый": "gray", "grau": "gray", "silber": "silver", "сереб": "silver", "rot": "red", "красный": "red", "blau": "blue", "синий": "blue", "grün": "green", "gruen": "green", "зеленый": "green", } class MobileDeMapper: """Map mobile.de search/detail payloads into the existing CarRecord schema.""" 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="NA", 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: title = self._text(detail.get("shortTitle") or detail.get("make") or "UNKNOWN") subtitle = self._text(detail.get("subTitle")) fake_listing = MobileDeListing( id=str(listing_id), url=MobileDeClient.build_detail_url(listing_id), title=title, subtitle=subtitle, price=self._text(detail.get("price") or detail.get("p")), raw=detail, ) return self.listing_to_car_record(fake_listing) @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: 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_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 _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))) 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('/')}"