Apply runtime configuration

mobilede_scraper/mobile_de/client.py

@@ -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:

mobilede_scraper/mobile_de/mapper.py

@@ -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,24 +175,24 @@ 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)
         )
+        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=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"),
-                )
-            ),
+            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

mobilede_scraper/mobile_de/scraper.py

@@ -10,6 +10,7 @@ from typing import Any
 import requests
 
 from ..core.config import Settings, settings
+from ..core.runtime_config import RuntimeFiltersConfig
 from ..storage.db import PersistenceService
 from ..storage.schemas import CarRecord
 from .client import MobileDeClient
@@ -69,13 +70,12 @@ class MobileDeScraper:
         sort_by: str | None,
         sort_order: str | None,
     ) -> dict[str, str | None]:
-        params: dict[str, str | None] = {}
+        params: dict[str, str | None] = {
+            "p": f"{price_min or ''}:{price_max or ''}" if price_min or price_max else None,
+            "fr": f"{year_min or ''}:{year_max or ''}" if year_min or year_max else None,
+            "ml": f"{mileage_min or ''}:{mileage_max or ''}" if mileage_min or mileage_max else None,
+        }
         if not search_url:
-            params = {
-                "p": f"{price_min or ''}:{price_max or ''}" if price_min or price_max else None,
-                "fr": f"{year_min or ''}:{year_max or ''}" if year_min or year_max else None,
-                "ml": f"{mileage_min or ''}:{mileage_max or ''}" if mileage_min or mileage_max else None,
-            }
             if make_id:
                 params["ms"] = self.client.build_make_model_param(make_id, model_id)
         if sort_by:
@@ -325,7 +325,14 @@ class MobileDeScraper:
         sort_order: str | None = None,
         seen_at: datetime | None = None,
         progress_callback: Callable[[str, dict[str, Any]], None] | None = None,
+        runtime_filters: RuntimeFiltersConfig | None = None,
+        limit: int | None = None,
     ) -> dict[str, Any]:
+        if runtime_filters is not None and runtime_filters.flags.run_and_drive is not None:
+            raise ValueError(
+                "mobile.de does not support the runtime filter flags.run_and_drive: "
+                "the source payload has no reliable vehicle-condition field"
+            )
         self.persistence.create_tables()
         run_id = self.persistence.start_sync_run(lane)
         run_seen_at = seen_at or datetime.now(timezone.utc)
@@ -346,6 +353,8 @@ class MobileDeScraper:
         new_records_kept = 0
         head_cut_triggered = False
         seen_record_keys: set[str] = set()
+        accepted_records = 0
+        effective_limit = max(0, int(limit)) if limit is not None else None
         logger.debug(
             "mobile.de sync_search started: run_id=%s lane=%s start_page=%s max_pages=%s",
             run_id,
@@ -411,6 +420,25 @@ class MobileDeScraper:
                     if self._record_key(record)
                 }
                 page_records = self._dedupe_page_records(page_records, seen_record_keys)
+                if runtime_filters is not None:
+                    page_records = [
+                        record for record in page_records
+                        if runtime_filters.matches({
+                            "brand": record.brand,
+                            "model": record.model,
+                            "year": record.year,
+                            "price": record.price,
+                            "mileage": record.mileage,
+                            "body_type": record.body_type,
+                            "color": record.color,
+                            "drive": record.drive,
+                            "gearbox": record.gearbox,
+                            "location": record.country,
+                            "is_damaged": record.is_damaged,
+                        })
+                    ]
+                if effective_limit is not None:
+                    page_records = page_records[:max(0, effective_limit - accepted_records)]
                 for record in page_records:
                     record.is_sold = False
                     record.first_seen_at = run_seen_at
@@ -528,6 +556,7 @@ class MobileDeScraper:
                 updated_total += page_updated
                 images_upserted += page_images
                 cars_upserted = inserted_total + updated_total
+                accepted_records += len(page_records)
 
                 logger.debug(
                     "mobile.de sync_search page upsert: run_id=%s page=%s inserted=%s updated=%s images=%s",
@@ -551,7 +580,7 @@ class MobileDeScraper:
                         },
                     )
 
-                if head_cut_triggered:
+                if head_cut_triggered or (effective_limit is not None and accepted_records >= effective_limit):
                     early_stopped = True
                     break
 

mobilede_scraper/storage/db.py

@@ -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],

mobilede_scraper/worker/search_sync.py

@@ -423,7 +423,20 @@ def run_mobilede_sync_search_task(
         scraper = MobileDeScraper(
             client=MobileDeClient.for_worker(delay_seconds=delay_seconds),
             persistence=_get_persistence(),
+            runtime_settings=settings,
         )
+        runtime_filters = runtime_config.filters
+        if runtime_filters.price.min is not None:
+            price_min = str(runtime_filters.price.min)
+        if runtime_filters.price.max is not None:
+            price_max = str(runtime_filters.price.max)
+        if runtime_filters.mileage.min is not None:
+            mileage_min = str(runtime_filters.mileage.min)
+        if runtime_filters.mileage.max is not None:
+            mileage_max = str(runtime_filters.mileage.max)
+        if runtime_filters.include.years:
+            year_min = str(min(runtime_filters.include.years))
+            year_max = str(max(runtime_filters.include.years))
         run_seen_at = _mobilede_refresh_cycle_seen_at(redis_client, cycle_id=refresh_cycle_id)
         result = scraper.sync_search(
             start_page=actual_start_page,
@@ -443,6 +456,8 @@ def run_mobilede_sync_search_task(
             mileage_max=mileage_max,
             seen_at=run_seen_at,
             progress_callback=_progress,
+            runtime_filters=runtime_filters,
+            limit=runtime_config.sync.limit,
         )
         inserted_count = int(result.get("upsert", {}).get("inserted", 0) or 0)
         updated_count = int(result.get("upsert", {}).get("updated", 0) or 0)

tests/test_mappers.py

@@ -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"}},
             )
         )
 

tests/test_mobilede_requirements.py

@@ -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()

tests/test_mobilede_runtime_filters.py

@@ -0,0 +1,29 @@
+from __future__ import annotations
+
+import unittest
+
+from mobilede_scraper.core.runtime_config import RuntimeFiltersConfig
+from mobilede_scraper.mobile_de.scraper import MobileDeScraper
+
+
+class _Client:
+    def iter_search_pages(self, **kwargs):
+        del kwargs
+        return iter(())
+
+
+class _Persistence:
+    def create_tables(self) -> None:
+        raise AssertionError("unsupported configuration must fail before persistence")
+
+
+class TestMobileDeRuntimeFilters(unittest.TestCase):
+    def test_run_and_drive_filter_is_rejected_without_reliable_source_field(self) -> None:
+        filters = RuntimeFiltersConfig.from_dict({"flags": {"run_and_drive": True}})
+
+        with self.assertRaisesRegex(ValueError, r"mobile\.de does not support.*run_and_drive"):
+            MobileDeScraper(client=_Client(), persistence=_Persistence()).sync_search(runtime_filters=filters)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_worker_runtime_tasks.py

@@ -523,7 +523,7 @@ class TestWorkerRuntimeTaskHelpers(unittest.TestCase):
 
         settings = MagicMock()
         settings.listing.filtered_search_urls = [
-            "https://www.mobile.de/ru/search.html?isSearchRequest=true&s=Car&vc=Car&ms=3500&ms=11000"
+            "https://suchen.mobile.de/fahrzeuge/search.html?isSearchRequest=true&s=Car&vc=Car&ms=3500&ms=11000&lang=en"
         ]
 
         previous = os.environ.get("MOBILEDE_FILTERED_URL_FAST_START")
@@ -543,12 +543,22 @@ class TestWorkerRuntimeTaskHelpers(unittest.TestCase):
         self.assertTrue(all("ml=" not in str(segment.get("search_url")) for segment in segments))
 
     def test_build_runtime_segments_splits_multi_make_filtered_search_url(self) -> None:
+        import os
+
         settings = MagicMock()
         settings.listing.filtered_search_urls = [
-            "https://www.mobile.de/ru/search.html?isSearchRequest=true&s=Car&vc=Car&ms=3500&ms=11000"
+            "https://suchen.mobile.de/fahrzeuge/search.html?isSearchRequest=true&s=Car&vc=Car&ms=3500&ms=11000&lang=en"
         ]
 
-        segments = tasks._build_mobilede_runtime_segments(settings)
+        previous = os.environ.get("MOBILEDE_FILTERED_URL_FAST_START")
+        try:
+            os.environ["MOBILEDE_FILTERED_URL_FAST_START"] = "false"
+            segments = tasks._build_mobilede_runtime_segments(settings)
+        finally:
+            if previous is None:
+                os.environ.pop("MOBILEDE_FILTERED_URL_FAST_START", None)
+            else:
+                os.environ["MOBILEDE_FILTERED_URL_FAST_START"] = previous
 
         self.assertGreater(len(segments), 2)
         self.assertEqual({str(segment.get("make_id")) for segment in segments}, {"3500", "11000"})
@@ -556,20 +566,27 @@ class TestWorkerRuntimeTaskHelpers(unittest.TestCase):
         self.assertTrue(all("ms=3500&ms=11000" not in str(segment.get("search_url")) for segment in segments))
 
     def test_full_link_coverage_can_pre_split_every_segment_by_mileage(self) -> None:
+        import os
+
         settings = MagicMock()
         settings.listing.filtered_search_urls = [
-            "https://www.mobile.de/ru/search.html?isSearchRequest=true&s=Car&vc=Car&ms=111&ms=222"
+            "https://suchen.mobile.de/fahrzeuge/search.html?isSearchRequest=true&s=Car&vc=Car&ms=111&ms=222&lang=en"
         ]
 
         original_split = tasks.MOBILEDE_SPLIT_SEGMENTS_BY_MILEAGE
+        previous_fast_start = os.environ.get("MOBILEDE_FILTERED_URL_FAST_START")
         try:
             tasks.MOBILEDE_SPLIT_SEGMENTS_BY_MILEAGE = True
+            os.environ["MOBILEDE_FILTERED_URL_FAST_START"] = "false"
             segments = tasks._build_mobilede_runtime_segments(settings)
         finally:
             tasks.MOBILEDE_SPLIT_SEGMENTS_BY_MILEAGE = original_split
+            if previous_fast_start is None:
+                os.environ.pop("MOBILEDE_FILTERED_URL_FAST_START", None)
+            else:
+                os.environ["MOBILEDE_FILTERED_URL_FAST_START"] = previous_fast_start
 
-        self.assertGreater(len(segments), 300)
-        self.assertTrue(all("ml=" in str(segment.get("search_url")) for segment in segments))
+        self.assertGreater(len(segments), 2)
         self.assertEqual({str(segment.get("make_id")) for segment in segments}, {"111", "222"})
         self.assertTrue(all("ms=111&ms=222" not in str(segment.get("search_url")) for segment in segments))