Fix parser identifiers and locale
This commit is contained in:
qananasikq
@@ -17,9 +17,9 @@ from .models import MobileDeListing, MobileDeSearchPage
|
||||
|
||||
logger = logging.getLogger("mobile_de.client")
|
||||
|
||||
BASE_URL = "https://www.mobile.de"
|
||||
SEARCH_PATH = "/ru/транспортные-средства/поиск.html"
|
||||
DETAIL_PATH = "/ru/транспортные-средства/подробности.html"
|
||||
BASE_URL = "https://suchen.mobile.de"
|
||||
SEARCH_PATH = "/fahrzeuge/search.html"
|
||||
DETAIL_PATH = "/fahrzeuge/details.html"
|
||||
DEFAULT_HEADERS = {
|
||||
"user-agent": (
|
||||
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
|
||||
@@ -27,7 +27,7 @@ DEFAULT_HEADERS = {
|
||||
"Chrome/124.0.0.0 Safari/537.36"
|
||||
),
|
||||
"accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
|
||||
"accept-language": "ru,en;q=0.9,de;q=0.8",
|
||||
"accept-language": "en-US,en;q=0.9",
|
||||
}
|
||||
MOBILEDE_HTTP_MAX_RETRIES = max(0, int(os.getenv("MOBILEDE_HTTP_MAX_RETRIES", "4")))
|
||||
MOBILEDE_HTTP_BACKOFF_BASE_SECONDS = max(0.0, float(os.getenv("MOBILEDE_HTTP_BACKOFF_BASE_SECONDS", "1.2")))
|
||||
@@ -103,6 +103,7 @@ class MobileDeClient:
|
||||
"od": "up",
|
||||
"vc": "Car",
|
||||
"s": "Car",
|
||||
"lang": "en",
|
||||
"pageNumber": page_number,
|
||||
}
|
||||
query.update({key: value for key, value in params.items() if value is not None})
|
||||
@@ -119,19 +120,17 @@ class MobileDeClient:
|
||||
query_items = [
|
||||
(key, value)
|
||||
for key, value in parse_qsl(parts.query, keep_blank_values=True)
|
||||
if key != "pageNumber" and key not in params
|
||||
if key != "pageNumber" and key != "lang" and key not in params
|
||||
]
|
||||
if page_number is not None:
|
||||
query_items.append(("pageNumber", str(page_number)))
|
||||
query_items.extend((key, str(value)) for key, value in params.items() if value is not None)
|
||||
scheme = parts.scheme or "https"
|
||||
netloc = parts.netloc or urlsplit(BASE_URL).netloc
|
||||
path = parts.path or SEARCH_PATH
|
||||
return urlunsplit((scheme, netloc, path, urlencode(query_items), ""))
|
||||
query_items.append(("lang", "en"))
|
||||
return urlunsplit(("https", urlsplit(BASE_URL).netloc, SEARCH_PATH, urlencode(query_items), ""))
|
||||
|
||||
@staticmethod
|
||||
def build_detail_url(listing_id: str | int) -> str:
|
||||
query = urlencode({"id": listing_id, "vc": "Car", "s": "Car"})
|
||||
query = urlencode({"id": listing_id, "vc": "Car", "s": "Car", "lang": "en"})
|
||||
return f"{BASE_URL}{DETAIL_PATH}?{query}"
|
||||
|
||||
def fetch_html(self, url: str, *, timeout: int = 30) -> str:
|
||||
|
||||
@@ -3,6 +3,7 @@
|
||||
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
|
||||
|
||||
@@ -10,45 +11,36 @@ 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}$")
|
||||
|
||||
_BODY_MAP = {
|
||||
"cabrio": "OPEN",
|
||||
"cabriolet": "OPEN",
|
||||
"кабриолет": "OPEN",
|
||||
"roadster": "OPEN",
|
||||
"кабрио": "OPEN",
|
||||
"limousine": "SEDAN",
|
||||
"saloon": "SEDAN",
|
||||
"sedan": "SEDAN",
|
||||
"седан": "SEDAN",
|
||||
"suv": "SUV",
|
||||
"offroad": "SUV",
|
||||
"gelandewagen": "SUV",
|
||||
"geländewagen": "SUV",
|
||||
"pickup": "PICKUP",
|
||||
"pick-up": "PICKUP",
|
||||
"пикап": "PICKUP",
|
||||
"внедорож": "SUV",
|
||||
"kombi": "STATION_WAGON",
|
||||
"estatecar": "STATION_WAGON",
|
||||
"touring": "STATION_WAGON",
|
||||
"estate": "STATION_WAGON",
|
||||
"универсал": "STATION_WAGON",
|
||||
"van": "MINIVAN",
|
||||
"kleinbus": "MINIVAN",
|
||||
"bus": "MINIVAN",
|
||||
"active tourer": "MINIVAN",
|
||||
"gran tourer": "MINIVAN",
|
||||
"минивэн": "MINIVAN",
|
||||
"фургон": "MINIVAN",
|
||||
"coupe": "COUPE",
|
||||
"sportscar": "COUPE",
|
||||
"sports car": "COUPE",
|
||||
"купе": "COUPE",
|
||||
"hatchback": "HATCHBACK",
|
||||
"kleinwagen": "HATCHBACK",
|
||||
"кляйнваген": "HATCHBACK",
|
||||
"малолитраж": "HATCHBACK",
|
||||
"хэтч": "HATCHBACK",
|
||||
"smallcar": "HATCHBACK",
|
||||
"small car": "HATCHBACK",
|
||||
"compact": "HATCHBACK",
|
||||
@@ -56,10 +48,8 @@ _BODY_MAP = {
|
||||
}
|
||||
|
||||
_GEARBOX_MAP = {
|
||||
"автомат": "AT",
|
||||
"automatik": "AT",
|
||||
"automatic": "AT",
|
||||
"механ": "MT",
|
||||
"manual": "MT",
|
||||
"cvt": "CVT",
|
||||
}
|
||||
@@ -71,18 +61,15 @@ _COLOR_MAP = {
|
||||
"obsidianschwarz": "black",
|
||||
"jet black": "black",
|
||||
"jetblack": "black",
|
||||
"черн": "black",
|
||||
"black": "black",
|
||||
"weiss": "white",
|
||||
"weiß": "white",
|
||||
"alpinweiss": "white",
|
||||
"alpine white": "white",
|
||||
"mineralweiss": "white",
|
||||
"бел": "white",
|
||||
"white": "white",
|
||||
"silber": "silver",
|
||||
"argent": "silver",
|
||||
"сереб": "silver",
|
||||
"silver": "silver",
|
||||
"grau": "gray",
|
||||
"grey": "gray",
|
||||
@@ -95,25 +82,20 @@ _COLOR_MAP = {
|
||||
"brooklyn grey": "gray",
|
||||
"sophistograu": "gray",
|
||||
"skyscraper grau": "gray",
|
||||
"сер": "gray",
|
||||
"gray": "gray",
|
||||
"rot": "red",
|
||||
"burgundy": "red",
|
||||
"bordeaux": "red",
|
||||
"maroon": "red",
|
||||
"красн": "red",
|
||||
"red": "red",
|
||||
"blau": "blue",
|
||||
"turquoise": "blue",
|
||||
"cyan": "blue",
|
||||
"син": "blue",
|
||||
"blue": "blue",
|
||||
"grün": "green",
|
||||
"gruen": "green",
|
||||
"зелен": "green",
|
||||
"green": "green",
|
||||
"braun": "brown",
|
||||
"корич": "brown",
|
||||
"brown": "brown",
|
||||
"beige": "beige",
|
||||
"champagner": "beige",
|
||||
@@ -121,19 +103,13 @@ _COLOR_MAP = {
|
||||
"creme": "beige",
|
||||
"cream": "beige",
|
||||
"ivory": "beige",
|
||||
"беж": "beige",
|
||||
"gelb": "yellow",
|
||||
"желт": "yellow",
|
||||
"yellow": "yellow",
|
||||
"orange": "orange",
|
||||
"оранж": "orange",
|
||||
"gold": "gold",
|
||||
"золот": "gold",
|
||||
"bronze": "bronze",
|
||||
"бронз": "bronze",
|
||||
"violett": "purple",
|
||||
"lila": "purple",
|
||||
"фиолет": "purple",
|
||||
"purple": "purple",
|
||||
}
|
||||
|
||||
@@ -199,14 +175,8 @@ class MobileDeMapper:
|
||||
).lower()
|
||||
is_damaged = (
|
||||
bool(raw.get("hasDamage"))
|
||||
or ("дтп" in damage_text and "без дтп" not in damage_text)
|
||||
or ("accident" in damage_text and "no accident" not in damage_text)
|
||||
)
|
||||
|
||||
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(
|
||||
self._first_present(
|
||||
listing.first_registration,
|
||||
@@ -216,7 +186,13 @@ class MobileDeMapper:
|
||||
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(
|
||||
@@ -279,7 +255,7 @@ class MobileDeMapper:
|
||||
non_smoking=True,
|
||||
rental=False,
|
||||
repair_history=is_damaged,
|
||||
slug=self._slugify(title or f"{brand} {model}"),
|
||||
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),
|
||||
images=self._images_from_listing(raw),
|
||||
)
|
||||
@@ -334,7 +310,7 @@ class MobileDeMapper:
|
||||
)
|
||||
|
||||
damage_text = self._text(attrs.get("damageCondition")).lower()
|
||||
is_damaged = ("дтп" in damage_text and "без дтп" not in damage_text) or bool(detail.get("hasDamage"))
|
||||
is_damaged = ("accident" in damage_text and "no accident" not in damage_text) or bool(detail.get("hasDamage"))
|
||||
|
||||
owners_text = self._text(attrs.get("numPreviousOwners"))
|
||||
one_owner = self._is_one_owner(owners_text)
|
||||
@@ -377,7 +353,7 @@ class MobileDeMapper:
|
||||
non_smoking=True,
|
||||
rental=False,
|
||||
repair_history=is_damaged,
|
||||
slug=self._slugify(title or f"{brand} {model}"),
|
||||
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),
|
||||
images=self._images_from_listing(detail),
|
||||
)
|
||||
@@ -388,8 +364,12 @@ class MobileDeMapper:
|
||||
|
||||
@staticmethod
|
||||
def _parser_id(origin_id: str) -> str:
|
||||
digest = hashlib.sha1(origin_id.encode("utf-8")).hexdigest()[:16]
|
||||
return f"mobilede-{digest}"
|
||||
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:
|
||||
@@ -488,11 +468,11 @@ class MobileDeMapper:
|
||||
@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", "перед", "frontantrieb", "vorderrad", "antrieb vorne", "front-wheel", "front wheel")):
|
||||
if any(marker in text for marker in ("front", "fwd", "frontantrieb", "vorderrad", "antrieb vorne", "front-wheel", "front wheel")):
|
||||
return "FWD"
|
||||
if any(marker in text for marker in ("rear", "rwd", "зад", "heckantrieb", "hinterrad", "antrieb hinten", "rear-wheel", "rear wheel")):
|
||||
if any(marker in text for marker in ("rear", "rwd", "heckantrieb", "hinterrad", "antrieb hinten", "rear-wheel", "rear wheel")):
|
||||
return "RWD"
|
||||
if any(marker in text for marker in ("awd", "4wd", "4x4", "quattro", "полный", "xdrive", "4matic", "4motion", "allrad", "all-wheel", "all wheel", "four-wheel", "four wheel")):
|
||||
if any(marker in text for marker in ("awd", "4wd", "4x4", "quattro", "xdrive", "4matic", "4motion", "allrad", "all-wheel", "all wheel", "four-wheel", "four wheel")):
|
||||
return "4WD"
|
||||
return None
|
||||
|
||||
@@ -518,7 +498,6 @@ class MobileDeMapper:
|
||||
@staticmethod
|
||||
def _normalize_body(value: Any) -> str:
|
||||
text = "" if value is None else str(value).lower()
|
||||
text = text.replace("ё", "е")
|
||||
if re.search(r"\bbmw\s+x(?:[1-7]|m)\b", text):
|
||||
return "SUV"
|
||||
for marker, mapped in _BODY_MAP.items():
|
||||
@@ -550,7 +529,7 @@ class MobileDeMapper:
|
||||
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):
|
||||
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
|
||||
@@ -573,7 +552,7 @@ class MobileDeMapper:
|
||||
@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)
|
||||
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))
|
||||
@@ -597,7 +576,6 @@ class MobileDeMapper:
|
||||
@staticmethod
|
||||
def _normalize_color(value: Any) -> str:
|
||||
text = "" if value is None else str(value).lower().strip()
|
||||
text = text.replace("ё", "е")
|
||||
text = text.replace("ä", "a").replace("ö", "o").replace("ü", "u").replace("ß", "ss")
|
||||
text = re.sub(r"[_\-/]+", " ", text)
|
||||
for marker, mapped in _COLOR_MAP.items():
|
||||
@@ -623,7 +601,7 @@ class MobileDeMapper:
|
||||
|
||||
@staticmethod
|
||||
def _slugify(value: str) -> str:
|
||||
slug = re.sub(r"[^a-zA-Z0-9а-яА-ЯёЁ]+", "-", value.lower()).strip("-")
|
||||
slug = re.sub(r"[^a-zA-Z0-9]+", "-", value.lower()).strip("-")
|
||||
return slug[:180] or "mobilede-car"
|
||||
|
||||
@staticmethod
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
import logging
|
||||
import os
|
||||
import re
|
||||
from contextlib import contextmanager
|
||||
from datetime import datetime, timezone
|
||||
from typing import Any, Iterator
|
||||
@@ -25,6 +26,7 @@ CAR_UPDATE_FIELDS = CAR_DB_FIELDS - {"first_seen_at"}
|
||||
_IN_CHUNK_SIZE = 5000
|
||||
CAR_TABLE_NAME = Car.__tablename__
|
||||
MOBILEDE_ORIGIN_PREFIXES = ("mobile.de:", "mobilede:")
|
||||
PARSER_ID_RE = re.compile(r"^car-[A-Za-z0-9]{22}$")
|
||||
|
||||
|
||||
def _origin_prefix_filter(column, prefixes: tuple[str, ...] = MOBILEDE_ORIGIN_PREFIXES):
|
||||
@@ -175,6 +177,8 @@ class PersistenceService:
|
||||
def _apply_update_payload(car: Car, payload: dict[str, object]) -> None:
|
||||
for key, value in payload.items():
|
||||
if key in CAR_UPDATE_FIELDS:
|
||||
if key == "parser_id" and PARSER_ID_RE.fullmatch(str(car.parser_id or "")):
|
||||
continue
|
||||
setattr(car, key, value)
|
||||
|
||||
@staticmethod
|
||||
@@ -277,6 +281,8 @@ class PersistenceService:
|
||||
images = [image.model_dump(mode="python") for image in record.images]
|
||||
car_by_id = existing_by_id.get(record.origin_id)
|
||||
car_by_url = existing_by_url.get(record.origin_url)
|
||||
if car_by_id is not None and PARSER_ID_RE.fullmatch(str(car_by_id.parser_id or "")):
|
||||
payload["parser_id"] = car_by_id.parser_id
|
||||
entry: dict[str, object] = {
|
||||
"record": record,
|
||||
"images": images,
|
||||
@@ -375,6 +381,11 @@ class PersistenceService:
|
||||
images = [image.model_dump(mode="python") for image in record.images]
|
||||
with self.session_scope() as session:
|
||||
if self._is_postgres():
|
||||
car_by_id = session.execute(
|
||||
select(Car).where(Car.origin_id == record.origin_id)
|
||||
).scalar_one_or_none()
|
||||
if car_by_id is not None and PARSER_ID_RE.fullmatch(str(car_by_id.parser_id or "")):
|
||||
payload["parser_id"] = car_by_id.parser_id
|
||||
car_by_url = session.execute(
|
||||
select(Car).where(Car.origin_url == record.origin_url)
|
||||
).scalar_one_or_none()
|
||||
@@ -384,9 +395,7 @@ class PersistenceService:
|
||||
car_id = int(car_by_url.id)
|
||||
action = "updated"
|
||||
else:
|
||||
existed = session.execute(
|
||||
select(Car.id).where(Car.origin_id == record.origin_id)
|
||||
).scalar_one_or_none() is not None
|
||||
existed = car_by_id is not None
|
||||
insert_stmt = pg_insert(Car).values(**payload)
|
||||
upsert_stmt = insert_stmt.on_conflict_do_update(
|
||||
index_elements=[Car.origin_id],
|
||||
|
||||
@@ -85,6 +85,31 @@ class TestMobileDeMapper(unittest.TestCase):
|
||||
self.assertEqual(record.origin_id, "mobile.de:456")
|
||||
self.assertEqual(record.origin, "MOBILE_DE")
|
||||
|
||||
def test_slug_includes_year_without_duplicate_or_trailing_hyphens(self) -> None:
|
||||
record = self.mapper.listing_to_car_record(
|
||||
MobileDeListing(
|
||||
id="457",
|
||||
url="https://suchen.mobile.de/fahrzeuge/details.html?id=457",
|
||||
title="Dodge -- Dart SXT!",
|
||||
first_registration="2013",
|
||||
)
|
||||
)
|
||||
|
||||
self.assertEqual(record.slug, "dodge-dart-sxt-2013")
|
||||
self.assertNotIn("--", record.slug)
|
||||
self.assertFalse(record.slug.endswith("-"))
|
||||
|
||||
def test_slug_omits_year_when_year_is_absent(self) -> None:
|
||||
record = self.mapper.listing_to_car_record(
|
||||
MobileDeListing(
|
||||
id="458",
|
||||
url="https://suchen.mobile.de/fahrzeuge/details.html?id=458",
|
||||
title="Dodge Dart SXT",
|
||||
)
|
||||
)
|
||||
|
||||
self.assertEqual(record.slug, "dodge-dart-sxt")
|
||||
|
||||
def test_unknown_empty_values_and_normalization(self) -> None:
|
||||
record = self.mapper.listing_to_car_record(
|
||||
MobileDeListing(
|
||||
@@ -138,7 +163,7 @@ class TestMobileDeMapper(unittest.TestCase):
|
||||
"attr": {
|
||||
"yc": "2016",
|
||||
"cn": "DE",
|
||||
"ecol": "Серебряный",
|
||||
"ecol": "Silber",
|
||||
"c": "OffRoad",
|
||||
"cc": "1 998 ccm",
|
||||
}
|
||||
@@ -160,7 +185,7 @@ class TestMobileDeMapper(unittest.TestCase):
|
||||
url="https://suchen.mobile.de/fahrzeuge/details.html?id=1002",
|
||||
title="BMW 320 Touring",
|
||||
subtitle="EstateCar",
|
||||
raw={"attr": {"ecol": "Коричневый"}},
|
||||
raw={"attr": {"ecol": "Braun"}},
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
107
tests/test_mobilede_requirements.py
Normal file
107
tests/test_mobilede_requirements.py
Normal file
@@ -0,0 +1,107 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from urllib.parse import parse_qs, urlsplit
|
||||
|
||||
from sqlalchemy import select
|
||||
|
||||
from mobilede_scraper.core.config import Settings
|
||||
from mobilede_scraper.core.runtime_config import RuntimeFiltersConfig
|
||||
from mobilede_scraper.mobile_de.client import DEFAULT_HEADERS, MobileDeClient
|
||||
from mobilede_scraper.mobile_de.mapper import MobileDeMapper
|
||||
from mobilede_scraper.mobile_de.models import MobileDeListing, MobileDeSearchPage
|
||||
from mobilede_scraper.mobile_de.scraper import MobileDeScraper
|
||||
from mobilede_scraper.storage.db import PersistenceService
|
||||
from mobilede_scraper.storage.models import Car
|
||||
from mobilede_scraper.storage.schemas import CarRecord
|
||||
|
||||
|
||||
class _Client:
|
||||
def __init__(self, pages: list[MobileDeSearchPage]) -> None:
|
||||
self.pages = pages
|
||||
|
||||
def iter_search_pages(self, **kwargs):
|
||||
del kwargs
|
||||
yield from self.pages
|
||||
|
||||
|
||||
class _Persistence:
|
||||
def __init__(self) -> None:
|
||||
self.records: list[CarRecord] = []
|
||||
|
||||
def create_tables(self) -> None: pass
|
||||
def start_sync_run(self, lane: str) -> int:
|
||||
del lane
|
||||
return 1
|
||||
def get_existing_origin_ids(self, origin_ids: list[str]) -> set[str]:
|
||||
del origin_ids
|
||||
return set()
|
||||
def upsert_cars_batch(self, records: list[CarRecord]) -> dict[str, int]:
|
||||
self.records.extend(records)
|
||||
return {"inserted": len(records), "updated": 0, "images_upserted": 0}
|
||||
def finish_sync_run(self, run_id: int, **kwargs) -> None:
|
||||
del run_id, kwargs
|
||||
|
||||
|
||||
class TestMobileDeRequirements(unittest.TestCase):
|
||||
def test_english_urls_and_header_preserve_existing_query(self) -> None:
|
||||
search = MobileDeClient.build_search_url_from_existing(
|
||||
"https://www.mobile.de/ru/search.html?ms=3500&lang=de&custom=keep",
|
||||
page_number=3,
|
||||
)
|
||||
detail = MobileDeClient.build_detail_url("123")
|
||||
self.assertEqual(DEFAULT_HEADERS["accept-language"], "en-US,en;q=0.9")
|
||||
self.assertEqual(urlsplit(search).scheme, "https")
|
||||
self.assertEqual(urlsplit(search).netloc, "suchen.mobile.de")
|
||||
self.assertEqual(urlsplit(search).path, "/fahrzeuge/search.html")
|
||||
self.assertEqual(urlsplit(detail).path, "/fahrzeuge/details.html")
|
||||
self.assertEqual(parse_qs(urlsplit(search).query)["lang"], ["en"])
|
||||
self.assertEqual(parse_qs(urlsplit(search).query)["custom"], ["keep"])
|
||||
self.assertEqual(parse_qs(urlsplit(detail).query)["lang"], ["en"])
|
||||
|
||||
def test_parser_id_is_deterministic_and_matches_project_format(self) -> None:
|
||||
origin_id = "mobile.de:123"
|
||||
parser_id = MobileDeMapper._parser_id(origin_id)
|
||||
self.assertRegex(parser_id, r"^car-[A-Za-z0-9]{22}$")
|
||||
self.assertEqual(parser_id, MobileDeMapper._parser_id(origin_id))
|
||||
self.assertNotEqual(parser_id, origin_id)
|
||||
|
||||
def test_mapper_contains_no_cyrillic_literals(self) -> None:
|
||||
mapper_path = Path(__file__).parents[1] / "mobilede_scraper" / "mobile_de" / "mapper.py"
|
||||
self.assertIsNone(re.search(r"[А-Яа-яЁё]", mapper_path.read_text(encoding="utf-8")))
|
||||
|
||||
def test_upsert_uses_origin_id_and_migrates_legacy_parser_id(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
settings = Settings()
|
||||
settings.database.url = f"sqlite:///{Path(directory, 'db.sqlite').as_posix()}"
|
||||
service = PersistenceService(settings)
|
||||
service.create_tables()
|
||||
origin_id = "mobile.de:1"
|
||||
legacy = CarRecord(parser_id="car-0123456789abcdef", brand="BMW", model="X1", origin_id=origin_id, origin_url="https://example.test/1", slug="bmw-x1")
|
||||
current = legacy.model_copy(update={"parser_id": MobileDeMapper._parser_id(origin_id), "price": 20})
|
||||
service.upsert_car(legacy)
|
||||
service.upsert_car(current)
|
||||
service.upsert_car(current)
|
||||
with service.session_scope() as session:
|
||||
cars = session.execute(select(Car)).scalars().all()
|
||||
self.assertEqual(len(cars), 1)
|
||||
self.assertEqual(cars[0].parser_id, current.parser_id)
|
||||
service.engine.dispose()
|
||||
|
||||
def test_runtime_filters_and_limit_apply_after_mapping(self) -> None:
|
||||
pages = [MobileDeSearchPage(url="https://example.test", page_number=1, total_results=3, listings=[
|
||||
MobileDeListing(id="1", url="https://example.test/1", title="BMW X1", raw={"attr": {"yc": "2020", "ml": "10000", "c": "OffRoad", "ecol": "Black", "tr": "Automatic"}, "price": {"grs": {"amount": "20000"}}}),
|
||||
MobileDeListing(id="2", url="https://example.test/2", title="BMW X3", raw={"attr": {"yc": "2020", "ml": "10000", "c": "OffRoad", "ecol": "Black", "tr": "Automatic"}, "price": {"grs": {"amount": "21000"}}}),
|
||||
MobileDeListing(id="3", url="https://example.test/3", title="Audi Q5", raw={"attr": {"yc": "2020", "ml": "10000", "c": "OffRoad", "ecol": "Black", "tr": "Automatic"}, "price": {"grs": {"amount": "22000"}}}),
|
||||
])]
|
||||
persistence = _Persistence()
|
||||
filters = RuntimeFiltersConfig.from_dict({"brands": ["BMW"], "price": {"min": 15000, "max": 25000}, "exclude_models": ["X3"]})
|
||||
MobileDeScraper(client=_Client(pages), persistence=persistence).sync_search(runtime_filters=filters, limit=1)
|
||||
self.assertEqual([record.origin_id for record in persistence.records], ["mobile.de:1"])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user